1. 智能体工程化的第一道坎把循环写清楚的编排层先说我看到的现象。现在很多人搭“智能体”用的是这种写法一个上下文变量一段while循环模型判断“要不要调工具”就调工具不然就返回答案。Demo 阶段完全没问题模型也知道自己在干什么跑起来效果不错。但这个东西一旦要接生产大概率会翻车。翻车点不是模型智商不够而是你根本说不清楚它现在跑到哪一步了、为什么跑到了这一步、这个状态能不能恢复、如果中途挂了能不能把上下文找回。我参考的这份 Deep Agents 参考实现里开篇第一件事就不是去调大模型而是在定义一个叫DeepAgentState的状态结构然后用 LangGraph 的StateGraph把节点和边画出来。这个顺序本身就说明了一件事生产级智能体的核心不是把“让模型思考”这件事做得更花哨而是把“决策循环的过程”变成可记录、可回放、可维护的工程结构。1.1 智能体与普通脚本的本质差异决策点在哪普通脚本的流程是写死的。比如数据处理程序先读文件、再清洗、再入库每一步的下一步是什么程序员在写代码的时候就完全确定了。运行时只是机械地执行而已即使有if/else分支分支条件和流程也都是预定义的。智能体不一样。同样是“查天气、比较价格、写报告”的任务不同的用户输入会产生完全不同的执行路径。模型可能会选择调天气工具可能会选择先搜索再调用某个分析工具也可能直接基于历史记忆回答。这些路径不是程序员预先枚举的而是模型根据当前状态动态决定的。问题就出在“动态决策”上。动态决策意味着同样的入口状态今天可能走 A 分支明天可能走 B 分支。你要 debug 的时候就必须知道它走的到底是哪条分支每个节点上看到了什么状态经过谁改成了什么样。如果代码只是一个大while循环套一个if分支这些信息全都藏在半结构化字符串里你只能在日志里猜。所以我在看 Deep Agents 参考实现时印象最深的一组源码设计是这样的每个节点都是一个纯函数输入是完整的运行状态输出是新的状态切片节点不直接访问外部变量节点之间没有隐藏依赖整个流程的“下一步做什么”被显式建模成了图上的边。这个造型带来的工程价值是巨大的你可以随便中断一次运行把当时的完整状态存储下来之后恢复接着跑你可以给任意一条边加条件判断判断逻辑肉眼可见你还可以把一次运行完整铺开看它访问了哪几个节点、每个节点耗时多久、消耗了多少 token。这本质上就是在用“有向图”替换“循环 串行调用链”。1.2 从 Deep Agents 源码看“图即程序”的编排视角LangGraph 里有一个非常核心的类叫StateGraph。它的工作方式非常像在我们公司里画业务流程架构图你把一个流程拆成几个阶段每个阶段是一个节点节点之间用有向边连接边可以带条件。我在参考实现里最常见的结构就是这样的from typing import TypedDict from langgraph.graph import StateGraph, END class DeepAgentState(TypedDict): user_query: str plan: list[str] results: list[str] current_step: int final_answer: str def planner(state: DeepAgentState) - DeepAgentState: # 让模型拆解用户当前请求形成 plan plan llm_with_planner.invoke(state[user_query]) return {plan: plan, current_step: 0} def executor(state: DeepAgentState) - DeepAgentState: # 执行当前步骤可能触发工具调用 step state[current_step] result agent_tools.run_step(state[plan][step]) return {results: state[results] [result], current_step: step 1} def evaluator(state: DeepAgentState) - DeepAgentState: # 检查是否完成若未完成则修改 plan 或继续执行 ... return {plan: new_plan, final_answer: answer} builder StateGraph(DeepAgentState) builder.add_node(planner, planner) builder.add_node(executor, executor) builder.add_node(evaluator, evaluator) builder.set_entry_point(planner) builder.add_edge(planner, executor) builder.add_edge(executor, evaluator) builder.add_conditional_edges( evaluator, lambda state: executor if not state.get(final_answer) else END, {executor: executor, done: END} ) graph builder.compile()这段代码的巧妙之处在于图上的每个节点都是一个纯状态转换函数。每个节点没有自己去调全局变量也不关心上一个节点内部是怎么实现的它只关心进来的状态长什么样、自己该返回什么状态变化。比如executor节点它会根据plan[current_step]去调工具然后把结果追加到results列表里再把步数加一。因为节点之间完全靠状态沟通所以在生产环境里我可以随时对这个图做三件事第一给任意节点加中断。比如某个工具需要人工确认我可以让执行到executor的时候暂停等人审批再继续。第二给整个图加持久化。运行到一半进程崩溃只要状态存下来了下次带着同一个thread_id就能从断点恢复。第三做并发。多个节点之间如果没有依赖关系LangGraph 会并行调度省时间也省钱。这些能力恰恰是“生产级”和“demo 级”之间的分界线。demo 级的智能体只关心模型能不能给出正确答案生产级的智能体关心的是这个正确答案是在什么路径上得到的、能不能复现、出错了能不能兜底。2. Deep Agents 源码骨架拆解StateGraph 的运行模型如果你刚接触 LangGraph看到StateGraph、add_node、add_edge、compile这一串 API 会觉得它像工作流引擎。这个直觉是对的。Star 也好、Airflow 也好、LangGraph 也好底层都是基于节点和边的 DAG/有环图调度器。只不过 LangGraph 的节点通常是大模型调用而不是 SQL 查询或数据清洗任务所以它还得额外处理模型输出质量波动的问题。看 Deep Agents 参考源码的时候不要只看它调用了哪些模型重点要看三个东西状态是哪个对象、节点是怎么组织的、条件边是怎么写的。2.1 一个 planner-executor-evaluator 骨架从入口到编译上节给出的代码里状态DeepAgentState用了TypedDict来定义。在工程上我建议尽量用pydantic的BaseModel或者带默认值的 typed dict因为 LangGraph 在合并各节点返回值的时候是基于字段粒度进行叠加覆盖的不是整个 dict 覆盖。这句话多解释两句。如果节点 A 返回{results: [r1]}节点 B 返回{results: [r2]}LangGraph 默认的叠加逻辑是新增字段不会把[r1]替换成[r2]。遇到 list 会做 append遇到 dict 会做深合并遇到普通字符串或数字才是覆盖。这个行为非常像 Redux 里的 reducer。我见过不少新手栽在这一点上他们在节点里给同一个字段反复赋值字符串结果发现后面节点的输出吞掉了前面的输出日志里只能看到最后一步的状态。所以我在这份参考实现里看到results设计成 list 而不是 string是刻意为之的。planner节点的职责是生成plan。这里有一个很关键的细节plan不应该只是自然语言句子最好让模型输出结构化的 JSON 数组。LangChain 里可以用with_structured_output(PydanticModel)强制约束输出结构。比如class PlanStep(BaseModel): step_id: str action: str parameters: dict[str, Any] {} class DeepAgentPlan(BaseModel): steps: list[PlanStep]用这种方式executor节点拿到plan之后可以直接遍历执行不用从文本里抠动作名。生产环境的稳定性主要靠强类型约束不太靠“prompt 写得足够严格”。evaluator节点的作用非常值得借鉴。它不只是看一眼“任务完成没”还要对执行结果做质量评估比如判断results是否满足用户原始意图、是否需要修改计划、是否需要多轮追问。如果发现 answer 已经满足要求就复制到final_answer字段设置条件边走向END如果不满足就带着 modified plan 重新进入executor。2.2 条件边是智能体的“判断回路”智能体之所以是智能体是因为它可以根据环境反馈调整下一步行动的路径。这在 LangGraph 里体现为add_conditional_edges。add_conditional_edges接收一个路由函数路由函数输入当前完整状态输出一个字符串 KeyLangGraph 根据这个 Key 选择要进入的节点。在我看的参考实现里至少有三类条件边执行完成与否的路由比如上面代码里evaluator判断是否需要继续执行。异常处理的兜底路由比如某个工具调用抛出异常状态里记录error信息路由到“修复节点”而不是直接崩溃。意图分流的路由比如用户输入是闲聊还是查数据直接决定进入哪个子图。写条件边最大的坑是路由进入死循环。模型如果判断“还没完成”就会重新走executor但如果模型能力不行或者指令描述不清可能来回走几十次浪费大量 token。所以参考实现对循环是有节制的。在planner生成第一次计划时就要求模型估算总步数每个节点执行前检查current_step是否超过上限evaluator里也会检查循环次数超过就直接生成“尽力答复”。我自己在生产环境加了这个保护逻辑MAX_EXECUTION_STEPS 5 def evaluator(state: DeepAgentState) - DeepAgentState: if state[current_step] MAX_EXECUTION_STEPS: return {final_answer: build_best_effort_answer(state)} ...看似很土但这个判断救过我好几次。模型在复杂 Agent 任务里漏判、错判是很常见的你不能赌它的判断一直稳定得有一种“程序兜底”机制把最坏情况拦在失控之前。3. 工具调用与状态管理生产级智能体的两个持久层细节我常跟同事说智能体做得省不省心看的不是模型回答得好不好而是状态和工具这两个地方乱不乱。模型偶尔回答不好可以加 prompt、换模型调优但状态一旦乱了整个流程逻辑就是不可预测的线上排查都无从下手。工具调用一旦出故障轻则返回错误重则引起资金风险。这两块是生产级和 demo 级的本质区别所在。3.1 状态不是一次性变量checkpoint 与线程隔离在 demo 里运行状态就是你 Python 函数里的一个局部变量函数返回就没了。但生产级智能体往往要跑在异步环境里一次用户请求可能经过多个 HTTP 循环、多个 Worker 进程甚至隔了很长一段时间等用户反馈。这个时候你必须在每次模型节点执行前后把状态持久化并且记录在哪个thread_id下。LangGraph 的checkpointer解决的就是这个。我在参考实现里看到统一用的是异步存储持久化而在原型阶段很多人会直接用内存from langgraph.checkpoint.memory import MemorySaver compiled_graph graph.compile(checkpointerMemorySaver())MemorySaver把状态存在进程内存里开发够用生产重启数据就没了。我在生产环境一般接 PostgreSQL 或者 Redis 的持久化实现。LangGraph 社区有许多 checkpoint 后端选型配好之后效果就是同一个thread_id下跑了一次没跑完进程重启后还能从上次节点继续而不会从planner整个重来。线程隔离也很重要。所谓thread_id你可以理解为“一段对话的会话 ID”。不同用户、不同 session各自用独立的thread_id跑图互相不干扰。如果多个请求共用一个 thread状态会被后一个覆盖那排查线上问题的时候所有状态都是串的根本没法看。我在写生产架构时习惯在请求一开始就把thread_id绑到用户会话上然后在所有图调用里显式传入config {configurable: {thread_id: user_session_id}} result compiled_graph.invoke(initial_state, configconfig)这样不仅做了隔离还能在用户翻历史记录时快速定位他当时到底跑到哪一步。3.2 工具调用协议化从自由文本到结构化落执行Demo 智能体调工具的方式通常是让模型自由输出一段字符串比如调用query_order(order_id123)然后主程序用正则把 order_id 抠出来再执行。这个方案在演示时装模作样但生产环境就是灾难。模型一旦输出格式稍有偏差正则匹配失败、业务数据拉不出来用户看到的就是答非所问或者报错。Deep Agents 参考实现里的做法是每个工具定义 Pydantic Schema模型在工具调用前先输出结构化参数框架负责做 JSON Schema 校验。我摘一段典型的等价实现from pydantic import BaseModel class QueryOrderParams(BaseModel): order_id: str customer_id: str | None None def query_order(params: QueryOrderParams) - dict: return order_api.query(order_idparams.order_id, ...)然后在 LangChain 里把函数绑定给模型from langchain_openai import ChatOpenAI llm ChatOpenAI(model...) llm_with_tools llm.bind_tools([query_order])模型在需要调工具时会返回一个结构化的tool_calls请求里面已经分好了函数名和参数 JSON。你在图节点里只要做一个统一分发器def call_tool(state: DeepAgentState) - DeepAgentState: tool_calls state[pending_tool_calls] results [] for call in tool_calls: try: result tool_registry.invoke(call[name], call[arguments]) except Exception as e: result {error: str(e)} results.append({tool: call[name], output: result}) return {tool_results: results}注意我把result直接变成可序列化的字典返回而不是 Python 对象。原因很简单状态要能持久化、要能打印、要能作为下一步模型的上下文它必须是可以进 JSON 的。如果你把 DataFrame、自定义对象塞进状态后面的落库和恢复都会特别麻烦。关于工具调用我还想强调一个点工具层的失败兜底很重要。线上工具很可能超时、下游服务报 500、数据库连接断开。不要指望模型来处理这些异常模型看到异常它可能崩溃或开始胡言乱语。更可靠的做法是在工具分发层就把异常捕获住返回一个结构化错误对象交给模型作为“这条路径不通”的观察结果让它根据这个信息调整计划。我在这种设计里还加了一个原则工具本身必须幂等。很多业务操作比如“发送优惠券”“标记订单已发货”重复执行会有副作用。智能体会为了重试而重试所以工具的雪花 ID、请求幂等键这些是必需品。不然模型一个重试动作就可能给用户重复发两次券。4. LangChain 与 LangGraph 到底怎么分工组件库和调度器经常有同学问LangChain 和 LangGraph 是不是重复了都用 LangChain 了为啥还要用 LangGraph甚至面试题里都开始出现这两个框架的区别。我在源码里看了很多遍之后得出的结论其实很简单LangChain 是“模型与工具的组件库”LangGraph 是“流程编排的运行时”。两者的关注点是不同层级的。4.1 LangChain 负责“模型侧”LangGraph 负责“流程侧”LangChain 给你的是各种现成的零件模型封装、Prompt 模板、输出解析器、向量存储、检索器、工具绑定。它帮你把“和模型打交道”这件事做得更顺手比如你切换不同模型厂商时接口不用大改你想让模型输出 JSON有with_structured_output帮你约束你想给 prompt 加历史消息有现成的 Message 封装。LangGraph 给你的是把这些零件组装成可管理系统的“骨架”。它强调状态流转和过程可控。到底先调哪个模型、再调哪个工具、结果怎么合并、错误怎么处理、走哪条分支这些都是 LangGraph 负责的事。很多人误以为 LangGraph 只是“可以做更复杂的 Agent”其实更准确的说法是LangGraph 把智能体从“一段不透明的模型调用”变成了“一组可打断、可回放、可审计的模块化步骤”。在 Deep Agents 参考实现里节点里大量使用 LangChain 的组件但整个运行流程是 LangGraph 图。两家是合作分工关系。4.2 什么时候该用 Chain什么时候该上 Graph我现在的判断标准一句话固定顺序就用 LangChain 的 Chain有循环、分支、并行、人工审批等复杂流程控制需求就上 LangGraph。举个例子一个简单的“总结邮件 → 生成回复草稿”功能串行两步就完事你完全可以在 LangChain 里写两个 Prompt 模板链式调用。但如果你要做的是一个“用户提需求 → 规划 → 调多个工具 → 根据结果反思 → 修正计划 → 最终答复”的客服智能体串行链根本表达不了。原因很简单回调次数不确定下一步的节点选择取决于上一步的工具输出。我见过一个折中的架构在一个 LangGraph 的节点内部用小型的 LangChain Chain 完成局部任务。这很合理。比如规划节点里我可以先让模型输出 JSON 计划再用一个小的输出解析 Chain 把 JSON 转成 Pydantic 对象整个过程是一个节点。图负责大流程Chain 负责节点内的局部流水线。4.3 LangChain 抽象的重与轻节点内尽量保留最少封装用 LangChain 太熟练也有副作用。开发员容易陷入“什么都要套 Chain、什么都要套 Agent”的写法结果每个节点都被封装得很重状态流转反而看不清。我个人的习惯是节点之间不要依赖 LangChain 的内部状态节点函数的签名严格只认状态字段。参考实现里节点接受state返回 dict 切片没有把 LangChain 对象直接塞进状态。这是非常有节制的用法。因为一旦你把一个 LLM 实例放进状态下次序列化恢复时就卡住了——LLM 对象默认是不可 JSON 序列化的。状态里放的全部是普通数据类型字符串、列表、字典、整数、浮点数。所以你在读源码的时候可以多看几眼数据类型的边界凡是进入状态一定是可序列化的。凡是状态之外才是模型对象、工具客户端、数据库连接等资源。这个边界守好了你的智能体才真正具备“可持久化、可恢复、可观测”的生产基础。5. 从源码走向生产我在落地 Deep Agents 时踩过的五个坑上面讲的都是架构层面的方法论真落到生产环境还会有一套和“代码本身”无关但很要命的坑。我用 Deep Agents 这套思路重构公司内部的客服智能体和账单归因智能体时前后踩了不少。挑五个最典型的按踩坑频率从上往下排。5.1 循环必须限量不只是 max_iterations 字段我在前面提过MAX_EXECUTION_STEPS但真正让我写死这个字段的是一次线上事故。当时有个账单查询智能体用户反复问“为什么账单金额不对”模型进入“查账单 → 用户不满意 → 再查账单”的死循环每轮都调数据库一次跑了几十轮。等到我发现的时候几千 token 已经花了数据库也被压出一排慢查询。教训是循环上限要在图编译之前就定死并且必须留在状态里可观测。每次执行到evaluator时都把当前步数打印出来超过阈值直接走“尽力答复”分支。宁可给用户一个未彻底解决的答复也不能让循环失控。5.2 日志和可观测性要跟着状态走而不是跟着行号走普通代码排查问题靠堆栈智能体排查问题靠状态快照。你有一次回复得奇怪最想知道的是它当时看到的状态是什么工具结果是什么模型基于什么信息得出这个结论所以我建议在节点入口和出口各打一条结构化日志def executor(state: DeepAgentState) - DeepAgentState: logger.info(executor:start, extra{current_step: state[current_step]}) ...运行节点... logger.info(executor:end, extra{results_len: len(state[results])})日志系统里能把每个节点的输入输出存起来出问题直接翻当时的快照比在那猜 Prompt 强得多。再进一步可以通过 LangGraph 自带的 trace 工具把整张图的节点流转、耗时、token 用量输出这个在调优时特别有用。我现在每次新增工具或者改 Prompt都会先跑一组回归用例看 trace再放上线。5.3 评估集不是上线后才做的智能体有一个天然劣势没有人能保证模型对同一个用户问题的回答是一致的。所以生产团队必须建一个回归评估集每次改动 Prompt、换模型、加工具前都跑一遍。我数据库里现在有一张 eval 表大概存了 200 多条用户问题每条问题都标注了预期行为比如预期调用某个工具、预期最终的答案包含某个关键词。在参考实现里评估节点也承担类似的功能图每跑一遍如果evaluator发现结果质量不达标就会返回一个“需要修正”的信号。这些信号在生产环境里都是最珍贵的 auto-label 数据。5.4 并发写 checkpoint 的冲突问题LangGraph 的持久化基于状态快照而智能体在高并发场景下同一个thread_id可能被多个请求同时写入。最常见的是用户连续发两条消息两个请求同时读到了同一个初始状态后面各自更新后写的覆盖先写的导致用户看到的状态和日志里不一致。我的处理方案很土但有效给thread_id加锁同一个 thread 的请求串行执行如果遇到连接超时或网络重试保证每个工具调用都带着幂等键。你可以在 Redis 里给每个 thread 放一个轻量锁比如SET NX语义锁的过期时间设成明显大于单次图执行时长。如果拿不到锁直接返回“正在处理请稍候”而不是把请求塞进去。5.5 别把所有全局能力都塞进智能体状态最后一个坑来自我自己的贪心。一开始我觉得状态越丰富模型决策越准确于是把用户历史、知识库摘要、工具权限、上下文全塞进DeepAgentState。结果状态暴涨每次持久化和恢复都要序列化几 KB 甚至几十 KB 的 JSON模型每次看到的上下文也会越来越长token 成本水涨船高。后来我把状态切成了三层核心业务字段放图状态作持久化会话上下文放 Redis 或数据库按thread_id读取大文件内容直接放对象存储状态里只放文件引用。模型真正需要“看”的内容在节点里动态载荷出来而不是全部塞进状态。这样既保证了节点间流程可控又不会让状态爆炸。我近期手里这个新系统就是把知识库摘要放在state[context]里做关键词命中但正文全部走外部检索只在节点执行时按需拉取。跑了一个月token 费用降了大概四成状态体积也小了一个量级。总的来说生产级智能体从来不是模型一个点上的胜利而是状态管理、工具调用、流程编排、可观测性、评估回测这些工程能力的集合体。Deep Agents 这套参考实现的思路对我最大的启发是把“让模型做决策”和“让流程可治理”这两件事清晰分开。你不需要把 LangChain 的所有能力都用上也不需要把 LangGraph 的图设计得无比复杂只要守住状态边界、控制循环、做好埋点和持久化生产环境的安全感和维护体验就会完全不一样。