Appearance
LangChain 从入门到实战精通
导航目录
- 一、LangChain 是什么
- 二、为什么项目里会引入 LangChain
- 三、核心包与安装建议
- 四、模型、消息与 Prompt 模板
- 五、结构化输出与解析器
- 六、Runnable 与 LCEL
- 七、链式调用与组合模式
- 八、向量、Embedding 与 Retriever
- 九、RAG 的完整实现思路
- 十、工具调用、Tool Calling 与 MCP
- 十一、Agent、ReAct 与多 Agent 协作
- 十二、记忆、状态与 LangGraph 的边界
- 十三、生产实践与常见坑
- 十四、面试高频问题
一、LangChain 是什么
1.1 核心说明
LangChain 是一个用于构建大模型应用的开发框架,它把 Prompt、模型调用、工具调用、检索、记忆、工作流编排等能力标准化,帮助你快速搭建可维护的 AI 应用。
典型应用场景:
- 智能对话机器人
- 知识库问答(RAG)
- 文档分析与摘要
- 多工具联动自动化
- API 化对外服务
1.2 最小可运行示例
python
# 基础链搭建代码示例
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
os.environ["OPENAI_API_KEY"] = "你的API_KEY"
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_template("请用一句话解释:{topic}")
chain = prompt | model
resp = chain.invoke({"topic": "什么是LangChain"})
print(resp.content)
# 效果说明:返回一个可读的解释文本1.3 关键注意点
- LangChain 不替代模型,它是模型应用编排层。
- 先学会
Prompt -> Model -> Output三件套,再上 RAG/Agent。
二、为什么项目里会引入 LangChain
2.1 核心价值
- 统一接口:多模型(OpenAI/Anthropic/本地模型)切换成本低
- 组件化:Prompt、Retriever、Tool、Memory 可插拔
- 可观测:支持 tracing、日志、错误恢复
- 工程化友好:适合 API 服务化与持续迭代
2.2 基础开发流程(完整)
- 环境搭建
- 安装依赖
- 配置 API Key / Base URL
- 初始化模型
- 组装 Prompt/Chain
- 联调与测试
- 增加检索、工具、记忆
- 服务化与监控
2.3 完整入门流程代码
python
# 基础开发流程代码示例
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
# 1) 配置环境变量
os.environ["OPENAI_API_KEY"] = "你的API_KEY"
# 2) 初始化模型
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0.2, # 输出稳定性更高
max_tokens=300, # 控制最大输出长度
timeout=30, # 网络超时
)
# 3) Prompt 模板
prompt = ChatPromptTemplate.from_template(
"你是资深工程师。请用3点解释 {topic},每点不超过20字。"
)
# 4) 组装链
chain = prompt | llm | StrOutputParser()
# 5) 执行测试
result = chain.invoke({"topic": "LangChain的核心价值"})
print(result)
# 效果说明:输出格式稳定、可直接展示到前端三、核心包与安装建议
3.1 Python 版本与核心依赖
- Python:建议
3.10+ - 核心包:
langchainlangchain-corelangchain-openailangchain-community
- RAG 常用:
chromadbfaiss-cpupypdf
3.2 安装命令(pip + 镜像)
bash
# 标准安装
pip install -U langchain langchain-core langchain-openai langchain-community
pip install -U chromadb faiss-cpu pypdf fastapi uvicorn python-dotenv
# 国内镜像(避免下载失败)
pip install -U langchain langchain-openai -i https://pypi.tuna.tsinghua.edu.cn/simple3.3 标准导入与验证
python
# 导入验证代码示例
import langchain
import langchain_core
import langchain_openai
print("langchain:", langchain.__version__)
print("langchain_core:", langchain_core.__version__)
print("langchain_openai:", langchain_openai.__version__)
# 效果说明:能输出版本号即环境可用3.4 新手常见误区与避坑
- 依赖冲突:统一升级到最新版,必要时新建虚拟环境
- API Key 未生效:确认
OPENAI_API_KEY在当前进程可见 - 网络超时:设置
timeout,并加重试机制
四、模型、消息与 Prompt 模板
4.1 核心概念速记
- LLM / ChatModel:负责生成内容
- PromptTemplate:参数化提示词
- Message:system/human/ai 消息结构
- Chain:把多个步骤串起来
- Tool:可调用的函数能力
4.2 PromptTemplate 基础与动态模板
python
# PromptTemplate 基础代码示例
from langchain_core.prompts import PromptTemplate
tpl = PromptTemplate.from_template(
"请将下面内容改写为更专业语气:\n{text}"
)
print(tpl.format(text="这个功能挺好用的。"))
# 效果说明:输出渲染后的 prompt 文本4.3 条件模板(通过分支实现)
python
# 条件模板代码示例
from langchain_core.prompts import ChatPromptTemplate
def build_prompt(level: str):
if level == "beginner":
return ChatPromptTemplate.from_template("请用新手能懂的方式解释:{topic}")
return ChatPromptTemplate.from_template("请从工程实践角度深入解释:{topic}")
prompt = build_prompt("beginner")
print(prompt.format_messages(topic="向量数据库"))
# 效果说明:根据 level 返回不同复杂度提示词4.4 批量模板生成
python
# 批量模板调用示例
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_template("一句话总结:{topic}")
chain = prompt | llm
inputs = [{"topic": "RAG"}, {"topic": "Agent"}, {"topic": "Tool Calling"}]
responses = chain.batch(inputs)
for r in responses:
print(r.content)
# 效果说明:一次请求批量处理多个输入,效率更高4.5 参数说明(常用)
- model:模型名称,如
gpt-4o-mini - temperature:随机性,
0~2,越低越稳定 - max_tokens:单次最大输出长度
- timeout:请求超时秒数
五、结构化输出与解析器
5.1 为什么要结构化输出
在生产中,返回 JSON 比自然语言更易存储、校验、接口传输。
5.2 Pydantic 结构化输出
python
# 结构化输出代码示例
from typing import List
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
class Plan(BaseModel):
title: str = Field(description="任务标题")
steps: List[str] = Field(description="执行步骤")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(Plan)
prompt = ChatPromptTemplate.from_template("给我一个学习{topic}的3步计划")
chain = prompt | structured_llm
data = chain.invoke({"topic": "LangChain"})
print(data)
# 效果说明:得到 Plan 对象而不是纯文本5.3 常见问题
- 输出字段缺失:在提示词中补充字段约束
- 解析失败:降低温度并强化格式要求
六、Runnable 与 LCEL
6.1 核心说明
LCEL(LangChain Expression Language)通过 | 把组件串联:prompt | model | parser
6.2 Runnable 组合示例
python
# Runnable 与 LCEL 代码示例
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_template("把这句话翻译成英文:{text}")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()
chain = prompt | llm | parser
print(chain.invoke({"text": "今天心情很好"}))
# 效果说明:输出纯字符串结果6.3 并行与映射
python
# 并行执行示例
from langchain_core.runnables import RunnableParallel
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()
zh = ChatPromptTemplate.from_template("翻译为中文:{text}") | llm | parser
en = ChatPromptTemplate.from_template("翻译为英文:{text}") | llm | parser
parallel = RunnableParallel(to_zh=zh, to_en=en)
print(parallel.invoke({"text": "LangChain improves LLM engineering."}))
# 效果说明:一次输入,同时得到两个方向的结果七、链式调用与组合模式
7.1 LLMChain / SequentialChain 说明
- LLMChain:单步骤处理
- SimpleSequentialChain:多个步骤串行
- SequentialChain:多输入多输出串行
7.2 对比代码示例
python
# Chain 对比代码示例
from langchain_openai import OpenAI
from langchain.chains import LLMChain, SimpleSequentialChain
from langchain.prompts import PromptTemplate
llm = OpenAI(temperature=0)
p1 = PromptTemplate(input_variables=["topic"], template="给出{topic}的一个标题")
c1 = LLMChain(llm=llm, prompt=p1)
p2 = PromptTemplate(input_variables=["title"], template="根据标题写一句摘要:{title}")
c2 = LLMChain(llm=llm, prompt=p2)
pipeline = SimpleSequentialChain(chains=[c1, c2], verbose=True)
print(pipeline.run("LangChain学习路线"))
# 效果说明:先生成标题,再生成摘要7.3 复杂 Chain(分支/循环/多模型)
python
# 分支 + 多模型联动思路示例
# 注意:示例展示架构思想,可在 LCEL 中通过 RunnableBranch/自定义函数实现
"""
1) 输入问题
2) 分类器判断:技术类/业务类
3) 技术类走模型A,业务类走模型B
4) 统一后处理并输出
"""
# 效果说明:复杂业务中可按任务类型切模型和处理路径八、向量、Embedding 与 Retriever
8.1 核心说明
RAG 基础链路:文档 -> 向量化 -> 存储 -> 检索 -> 生成
8.2 Chroma 向量库示例
python
# Embedding + Chroma 代码示例
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
emb = OpenAIEmbeddings(model="text-embedding-3-small")
docs = [
Document(page_content="LangChain可以构建Agent系统", metadata={"source": "a"}),
Document(page_content="RAG用于外部知识增强", metadata={"source": "b"}),
]
vs = Chroma.from_documents(documents=docs, embedding=emb, collection_name="demo_docs")
hits = vs.similarity_search("什么是RAG", k=2)
for h in hits:
print(h.page_content, h.metadata)
# 效果说明:返回最相近的文档片段8.3 FAISS 示例
python
# Embedding + FAISS 代码示例
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
emb = OpenAIEmbeddings(model="text-embedding-3-small")
texts = ["苹果是水果", "汽车用于出行", "数据库用于存储数据"]
vs = FAISS.from_texts(texts, emb)
print(vs.similarity_search("什么用于存储信息", k=1)[0].page_content)
# 效果说明:检索到“数据库用于存储数据”8.4 参数说明
- k:召回文档数量,常用
3~8 - score_threshold:相似度阈值(按库支持)
- chunk_size/chunk_overlap:影响检索粒度与上下文完整性
九、RAG 的完整实现思路
9.1 全流程
- 文档加载(PDF/TXT/Word)
- 文本切分
- 向量化入库
- Retriever 检索
- 组装问答链
9.2 可直接复用代码
python
# 知识库问答完整代码示例
import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
os.environ["OPENAI_API_KEY"] = "你的API_KEY"
# 1) 文档加载(按你的文件类型选择)
docs = TextLoader("kb.txt", encoding="utf-8").load()
# docs = PyPDFLoader("kb.pdf").load()
# 2) 切分
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=80)
chunks = splitter.split_documents(docs)
# 3) 向量存储
emb = OpenAIEmbeddings(model="text-embedding-3-small")
vs = Chroma.from_documents(chunks, emb, collection_name="kb_collection")
retriever = vs.as_retriever(search_kwargs={"k": 4})
# 4) 问答链
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_template(
"你是知识库问答助手。仅基于上下文回答。\n\n上下文:\n{context}\n\n问题:{question}"
)
parser = StrOutputParser()
def ask(question: str) -> str:
docs = retriever.invoke(question)
context = "\n\n".join(d.page_content for d in docs)
chain = prompt | llm | parser
return chain.invoke({"context": context, "question": question})
print(ask("这份知识库主要讲了什么?"))
# 效果说明:回答内容受知识库约束,幻觉显著减少9.3 注意事项
- 检索不准先调
chunk_size/k - 对高精度场景加
rerank - 答案要“带来源”便于追溯
十、工具调用、Tool Calling 与 MCP
10.1 内置 Tool 与自定义 Tool
python
# 自定义 Tool 代码示例
from langchain_core.tools import tool
@tool
def add_numbers(a: int, b: int) -> int:
"""计算两个整数之和"""
return a + b
print(add_numbers.invoke({"a": 3, "b": 5}))
# 效果说明:返回 810.2 Tool 与 Chain 联动逻辑
- Chain 负责语言理解
- Tool 负责外部动作(查库/查天气/调用业务系统)
- Agent 负责在二者之间“决策调度”
10.3 工具权限控制建议
- 只暴露必要工具
- 为工具输入做 schema 校验
- 对高风险工具(写操作)增加白名单和审批
10.4 MCP 说明
MCP 可把外部系统能力标准化为可调用工具资源,LangChain 可通过工具层接入,形成统一调度链路。
十一、Agent、ReAct 与多 Agent 协作
11.1 Agent 类型与适用场景
- 基础 Agent:简单问答 + 工具调用
- ReAct Agent:边思考边行动,适合多步推理
- Self-Ask:先拆问题再求解,适合复杂问答
- Plan-and-Execute:先规划后执行,适合流程任务
11.2 ReAct Agent 示例
python
# ReAct Agent 代码示例
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate
from langchain_core.tools import tool
@tool
def multiply(a: int, b: int) -> int:
"""计算乘法"""
return a * b
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [multiply]
prompt = PromptTemplate.from_template(
"""你可以使用工具解决问题。
问题: {input}
可用工具: {tools}
工具名称: {tool_names}
{agent_scratchpad}"""
)
agent = create_react_agent(llm=llm, tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
print(executor.invoke({"input": "7乘以9是多少?"}))
# 效果说明:Agent 会自动调用 multiply 工具11.3 Plan-and-Execute 思路
python
# Plan-and-Execute 架构示例(伪代码)
"""
planner: 负责把目标拆成子任务
executor: 逐个执行子任务(可调用多个工具)
supervisor: 汇总结果并给出最终答复
"""
# 注意事项:计划要可观测、可中断、可重试11.4 多 Agent 协作建议
- 角色分工明确:规划/执行/审查
- 共享上下文最小化:避免上下文污染
- 每个 Agent 限制工具权限
十二、记忆、状态与 LangGraph 的边界
12.1 多轮对话记忆(机器人场景)
python
# 多轮对话记忆代码示例(简化)
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
history = [SystemMessage(content="你是专业客服助手。")]
def chat(user_text: str) -> str:
history.append(HumanMessage(content=user_text))
resp = llm.invoke(history)
history.append(AIMessage(content=resp.content))
return resp.content
print(chat("我想了解你们的退款规则"))
print(chat("请结合我刚才的问题,再给一个简短版本"))
# 效果说明:第二轮可引用第一轮上下文12.2 LangChain 与 LangGraph 边界
- LangChain:线性/轻量编排快
- LangGraph:复杂状态机、多分支回退、长流程更强
12.3 何时升级到 LangGraph
- 需要强状态控制
- 需要人工中断与恢复
- 需要多 Agent 工作流编排
十三、生产实践与常见坑
13.1 错误处理、日志、重试(可复用)
python
# 稳定性增强代码示例
import time
import logging
from langchain_openai import ChatOpenAI
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("llm_service")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, timeout=20)
def safe_invoke(messages, retry=3):
last_err = None
for i in range(retry):
try:
return llm.invoke(messages)
except Exception as e:
last_err = e
logger.warning("调用失败,第%s次重试,错误=%s", i + 1, str(e))
time.sleep(2 ** i)
raise RuntimeError(f"模型调用失败: {last_err}")13.2 FastAPI 封装(API 接口开发)
python
# FastAPI 接口实战代码示例
import os
from fastapi import FastAPI
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
os.environ["OPENAI_API_KEY"] = "你的API_KEY"
app = FastAPI(title="LangChain API Demo")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
prompt = ChatPromptTemplate.from_template("请回答问题:{question}")
chain = prompt | llm | StrOutputParser()
class AskReq(BaseModel):
question: str
@app.post("/ask")
def ask(req: AskReq):
answer = chain.invoke({"question": req.question})
return {"answer": answer}
# 运行方式:
# uvicorn app:app --host 0.0.0.0 --port 8000 --reload13.3 本地化部署(离线可控)
python
# 本地模型接入示例(以本地OpenAI兼容网关为例)
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="qwen2.5-7b-instruct",
api_key="EMPTY",
base_url="http://127.0.0.1:8000/v1", # 本地兼容OpenAI接口服务
temperature=0.2,
)
print(llm.invoke("请用一句话介绍你自己").content)
# 效果说明:脱离公网API,走本地推理服务13.4 批量处理与并发优化
python
# 批量并发调用示例
from concurrent.futures import ThreadPoolExecutor
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
questions = ["什么是RAG", "什么是Agent", "什么是Embedding"]
def run(q):
return llm.invoke(q).content
with ThreadPoolExecutor(max_workers=4) as ex:
results = list(ex.map(run, questions))
print(results)
# 注意事项:并发过高可能触发限流,需配合重试和速率控制13.5 Docker 部署模板
dockerfile
# Dockerfile 示例
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]13.6 常见问题(错误代码 + 修正代码)
问题1:API 密钥未配置
python
# 错误代码
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
print(llm.invoke("hello").content)
# 原因说明:未设置 OPENAI_API_KEY,调用会报认证错误python
# 修正代码
import os
from langchain_openai import ChatOpenAI
os.environ["OPENAI_API_KEY"] = "你的API_KEY"
llm = ChatOpenAI(model="gpt-4o-mini")
print(llm.invoke("hello").content)问题2:向量检索不准确
python
# 常见原因:chunk_size 太大,语义混杂;k 太小召回不足
# 修正方向:减小 chunk_size、增加 k、增加 rerank问题3:Agent 逻辑混乱
python
# 常见原因:工具描述不清,Prompt 未约束动作顺序
# 修正方向:明确工具输入输出、限制最大迭代、增加中间日志13.7 高级避坑技巧
- 降低幻觉:检索增强 + 输出引用来源 + 低温度
- 降低成本:缓存常见问答、优先小模型、批处理
- 提升速度:并发 + 流式输出 + 预热连接池
- 兼容多模型:抽象模型层,统一请求参数
十四、面试高频问题
14.1 LangChain 核心组件有哪些?
- Prompt、Model、OutputParser、Retriever、Tool、Agent、Memory、Runnable
14.2 LLMChain 和 Agent 的差异?
- LLMChain 是固定流程,Agent 是动态决策流程。
14.3 RAG 关键优化点?
- 文档切分、召回质量、重排、上下文压缩、答案引用。
14.4 什么时候用 LangGraph?
- 多状态、多分支、可恢复流程、人工介入场景。
14.5 3个完整实战案例汇总(从需求到部署)
案例A:智能客服机器人
- 需求:支持多轮对话与FAQ
- 实现:ChatModel + 历史记忆 + 工具查询订单
- 优化:加入敏感词规则、重试和日志
- 部署:FastAPI + Docker
案例B:企业知识库问答
- 需求:PDF/TXT 文档检索问答
- 实现:Loader + Splitter + Chroma + Retriever + QA Chain
- 优化:chunk 参数调优 + rerank + 来源引用
- 部署:定时增量索引 + 监控检索命中率
案例C:文档分析工具
- 需求:批量总结文档并导出报告
- 实现:批量读取 + LCEL batch + 结构化输出
- 优化:并发池 + 缓存 + 失败重试
- 部署:任务队列 + API 查询任务状态
案例D:本地化离线助手
- 需求:内网可用、数据不出域
- 实现:本地 LLM(OpenAI 兼容网关)+ 本地向量库
- 优化:模型量化、索引预构建
- 部署:单机 Docker Compose
案例E:多工具协同 Agent
- 需求:自动查询数据并生成日报
- 实现:ReAct Agent + SQL Tool + Report Tool
- 优化:权限控制 + 审计日志 + 输出模板
- 部署:定时任务 + Webhook 推送
核心参数手册(速查)
模型参数
- model:模型名称,决定能力与成本
- temperature:随机性,低值更稳定
- max_tokens:输出长度上限
- timeout:请求超时
- top_p:核采样范围(部分模型支持)
检索参数
- k:召回条数
- chunk_size:切块大小
- chunk_overlap:切块重叠
- search_type:检索策略(similarity/mmr)
Agent 参数
- max_iterations:最大推理步数
- verbose:打印中间步骤
- handle_parsing_errors:解析失败时容错
优质模板(可直接复制改造)
1) 通用问答模板
python
TEMPLATE = """你是专业助手,请基于事实回答:
问题:{question}
要求:
1. 结论先行
2. 分点说明
3. 无依据时明确说明不知道
"""2) RAG 问答模板
python
RAG_TEMPLATE = """你是知识库助手,只能依据上下文作答。
上下文:
{context}
问题:{question}
请输出:
1) 简短答案
2) 依据片段
"""3) Agent 工具调用模板
python
AGENT_TEMPLATE = """你可以使用工具解决问题。
策略:
1) 先判断是否需要工具
2) 需要时仅调用最合适工具
3) 输出最终结论与依据
问题:{input}
"""学习资源推荐
- 官方文档:https://python.langchain.com/
- LangSmith(可观测平台):https://smith.langchain.com/
- Chroma 文档:https://docs.trychroma.com/
- FAISS 文档:https://github.com/facebookresearch/faiss
- FastAPI 文档:https://fastapi.tiangolo.com/
附录:依赖清单示例
txt
langchain
langchain-core
langchain-openai
langchain-community
chromadb
faiss-cpu
pypdf
fastapi
uvicorn
python-dotenv
pydantic附录:运行步骤(统一)
- 创建虚拟环境并安装依赖
- 配置
OPENAI_API_KEY(或本地兼容网关) - 运行单文件示例验证模型可用
- 再集成 RAG / Tool / Agent
- 最后封装 FastAPI 并容器化上线