1. 从零搭建智能体系统:为什么选择LangChain生态
1.1 三个框架到底解决什么问题
很多刚接触智能体开发的朋友会有一个困惑:LangChain、LangGraph、Deep Agents这三个东西名字都带“Lang”,到底该用哪个?我刚开始也绕了不少弯路,后来在几个实际项目中反复切换之后才理清楚它们各自的定位。
LangChain本质上是一套组件库和抽象层。它把大模型调用、提示词模板、输出解析、工具调用、记忆管理这些零散的能力封装成统一接口。你可以把它理解成一个“乐高积木箱”,里面什么零件都有,但怎么拼、拼成什么样,完全取决于你自己。它的优势在于生态成熟、文档丰富、社区案例多,几乎你遇到的任何模型提供商、向量数据库、工具服务,LangChain都有对应的集成包。
LangGraph则是在LangChain基础上长出来的编排引擎。当你需要构建一个有状态、有分支、有循环的复杂流程时,单纯用LangChain的Chain串联会非常吃力。LangGraph把整个执行过程建模成一张有向图,节点是计算步骤,边是流转条件,状态在节点之间传递和累积。这个设计让“让模型自己决定下一步走哪条路”变得非常自然,也是构建多轮推理、工具调用循环、多角色协作的基础。
Deep Agents是更高层的智能体模式封装。它把一些经过验证的智能体架构(比如规划-执行-反思循环、子智能体委派、长期记忆管理)做成可复用的模板。你不需要从零设计提示词和状态机,直接套用这些模式就能得到一个行为相对可靠的智能体。适合快速验证想法,或者在标准场景下减少重复劳动。
注意:这三个框架不是替代关系,而是层次关系。LangChain提供基础能力,LangGraph提供编排能力,Deep Agents提供模式能力。实际项目中经常是三者混用。
1.2 什么场景该用哪一层
我自己的判断标准很简单:看你的流程是否需要“模型自主决策下一步”。
如果流程是固定的——比如“用户提问→检索知识库→拼接提示词→调用模型→返回答案”,这种线性管道用LangChain的LCEL表达式就够了,写起来快,调试也直观。没必要上LangGraph,反而增加复杂度。
如果流程中存在条件分支、循环重试、多步推理、工具调用后根据结果决定下一步,那就必须用LangGraph。比如一个客服智能体,需要先判断用户意图,再决定是查订单、退换货还是转人工,每一步的结果都可能影响后续走向,这种场景用LangGraph的状态图来表达最清晰。
如果你要构建的是一个“通用型助手”,需要它自己拆解任务、自己调用工具、自己检查结果并修正,那Deep Agents的模式封装能帮你省掉大量提示词工程。但代价是灵活性降低,遇到特殊需求时还是得回到LangGraph层面去改。
1.3 环境准备与依赖管理
我习惯用uv来管理Python环境,比pip快很多,依赖解析也更稳。下面是我在一个新项目里初始化环境的完整流程:
# 创建项目目录 mkdir agent-demo && cd agent-demo # 初始化uv环境 uv init # 添加核心依赖 uv add langchain langchain-openai langgraph deepagents # 添加工具类依赖 uv add tavily-python python-dotenv rich这里有几个选型考量。langchain-openai是官方维护的OpenAI集成包,更新最及时。tavily-python是给智能体用的搜索工具,免费额度够开发测试。rich用来在终端里打印漂亮的调试信息,看状态流转时特别有用。
环境变量统一放在.env文件里:
OPENAI_API_KEY=your_key_here TAVILY_API_KEY=your_key_here实操心得:不要把密钥硬编码在代码里,也不要在调试时把密钥打印到日志。我见过有人把带密钥的报错截图发到群里,后果很麻烦。用
.env加.gitignore是最基本的习惯。
依赖装好之后,先跑一个最小验证,确认模型能通:
from langchain_openai import ChatOpenAI from dotenv import load_dotenv load_dotenv() llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) print(llm.invoke("用一句话解释什么是状态图").content)这一步能跑通,后面的工作才有基础。如果这里就报错,先检查密钥、网络、模型名称这三项,别急着往下写。
2. LangChain核心组件拆解与实操要点
2.1 提示词模板与输出解析的配合
LangChain里最基础也最容易被低估的就是提示词模板。很多人直接写f-string拼接,觉得模板是多余的一层。但在实际项目里,模板带来的结构化和可复用性非常关键。
我拿一个“从用户反馈中提取结构化信息”的任务举例。原始做法是拼字符串然后让模型返回JSON,再手动解析。用LangChain的模板加解析器是这样的:
from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field class Feedback(BaseModel): sentiment: str = Field(description="情感倾向:正面/负面/中性") category: str = Field(description="问题分类:功能/性能/界面/其他") summary: str = Field(description="一句话总结") parser = JsonOutputParser(pydantic_object=Feedback) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个用户反馈分析助手。\n{format_instructions}"), ("human", "分析以下反馈:\n{feedback}") ]).partial(format_instructions=parser.get_format_instructions()) chain = prompt | llm | parser result = chain.invoke({"feedback": "这个页面加载太慢了,等半天打不开"}) print(result)这里的关键在于get_format_instructions()会自动把Pydantic模型的字段说明注入到提示词里,模型知道该返回什么结构。解析器负责把模型输出转成Python字典,如果格式不对会抛异常,方便你捕获处理。
注意事项:即使有格式指令,模型偶尔还是会返回多余的文字。生产环境里一定要给解析加try-except,并准备一个降级方案,比如重新调用一次或者返回默认值。
2.2 工具调用的绑定与执行
工具调用是智能体的手脚。LangChain里用@tool装饰器把一个普通函数变成模型可调用的工具:
from langchain_core.tools import tool @tool def query_order(order_id: str) -> str: """根据订单号查询订单状态。输入应该是纯数字的订单号。""" # 实际项目中这里会查数据库 mock_data = {"12345": "已发货", "67890": "待付款"} return mock_data.get(order_id, "未找到该订单") @tool def calculate_refund(amount: float, days: int) -> str: """计算退款金额。amount是原价,days是购买天数。""" if days <= 7: return f"全额退款:{amount}元" else: return f"扣除手续费后退款:{amount * 0.9}元"工具函数的文档字符串非常重要,模型就是靠这段描述来判断什么时候该调用哪个工具。我踩过的坑是文档写得太模糊,比如只写“查询订单”,模型分不清是查物流还是查支付状态。后来改成“根据订单号查询订单的当前状态,包括待付款、已发货、已完成”,准确率明显提升。
绑定工具并执行:
llm_with_tools = llm.bind_tools([query_order, calculate_refund]) response = llm_with_tools.invoke("帮我查一下订单12345的状态") print(response.tool_calls)返回的tool_calls里包含工具名和参数,你需要自己写循环去执行并把结果喂回模型。这个循环在LangGraph里会被自动化,但在LangChain层面需要手动处理。
2.3 记忆管理的几种方案对比
智能体需要记住对话历史,否则每轮都是失忆状态。LangChain提供了几种记忆方案,我整理了一个对比表:
| 方案 | 存储位置 | 适用场景 | 注意事项 |
|---|---|---|---|
| ConversationBufferMemory | 内存 | 短对话、原型验证 | 对话长了会撑爆上下文 |
| ConversationSummaryMemory | 内存 | 长对话、成本敏感 | 摘要会丢失细节 |
| ConversationBufferWindowMemory | 内存 | 需要最近N轮 | 窗口大小要调 |
| 向量存储检索记忆 | 外部数据库 | 跨会话、大量历史 | 需要嵌入模型和检索逻辑 |
我个人的经验是:开发阶段用BufferWindow,窗口设10轮左右;生产环境用向量存储加摘要的混合方案。纯摘要会丢细节,纯向量检索又可能召回不相关内容,两者结合效果最稳。
from langchain.memory import ConversationBufferWindowMemory memory = ConversationBufferWindowMemory(k=10, return_messages=True) memory.save_context({"input": "我叫张三"}, {"output": "你好张三"}) print(memory.load_memory_variables({}))实操心得:记忆不是越多越好。我做过一个测试,把20轮历史全部塞进上下文,模型反而更容易跑偏,因为它被早期无关信息干扰了。控制在最近5到10轮,加上关键信息的摘要,效果最好。
3. LangGraph状态图编排实战
3.1 状态定义与节点设计
LangGraph的核心概念是状态和节点。状态是一个共享的数据结构,所有节点都能读写;节点是处理函数,接收状态返回更新。
先定义状态。我用TypedDict来声明,清晰且类型安全:
from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_intent: str order_info: str final_answer: stradd_messages是一个reducer函数,它定义了当多个节点都想更新messages时该怎么合并——这里是追加而不是覆盖。这个细节很关键,如果不加Annotated,后一个节点的更新会直接把前面的消息冲掉。
节点就是普通函数:
def classify_intent(state: AgentState): last_msg = state["messages"][-1].content # 实际项目中这里调用模型判断意图 if "订单" in last_msg: intent = "query_order" elif "退款" in last_msg: intent = "refund" else: intent = "general" return {"user_intent": intent} def handle_order(state: AgentState): # 处理订单查询逻辑 return {"order_info": "订单12345已发货"} def generate_answer(state: AgentState): answer = f"根据您的{state['user_intent']}请求,结果是:{state.get('order_info', '无相关信息')}" return {"final_answer": answer, "messages": [("ai", answer)]}3.2 条件边与循环控制
LangGraph真正强大的地方在于条件边。你可以根据状态里的值决定下一步走哪个节点:
from langgraph.graph import StateGraph, END def route_by_intent(state: AgentState): if state["user_intent"] == "query_order": return "handle_order" elif state["user_intent"] == "refund": return "handle_refund" else: return "generate_answer" builder = StateGraph(AgentState) builder.add_node("classify", classify_intent) builder.add_node("handle_order", handle_order) builder.add_node("handle_refund", handle_refund) builder.add_node("answer", generate_answer) builder.set_entry_point("classify") builder.add_conditional_edges("classify", route_by_intent, { "handle_order": "handle_order", "handle_refund": "handle_refund", "generate_answer": "answer" }) builder.add_edge("handle_order", "answer") builder.add_edge("handle_refund", "answer") builder.add_edge("answer", END) graph = builder.compile()这段代码构建了一个根据意图分流的图。add_conditional_edges的第三个参数是路径映射,把路由函数的返回值映射到实际节点名。这个映射必须完整,否则运行时会报错。
循环控制也是类似思路。比如工具调用场景,模型返回工具调用请求后进入工具执行节点,执行完再回到模型节点判断是否还需要继续调用,直到模型返回最终答案才结束。这个循环用条件边实现:
def should_continue(state: AgentState): last_msg = state["messages"][-1] if hasattr(last_msg, "tool_calls") and last_msg.tool_calls: return "tools" return END builder.add_conditional_edges("agent", should_continue, { "tools": "tools", END: END }) builder.add_edge("tools", "agent")注意事项:循环一定要有终止条件。我见过有人写的图在工具调用失败时无限循环,把额度跑光了。建议在状态里加一个计数器,超过阈值强制结束。
3.3 状态持久化与断点恢复
LangGraph支持把状态存到外部存储,实现断点恢复和人工介入。用MemorySaver做内存持久化:
from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() graph = builder.compile(checkpointer=memory) config = {"configurable": {"thread_id": "user-001"}} result = graph.invoke({"messages": [("human", "查订单12345")]}, config)thread_id是会话标识,同一个id的多次调用会共享状态。这意味着你可以先执行一半,过一会儿再用同样的id继续,状态不会丢。
生产环境建议换成数据库持久化,比如Postgres或SQLite。LangGraph提供了对应的checkpointer实现,切换成本很低。这个能力在做“人工审核”流程时特别有用:智能体执行到某个节点暂停,等人工确认后再继续。
4. Deep Agents模式与多智能体协作
4.1 规划-执行-反思循环的实现
Deep Agents封装的核心模式之一是规划-执行-反思。智能体先拆解任务生成计划,然后逐步执行,每步执行后反思结果是否符合预期,不符合就调整计划。
这个模式用LangGraph实现大概是这样的结构:
class DeepAgentState(TypedDict): task: str plan: list current_step: int results: list reflection: str def planner(state: DeepAgentState): # 调用模型生成任务计划 plan = ["步骤1:收集信息", "步骤2:分析数据", "步骤3:生成报告"] return {"plan": plan, "current_step": 0} def executor(state: DeepAgentState): step = state["plan"][state["current_step"]] # 执行当前步骤 result = f"完成:{step}" return {"results": state["results"] + [result]} def reflector(state: DeepAgentState): # 反思执行结果 reflection = "结果符合预期,继续下一步" return {"reflection": reflection, "current_step": state["current_step"] + 1} def should_continue(state: DeepAgentState): if state["current_step"] >= len(state["plan"]): return END return "executor"这个循环的关键在于反思节点的提示词设计。我试过几种写法,最有效的是让模型回答三个问题:当前结果是否完成了预期目标?有没有遗漏或错误?下一步需要调整吗?把这三个问题的答案作为反思输出,比笼统地让模型“评价一下”要具体得多。
4.2 子智能体委派与任务分解
复杂任务单靠一个智能体容易顾此失彼。Deep Agents支持把子任务委派给专门的子智能体,每个子智能体有自己的工具集和提示词。
比如一个“市场调研”任务,可以拆成三个子智能体:一个负责搜索公开信息,一个负责分析数据,一个负责撰写报告。主智能体只负责协调和汇总。
def delegate_task(state: DeepAgentState): task = state["task"] if "搜索" in task: return {"next_agent": "search_agent"} elif "分析" in task: return {"next_agent": "analysis_agent"} else: return {"next_agent": "writer_agent"}每个子智能体可以是一个独立的LangGraph图,主图通过节点调用子图。这种嵌套结构让系统模块化,每个子智能体可以单独测试和优化。
实操心得:子智能体之间的接口约定非常重要。我建议统一用字典格式传递数据,字段名提前定好,不要一个用
result一个用output。否则调试时会花大量时间在字段对齐上。
4.3 长期记忆与知识沉淀
Deep Agents的另一个特点是长期记忆管理。短期记忆是对话历史,长期记忆是跨会话的知识积累。
实现方式通常是把重要信息抽取出来存到向量数据库,下次遇到相关任务时检索出来注入上下文。LangChain提供了VectorStoreRetrieverMemory来简化这个流程:
from langchain.memory import VectorStoreRetrieverMemory from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings embeddings = OpenAIEmbeddings() vectorstore = FAISS.from_texts(["用户偏好简洁回答", "用户是技术背景"], embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 2}) memory = VectorStoreRetrieverMemory(retriever=retriever)什么信息值得存长期记忆?我的经验是三类:用户偏好(喜欢什么风格、什么格式)、领域知识(项目特有的术语、规则)、历史决策(之前做过什么选择、为什么)。其他临时信息没必要存,存多了反而干扰检索。
5. 常见问题与排查技巧实录
5.1 模型不调用工具怎么办
这是最常见的问题。模型明明绑定了工具,但就是不用,直接自己编答案。排查思路按顺序来:
第一,检查工具描述是否清晰。模型判断是否调用工具,完全依赖工具的名称和文档字符串。如果描述太短或太模糊,模型会倾向于自己回答。我一般要求工具描述至少包含“做什么、输入是什么、返回什么”三个要素。
第二,检查提示词是否给了调用工具的理由。有时候需要在系统提示词里明确说“当需要查询实时信息时,必须使用提供的工具,不要自己编造”。这句话能显著提升工具调用率。
第三,检查模型能力。小模型在工具调用上的表现明显弱于大模型。如果用的是轻量模型,可能需要换一个或者增加few-shot示例。
第四,检查工具参数定义。如果参数类型复杂或者有嵌套结构,模型容易搞错。尽量用简单类型,字符串和数字最稳。
5.2 状态更新不生效的排查
LangGraph里状态更新不生效,九成是reducer的问题。默认情况下,节点返回的字典会覆盖对应字段。如果你希望追加而不是覆盖,必须用Annotated加reducer函数。
我踩过的一个坑是:在多个节点里都返回messages字段,但忘了加add_messages,结果只有最后一个节点的消息留下了。排查方法很简单,在节点函数里打印返回的字典,再在图的出口打印最终状态,对比一下就知道哪一步丢了。
另一个常见问题是状态字段名拼写不一致。Python字典不会报错,但图会静默忽略不存在的字段。建议用TypedDict定义状态,这样IDE和类型检查能帮你发现拼写错误。
5.3 循环失控与成本控制
智能体循环调用工具时,如果工具一直返回错误或者模型一直不满意,就会无限循环。防护措施有三层:
第一层,在状态里加计数器,每次循环加一,超过阈值强制走END。阈值设多少取决于任务复杂度,我一般设5到10。
第二层,在工具执行节点加异常捕获,工具报错时返回一个明确的错误信息而不是抛异常,让模型知道失败了并决定是否重试。
第三层,设置API调用的超时和重试上限。LangChain的模型调用支持max_retries参数,设成2或3就够了,再多也是浪费。
成本控制方面,我习惯在开发阶段用便宜的小模型跑通流程,确认逻辑没问题后再换成大模型。另外,把temperature设成0能减少随机性,也间接降低重试概率。
5.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 模型不调用工具 | 工具描述模糊 | 打印工具schema检查 | 完善文档字符串 |
| 状态字段丢失 | 缺少reducer | 对比节点输入输出 | 加Annotated和reducer |
| 图执行报错 | 条件边映射不全 | 检查路由函数返回值 | 补全映射字典 |
| 循环不终止 | 缺少终止条件 | 打印每轮状态 | 加计数器或阈值 |
| 记忆不生效 | thread_id不一致 | 检查config参数 | 统一会话id |
| 输出解析失败 | 模型返回多余文字 | 打印原始输出 | 加异常捕获和降级 |
避坑技巧:调试LangGraph时,把
graph.stream()用起来,它能逐步输出每个节点的状态变化,比invoke一次性返回全部结果要直观得多。配合rich库的pprint,状态流转一目了然。
6. 从原型到生产的工程化建议
6.1 配置与代码分离
原型阶段把模型名、温度、提示词都写在代码里没问题,但上生产前一定要抽出来。我用pydantic-settings来管理配置:
from pydantic_settings import BaseSettings class Settings(BaseSettings): model_name: str = "gpt-4o-mini" temperature: float = 0.0 max_retries: int = 3 max_loops: int = 8 class Config: env_file = ".env" settings = Settings()这样切换环境只需要改环境变量,不用动代码。提示词也可以存到外部文件或数据库,方便非技术人员调整。
6.2 可观测性建设
智能体系统的调试比传统程序难,因为中间步骤多且不确定。我建议至少记录三类日志:每次模型调用的输入输出、每个节点的状态变化、工具调用的参数和结果。
LangChain生态里有LangSmith可以做全链路追踪,开发阶段用免费额度足够。如果不想用外部服务,自己用Python的logging模块记录到文件也行,关键是格式要结构化,方便后续检索。
import logging import json logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") def log_node(node_name, state): logging.info(json.dumps({ "node": node_name, "state_keys": list(state.keys()), "timestamp": time.time() }, ensure_ascii=False))6.3 测试策略
智能体系统的测试不能只靠单元测试,因为模型输出有随机性。我的做法是分三层:
第一层,组件测试。单独测试每个工具函数、每个解析器,确保输入输出符合预期。这部分可以用传统单元测试。
第二层,流程测试。用固定的输入跑整个图,检查最终输出是否在可接受范围内。这里不要求完全一致,但要求关键信息正确。
第三层,回归测试。收集一批真实场景的输入输出对,每次改动后跑一遍,看通过率有没有下降。这个最花时间但最有价值。
实操心得:给智能体写测试时,断言要宽松。比如不要断言“输出等于某句话”,而是断言“输出包含订单号”或“输出情感分类为负面”。太严格的断言会让你频繁改测试,反而失去意义。
6.4 性能优化方向
智能体系统的延迟主要来自模型调用次数。优化思路有几个:
减少不必要的模型调用。比如意图分类可以用小模型或者规则引擎先过滤,只有复杂情况才走大模型。
并行化独立节点。LangGraph支持并行执行没有依赖关系的节点,把能并行的搜索、查询操作放在同一层,能省不少时间。
缓存重复调用。同样的输入和参数,结果可以直接缓存。LangChain提供了set_llm_cache接口,开发阶段用内存缓存,生产环境用Redis。
流式输出。对于面向用户的场景,用stream模式让用户尽早看到部分结果,体验上比等全部完成再显示要好得多。
这些优化不需要一次性全做,根据实际瓶颈逐步加就行。我一般先上线跑一段时间,看日志里哪个环节最慢,再针对性优化。