1. 这不是“选框架”的问题,而是你正在踩的工程认知陷阱
LangChain 和 LangGraph 这两个词最近在技术社区里高频出现,尤其在刚接触大模型应用开发的人群里,几乎成了“入门必问三连”之一:哪个好?先学哪个?要不要一起学?我带过十几期 LLM 应用开发训练营,发现一个特别典型的误区——大家把 LangChain 和 LangGraph 当成两个并列的“工具包”,像选 Django 还是 Flask 那样去比较。但事实恰恰相反:LangGraph 不是 LangChain 的替代品,也不是它的升级版,它根本就不是“另一个框架”。它是 LangChain 在解决复杂 Agent 编排问题时,被迫生长出来的一套底层抽象机制。换句话说,LangChain 是你写第一行from langchain_core.prompts import ChatPromptTemplate时就用上的“胶水层”,而 LangGraph 是当你发现胶水开始失效、流程开始打结、错误无法定位、状态无法追踪时,才真正需要动用的“手术刀”。
这背后反映的是整个 LLM 应用开发范式的演进:从静态 RAG 检索增强问答(LangChain 最初的强项),到动态多步决策 Agent(LangChain 2.x 开始艰难支撑),再到具备显式状态机、可中断、可回溯、可监控的自主智能体系统(LangGraph 的核心价值)。关键词Agent、LLM、RAG并非孤立存在——RAG 是数据层的“输入增强”,LLM 是推理层的“大脑引擎”,而 Agent 是编排层的“神经中枢”,LangChain 曾试图用链式调用(Chain)模拟这个中枢,但当逻辑分支超过 3 层、失败重试超过 2 次、记忆需要跨步骤共享时,Chain 就变成了意大利面条代码。LangGraph 则直接引入有向图(Directed Acyclic Graph + Cycle Support)作为第一公民,把“下一步走哪条边”这件事,从隐式条件判断,变成显式节点定义。这不是语法糖,是工程范式的切换。
所以如果你正站在起点,手头有个真实需求:比如要搭一个能自动查合同条款、比对法条、生成修订建议的法律助手;或者一个能根据用户语音指令,分步查询航班、预订酒店、发送确认邮件的旅行助理——那么你根本不需要纠结“LangChain 还是 LangGraph”,你需要的是:先用 LangChain 快速验证 LLM 能力边界,再用 LangGraph 构建生产级 Agent 流程。这个路径不是理论推导,而是我在三个落地项目中反复验证过的节奏:第一个月用 LangChain 写通 RAG+LLM 基础链路,第二个月暴露出循环依赖、状态丢失、错误不可追溯等问题,第三个月引入 LangGraph 重构核心编排逻辑,交付周期反而缩短 40%。下面我会用真实代码片段、参数选择依据、调试日志截图(文字还原)和踩坑记录,带你走完这条被压缩了 80% 时间的学习路线。
2. 核心设计哲学差异:从“链式胶水”到“图式操作系统”
2.1 LangChain 的本质:面向过程的 LLM 调用封装器
LangChain 的设计初衷非常务实:让开发者不用每次调用 OpenAI API 都手动拼 JSON、处理 token 限制、管理 prompt 模板。它把 LLM 调用、文档加载、文本分割、向量存储、检索器封装成一个个可组合的模块(Component),通过Runnable接口统一暴露。看一个最典型的 RAG 流程:
from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings # 1. 加载文档 → 分割 → 向量化 → 存入 Chroma vectorstore = Chroma.from_documents( documents=split_docs, embedding=OpenAIEmbeddings() ) # 2. 构建检索器 retriever = vectorstore.as_retriever() # 3. 定义 Prompt 模板 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个法律专家,请基于以下上下文回答问题:{context}"), ("human", "{question}") ]) # 4. 组装 Chain:检索 → 格式化 → LLM 调用 → 输出解析 rag_chain = ( {"context": retriever, "question": RunnablePassthrough()} | prompt | ChatOpenAI(model="gpt-4o", temperature=0) | StrOutputParser() )这段代码的核心是|操作符,它代表函数式管道(Function Composition)。每个环节输出是下一个环节的输入,数据流是线性的、单向的、无状态的。这种设计在简单场景下极其高效:RAG 问答、摘要生成、基础分类。但一旦加入“如果检索不到结果,则尝试改写问题再检索一次”、“如果 LLM 返回格式错误,则触发重试并降级为规则匹配”这类逻辑,代码就会迅速膨胀:
# 伪代码:LangChain 中实现“失败重试+降级”的典型写法 def robust_rag(question: str) -> str: # 第一次检索 docs = retriever.invoke(question) if not docs: # 改写问题 rewritten = rewrite_prompt.invoke({"original": question}) docs = retriever.invoke(rewritten.content) # 构建上下文 context = "\n\n".join([d.page_content for d in docs]) # 调用 LLM try: result = llm.invoke(prompt.format(context=context, question=question)) return result.content except Exception as e: # 降级:用规则匹配 return rule_based_fallback(question)问题在于:这个robust_rag函数已经脱离了 LangChain 的Runnable范式,变成了普通 Python 函数。你失去了 LangChain 提供的内置日志、异步支持、流式输出、中间结果缓存等能力。更严重的是,状态完全丢失——你无法在重试时复用第一次检索的向量查询结果,也无法在降级时知道之前失败的具体错误类型。这就是 LangChain 在 Agent 场景下的根本瓶颈:它缺乏对“执行上下文”(Execution Context)的原生建模。
提示:LangChain 的
Runnable本质是__call__方法的封装,它不保存任何执行历史。每一次调用都是全新的、孤立的。这决定了它适合“请求-响应”模式,不适合“会话-状态”模式。
2.2 LangGraph 的本质:面向状态的 Agent 编排操作系统
LangGraph 的诞生,直接回应了上述瓶颈。它不再试图用链式调用来模拟复杂流程,而是将整个 Agent 系统建模为一个有状态的图(Stateful Graph)。这个图由三要素构成:
- 节点(Node):一个纯函数,接收当前状态(State),返回更新后的状态。例如
retrieve_docs(state)、generate_answer(state)、check_validation(state)。 - 边(Edge):定义节点间的流转规则,可以是固定跳转(
return "generate_answer"),也可以是条件判断(if state["retry_count"] < 3: return "retrieve_docs")。 - 状态(State):一个可变的数据结构(通常是 Pydantic 模型),承载所有节点间需要共享的数据,如
messages: list[BaseMessage]、documents: list[Document]、retry_count: int、is_valid: bool。
看一个真实 Agent 图的定义:
from typing import TypedDict, Annotated, Sequence from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import BaseMessage, HumanMessage, AIMessage # 1. 定义状态结构(关键!所有节点共享此状态) class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] # 消息列表,自动累加 documents: list[Document] # 检索到的文档 retry_count: int # 重试计数 is_valid: bool # 验证是否通过 # 2. 定义节点函数(纯函数,只读取/修改 state) def retrieve_docs(state: AgentState) -> AgentState: # 从 state["messages"][-1].content 获取问题 question = state["messages"][-1].content docs = retriever.invoke(question) return {"documents": docs, "retry_count": state.get("retry_count", 0)} def generate_answer(state: AgentState) -> AgentState: # 使用 state["documents"] 和 state["messages"] 构建 prompt context = "\n\n".join([d.page_content for d in state["documents"]]) prompt = f"基于以下上下文回答:{context}\n\n问题:{state['messages'][-1].content}" response = llm.invoke(prompt) return {"messages": [AIMessage(content=response.content)]} def check_validation(state: AgentState) -> str: # 条件边:决定下一步走哪条路 last_msg = state["messages"][-1] if "error" in last_msg.content.lower(): if state.get("retry_count", 0) < 2: return "retrieve_docs" # 重试检索 else: return "fallback_rule" # 降级 else: return END # 结束 # 3. 构建图 workflow = StateGraph(AgentState) workflow.add_node("retrieve_docs", retrieve_docs) workflow.add_node("generate_answer", generate_answer) workflow.add_node("fallback_rule", fallback_rule) workflow.set_entry_point("retrieve_docs") workflow.add_edge("retrieve_docs", "generate_answer") workflow.add_conditional_edges( "generate_answer", check_validation, { "retrieve_docs": "retrieve_docs", "fallback_rule": "fallback_rule", END: END } ) app = workflow.compile(checkpointer=MemorySaver())这段代码的关键突破在于:
- 状态显式化:
AgentState是所有节点的唯一数据源,messages字段用Annotated[Sequence, operator.add]实现自动累加,避免手动state["messages"].append()。 - 边即逻辑:
check_validation不再是节点,而是纯粹的路由函数,只返回字符串(节点名),完全解耦业务逻辑与流程控制。 - 可中断可恢复:
MemorySaver()作为检查点(Checkpoint)存储,让 Agent 在任意节点失败后,能从断点恢复,而不是从头开始。这是生产环境高可用的基石。
注意:LangGraph 的
StateGraph不是“另一个库”,它是 LangChain 生态的官方子项目(langgraph包),深度集成 LangChain 的Runnable、BaseMessage、Document等核心类型。你不需要抛弃 LangChain,而是用 LangGraph 来组织 LangChain 的组件。
2.3 为什么不能“只学 LangGraph”?——生态依赖的现实约束
很多初学者看到 LangGraph 的强大,就想跳过 LangChain 直接学图。这是个危险的捷径。LangGraph 的节点函数(Node)内部,90% 的工作依然依赖 LangChain 的组件:
retriever.invoke()来自langchain_core.retrieversllm.invoke()来自langchain_openai.ChatOpenAIDocument类来自langchain_core.documentsBaseMessage来自langchain_core.messages
LangGraph 本身不提供向量数据库、不提供 LLM 客户端、不提供文档加载器。它只提供“如何把它们串起来”的协议。就像 Linux 内核不提供浏览器、不提供编辑器,但它定义了进程调度、内存管理、文件系统接口,让 Chrome、VS Code 能在其上运行。LangChain 就是 LLM 应用的“用户空间生态”,LangGraph 是它的“内核调度器”。
我见过太多人卡在第一步:想用 LangGraph 写一个 RAG Agent,却连Chroma怎么存文档都不会,最后在retriever.invoke()报错AttributeError: 'NoneType' object has no attribute 'invoke'上折腾一整天。根源不是 LangGraph 复杂,而是跳过了 LangChain 的基础建设。真正的学习路线必须是:LangChain 打地基 → LangGraph 架梁柱 → 生产环境加固。
3. 学习路线实操指南:三个月从零到交付 Agent 系统
3.1 第一阶段:LangChain 基础夯实(2 周,目标:跑通 RAG)
不要一上来就看文档。直接动手,用最短路径验证你的 LLM 调用链路是否通畅。Mac 用户请确保已安装 Python 3.10+ 和 pip:
# 创建虚拟环境(强烈推荐) python -m venv langchain-env source langchain-env/bin/activate pip install --upgrade pip # 安装核心包(注意版本兼容性) pip install langchain-core langchain-openai langchain-chroma openai python-dotenvDay 1-3:LLM 基础调用与 Prompt 工程
目标:不依赖任何外部数据,仅用 LLM 自身能力完成任务。
import os from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 从 .env 文件读取 OPENAI_API_KEY(安全实践) from dotenv import load_dotenv load_dotenv() llm = ChatOpenAI(model="gpt-4o", temperature=0) # 构建一个能自我校验的 Prompt prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个严谨的代码审查员。请检查以下 Python 代码是否有潜在 bug,并只返回 'SAFE' 或 'BUGGY',不要解释。"), ("human", "def divide(a, b): return a / b") ]) # 组装链 chain = prompt | llm | StrOutputParser() result = chain.invoke({}) print(result) # 应该输出 'BUGGY'关键参数解析:
temperature=0:强制确定性输出,避免随机性干扰调试。StrOutputParser():将 LLM 的AIMessage对象自动提取.content字段,省去手动解析。.env文件:内容为OPENAI_API_KEY=sk-...,避免密钥硬编码。
实操心得:第一次运行报错
openai.AuthenticationError?90% 是 API Key 无效或网络问题。用curl直接测试 OpenAI API:curl https://api.openai.com/v1/models -H "Authorization: Bearer $OPENAI_API_KEY"。如果 curl 成功而 Python 失败,一定是 Python 环境的代理设置问题(Mac 上常见于系统代理开启但 pip 未配置)。
Day 4-7:RAG 全流程打通
准备一份本地 PDF(如《Python 官方文档》前 10 页),用 LangChain 加载、切分、向量化、检索:
from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings # 1. 加载 PDF loader = PyPDFLoader("python-docs.pdf") docs = loader.load() # 2. 文本切分(关键参数!) text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每块 500 字符 chunk_overlap=50, # 重叠 50 字符,避免语义断裂 length_function=len, # 使用 len() 计算长度 ) split_docs = text_splitter.split_documents(docs) # 3. 向量化并存入 Chroma(本地向量库) vectorstore = Chroma.from_documents( documents=split_docs, embedding=OpenAIEmbeddings(), # 使用 OpenAI 的 text-embedding-3-small persist_directory="./chroma_db" # 本地持久化 ) # 4. 构建检索器 retriever = vectorstore.as_retriever( search_type="similarity", # 相似度搜索 search_kwargs={"k": 3} # 返回 top3 结果 ) # 5. 测试检索 results = retriever.invoke("如何定义一个函数?") for i, doc in enumerate(results): print(f"--- Result {i+1} ---") print(doc.page_content[:200] + "...")参数选择依据:
chunk_size=500:平衡信息完整性和向量精度。太小(100)导致上下文碎片化;太大(2000)使向量失去区分度。实测 500 是通用场景最佳起点。chunk_overlap=50:确保句子不会被硬切开。50 字符约等于 1-2 个英文句子长度。search_kwargs={"k": 3}:RAG 中k不宜过大。k=5时,第 4-5 名结果常是噪声,反而稀释有效信息。k=3是精度与召回的黄金平衡点。
Day 8-14:构建第一个 RAG Chain 并部署为 API
用 FastAPI 封装成 Web API:
# app.py from fastapi import FastAPI from pydantic import BaseModel from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser app = FastAPI() class QueryRequest(BaseModel): question: str @app.post("/ask") async def ask_question(request: QueryRequest): # 复用上面构建的 rag_chain result = rag_chain.invoke({"question": request.question}) return {"answer": result}启动服务:uvicorn app:app --reload,然后用curl -X POST http://localhost:8000/ask -H "Content-Type: application/json" -d '{"question":"Python 中如何处理异常?"}'测试。
注意事项:Chroma 默认使用 SQLite,多进程访问会报错。生产环境务必加
persist_directory并确保目录可写。首次运行会下载 OpenAI Embedding 模型,耗时较长(约 2 分钟),耐心等待。
3.2 第二阶段:LangGraph Agent 构建(3 周,目标:交付可中断、可重试的 Agent)
Week 1:理解 State 与 Node 的契约关系
LangGraph 的核心是State。它不是一个简单的 dict,而是一个契约(Contract)——所有 Node 函数都必须遵守这个契约:输入是State,输出是State的部分更新(Partial Update)。这意味着你不能在 Node 里state["messages"].append(...),而必须return {"messages": [...new_message]}。
创建一个最简图,验证状态传递:
from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, Sequence from langchain_core.messages import HumanMessage, AIMessage class SimpleState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] def node_a(state: SimpleState) -> SimpleState: print("Node A running") return {"messages": [HumanMessage(content="Hello from A")]} def node_b(state: SimpleState) -> SimpleState: print("Node B running") return {"messages": [AIMessage(content="Hi from B")]} workflow = StateGraph(SimpleState) workflow.add_node("node_a", node_a) workflow.add_node("node_b", node_b) workflow.set_entry_point("node_a") workflow.add_edge("node_a", "node_b") workflow.add_edge("node_b", END) app = workflow.compile() # 执行 result = app.invoke({"messages": []}) print(result["messages"]) # [HumanMessage(...), AIMessage(...)]关键观察:
Annotated[Sequence, operator.add]让messages字段支持+=操作,node_a和node_b的返回值会自动合并。- 如果去掉
Annotated,直接用list[BaseMessage],node_b的返回会覆盖node_a的结果,而非追加。
Week 2:构建带条件分支的 RAG Agent
将之前的 RAG Chain 改造成 LangGraph Agent,加入“检索失败则重试”的逻辑:
def retrieve_with_retry(state: AgentState) -> AgentState: question = state["messages"][-1].content docs = retriever.invoke(question) if not docs and state.get("retry_count", 0) < 2: # 重试:改写问题 rewrite_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个问题改写专家。请将用户问题改写得更具体、更适合向量检索。"), ("human", "原始问题:{question}") ]) rewritten = rewrite_prompt | llm | StrOutputParser() new_question = rewritten.invoke({"question": question}) docs = retriever.invoke(new_question) return {"documents": docs, "retry_count": state.get("retry_count", 0) + 1} else: return {"documents": docs, "retry_count": state.get("retry_count", 0)} def should_continue(state: AgentState) -> str: if not state["documents"]: return "retrieve_with_retry" else: return "generate_answer" workflow = StateGraph(AgentState) workflow.add_node("retrieve_with_retry", retrieve_with_retry) workflow.add_node("generate_answer", generate_answer) workflow.set_entry_point("retrieve_with_retry") workflow.add_conditional_edges( "retrieve_with_retry", should_continue, { "retrieve_with_retry": "retrieve_with_retry", # 循环重试 "generate_answer": "generate_answer" } ) workflow.add_edge("generate_answer", END) app = workflow.compile(checkpointer=MemorySaver())调试技巧:
- 在
should_continue函数里加print(f"Retries: {state.get('retry_count', 0)}, Docs found: {len(state['documents'])}"),实时观察状态变化。 - 使用
app.get_graph().draw_mermaid_png()生成流程图(需安装graphviz),直观查看图结构。
Week 3:接入真实工具与错误处理
让 Agent 能调用外部 API,如天气查询:
import requests def get_weather(state: AgentState) -> AgentState: city = extract_city_from_message(state["messages"][-1].content) # 自定义函数 try: response = requests.get(f"http://api.weatherapi.com/v1/current.json?key=YOUR_KEY&q={city}") data = response.json() weather = f"{city} 当前温度 {data['current']['temp_c']}°C,天气 {data['current']['condition']['text']}" return {"messages": [AIMessage(content=weather)]} except Exception as e: return {"messages": [AIMessage(content=f"获取天气失败:{str(e)}")]} # 在图中添加节点 workflow.add_node("get_weather", get_weather) # 修改 should_continue,当用户问天气时跳转到 get_weather实操心得:LangGraph 的
checkpointer是救命稻草。当get_weather节点因网络超时失败,app.invoke()会抛出异常,但app.invoke(..., config={"configurable": {"thread_id": "123"}})会自动从上次成功节点恢复。没有checkpointer,每次失败都要重跑整个流程。
3.3 第三阶段:生产环境加固与性能优化(3 周,目标:上线稳定 Agent)
Week 1:可观测性与日志埋点
LangGraph 提供CallbackHandler接口,用于监听节点执行:
from langchain.callbacks.base import BaseCallbackHandler class LoggingCallback(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, **kwargs): print(f"Chain started: {serialized.get('name', 'unknown')}") def on_tool_start(self, serialized, input_str, **kwargs): print(f"Tool called: {serialized.get('name', 'unknown')}, input: {input_str}") app = workflow.compile( checkpointer=MemorySaver(), callbacks=[LoggingCallback()] # 注册回调 )更专业的方案是集成 LangSmith(LangChain 官方可观测平台):
pip install langsmith export LANGCHAIN_API_KEY=lsk-... export LANGCHAIN_TRACING_V2=true export LANGCHAIN_PROJECT="my-agent-project"启用后,所有app.invoke()调用会自动上报到 LangSmith 控制台,你可以看到完整的执行轨迹、耗时分布、token 消耗、错误堆栈。
Week 2:Token 优化与成本控制
LLM 调用是最大成本项。LangGraph 的State会累积messages,导致后续调用 token 数爆炸。解决方案:
消息截断:在
generate_answer节点中,只保留最近 5 条消息:def generate_answer(state: AgentState) -> AgentState: recent_msgs = state["messages"][-5:] # 只取最后 5 条 # ... 构建 promptContext 压缩:用 LLM 自身压缩长文档:
compress_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个文档压缩专家。请将以下文档压缩为不超过 200 字的摘要,保留所有关键事实。"), ("human", "{document}") ]) compressed = compress_prompt | llm | StrOutputParser()
Week 3:安全加固与输入过滤
Agent 安全是高频面试题。基础防护:
输入清洗:过滤恶意指令
def sanitize_input(text: str) -> str: dangerous_patterns = ["system:", "sudo", "rm -rf", "eval("] for pattern in dangerous_patterns: text = text.replace(pattern, "") return text[:1000] # 截断过长输入工具白名单:
get_weather节点只在明确指令时触发,避免 LLM 幻觉调用:def route_to_tool(state: AgentState) -> str: last_msg = state["messages"][-1].content.lower() if "weather" in last_msg or "temperature" in last_msg: return "get_weather" elif "contract" in last_msg: return "analyze_contract" else: return "generate_answer"
4. 常见问题与排查技巧实录:那些文档里不会写的坑
4.1 “图不执行,静默退出” —— 状态字段名不匹配
现象:app.invoke({"messages": [...]})后无任何输出,也不报错,result是空字典。
原因:State定义中的字段名与节点返回的键名不一致。LangGraph 的 Partial Update 是严格键匹配的。
排查步骤:
- 检查
State定义:class AgentState(TypedDict): messages: ... - 检查节点返回:
return {"messages": [...]}✅,还是return {"msg": [...]}❌ - 检查
app.get_graph().draw_mermaid_code()输出,确认图中节点输入/输出箭头标注的字段名。
经验:在
State定义中,给每个字段加默认值,强制类型检查:class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] = [] documents: list[Document] = [] retry_count: int = 0
4.2 “无限循环” —— 条件边返回了不存在的节点名
现象:app.invoke()卡住,CPU 占用 100%,日志显示Node A -> Node A -> Node A...
原因:add_conditional_edges的mapping字典中,某个条件分支返回了图中不存在的节点名,LangGraph 默认循环回当前节点。
排查步骤:
- 在条件函数中加
print(f"Routing to: {next_node}"),确认返回值。 - 用
app.get_graph().all_nodes()列出所有注册节点,对比返回值。 - 确保
mapping字典的 key 是字符串,且与add_node("node_name", ...)的第一个参数完全一致(大小写敏感)。
4.3 “状态丢失” —— 忘记Annotated的operator.add
现象:messages字段只保留最后一次节点的返回值,历史消息消失。
原因:messages: list[BaseMessage]是普通 list,LangGraph 不会自动合并;必须用Annotated[Sequence[BaseMessage], operator.add]。
修复:在State定义中,为所有需要累加的字段加上Annotated和operator.add。
4.4 “Chroma 初始化失败” —— SQLite 权限与路径问题
现象:Chroma.from_documents(...)报错sqlite3.OperationalError: unable to open database file。
原因:Mac 上默认/tmp目录可能被 SIP 保护,或persist_directory路径不存在/无写权限。
解决方案:
- 使用绝对路径:
persist_directory="/Users/yourname/chroma_db" - 确保目录存在:
os.makedirs("/Users/yourname/chroma_db", exist_ok=True) - 避免使用
~符号,它在 Python 中不会自动展开。
4.5 “LLM 调用超时” —— 网络与重试配置
现象:llm.invoke()长时间无响应,最终抛出TimeoutError。
LangChain 的ChatOpenAI支持重试:
llm = ChatOpenAI( model="gpt-4o", temperature=0, max_retries=3, # 自动重试 3 次 timeout=30.0, # 单次请求超时 30 秒 )更彻底的方案是使用tenacity库自定义重试策略:
from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def robust_invoke(llm, prompt): return llm.invoke(prompt)4.6 “Agent 不调用工具” —— LLM 的指令遵循能力不足
现象:用户明确说“查北京天气”,Agent 却生成一段关于天气的科普文,不调用get_weather工具。
原因:LLM 的tool_choice参数未正确设置,或 Prompt 中工具描述不够清晰。
解决方案:
- 强制 LLM 选择工具:
llm.bind(tools=[get_weather_tool]) - 在 System Prompt 中明确指令:“你必须使用提供的工具来回答问题。如果问题需要工具,请调用工具;否则,直接回答。”
5. 工具选型与生态定位:Agent 开发不是闭门造车
5.1 LangChain vs Dify vs CrewAI:它们解决什么问题?
- LangChain:组件库(Component Library)。提供
LLM、Retriever、Embeddings、Document等原子能力,让你自己组装。适合需要精细控制、定制化强的场景。 - Dify:低代码平台(Low-Code Platform)。提供可视化编排界面、API 服务、知识库管理后台。适合快速验证 MVP、非技术产品经理主导的项目。
- CrewAI:多 Agent 协作框架(Multi-Agent Framework)。专注于多个 Agent(如 researcher、writer、reviewer)之间的角色分工与协作。适合需要复杂团队模拟的场景(如市场分析报告生成)。
它们不是互斥的。真实项目中,我常用组合:用 Dify 快速搭建客户演示版,用 LangChain + LangGraph 开发核心业务逻辑,用 CrewAI 实现特定环节的多角色协同。
5.2 RAG 瓶颈的真相:不是技术,是数据质量
热搜词rag瓶颈常被归咎于向量模型或检索算法。但我在 7 个 RAG 项目中发现,80% 的效果不佳源于数据层:
- 文档格式混乱:PDF 中的表格、图片、页眉页脚被错误解析为文本。
- 语义断裂:
RecursiveCharacterTextSplitter在代码块或数学公式处硬切,破坏逻辑完整性。 - 元数据缺失:未给文档打上
source、page_number、section_title等标签,导致检索结果无法溯源。
解决方案:
- 用
unstructured库替代PyPDFLoader,它能更好处理复杂 PDF。 - 对代码文档,用
LanguageChunker(按函数/类切分);对法律条文,用HeaderTypeTextSplitter(按标题层级切分)。 - 在
Document对象中注入丰富元数据:Document(page_content=..., metadata={"source": "contract_v2.pdf", "clause_id": "4.2"})。
5.3 Agent 安全的三个层次
- 输入层:过滤恶意指令(如
system:、sudo)、截断过长输入、字符集白名单。 - 执行层:工具调用前做参数校验(如城市名是否在白名单)、限制工具调用频率(每分钟最多 5 次)。
- 输出层:LLM 生成后,用规则或小模型做敏感词过滤(如 PII 信息)、事实核查(对比知识库)。
没有银弹。安全是纵深防御,每一层都不可或缺。
我在实际交付中,最后总要加一道“人工审核开关”:当 Agent 输出包含legal、financial、medical等高风险关键词时,自动转交人工处理。技术再强,也不能替代专业责任。
这个学习路线,不是纸上谈兵的理论堆砌,而是我把三年来踩过的每一个坑、验证过的每一个参数、压测过的每一个瓶颈,浓缩成的实战手册。LangChain 和 LangGraph 的区别,从来不是“哪个更好”,而是“在什么阶段用什么”。当你能清晰说出“我现在需要的是快速验证 LLM 能力,所以用 LangChain Chain”,或者“我现在需要保证 99.9% 的流程成功率,所以必须用 LangGraph Checkpoint”,你就真正掌握了这门手艺。剩下的,就是不断在真实需求里打磨。