news 2026/10/14 20:50:52

LangChain、LangGraph、Deep Agents智能体开发实战指南

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
LangChain、LangGraph、Deep Agents智能体开发实战指南

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: str

add_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模式让用户尽早看到部分结果,体验上比等全部完成再显示要好得多。

这些优化不需要一次性全做,根据实际瓶颈逐步加就行。我一般先上线跑一段时间,看日志里哪个环节最慢,再针对性优化。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/10/14 20:50:10

智慧校园供应商考察报告怎么写?关键要素与评估标准全解析

去年帮一所高校做智慧校园数据中台的选型&#xff0c;前后面谈了六家供应商&#xff0c;每家都拍着胸脯说自家平台架构先进、接口开放、实施经验丰富。可等到考察报告写完、提交评审会的时候&#xff0c;我心里清楚&#xff0c;六家里面至少有两家在“先进架构”四个字上掺了水…

作者头像 李华
网站建设 2026/10/14 20:49:59

3天从85%降到20%!这3个降AIGC平台让我导师直夸原创

还记得上周三凌晨两点&#xff0c;当我第三次收到知网AIGC检测报告时&#xff0c;手心都在冒汗——85%的AI相似度&#xff0c;40%的查重率&#xff0c;这意味着我的毕业论文根本达不到盲审要求。导师直接在我的初稿上批注“学术合规性存疑&#xff0c;建议重写”&#xff0c;那…

作者头像 李华
网站建设 2026/10/14 20:49:19

Fortran实现卡尔曼滤波:从矩阵公式到工程落地避坑指南

简介&#xff1a;这是一份用Fortran实现卡尔曼滤波算法的完整程序包&#xff0c;面向控制系统设计与导航定位、信号处理等领域的开发者、研究生和本科生&#xff0c;可帮助理解噪声环境中状态估计与预测更新机制。压缩包共38个文件&#xff0c;rar格式&#xff0c;体积约959KB&…

作者头像 李华
网站建设 2026/10/14 20:48:42

Superpowers开发者增强工具链:Cursor+Claude+Antigravity+Codex实战指南

1. 项目概述&#xff1a;Superpowers 不是超能力&#xff0c;而是开发者效率的“质变杠杆”你搜“superpowers”时&#xff0c;看到的不是漫威电影里的雷神之锤&#xff0c;也不是DC宇宙里的氪星血脉——而是一群人在深夜调试接口时突然拍桌惊呼&#xff1a;“这功能怎么敢这么…

作者头像 李华
网站建设 2026/10/14 20:48:33

宠物论坛小程序V4实战:云开发架构、内容审核与性能优化

做微信小程序宠物论坛这个项目&#xff0c;我接手时已经走到第4版了。前3版其实问题不少&#xff1a;代码写到后面越来越乱&#xff0c;帖子列表一多就开始白屏&#xff0c;用户反馈“刷新慢、评论要等半天”&#xff0c;后台审核基本靠人工一个字一个字看。到了V4&#xff0c;…

作者头像 李华
网站建设 2026/10/14 20:47:05

报文交换 vs 分组交换:串行与并行如何决定传输效率

前阵子帮朋友调一个内网传输的老问题&#xff1a;同样的百兆链路&#xff0c;同时传二十个小文件居然比传一个几百兆的大文件还要快。乍一听不合直觉&#xff0c;小文件多路并发&#xff0c;在网络里是典型的“分组并行”&#xff1b;大文件虽然被TCP拆成了段&#xff0c;但应用…

作者头像 李华