1. 从链式调用到图式编排为什么我最终把三个框架揉在一起用去年下半年开始我陆续接手了几个基于大模型的自动化项目从最早的简单问答机器人到后来的多步骤文档处理流水线再到需要动态决策的复杂任务代理。一开始我只用LangChain觉得它的链式调用已经够用了——把提示词模板、模型调用、输出解析器串起来跑通一个流程并不难。但项目稍微复杂一点问题就来了我需要根据中间结果决定下一步走哪个分支需要在某个环节失败时回退重试还需要让多个子任务并行执行再汇总。这些需求用LangChain的线性链去硬凑代码很快就变成了一团乱麻各种if-else嵌套和手动状态管理维护成本极高。后来我接触到了LangGraph它把工作流建模成一张有向图节点是具体的处理单元边定义了流转逻辑还支持条件边、循环、并行分支和持久化状态。这让我眼前一亮——终于可以用图的方式去描述复杂流程了。但LangGraph本身偏底层很多能力需要自己搭比如工具调用、记忆管理、多代理协作这些如果每个项目都从头写一遍效率太低。再后来Deep Agents出现了它把深度代理的常见模式封装成了更高层的抽象比如规划、执行、反思、工具使用等开箱即用。但Deep Agents也不是万能的遇到高度定制化的流程还是得回到LangGraph去手搓图结构。所以我的结论是这三个东西不是替代关系而是分层协作的关系。LangChain提供基础的模型、工具、提示词、解析器等原子能力LangGraph负责把这些原子能力编排成可控的工作流Deep Agents则在LangGraph之上封装了代理的常见行为模式让你少写重复代码。把它们揉在一起用才能既保证灵活性又保证开发效率。这篇文章我就把这套组合拳的实战经验完整拆开从环境搭建到核心概念再到一个完整的项目案例最后聊聊我踩过的坑和性能调优的心得。2. 环境准备与版本兼容别让依赖冲突毁掉你的周末2.1 三个框架的安装顺序与版本锁定很多人上来就pip install langchain langgraph deepagents一把梭结果跑起来各种ImportError。我踩过最坑的一次是LangChain的某个子包版本和LangGraph要求的核心库版本不兼容导致图编译时报了一个完全看不懂的序列化错误。后来我学乖了安装顺序和版本锁定必须严格对待。我的建议是先用虚拟环境隔离然后按以下顺序安装python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate # 先装核心依赖锁定大版本 pip install langchain0.3,0.4 pip install langchain-core0.3,0.4 pip install langgraph0.2,0.3 pip install deepagents0.1,0.2 # 再装模型提供商和工具库 pip install langchain-openai0.2,0.3 pip install langchain-community0.3,0.4为什么要先装langchain-core因为LangGraph和Deep Agents都依赖它如果先装LangGraph它可能会拉一个较老的core版本后面再装LangChain时又会被覆盖导致版本错乱。先锁定core的大版本再装上层框架能最大程度避免这种问题。注意不要用pip install -U无脑升级所有包大版本升级经常带来破坏性变更。我一般会在requirements.txt里写死小版本号比如langgraph0.2.60确保团队里每个人跑出来的结果一致。2.2 API Key与模型配置的集中管理三个框架都支持多种模型提供商但配置方式各有不同。我的做法是统一用环境变量管理密钥然后在代码里通过一个工厂函数创建模型实例。这样切换模型时只需要改一个地方。import os from langchain_openai import ChatOpenAI def get_llm(model_name: str gpt-4o-mini, temperature: float 0.0): return ChatOpenAI( modelmodel_name, temperaturetemperature, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), # 可选用于兼容其他服务 )这里有个细节temperature设成0.0对于需要稳定输出的工作流非常重要。我在做文档分类任务时temperature0.7导致同样的输入有时分到A类有时分到B类排查了半天才发现是随机性在作祟。对于代理类任务建议默认用0.0需要创意生成时再单独调高。2.3 目录结构规划让项目从第一天就整洁我见过太多项目把所有代码堆在一个main.py里到后面根本没法维护。我的习惯是按职责分层project/ ├── config/ │ └── settings.py # 模型、工具、常量配置 ├── core/ │ ├── llm_factory.py # 模型实例工厂 │ ├── tools.py # 自定义工具定义 │ └── state.py # 图状态定义 ├── graphs/ │ ├── simple_chain.py # 基础链式流程 │ ├── conditional_graph.py # 带条件分支的图 │ └── multi_agent.py # 多代理协作图 ├── agents/ │ └── deep_agent.py # Deep Agents封装 ├── utils/ │ └── helpers.py # 通用工具函数 └── main.py # 入口这个结构的好处是当你的图越来越复杂时每个图独立一个文件状态定义集中管理工具函数复用方便。我后来把某个图从单文件拆成三个模块后调试效率至少提升了一倍。3. LangChain的定位它到底该干哪些活不该干哪些活3.1 提示词模板与输出解析把不确定性关进笼子LangChain最核心的价值之一是它提供了一套标准化的提示词管理和输出解析机制。很多人觉得提示词就是拼字符串没必要用框架。但当你需要动态注入变量、处理多轮对话历史、强制模型输出JSON格式时手写字符串很快就会失控。我常用的组合是ChatPromptTemplate加PydanticOutputParser。前者负责把系统指令、历史消息、用户输入组装成模型能理解的格式后者负责把模型的自由文本输出解析成结构化对象。from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field class TaskPlan(BaseModel): steps: list[str] Field(description执行步骤列表) priority: str Field(description优先级high/medium/low) parser PydanticOutputParser(pydantic_objectTaskPlan) prompt ChatPromptTemplate.from_messages([ (system, 你是一个任务规划助手。请根据用户需求生成执行计划。\n{format_instructions}), (human, {user_input}), ]).partial(format_instructionsparser.get_format_instructions()) chain prompt | get_llm() | parser result chain.invoke({user_input: 帮我整理一份季度报告}) print(result.steps, result.priority)这里的关键是get_format_instructions()它会自动生成一段描述JSON schema的文本告诉模型应该输出什么格式。实测下来加上这段指令后模型输出格式错误的概率从大约15%降到了2%以下。如果还是偶尔出错可以再加一层重试逻辑用with_retry或者自己写个循环。3.2 工具定义让模型能动手而不是只动嘴LangChain的工具抽象让模型可以调用外部函数这是构建代理的基础。定义一个工具最简单的方式是用tool装饰器from langchain_core.tools import tool tool def search_docs(query: str) - str: 根据关键词搜索内部文档库返回最相关的三条记录。 # 实际实现可能是向量检索或数据库查询 return f关于{query}的搜索结果... tool def calculate(expression: str) - float: 计算数学表达式支持加减乘除和括号。 return eval(expression, {__builtins__: {}}, {})注意docstring非常重要模型就是靠它来判断什么时候该调用这个工具。我踩过的坑是docstring写得太模糊比如只写搜索结果模型在需要计算的时候也去调搜索工具。后来我把每个工具的docstring都写成什么场景下用、输入是什么、返回什么的完整说明工具调用的准确率明显提升。另外eval那里我加了{__builtins__: {}}来限制命名空间这是安全实践。虽然在这个场景下输入来自模型但养成好习惯总没错。3.3 LangChain不擅长的事别硬撑LangChain的链式结构本质上是线性的虽然可以用RunnableBranch做条件分支但一旦分支嵌套超过两层代码可读性就急剧下降。我试过用LangChain实现一个根据文档类型走不同处理流程每个流程内部又有多个步骤某些步骤失败要回退的逻辑写了三百多行自己回头看都费劲。这种场景就应该交给LangGraph。LangChain负责的是原子操作——一次模型调用、一次工具执行、一次解析。至于这些原子操作怎么组合、怎么流转、怎么管理状态那是LangGraph的活。把职责分清楚代码才不会变成一锅粥。4. LangGraph核心机制状态、节点、边与检查点4.1 状态定义图的大脑LangGraph的核心概念是状态State它是一个贯穿整个图执行过程的数据结构。每个节点读取状态、处理、然后返回状态的更新部分。定义状态通常用TypedDictfrom typing import TypedDict, Annotated from langgraph.graph import add_messages class GraphState(TypedDict): messages: Annotated[list, add_messages] current_step: str retry_count: int result: str | NoneAnnotated[list, add_messages]这个写法很关键。它告诉LangGraph当多个节点都往messages里追加内容时用add_messages这个函数来合并而不是直接覆盖。我一开始没加这个注解结果并行分支的消息互相覆盖丢了数据排查了很久。状态设计的原则是只放需要跨节点共享的数据。临时变量放在节点内部就好不要什么都往状态里塞否则状态会变得臃肿调试时也很难看清哪些字段被谁改了。4.2 节点与边的编排把流程画出来节点就是普通的Python函数接收状态、返回更新字典。边定义了节点之间的流转关系。最简单的图是线性的from langgraph.graph import StateGraph, START, END def node_a(state: GraphState): return {current_step: a_done, messages: [(ai, 步骤A完成)]} def node_b(state: GraphState): return {current_step: b_done, messages: [(ai, 步骤B完成)]} builder StateGraph(GraphState) builder.add_node(a, node_a) builder.add_node(b, node_b) builder.add_edge(START, a) builder.add_edge(a, b) builder.add_edge(b, END) graph builder.compile() result graph.invoke({messages: [], current_step: start, retry_count: 0, result: None})这段代码跑起来后状态会依次经过a和b最后messages里有两条AI消息。看起来简单但这是所有复杂图的基础。条件边是LangGraph真正强大的地方。你可以根据状态里的某个字段决定下一步走哪个节点def route_after_a(state: GraphState) - str: if state[retry_count] 2: return give_up if 错误 in state[messages][-1].content: return retry return b builder.add_conditional_edges(a, route_after_a, { retry: a, b: b, give_up: END, })这个路由函数返回一个字符串LangGraph根据映射表决定跳转到哪个节点。我常用这种模式实现重试逻辑某个节点执行失败时retry_count加一然后条件边判断是否超过阈值没超过就回到该节点重试超过了就走到兜底节点。4.3 检查点与持久化让图能断点续跑LangGraph的检查点机制是我最喜欢的功能之一。它可以在每个节点执行后保存状态快照这样如果中途出错或者需要人工介入可以从上次的检查点恢复而不是从头再来。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: task-001}} result graph.invoke(initial_state, config)thread_id是会话标识同一个thread_id的多次调用会共享状态历史。这意味着你可以先跑一半停下来检查中间结果确认无误后再继续跑。我在做长文档处理时经常用这个机制先跑到摘要生成节点人工看一眼摘要质量没问题再继续后面的翻译和排版。生产环境建议用持久化检查点比如基于SQLite或Postgres的实现这样服务重启后状态不丢。MemorySaver只适合开发和测试。4.4 并行分支与扇入扇出LangGraph支持从一个节点扇出到多个节点并行执行然后再扇入汇总。这在需要同时调用多个工具或处理多个子任务时非常有用。builder.add_edge(split, task_a) builder.add_edge(split, task_b) builder.add_edge(split, task_c) builder.add_edge(task_a, merge) builder.add_edge(task_b, merge) builder.add_edge(task_c, merge)三个task节点会并行执行全部完成后才进入merge节点。注意并行节点对状态的更新必须是可合并的这就是为什么前面要用add_messages这种reducer。如果两个节点同时写同一个非reducer字段后写的会覆盖先写的可能丢数据。5. Deep Agents的封装价值少写重复代码但别指望它包治百病5.1 Deep Agents解决了什么痛点用LangGraph手搓一个代理你需要自己实现规划节点把大任务拆成小步骤、执行节点调用工具或模型、反思节点检查结果是否合格、路由逻辑决定下一步。这些模式在多个项目里反复出现每次重写既浪费时间又容易出错。Deep Agents把这些常见模式封装成了可配置的组件。你只需要定义好工具集和任务描述它就能自动生成一个包含规划、执行、反思循环的代理图。我实测下来一个原本需要两百多行LangGraph代码的代理用Deep Agents大概三十行就能跑起来。from deepagents import create_deep_agent agent create_deep_agent( tools[search_docs, calculate], instructions你是一个研究助手负责查找资料并计算相关数据。, modelget_llm(), ) result agent.invoke({messages: [(human, 查一下最近三个月的销售数据并计算增长率)]})5.2 什么时候该用Deep Agents什么时候该回到LangGraph我的判断标准很简单如果你的代理行为符合规划-执行-反思这个通用模式用Deep Agents如果需要高度定制化的流程控制比如特定的审批环节、复杂的状态机、非标准的重试策略那就用LangGraph自己搭。举个例子我做过一个合同审核代理流程是提取关键条款→比对标准模板→标记风险点→生成审核意见→人工确认→归档。其中人工确认环节需要暂停图执行等外部系统回调后再继续。这种带外部交互的流程Deep Agents的默认模式覆盖不了我就用LangGraph的检查点机制自己实现了暂停和恢复。另一个经验是Deep Agents的抽象层虽然方便但调试时不如LangGraph直观。当代理行为不符合预期时LangGraph可以逐节点打印状态看清楚每一步发生了什么Deep Agents内部封装了太多东西有时候只能靠日志猜。所以我的做法是先用Deep Agents快速搭原型验证可行性如果后续需要精细调优再把关键部分用LangGraph重写。5.3 工具设计对代理效果的影响不管用哪个框架工具设计的质量直接决定代理的上限。我总结了几个原则第一工具粒度要适中。太细的工具比如读取文件第一行会让代理花大量步骤在琐碎操作上太粗的工具比如处理整个文档又失去了灵活性。我一般按一个工具完成一个语义完整的动作来划分比如提取文档中的表格就是一个合适的粒度。第二工具返回值要包含足够的上下文。如果工具只返回一个数字代理可能不知道这个数字代表什么。我习惯让工具返回结构化的文本比如销售额12345元统计周期2024年Q3这样代理在后续推理时能直接引用。第三错误处理要友好。工具执行失败时不要直接抛异常中断整个图而是返回一个描述错误的字符串让代理有机会根据错误信息调整策略。比如搜索工具没找到结果时返回未找到相关记录建议更换关键词代理看到后可能会换个词再试。6. 实战案例搭建一个多代理协作的研究报告生成系统6.1 需求拆解与架构设计假设我们要做一个系统用户输入一个研究主题系统自动完成资料检索、数据整理、报告撰写、质量审核四个环节最后输出一份结构化的研究报告。这个需求用单个代理也能做但效果不稳定因为不同环节需要的思维方式不同——检索需要广撒网整理需要严谨撰写需要文采审核需要挑剔。所以我设计成四个专职代理用一个主管图来协调。架构是这样的主管图LangGraph负责流程控制四个子代理用Deep Agents创建分别负责各自环节。主管图的状态里保存原始主题、检索结果、整理后的数据、报告草稿、审核意见。每个子代理完成后结果写回主管图状态由条件边决定下一步。6.2 主管图的节点实现先定义状态class ResearchState(TypedDict): topic: str raw_materials: list[str] organized_data: dict draft: str review_feedback: str final_report: str retry_count: int检索节点调用检索代理def retrieve_node(state: ResearchState): agent create_deep_agent( tools[search_docs, fetch_url], instructions你是一个资料检索专家请围绕给定主题广泛搜集资料。, modelget_llm(), ) result agent.invoke({messages: [(human, f检索主题{state[topic]})]}) materials [m.content for m in result[messages] if m.type ai] return {raw_materials: materials}整理节点把原始资料结构化def organize_node(state: ResearchState): prompt ChatPromptTemplate.from_messages([ (system, 请把以下资料整理成结构化数据提取关键事实、数据和观点。), (human, {materials}), ]) chain prompt | get_llm() | StrOutputParser() organized chain.invoke({materials: \n.join(state[raw_materials])}) return {organized_data: {content: organized}}撰写和审核节点类似都是调用对应的子代理。审核节点返回的反馈如果包含需要修改条件边就路由回撰写节点retry_count加一超过两次则强制输出当前草稿。6.3 子代理的配置差异四个子代理虽然都用Deep Agents创建但配置不同。检索代理的temperature设成0.3允许一定的探索性整理代理设成0.0要求严格忠实原文撰写代理设成0.5让文字更流畅审核代理设成0.0保持客观。工具集也不同检索代理有搜索和网页抓取工具整理代理只有文本处理工具撰写代理没有外部工具纯靠模型生成审核代理有一个比对标准的工具用来检查报告是否符合格式要求。这种差异化配置是单个代理做不到的。如果用一个代理跑完全程它很难在不同阶段切换性格往往导致检索不够广、整理不够严、撰写不够活、审核不够狠。6.4 完整跑通与结果验证把图编译好后用graph.invoke跑一个测试主题比如2024年新能源汽车市场趋势。整个流程大概需要调用模型8到12次耗时取决于模型响应速度。我实测下来用中等规模的模型端到端大约40到60秒。验证结果时我主要看三点一是报告结构是否完整有没有缺失章节二是数据引用是否准确有没有编造数字三是审核意见是否被采纳修改后的草稿有没有改进。如果发现审核环节形同虚设我会检查审核代理的提示词是不是太宽松或者条件边的路由逻辑是不是有问题。7. 踩坑实录那些让我熬夜的报错与排查过程7.1 状态字段被意外覆盖有一次我定义了一个状态字段analysis_result没有加reducer注解。结果两个并行节点都往这个字段写数据后执行的把先执行的覆盖了导致最终结果只包含一半的分析。排查时我打印了每个节点执行后的状态发现字段值在某个节点后突然变短才定位到问题。解决方案是给需要合并的字段加Annotated[list, operator.add]或者自定义reducer函数。如果是字典类型的合并可以写一个merge_dicts函数用Annotated[dict, merge_dicts]。7.2 条件边路由函数返回了未定义的键LangGraph的条件边要求路由函数返回的字符串必须在映射表里有对应项。我有一次手误返回了end小写但映射表里写的是END结果图执行到那里直接报错说找不到目标节点。这个错误信息不太直观我一开始以为是节点定义有问题查了半天才发现是大小写不一致。现在的习惯是路由函数的返回值用常量定义比如ROUTE_RETRY retry映射表里也用常量避免手写字符串出错。7.3 Deep Agents的工具调用陷入死循环有一次我配置的检索代理反复调用同一个搜索工具每次返回的结果都差不多但它就是不停直到达到最大迭代次数才退出。排查后发现是工具的docstring写得太宽泛代理认为任何时候都可以调用它。后来我在docstring里加了限制条件比如仅当需要查找外部信息时调用如果已有足够信息请直接回答死循环问题就解决了。另一个经验是设置合理的最大迭代次数。Deep Agents默认可能允许很多轮我一般设成5到8轮超过就强制输出当前结果。这样即使代理陷入循环也不会无限跑下去。7.4 检查点恢复后状态不一致用MemorySaver做检查点时我遇到过恢复后状态和预期不符的情况。原因是检查点保存的是节点执行后的状态如果某个节点执行了一半就崩溃了恢复时会从该节点执行前的状态重新开始导致该节点的副作用比如已经发送的API请求被重复执行。解决方案是把有副作用的操作设计成幂等的或者在节点内部先检查状态里是否已经有该操作的结果有就跳过。对于关键操作我还会在状态里加一个completed_steps列表记录已经成功完成的步骤恢复时先检查这个列表。8. 性能与成本优化让代理跑得又快又省8.1 模型分级使用不是所有节点都需要用最强的模型。我的做法是分级规划、审核这类需要强推理的环节用大模型检索、整理这类相对机械的环节用小模型。实测下来整体成本能降低40%到60%而质量下降不明显。在代码里我给每个节点单独指定模型def retrieve_node(state): llm get_llm(model_namegpt-4o-mini) # 小模型 ... def review_node(state): llm get_llm(model_namegpt-4o) # 大模型 ...8.2 缓存重复调用代理执行过程中经常有重复的模型调用比如同样的提示词在重试时被再次发送。LangChain支持缓存层可以把模型响应缓存起来from langchain_core.caches import InMemoryCache from langchain_core.globals import set_llm_cache set_llm_cache(InMemoryCache())对于开发调试阶段这个缓存能省不少钱。生产环境可以用Redis做分布式缓存多个实例共享。8.3 并行化能并行的节点前面提到的扇出扇入模式能把原本串行的任务变成并行。比如检索阶段需要查三个不同的数据源串行执行要三次模型调用加三次工具调用并行执行只需要一次模型调用的时间假设工具调用本身是IO密集型的。我用这个模式把一个报告生成任务从90秒压到了45秒。但要注意并行节点对状态的写入必须可合并否则会丢数据。另外并行调用模型可能会触发速率限制需要根据API的配额调整并行度。8.4 控制上下文长度代理跑多轮之后消息历史会越来越长每次调用模型都要把全部历史传进去token消耗快速增长。我的做法是在状态里只保留最近N轮的消息更早的消息用摘要代替。LangGraph有内置的消息裁剪和摘要功能也可以自己写一个节点定期压缩历史。def trim_messages(state): messages state[messages] if len(messages) 10: # 保留最近5条其余摘要 recent messages[-5:] older messages[:-5] summary summarize(older) return {messages: [(system, f历史摘要{summary})] recent} return {}这个节点可以放在每轮循环的开始确保上下文不会无限膨胀。9. 我总结的几条实战原则第一先用LangChain把原子能力跑通再用LangGraph编排流程最后用Deep Agents封装通用模式。不要一上来就上最复杂的框架否则出了问题你都不知道是哪一层的问题。第二状态设计宁简勿繁。只放真正需要跨节点共享的数据临时变量留在节点内部。状态越简单调试越容易。第三每个节点只做一件事。一个节点里既调模型又调工具又写数据库出问题时很难定位。拆成多个节点每个节点职责单一配合检查点机制排查效率会高很多。第四工具docstring要写清楚使用场景。模型判断是否调用工具主要靠docstring。写得好代理行为就稳定写得模糊代理就会乱调工具。第五永远设置最大迭代次数和超时。代理可能因为各种原因陷入循环没有兜底机制的话轻则浪费token重则服务卡死。第六开发阶段用MemorySaver生产环境换持久化检查点。别等到服务重启后发现所有状态丢失才后悔。这套组合拳我用了大半年从简单的文档处理到复杂的多代理协作都跑过整体稳定性不错。当然框架本身也在快速迭代新版本可能会带来更简洁的API或者更好的性能保持关注、及时升级但升级前一定要在测试环境充分验证。
1. 从链式调用到图式编排为什么我最终把三个框架揉在一起用去年下半年开始我陆续接手了几个基于大模型的自动化项目从最早的简单问答机器人到后来的多步骤文档处理流水线再到需要动态决策的复杂任务代理。一开始我只用LangChain觉得它的链式调用已经够用了——把提示词模板、模型调用、输出解析器串起来跑通一个流程并不难。但项目稍微复杂一点问题就来了我需要根据中间结果决定下一步走哪个分支需要在某个环节失败时回退重试还需要让多个子任务并行执行再汇总。这些需求用LangChain的线性链去硬凑代码很快就变成了一团乱麻各种if-else嵌套和手动状态管理维护成本极高。后来我接触到了LangGraph它把工作流建模成一张有向图节点是具体的处理单元边定义了流转逻辑还支持条件边、循环、并行分支和持久化状态。这让我眼前一亮——终于可以用图的方式去描述复杂流程了。但LangGraph本身偏底层很多能力需要自己搭比如工具调用、记忆管理、多代理协作这些如果每个项目都从头写一遍效率太低。再后来Deep Agents出现了它把深度代理的常见模式封装成了更高层的抽象比如规划、执行、反思、工具使用等开箱即用。但Deep Agents也不是万能的遇到高度定制化的流程还是得回到LangGraph去手搓图结构。所以我的结论是这三个东西不是替代关系而是分层协作的关系。LangChain提供基础的模型、工具、提示词、解析器等原子能力LangGraph负责把这些原子能力编排成可控的工作流Deep Agents则在LangGraph之上封装了代理的常见行为模式让你少写重复代码。把它们揉在一起用才能既保证灵活性又保证开发效率。这篇文章我就把这套组合拳的实战经验完整拆开从环境搭建到核心概念再到一个完整的项目案例最后聊聊我踩过的坑和性能调优的心得。2. 环境准备与版本兼容别让依赖冲突毁掉你的周末2.1 三个框架的安装顺序与版本锁定很多人上来就pip install langchain langgraph deepagents一把梭结果跑起来各种ImportError。我踩过最坑的一次是LangChain的某个子包版本和LangGraph要求的核心库版本不兼容导致图编译时报了一个完全看不懂的序列化错误。后来我学乖了安装顺序和版本锁定必须严格对待。我的建议是先用虚拟环境隔离然后按以下顺序安装python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate # 先装核心依赖锁定大版本 pip install langchain0.3,0.4 pip install langchain-core0.3,0.4 pip install langgraph0.2,0.3 pip install deepagents0.1,0.2 # 再装模型提供商和工具库 pip install langchain-openai0.2,0.3 pip install langchain-community0.3,0.4为什么要先装langchain-core因为LangGraph和Deep Agents都依赖它如果先装LangGraph它可能会拉一个较老的core版本后面再装LangChain时又会被覆盖导致版本错乱。先锁定core的大版本再装上层框架能最大程度避免这种问题。注意不要用pip install -U无脑升级所有包大版本升级经常带来破坏性变更。我一般会在requirements.txt里写死小版本号比如langgraph0.2.60确保团队里每个人跑出来的结果一致。2.2 API Key与模型配置的集中管理三个框架都支持多种模型提供商但配置方式各有不同。我的做法是统一用环境变量管理密钥然后在代码里通过一个工厂函数创建模型实例。这样切换模型时只需要改一个地方。import os from langchain_openai import ChatOpenAI def get_llm(model_name: str gpt-4o-mini, temperature: float 0.0): return ChatOpenAI( modelmodel_name, temperaturetemperature, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), # 可选用于兼容其他服务 )这里有个细节temperature设成0.0对于需要稳定输出的工作流非常重要。我在做文档分类任务时temperature0.7导致同样的输入有时分到A类有时分到B类排查了半天才发现是随机性在作祟。对于代理类任务建议默认用0.0需要创意生成时再单独调高。2.3 目录结构规划让项目从第一天就整洁我见过太多项目把所有代码堆在一个main.py里到后面根本没法维护。我的习惯是按职责分层project/ ├── config/ │ └── settings.py # 模型、工具、常量配置 ├── core/ │ ├── llm_factory.py # 模型实例工厂 │ ├── tools.py # 自定义工具定义 │ └── state.py # 图状态定义 ├── graphs/ │ ├── simple_chain.py # 基础链式流程 │ ├── conditional_graph.py # 带条件分支的图 │ └── multi_agent.py # 多代理协作图 ├── agents/ │ └── deep_agent.py # Deep Agents封装 ├── utils/ │ └── helpers.py # 通用工具函数 └── main.py # 入口这个结构的好处是当你的图越来越复杂时每个图独立一个文件状态定义集中管理工具函数复用方便。我后来把某个图从单文件拆成三个模块后调试效率至少提升了一倍。3. LangChain的定位它到底该干哪些活不该干哪些活3.1 提示词模板与输出解析把不确定性关进笼子LangChain最核心的价值之一是它提供了一套标准化的提示词管理和输出解析机制。很多人觉得提示词就是拼字符串没必要用框架。但当你需要动态注入变量、处理多轮对话历史、强制模型输出JSON格式时手写字符串很快就会失控。我常用的组合是ChatPromptTemplate加PydanticOutputParser。前者负责把系统指令、历史消息、用户输入组装成模型能理解的格式后者负责把模型的自由文本输出解析成结构化对象。from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field class TaskPlan(BaseModel): steps: list[str] Field(description执行步骤列表) priority: str Field(description优先级high/medium/low) parser PydanticOutputParser(pydantic_objectTaskPlan) prompt ChatPromptTemplate.from_messages([ (system, 你是一个任务规划助手。请根据用户需求生成执行计划。\n{format_instructions}), (human, {user_input}), ]).partial(format_instructionsparser.get_format_instructions()) chain prompt | get_llm() | parser result chain.invoke({user_input: 帮我整理一份季度报告}) print(result.steps, result.priority)这里的关键是get_format_instructions()它会自动生成一段描述JSON schema的文本告诉模型应该输出什么格式。实测下来加上这段指令后模型输出格式错误的概率从大约15%降到了2%以下。如果还是偶尔出错可以再加一层重试逻辑用with_retry或者自己写个循环。3.2 工具定义让模型能动手而不是只动嘴LangChain的工具抽象让模型可以调用外部函数这是构建代理的基础。定义一个工具最简单的方式是用tool装饰器from langchain_core.tools import tool tool def search_docs(query: str) - str: 根据关键词搜索内部文档库返回最相关的三条记录。 # 实际实现可能是向量检索或数据库查询 return f关于{query}的搜索结果... tool def calculate(expression: str) - float: 计算数学表达式支持加减乘除和括号。 return eval(expression, {__builtins__: {}}, {})注意docstring非常重要模型就是靠它来判断什么时候该调用这个工具。我踩过的坑是docstring写得太模糊比如只写搜索结果模型在需要计算的时候也去调搜索工具。后来我把每个工具的docstring都写成什么场景下用、输入是什么、返回什么的完整说明工具调用的准确率明显提升。另外eval那里我加了{__builtins__: {}}来限制命名空间这是安全实践。虽然在这个场景下输入来自模型但养成好习惯总没错。3.3 LangChain不擅长的事别硬撑LangChain的链式结构本质上是线性的虽然可以用RunnableBranch做条件分支但一旦分支嵌套超过两层代码可读性就急剧下降。我试过用LangChain实现一个根据文档类型走不同处理流程每个流程内部又有多个步骤某些步骤失败要回退的逻辑写了三百多行自己回头看都费劲。这种场景就应该交给LangGraph。LangChain负责的是原子操作——一次模型调用、一次工具执行、一次解析。至于这些原子操作怎么组合、怎么流转、怎么管理状态那是LangGraph的活。把职责分清楚代码才不会变成一锅粥。4. LangGraph核心机制状态、节点、边与检查点4.1 状态定义图的大脑LangGraph的核心概念是状态State它是一个贯穿整个图执行过程的数据结构。每个节点读取状态、处理、然后返回状态的更新部分。定义状态通常用TypedDictfrom typing import TypedDict, Annotated from langgraph.graph import add_messages class GraphState(TypedDict): messages: Annotated[list, add_messages] current_step: str retry_count: int result: str | NoneAnnotated[list, add_messages]这个写法很关键。它告诉LangGraph当多个节点都往messages里追加内容时用add_messages这个函数来合并而不是直接覆盖。我一开始没加这个注解结果并行分支的消息互相覆盖丢了数据排查了很久。状态设计的原则是只放需要跨节点共享的数据。临时变量放在节点内部就好不要什么都往状态里塞否则状态会变得臃肿调试时也很难看清哪些字段被谁改了。4.2 节点与边的编排把流程画出来节点就是普通的Python函数接收状态、返回更新字典。边定义了节点之间的流转关系。最简单的图是线性的from langgraph.graph import StateGraph, START, END def node_a(state: GraphState): return {current_step: a_done, messages: [(ai, 步骤A完成)]} def node_b(state: GraphState): return {current_step: b_done, messages: [(ai, 步骤B完成)]} builder StateGraph(GraphState) builder.add_node(a, node_a) builder.add_node(b, node_b) builder.add_edge(START, a) builder.add_edge(a, b) builder.add_edge(b, END) graph builder.compile() result graph.invoke({messages: [], current_step: start, retry_count: 0, result: None})这段代码跑起来后状态会依次经过a和b最后messages里有两条AI消息。看起来简单但这是所有复杂图的基础。条件边是LangGraph真正强大的地方。你可以根据状态里的某个字段决定下一步走哪个节点def route_after_a(state: GraphState) - str: if state[retry_count] 2: return give_up if 错误 in state[messages][-1].content: return retry return b builder.add_conditional_edges(a, route_after_a, { retry: a, b: b, give_up: END, })这个路由函数返回一个字符串LangGraph根据映射表决定跳转到哪个节点。我常用这种模式实现重试逻辑某个节点执行失败时retry_count加一然后条件边判断是否超过阈值没超过就回到该节点重试超过了就走到兜底节点。4.3 检查点与持久化让图能断点续跑LangGraph的检查点机制是我最喜欢的功能之一。它可以在每个节点执行后保存状态快照这样如果中途出错或者需要人工介入可以从上次的检查点恢复而不是从头再来。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: task-001}} result graph.invoke(initial_state, config)thread_id是会话标识同一个thread_id的多次调用会共享状态历史。这意味着你可以先跑一半停下来检查中间结果确认无误后再继续跑。我在做长文档处理时经常用这个机制先跑到摘要生成节点人工看一眼摘要质量没问题再继续后面的翻译和排版。生产环境建议用持久化检查点比如基于SQLite或Postgres的实现这样服务重启后状态不丢。MemorySaver只适合开发和测试。4.4 并行分支与扇入扇出LangGraph支持从一个节点扇出到多个节点并行执行然后再扇入汇总。这在需要同时调用多个工具或处理多个子任务时非常有用。builder.add_edge(split, task_a) builder.add_edge(split, task_b) builder.add_edge(split, task_c) builder.add_edge(task_a, merge) builder.add_edge(task_b, merge) builder.add_edge(task_c, merge)三个task节点会并行执行全部完成后才进入merge节点。注意并行节点对状态的更新必须是可合并的这就是为什么前面要用add_messages这种reducer。如果两个节点同时写同一个非reducer字段后写的会覆盖先写的可能丢数据。5. Deep Agents的封装价值少写重复代码但别指望它包治百病5.1 Deep Agents解决了什么痛点用LangGraph手搓一个代理你需要自己实现规划节点把大任务拆成小步骤、执行节点调用工具或模型、反思节点检查结果是否合格、路由逻辑决定下一步。这些模式在多个项目里反复出现每次重写既浪费时间又容易出错。Deep Agents把这些常见模式封装成了可配置的组件。你只需要定义好工具集和任务描述它就能自动生成一个包含规划、执行、反思循环的代理图。我实测下来一个原本需要两百多行LangGraph代码的代理用Deep Agents大概三十行就能跑起来。from deepagents import create_deep_agent agent create_deep_agent( tools[search_docs, calculate], instructions你是一个研究助手负责查找资料并计算相关数据。, modelget_llm(), ) result agent.invoke({messages: [(human, 查一下最近三个月的销售数据并计算增长率)]})5.2 什么时候该用Deep Agents什么时候该回到LangGraph我的判断标准很简单如果你的代理行为符合规划-执行-反思这个通用模式用Deep Agents如果需要高度定制化的流程控制比如特定的审批环节、复杂的状态机、非标准的重试策略那就用LangGraph自己搭。举个例子我做过一个合同审核代理流程是提取关键条款→比对标准模板→标记风险点→生成审核意见→人工确认→归档。其中人工确认环节需要暂停图执行等外部系统回调后再继续。这种带外部交互的流程Deep Agents的默认模式覆盖不了我就用LangGraph的检查点机制自己实现了暂停和恢复。另一个经验是Deep Agents的抽象层虽然方便但调试时不如LangGraph直观。当代理行为不符合预期时LangGraph可以逐节点打印状态看清楚每一步发生了什么Deep Agents内部封装了太多东西有时候只能靠日志猜。所以我的做法是先用Deep Agents快速搭原型验证可行性如果后续需要精细调优再把关键部分用LangGraph重写。5.3 工具设计对代理效果的影响不管用哪个框架工具设计的质量直接决定代理的上限。我总结了几个原则第一工具粒度要适中。太细的工具比如读取文件第一行会让代理花大量步骤在琐碎操作上太粗的工具比如处理整个文档又失去了灵活性。我一般按一个工具完成一个语义完整的动作来划分比如提取文档中的表格就是一个合适的粒度。第二工具返回值要包含足够的上下文。如果工具只返回一个数字代理可能不知道这个数字代表什么。我习惯让工具返回结构化的文本比如销售额12345元统计周期2024年Q3这样代理在后续推理时能直接引用。第三错误处理要友好。工具执行失败时不要直接抛异常中断整个图而是返回一个描述错误的字符串让代理有机会根据错误信息调整策略。比如搜索工具没找到结果时返回未找到相关记录建议更换关键词代理看到后可能会换个词再试。6. 实战案例搭建一个多代理协作的研究报告生成系统6.1 需求拆解与架构设计假设我们要做一个系统用户输入一个研究主题系统自动完成资料检索、数据整理、报告撰写、质量审核四个环节最后输出一份结构化的研究报告。这个需求用单个代理也能做但效果不稳定因为不同环节需要的思维方式不同——检索需要广撒网整理需要严谨撰写需要文采审核需要挑剔。所以我设计成四个专职代理用一个主管图来协调。架构是这样的主管图LangGraph负责流程控制四个子代理用Deep Agents创建分别负责各自环节。主管图的状态里保存原始主题、检索结果、整理后的数据、报告草稿、审核意见。每个子代理完成后结果写回主管图状态由条件边决定下一步。6.2 主管图的节点实现先定义状态class ResearchState(TypedDict): topic: str raw_materials: list[str] organized_data: dict draft: str review_feedback: str final_report: str retry_count: int检索节点调用检索代理def retrieve_node(state: ResearchState): agent create_deep_agent( tools[search_docs, fetch_url], instructions你是一个资料检索专家请围绕给定主题广泛搜集资料。, modelget_llm(), ) result agent.invoke({messages: [(human, f检索主题{state[topic]})]}) materials [m.content for m in result[messages] if m.type ai] return {raw_materials: materials}整理节点把原始资料结构化def organize_node(state: ResearchState): prompt ChatPromptTemplate.from_messages([ (system, 请把以下资料整理成结构化数据提取关键事实、数据和观点。), (human, {materials}), ]) chain prompt | get_llm() | StrOutputParser() organized chain.invoke({materials: \n.join(state[raw_materials])}) return {organized_data: {content: organized}}撰写和审核节点类似都是调用对应的子代理。审核节点返回的反馈如果包含需要修改条件边就路由回撰写节点retry_count加一超过两次则强制输出当前草稿。6.3 子代理的配置差异四个子代理虽然都用Deep Agents创建但配置不同。检索代理的temperature设成0.3允许一定的探索性整理代理设成0.0要求严格忠实原文撰写代理设成0.5让文字更流畅审核代理设成0.0保持客观。工具集也不同检索代理有搜索和网页抓取工具整理代理只有文本处理工具撰写代理没有外部工具纯靠模型生成审核代理有一个比对标准的工具用来检查报告是否符合格式要求。这种差异化配置是单个代理做不到的。如果用一个代理跑完全程它很难在不同阶段切换性格往往导致检索不够广、整理不够严、撰写不够活、审核不够狠。6.4 完整跑通与结果验证把图编译好后用graph.invoke跑一个测试主题比如2024年新能源汽车市场趋势。整个流程大概需要调用模型8到12次耗时取决于模型响应速度。我实测下来用中等规模的模型端到端大约40到60秒。验证结果时我主要看三点一是报告结构是否完整有没有缺失章节二是数据引用是否准确有没有编造数字三是审核意见是否被采纳修改后的草稿有没有改进。如果发现审核环节形同虚设我会检查审核代理的提示词是不是太宽松或者条件边的路由逻辑是不是有问题。7. 踩坑实录那些让我熬夜的报错与排查过程7.1 状态字段被意外覆盖有一次我定义了一个状态字段analysis_result没有加reducer注解。结果两个并行节点都往这个字段写数据后执行的把先执行的覆盖了导致最终结果只包含一半的分析。排查时我打印了每个节点执行后的状态发现字段值在某个节点后突然变短才定位到问题。解决方案是给需要合并的字段加Annotated[list, operator.add]或者自定义reducer函数。如果是字典类型的合并可以写一个merge_dicts函数用Annotated[dict, merge_dicts]。7.2 条件边路由函数返回了未定义的键LangGraph的条件边要求路由函数返回的字符串必须在映射表里有对应项。我有一次手误返回了end小写但映射表里写的是END结果图执行到那里直接报错说找不到目标节点。这个错误信息不太直观我一开始以为是节点定义有问题查了半天才发现是大小写不一致。现在的习惯是路由函数的返回值用常量定义比如ROUTE_RETRY retry映射表里也用常量避免手写字符串出错。7.3 Deep Agents的工具调用陷入死循环有一次我配置的检索代理反复调用同一个搜索工具每次返回的结果都差不多但它就是不停直到达到最大迭代次数才退出。排查后发现是工具的docstring写得太宽泛代理认为任何时候都可以调用它。后来我在docstring里加了限制条件比如仅当需要查找外部信息时调用如果已有足够信息请直接回答死循环问题就解决了。另一个经验是设置合理的最大迭代次数。Deep Agents默认可能允许很多轮我一般设成5到8轮超过就强制输出当前结果。这样即使代理陷入循环也不会无限跑下去。7.4 检查点恢复后状态不一致用MemorySaver做检查点时我遇到过恢复后状态和预期不符的情况。原因是检查点保存的是节点执行后的状态如果某个节点执行了一半就崩溃了恢复时会从该节点执行前的状态重新开始导致该节点的副作用比如已经发送的API请求被重复执行。解决方案是把有副作用的操作设计成幂等的或者在节点内部先检查状态里是否已经有该操作的结果有就跳过。对于关键操作我还会在状态里加一个completed_steps列表记录已经成功完成的步骤恢复时先检查这个列表。8. 性能与成本优化让代理跑得又快又省8.1 模型分级使用不是所有节点都需要用最强的模型。我的做法是分级规划、审核这类需要强推理的环节用大模型检索、整理这类相对机械的环节用小模型。实测下来整体成本能降低40%到60%而质量下降不明显。在代码里我给每个节点单独指定模型def retrieve_node(state): llm get_llm(model_namegpt-4o-mini) # 小模型 ... def review_node(state): llm get_llm(model_namegpt-4o) # 大模型 ...8.2 缓存重复调用代理执行过程中经常有重复的模型调用比如同样的提示词在重试时被再次发送。LangChain支持缓存层可以把模型响应缓存起来from langchain_core.caches import InMemoryCache from langchain_core.globals import set_llm_cache set_llm_cache(InMemoryCache())对于开发调试阶段这个缓存能省不少钱。生产环境可以用Redis做分布式缓存多个实例共享。8.3 并行化能并行的节点前面提到的扇出扇入模式能把原本串行的任务变成并行。比如检索阶段需要查三个不同的数据源串行执行要三次模型调用加三次工具调用并行执行只需要一次模型调用的时间假设工具调用本身是IO密集型的。我用这个模式把一个报告生成任务从90秒压到了45秒。但要注意并行节点对状态的写入必须可合并否则会丢数据。另外并行调用模型可能会触发速率限制需要根据API的配额调整并行度。8.4 控制上下文长度代理跑多轮之后消息历史会越来越长每次调用模型都要把全部历史传进去token消耗快速增长。我的做法是在状态里只保留最近N轮的消息更早的消息用摘要代替。LangGraph有内置的消息裁剪和摘要功能也可以自己写一个节点定期压缩历史。def trim_messages(state): messages state[messages] if len(messages) 10: # 保留最近5条其余摘要 recent messages[-5:] older messages[:-5] summary summarize(older) return {messages: [(system, f历史摘要{summary})] recent} return {}这个节点可以放在每轮循环的开始确保上下文不会无限膨胀。9. 我总结的几条实战原则第一先用LangChain把原子能力跑通再用LangGraph编排流程最后用Deep Agents封装通用模式。不要一上来就上最复杂的框架否则出了问题你都不知道是哪一层的问题。第二状态设计宁简勿繁。只放真正需要跨节点共享的数据临时变量留在节点内部。状态越简单调试越容易。第三每个节点只做一件事。一个节点里既调模型又调工具又写数据库出问题时很难定位。拆成多个节点每个节点职责单一配合检查点机制排查效率会高很多。第四工具docstring要写清楚使用场景。模型判断是否调用工具主要靠docstring。写得好代理行为就稳定写得模糊代理就会乱调工具。第五永远设置最大迭代次数和超时。代理可能因为各种原因陷入循环没有兜底机制的话轻则浪费token重则服务卡死。第六开发阶段用MemorySaver生产环境换持久化检查点。别等到服务重启后发现所有状态丢失才后悔。这套组合拳我用了大半年从简单的文档处理到复杂的多代理协作都跑过整体稳定性不错。当然框架本身也在快速迭代新版本可能会带来更简洁的API或者更好的性能保持关注、及时升级但升级前一定要在测试环境充分验证。
