ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

生产级Agent开发实战:从LangChain到Deep Agents的工程化落地

生产级Agent开发实战:从LangChain到Deep Agents的工程化落地 智能体开发这件事从2024年火到2025年真正落到生产环境的项目依然不多。我见过太多团队用 LangChain 搭了个能跑通的 Demo一到线上就各种翻车工具调用超时没人管、多轮对话状态丢失、Agent 陷入死循环烧掉几百刀 token、错误处理全靠 try-except 兜底。问题出在哪不是框架不好用而是大多数人只学了怎么调 API没搞明白生产级 Agent 到底需要哪些工程能力。这篇内容围绕 Deep Agents、LangChain 和 LangGraph 三个核心组件展开重点拆解 Deep Agents Code 这个开源项目的源码设计思路把 Agent 从能跑到能上生产之间缺失的那层工程实践补上。适合已经用 LangChain 写过基础 Agent、但对状态管理、错误恢复、工具编排、可观测性这些生产级需求还比较模糊的开发者。读完你至少能搞清楚三件事LangGraph 的状态图到底解决了什么问题、Deep Agents 在 LangChain 之上加了哪些工程层、以及一个生产级 Agent 项目的代码结构应该长什么样。1. 先搞清楚 LangChain、LangGraph 和 Deep Agents 各自站在哪一层很多人一上来就纠结LangGraph 和 LangChain 有什么区别其实这个问题本身就问偏了。它们不是替代关系而是不同抽象层级的东西。我用一个类比来解释LangChain 像是给你一套标准化的厨房设备——锅碗瓢盆、烤箱、搅拌机都有你按需组合就能做菜。LangGraph 则像是厨房的动线设计和工作流编排——先切菜还是先烧水、哪些步骤可以并行、哪个环节出错了要回退到哪一步。Deep Agents 更像是整套中央厨房解决方案它把设备、动线、食材管理、出餐流程全部打包好了你只需要定义菜单。1.1 LangChain 的核心抽象与它的边界LangChain 最核心的价值在于它定义了一套 LLM 应用开发的通用接口。ChatModel、PromptTemplate、OutputParser、Retriever、Tool这些抽象让开发者不用关心底层是 OpenAI 还是 Anthropic切换模型只需要改一行配置。它的AgentExecutor也确实能跑通LLM 决定调哪个工具→执行工具→把结果喂回 LLM这个基本循环。但 LangChain 的AgentExecutor在生产环境有几个硬伤。第一它的执行流程是隐式的你很难在中间插入自定义逻辑比如工具调用超过3次就强制中断或者某个工具返回错误时走降级路径。第二它的状态管理比较粗糙多轮对话的上下文靠Memory组件维护但Memory本质上就是一个消息列表的增删改查没法表达更复杂的状态转换。第三错误处理基本靠max_iterations和early_stopping_method这两个参数粒度太粗。我实际项目中遇到过这样的情况一个客服 Agent 需要先查订单、再查物流、最后根据结果决定是否转人工。用AgentExecutor写LLM 有时候会跳过查订单直接查物流有时候查完订单忘了查物流就开始回复用户。你没法在框架层面强制这个顺序只能靠 prompt 里反复强调但 prompt 的遵循率你懂的不可能100%。1.2 LangGraph 用状态图重新定义了 Agent 的执行模型LangGraph 的核心创新是把 Agent 的执行流程建模成一张有向图。图中的每个节点是一个计算步骤可以是一次 LLM 调用、一次工具执行、或者一段纯代码逻辑边定义了节点之间的流转条件。整个图的执行状态由一个共享的State对象维护每个节点读取 State、执行逻辑、返回 State 的更新。这个模型的好处是显式化。你不再需要猜测 Agent 内部发生了什么打开图定义就能看到完整的执行路径。更重要的是你可以在任意节点之间插入条件边实现如果工具调用失败则走重试节点如果置信度低于阈值则走人工审核节点这类生产环境必需的控制逻辑。LangGraph 的另一个关键能力是持久化。它内置了 Checkpointer 机制每一步执行完都会把 State 快照存下来。这意味着 Agent 可以在任意步骤中断后恢复也支持时间旅行式的调试——你可以回到任意一个历史状态重新执行。对于需要长时间运行的任务比如一个需要跑几小时的调研 Agent这个能力是刚需。1.3 Deep Agents 在两者之上补了哪些工程层Deep Agents 是 LangChain 官方团队推出的一个更高层抽象它的定位是开箱即用的生产级 Agent 模板。从 Deep Agents Code 的源码结构来看它在 LangGraph 的基础上主要补了四层东西第一层是规划与任务分解。Deep Agents 内置了一个 Planner 节点会把用户的复杂请求自动拆解成子任务列表然后逐个执行。这个 Planner 不是简单的 prompt 工程它有一套完整的任务状态管理机制支持子任务之间的依赖关系。第二层是子 Agent 编排。一个复杂任务往往需要多种能力的 Agent 协作比如一个负责搜索、一个负责代码执行、一个负责文件操作。Deep Agents 定义了一套 SubAgent 协议让主 Agent 可以把子任务委派给专门的子 Agent并管理它们的生命周期。第三层是文件系统抽象。Deep Agents 内置了一个虚拟文件系统Agent 可以在执行过程中读写文件、保存中间结果、维护工作记忆。这个设计借鉴了人类处理复杂任务时把想法写下来的习惯实测对长任务的稳定性提升很明显。第四层是中间件机制。Deep Agents 允许你在 Agent 执行的关键节点插入中间件比如每次工具调用前检查权限每次 LLM 调用后记录 token 消耗检测到敏感操作时暂停等待人工确认。这层机制是生产环境合规和可观测性的基础。下面这张表可以帮你快速定位三个组件的职责边界维度LangChainLangGraphDeep Agents核心抽象Chain / AgentExecutorStateGraph / Node / EdgePlanner / SubAgent / Middleware状态管理Memory消息列表共享 State Checkpointer虚拟文件系统 任务状态流程控制隐式循环显式图 条件边规划驱动 动态委派错误恢复max_iterations断点续跑 时间旅行子任务级重试 降级策略适用场景简单链式调用复杂流程编排生产级多 Agent 系统2. 从 Deep Agents Code 源码看生产级 Agent 的目录结构设计拿到一个开源项目我习惯先看它的目录结构因为目录结构往往比文档更能反映作者的工程思维。Deep Agents Code 的源码组织方式很值得借鉴它把Agent 逻辑和工程基础设施做了清晰的分离。2.1 核心模块划分与职责边界Deep Agents Code 的源码大致分为这几个目录agents/存放 Agent 的定义包括主 Agent 和各个 SubAgent 的配置。每个 Agent 一个文件里面定义它的 system prompt、可用工具列表、使用的模型、以及绑定的中间件。graph/存放 LangGraph 的图定义。包括节点函数、条件边逻辑、State 的 schema 定义。这个目录是纯编排逻辑不包含具体的业务实现。tools/存放所有工具的定义。每个工具一个文件包含工具的 schema 描述、执行函数、错误处理逻辑。middleware/存放中间件。比如日志中间件、权限检查中间件、token 统计中间件。memory/存放记忆和状态管理相关的代码。包括虚拟文件系统的实现、对话历史的压缩策略、长期记忆的存储接口。config/存放配置。模型配置、工具配置、中间件配置全部通过配置文件管理不硬编码在代码里。这个划分的核心原则是关注点分离。graph/只管流程怎么走tools/只管每个工具怎么执行agents/只管每个 Agent 的角色定义。当你想改一个工具的实现时不需要动图定义当你想调整执行流程时不需要动工具代码。我见过很多项目把所有逻辑塞在一个agent.py里几百行代码混在一起改一个工具的参数要翻半天。Deep Agents Code 这种组织方式在项目规模超过10个工具、3个 Agent 之后优势会非常明显。2.2 State Schema 的设计Agent 的工作记忆该怎么定义LangGraph 的 State 是整个 Agent 的共享内存所有节点都能读写。State 的 schema 设计直接决定了 Agent 能表达多复杂的状态。Deep Agents Code 里的 State 定义大致包含这几类字段from typing import TypedDict, Annotated from langgraph.graph import add_messages class AgentState(TypedDict): # 消息历史用 add_messages reducer 自动合并 messages: Annotated[list, add_messages] # 当前任务规划 plan: list[dict] # [{task: ..., status: pending/done/failed}] # 虚拟文件系统的文件映射 files: dict[str, str] # 当前活跃的子 Agent active_subagent: str | None # 错误计数用于触发降级 error_count: int # 人工审核标记 needs_human_review: bool这里有几个设计细节值得说。messages字段用了add_messages这个 reducer意味着每个节点返回的新消息会自动追加到列表里而不是覆盖。这是 LangGraph 的一个核心机制——reducer 定义了状态更新的合并策略。如果你不用 reducer每个节点返回的messages会直接替换掉原来的历史就丢了。plan字段用列表存储任务规划每个任务有状态标记。这样 Planner 节点生成计划后Executor 节点可以逐个执行并更新状态整个过程中断后恢复也能知道执行到哪了。files字段是虚拟文件系统的核心。Agent 在执行过程中可以把中间结果写进文件后续节点读取文件而不是依赖消息历史。这个设计对长任务特别有用因为消息历史会随着轮次增加越来越长最终超出 context window而文件是按需读取的。2.3 图定义中的条件边生产环境必需的分支逻辑Deep Agents Code 的图定义里条件边是生产级能力的集中体现。一个典型的图结构大致是这样的from langgraph.graph import StateGraph, END def should_continue(state: AgentState) - str: 决定下一步走向 last_message state[messages][-1] # 如果 LLM 没有调用工具说明它认为任务完成 if not last_message.tool_calls: return end # 如果错误次数超过阈值走降级 if state[error_count] 3: return fallback # 如果需要人工审核暂停 if state[needs_human_review]: return human_review return tools graph StateGraph(AgentState) graph.add_node(planner, planner_node) graph.add_node(agent, agent_node) graph.add_node(tools, tool_node) graph.add_node(fallback, fallback_node) graph.add_node(human_review, human_review_node) graph.set_entry_point(planner) graph.add_edge(planner, agent) graph.add_conditional_edges(agent, should_continue, { end: END, tools: tools, fallback: fallback, human_review: human_review, }) graph.add_edge(tools, agent)这段代码里最关键的是should_continue这个路由函数。它检查三个条件任务是否完成、错误是否超限、是否需要人工介入。这三个条件覆盖了生产环境最常见的三种分支场景。对比一下 LangChain 的AgentExecutor你只能设置max_iterations来防止死循环但没法在错误超限时走一条完全不同的处理路径。LangGraph 的条件边让这种精细控制成为可能。3. 工具调用与错误恢复生产环境最容易翻车的地方工具调用是 Agent 和外部世界交互的唯一通道也是生产环境出问题最多的地方。我统计过自己经手的项目线上事故里大概有六成和工具调用有关超时、返回格式不对、权限不足、第三方服务挂了、返回内容超出预期长度。Deep Agents Code 在工具层做了不少工程处理值得逐个拆解。3.1 工具定义的 schema 设计与参数校验LangChain 的工具定义用tool装饰器从函数的类型注解自动生成 schema。这个机制很方便但生产环境需要更严格的校验。Deep Agents Code 里的工具定义通常包含这几层from pydantic import BaseModel, Field, field_validator from langchain_core.tools import tool class SearchInput(BaseModel): query: str Field(description搜索关键词, min_length1, max_length200) max_results: int Field(default5, ge1, le20) field_validator(query) classmethod def sanitize_query(cls, v: str) - str: # 去掉首尾空白过滤特殊字符 return v.strip() tool(args_schemaSearchInput) def search_tool(query: str, max_results: int 5) - str: 搜索互联网获取信息 try: results do_search(query, max_results) return format_results(results) except TimeoutError: return 搜索超时请稍后重试或换一个关键词 except Exception as e: return f搜索失败{type(e).__name__}这里有几个关键点。第一用 Pydantic 模型定义参数 schema而不是只靠函数签名。Pydantic 提供了min_length、ge、le这些约束能在参数进入函数之前就拦截非法输入。第二用field_validator做参数清洗比如去掉首尾空白、过滤特殊字符。第三工具函数内部捕获异常并返回人类可读的错误信息而不是直接抛出。第三点特别重要。如果工具直接抛异常LangGraph 的执行会中断整个 Agent 挂掉。如果返回错误信息字符串LLM 会看到这个信息并决定下一步怎么做——可能是重试、可能是换个工具、可能是告诉用户失败了。这给了 Agent 自我修复的机会。3.2 超时控制与重试策略的工程实现工具调用超时是生产环境最常见的问题。一个 HTTP 请求卡住30秒整个 Agent 就卡住30秒。Deep Agents Code 的做法是给每个工具设置独立的超时并且用异步执行避免阻塞。import asyncio from functools import wraps def with_timeout(seconds: int): def decorator(func): wraps(func) async def wrapper(*args, **kwargs): try: return await asyncio.wait_for( func(*args, **kwargs), timeoutseconds ) except asyncio.TimeoutError: return f工具执行超时{seconds}秒请尝试其他方式 return wrapper return decorator tool with_timeout(15) async def fetch_url(url: str) - str: 获取网页内容 async with aiohttp.ClientSession() as session: async with session.get(url) as resp: return await resp.text()超时时间怎么定我的经验是搜索类工具10-15秒数据库查询5秒文件操作3秒LLM 调用30-60秒。这些数字不是拍脑袋来的而是根据 P99 延迟加一定余量。你可以先上线观察一段时间统计每个工具的实际耗时分布再调整超时阈值。重试策略要区分错误类型。网络抖动导致的超时值得重试参数错误导致的失败重试多少次都没用。Deep Agents Code 里通常用指数退避加最大重试次数async def retry_with_backoff(func, max_retries3, base_delay1): for attempt in range(max_retries): try: return await func() except (TimeoutError, ConnectionError) as e: if attempt max_retries - 1: raise delay base_delay * (2 ** attempt) await asyncio.sleep(delay)3.3 工具返回结果的长度控制与截断这个问题很多人会忽略。一个搜索工具返回了5万字的网页内容直接塞进消息历史下一次 LLM 调用的 token 消耗直接爆炸。Deep Agents Code 在工具层做了结果截断def truncate_result(text: str, max_chars: int 4000) - str: if len(text) max_chars: return text # 保留开头和结尾中间用省略标记 head text[:max_chars // 2] tail text[-(max_chars // 2):] return f{head}\n\n... [内容过长已截断 {len(text) - max_chars} 字符] ...\n\n{tail}保留开头和结尾而不是只保留开头是因为很多文档的关键信息在结尾比如结论、总结。截断标记要明确告诉 LLM 内容被截断了否则它会以为这就是全部信息。更进阶的做法是把完整结果存进虚拟文件系统只把摘要和文件路径返回给 LLM。LLM 如果需要更多细节可以主动调用文件读取工具。这个模式在 Deep Agents 里叫渐进式信息披露对处理长文档特别有效。4. 多 Agent 协作与任务委派的实际落地单 Agent 能处理的任务复杂度是有上限的。当任务涉及多个专业领域比如既要写代码又要查资料还要操作数据库把所有工具塞给一个 Agent 会导致 prompt 过长、工具选择准确率下降。Deep Agents 的解法是主 Agent 加子 Agent 的架构。4.1 主 Agent 如何决定委派给哪个子 Agent主 Agent 的核心职责是任务分解和路由它自己不做具体执行。Deep Agents Code 里主 Agent 的 system prompt 大致是这样的结构你是一个任务协调者。你的职责是 1. 分析用户请求判断需要哪些能力 2. 将任务分解为子任务委派给对应的子 Agent 3. 收集子 Agent 的结果整合后回复用户 可用的子 Agent - research_agent负责信息搜索和资料整理 - code_agent负责代码编写和执行 - data_agent负责数据查询和分析 委派时使用 delegate_task 工具指定 agent_name 和 task_description。主 Agent 通过一个delegate_task工具来委派任务。这个工具的输入是子 Agent 名称和任务描述输出是子 Agent 的执行结果。从 LangGraph 的角度看委派本质上就是一次特殊的工具调用只不过执行工具的不是外部 API而是另一个 Agent 图。4.2 子 Agent 的隔离与上下文传递子 Agent 和主 Agent 之间需要做好隔离。子 Agent 有自己的消息历史、自己的工具集、自己的 system prompt。主 Agent 的消息历史不会自动传给子 Agent只传递任务描述。这个隔离设计有两个好处。第一子 Agent 的 context window 不会被主 Agent 的无关历史占用。第二子 Agent 的行为更可控不会因为看到了主 Agent 的历史而产生意外行为。但隔离也带来一个问题子 Agent 执行任务需要的背景信息怎么传Deep Agents Code 的做法是在任务描述里显式包含必要的上下文。比如主 Agent 委派代码任务时会把需求描述、输入数据格式、预期输出格式都写进task_description。delegate_task( agent_namecode_agent, task_description 编写一个 Python 函数实现以下功能 - 输入一个包含订单信息的 JSON 列表 - 输出按用户 ID 分组的订单统计 - 要求处理空列表和缺失字段的情况 - 输入示例[{user_id: u1, amount: 100}, ...] )4.3 子 Agent 失败时的降级与兜底子 Agent 执行失败是常态必须有降级策略。Deep Agents Code 里的处理逻辑大致是子 Agent 返回明确失败信息时主 Agent 尝试重新委派但会调整任务描述比如缩小范围、提供更多上下文。重试超过2次仍失败主 Agent 尝试自己用基础工具完成或者委派给备选子 Agent。所有路径都失败主 Agent 向用户返回明确的失败原因和已尝试的方案。这个降级链路需要在图定义里用条件边实现。关键是要设置全局错误预算防止无限重试。我通常设置单个子任务最多重试2次整个会话最多触发3次降级超过就终止并报告。5. 可观测性Agent 上线后你怎么知道它跑得好不好Agent 和传统软件最大的区别是它的行为不确定。同样的输入两次执行可能走完全不同的路径。没有可观测性你根本不知道线上发生了什么。Deep Agents Code 在中间件层提供了几个关键的观测点。5.1 关键指标的采集与埋点位置生产级 Agent 至少要采集这几类指标指标类别具体指标采集位置性能LLM 调用延迟、工具执行延迟、端到端延迟中间件包裹每次调用成本每次会话 token 消耗、单次调用 token 数LLM 调用后统计质量工具调用成功率、任务完成率、重试次数节点执行后记录行为执行路径、工具调用序列、分支走向图执行时记录埋点的最佳位置是中间件。Deep Agents 的中间件机制允许你在不改动业务逻辑的情况下插入埋点代码class MetricsMiddleware: async def on_llm_start(self, state, config): self.start_time time.time() async def on_llm_end(self, state, response): latency time.time() - self.start_time tokens response.usage_metadata[total_tokens] metrics.record(llm_latency, latency) metrics.record(llm_tokens, tokens)5.2 执行轨迹的记录与回放LangGraph 的 Checkpointer 天然支持执行轨迹记录。每一步的 State 快照都存下来了你可以完整回放一次会话的执行过程。这对调试特别有用——当用户报告Agent 回答错了你可以调出那次会话的完整轨迹看到它在哪一步走偏了。我通常会把轨迹存到数据库并且提供一个查询界面。关键字段包括会话 ID、每步的节点名、输入 State 摘要、输出 State 摘要、耗时、token 消耗。有了这些数据你可以分析出很多问题哪个工具最常失败、哪个节点最耗时、哪类任务最容易触发重试。5.3 异常告警的阈值设置经验告警阈值不能拍脑袋定要基于实际数据。我的做法是上线后先观察一周统计各项指标的 P50、P95、P99然后按 P99 的1.5倍设置告警阈值。比如 LLM 调用延迟 P99 是8秒那告警阈值设12秒。几个必须设告警的指标单次会话 token 消耗超过阈值防止死循环烧钱、工具调用失败率超过10%、端到端延迟超过用户可接受上限、错误降级触发次数突增。告警要带上下文比如会话 xxx 的 token 消耗达到 50000执行路径为 planner→agent→tools→agent→...这样收到告警能快速定位。6. 从 Demo 到生产的几个关键决策点最后聊几个我在实际项目中踩过坑、后来形成固定做法的决策点。这些不是框架层面的东西而是工程经验。6.1 模型选择的权衡能力、成本、延迟的三角生产环境选模型不是越强越好。GPT-4 级别的模型能力强但贵且慢小模型便宜快但复杂任务搞不定。我的做法是分层用模型主 Agent 的规划节点用强模型保证任务分解质量子 Agent 的执行节点用中等模型平衡成本和能力简单的格式化、分类任务用小模型。Deep Agents Code 支持给每个节点单独配置模型这个灵活性很重要。实测下来一个规划用强模型加执行用中等模型的组合成本能比全用强模型降低60%以上而任务完成率只下降不到5%。6.2 上下文窗口管理什么时候该压缩、什么时候该丢弃长会话的上下文管理是个难题。消息历史越来越长最终会超出 context window。Deep Agents Code 提供了几种策略滑动窗口只保留最近 N 轮对话更早的丢弃。简单但会丢失早期信息。摘要压缩把早期对话用 LLM 总结成一段摘要替换原始消息。保留信息但增加一次 LLM 调用。文件外置把重要信息写进虚拟文件系统消息历史里只保留文件引用。按需读取最灵活。我的经验是组合使用最近5轮保留原文5-20轮做摘要压缩20轮以上的重要信息外置到文件。这个策略在实测中能把 context 长度控制在合理范围同时不丢失关键信息。6.3 人工介入的触发条件设计不是所有任务都适合全自动。涉及资金操作、敏感数据、不可逆动作的场景必须有人工确认环节。Deep Agents Code 的needs_human_review标记就是干这个的。触发人工介入的条件要设计得精准。太宽松会导致频繁打断用户体验差太严格又起不到保护作用。我通常设置这几类触发条件工具调用涉及写操作且影响范围超过阈值、Agent 连续两次执行结果矛盾、置信度评分低于阈值、用户显式要求人工服务。人工介入的实现方式有两种同步等待Agent 暂停等人确认后继续和异步通知Agent 先做其他任务人确认后再回来处理。同步适合紧急操作异步适合批量处理。LangGraph 的 Checkpointer 让同步等待变得可行——暂停时 State 已经持久化了人确认后从断点恢复即可。6.4 版本迭代与灰度发布的工程实践Agent 的 prompt、工具、图结构都会随版本迭代变化。直接全量发布风险很大因为 Agent 的行为变化很难通过单元测试完全覆盖。我的做法是灰度发布加 A/B 对比。具体操作新版本先对5%的流量生效同时记录新旧版本的关键指标任务完成率、平均轮次、token 消耗、用户满意度。观察24小时如果新版本指标不劣于旧版本逐步扩大比例到20%、50%、100%。如果指标恶化立即回滚。这套机制的前提是配置与代码分离。prompt、模型配置、工具开关都放在配置文件或配置中心改配置不需要重新部署。Deep Agents Code 的config/目录设计就是为此服务的。7. 一个最小可用的生产级 Agent 骨架把上面的东西串起来一个最小可用的生产级 Agent 项目骨架大致是这样的project/ ├── config/ │ ├── models.yaml # 模型配置 │ ├── agents.yaml # Agent 定义 │ └── middleware.yaml # 中间件配置 ├── agents/ │ ├── main_agent.py # 主 Agent │ └── sub_agents/ # 子 Agent ├── graph/ │ ├── state.py # State schema │ ├── nodes.py # 节点函数 │ └── builder.py # 图构建 ├── tools/ │ ├── search.py │ ├── code_exec.py │ └── file_ops.py ├── middleware/ │ ├── metrics.py │ ├── logging.py │ └── guardrails.py ├── memory/ │ ├── checkpointer.py │ └── file_system.py └── main.py # 入口这个骨架的核心思想是每一层都可以独立替换和测试。工具层可以单独写单元测试图定义可以用 mock 节点测试流转逻辑中间件可以独立验证埋点是否正确。这种可测试性是从 Demo 走向生产的关键。我在实际项目中的体会是Agent 开发的难点从来不在怎么调 LLM而在怎么让它在各种异常情况下依然稳定工作。Deep Agents、LangChain、LangGraph 这三个组件本质上是在不同抽象层级上帮你处理这些工程问题。LangChain 解决了接口标准化LangGraph 解决了流程可控Deep Agents 解决了生产级能力的开箱即用。理解它们各自站在哪一层比记住具体 API 更重要。最后分享一个小技巧每次 Agent 线上出问题我都会把那次会话的完整执行轨迹导出来逐节点分析。坚持做这件事三个月你会对Agent 会在哪里出问题形成直觉这种直觉比任何文档都有价值。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进