1. DeepAgent 到底是个什么东西
第一次听到 DeepAgent 这个名字,很多人会下意识觉得它又是一个套壳的 Agent 框架,无非是把 LangChain 的AgentExecutor换个包装再发一遍。我一开始也是这么想的,直到把它和 LangGraph 放在一起跑通一个多步任务之后,才意识到它想解决的问题其实很具体:让 Agent 在长链路、多工具、多轮次的任务里,不再靠一个巨大的 prompt 硬撑,而是把"思考"和"执行"拆成可编排、可观测、可中断的图结构。
DeepAgent 是构建在 LangChain 与 LangGraph 之上的 Agent 编排方案,核心关键词就是这几个:DeepAgent、LangGraph、LangChain、中间件、Agent。它面向的是需要处理复杂任务的开发者——比如你要做一个能读测试用例、自动生成 UI 自动化脚本的 Agent,或者做一个带记忆、带工具调用、带人工审核节点的本地知识库助手。这类任务用最原始的initialize_agent写出来,代码会迅速膨胀成一坨难以维护的意大利面,而 DeepAgent 提供的是一套"分层 + 中间件 + 图编排"的组织方式。
这篇文章我会从整体设计思路讲起,把 DeepAgent 和 LangGraph、LangChain 的关系理清楚,然后拆解它的核心机制,包括中间件、工具调用、状态管理、记忆这几块,再给出一套可以直接抄的实操流程,最后把我踩过的坑和排查经验整理成速查表。不管你是刚接触 Agent 开发的新手,还是已经用 LangChain 写过几个 demo 想进阶的人,应该都能从里面拿到能直接用的东西。
2. 整体设计与思路拆解
2.1 为什么不是"一个 Agent 打天下"
先说清楚一个背景。LangChain 早期的 Agent 抽象非常直接:给它一个 LLM、一堆工具、一段系统提示,它就会进入Thought -> Action -> Observation的循环,直到给出Final Answer。这个模式在工具少于 5 个、任务步数少于 5 步的时候很好用,代码也短。
但真实项目里,任务复杂度一上来,问题就暴露了:
- 上下文爆炸:每一步的 Observation 都塞回 prompt,几十步之后上下文直接顶到模型窗口上限,成本和延迟都失控。
- 不可中断:跑到一半想让人工确认一下,或者想暂停保存状态,原生 Agent 做不到。
- 不可观测:出错了你只能看到一长串日志,很难定位是哪一步、哪个工具出的问题。
- 难以复用:不同任务要重写一遍循环逻辑,工具和提示词耦合在一起。
DeepAgent 的设计思路,本质上是把"一个 Agent 打天下"换成"一个可编排的执行图 + 若干可插拔的中间件"。LangGraph 负责图结构和状态流转,LangChain 负责模型、工具、检索这些基础组件,DeepAgent 则在这两者之上定义了一套更贴近"深度任务"的编排约定。
提示:理解 DeepAgent 的关键,是先接受"Agent 不是一个对象,而是一张图"这个心智模型。想通这一点,后面所有设计都顺了。
2.2 LangChain、LangGraph、DeepAgent 三者的分工
很多人分不清 LangChain 和 LangGraph 的区别,面试也常问。我用一句话概括:LangChain 提供零件,LangGraph 提供装配线,DeepAgent 提供的是针对复杂任务的装配工艺。
| 层级 | 代表组件 | 职责 | 类比 |
|---|---|---|---|
| LangChain | ChatModel、Tool、Retriever、Prompt | 提供模型调用、工具封装、检索等原子能力 | 汽车零件 |
| LangGraph | StateGraph、Node、Edge、Checkpointer | 把原子能力编排成有状态、可循环、可中断的图 | 装配流水线 |
| DeepAgent | 中间件、规划节点、执行节点、记忆层 | 针对深度任务定义编排模式与扩展点 | 装配工艺标准 |
这个分层带来的直接好处是:你换模型、换工具、换检索库,都不需要动编排逻辑;你改编排逻辑,也不需要重写工具。这就是"中间件"这个词在这里的意义——它把横切关注点(日志、重试、权限、人工审核、记忆写入)从业务节点里抽出来,做成可插拔的一层。
2.3 中间件机制为什么是核心
中间件(middleware)在 DeepAgent 里不是可有可无的装饰,而是整个架构的骨架。它的工作方式很像 Web 框架里的请求中间件:一个请求(这里是一次 Agent 步骤)从进入节点到离开节点,会依次穿过若干中间件,每个中间件可以在前置、后置、异常三个时机介入。
我实际用下来,中间件主要解决四类问题:
- 可观测性:记录每一步的输入输出、耗时、token 消耗,方便排查和成本核算。
- 可靠性:工具调用失败时自动重试、降级、熔断,避免整个图因为一次网络抖动崩掉。
- 安全与合规:在工具真正执行前做参数校验、权限判断,拦截危险操作。
- 人机协同:在关键节点插入人工确认,确认通过才继续往下走。
这四类需求几乎每个生产级 Agent 都会遇到,如果不用中间件,你就得在每个节点里重复写一遍,代码会迅速腐化。DeepAgent 把它们收敛成统一扩展点,这是它相比裸写 LangGraph 最实在的价值。
2.4 状态设计:图能跑起来的地基
LangGraph 的图是围绕一个共享的State对象运转的,所有节点读写同一个状态。DeepAgent 在这个基础上,通常会把状态拆成几个逻辑区:
- messages:对话与工具调用历史,这是最核心的部分。
- plan:当前任务的规划结果,比如待办步骤列表。
- scratchpad:中间推理草稿,不一定要回传给模型。
- artifacts:任务产出的结构化结果,比如生成的脚本、报告。
- meta:轮次计数、错误计数、当前阶段等控制信息。
把状态分区的好处是,你可以精确控制"哪些信息回传给模型、哪些只留在本地"。比如 scratchpad 里的草稿没必要每轮都塞进 prompt,artifacts 只在最后汇总时读取。这一步做得好不好,直接决定了你的 Agent 能不能跑长任务而不爆上下文。
3. 核心细节解析与实操要点
3.1 规划与执行分离:DeepAgent 的"深度"体现在哪
DeepAgent 名字里的"Deep",我理解主要体现在规划与执行分离上。普通 Agent 是边想边做,一步一个 Thought;DeepAgent 倾向于先产出一个显式的计划,再逐步执行,执行过程中还能回头修订计划。
这个模式的好处很直观。举个我做过的例子:让 Agent 读取一批测试用例,自动生成 Playwright 的 UI 自动化脚本。如果边想边做,模型很容易在第三个用例就忘了前面的命名规范;而先规划出"解析用例 -> 提取元素定位 -> 生成脚本骨架 -> 填充断言 -> 校验语法"这几个阶段,再逐阶段执行,产出的脚本一致性会高很多。
规划节点通常这样设计:
from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] plan: list[str] current_step: int artifacts: dict def planner_node(state: AgentState): # 调用模型生成结构化计划 plan = llm_plan(state["messages"]) return {"plan": plan, "current_step": 0} def executor_node(state: AgentState): step = state["plan"][state["current_step"]] result = run_step(step, state) return { "messages": [result], "current_step": state["current_step"] + 1, } def should_continue(state: AgentState): if state["current_step"] >= len(state["plan"]): return "end" return "continue"这段代码是骨架,重点在于planner_node和executor_node的职责是分开的。规划节点只负责"想",执行节点只负责"做",两者通过plan和current_step这两个状态字段通信。这种解耦让每一步都可以单独测试、单独加中间件。
注意:规划节点产出的计划一定要做结构化约束(比如用 JSON schema 或 Pydantic 模型),否则模型会给你返回一段自然语言,后续解析会非常痛苦。
3.2 工具调用的三种典型模式
工具调用是 Agent 的手脚,DeepAgent 里常见的调用模式有三种,选错了会直接影响稳定性。
第一种:模型自主决定调用哪个工具。这是最经典的 function calling 模式,把工具 schema 交给模型,模型返回tool_calls,框架负责执行并把结果回填。优点是灵活,缺点是模型可能选错工具或传错参数。
第二种:图结构硬编码调用顺序。某些任务步骤是确定的,比如"先检索再总结",那就没必要让模型决定,直接在图上连边。优点是稳定可控,缺点是不够灵活。
第三种:混合模式。大阶段由图标定,阶段内部由模型自主选工具。这是我在实际项目里用得最多的方式,兼顾了可控性和灵活性。
工具定义本身也有讲究。下面是一个规范的 LangChain 工具定义:
from langchain_core.tools import tool @tool def read_test_case(file_path: str) -> str: """读取指定路径的测试用例文件内容。 Args: file_path: 测试用例文件的绝对路径,必须以 .yaml 或 .json 结尾。 """ if not file_path.endswith((".yaml", ".json")): raise ValueError("仅支持 yaml 或 json 格式的用例文件") with open(file_path, "r", encoding="utf-8") as f: return f.read()这里有几个细节值得说。docstring 不是装饰,是给模型看的说明书,写得越清楚,模型选错工具的概率越低。参数类型标注要准确,LangChain 会据此生成 JSON schema。工具内部要做参数校验,别指望模型每次都传对。
3.3 记忆层:短期、长期、工作记忆
Agent 记忆是热词里出现频率很高的一个,也是新手最容易做糊的地方。我习惯把记忆分成三层:
- 短期记忆:就是当前会话的 messages,靠 LangGraph 的 checkpointer 持久化,支持中断恢复。
- 长期记忆:跨会话的知识,通常落到向量库(比如 Chroma)或关系库,通过检索注入。
- 工作记忆:当前任务内的临时状态,比如 scratchpad、中间变量,任务结束就丢弃。
三层记忆的读写时机完全不同。短期记忆每轮自动更新;长期记忆在任务开始和关键节点检索;工作记忆只在任务生命周期内有效。把它们混在一起,就会出现"上个任务的草稿污染这个任务"的经典 bug。
用 Chroma 做长期记忆的检索注入,大致是这样:
from langchain_chroma import Chroma from langchain_ollama import OllamaEmbeddings vectorstore = Chroma( collection_name="agent_memory", embedding_function=OllamaEmbeddings(model="nomic-embed-text"), persist_directory="./chroma_db", ) def retrieve_memory(query: str, k: int = 3): docs = vectorstore.similarity_search(query, k=k) return "\n".join(d.page_content for d in docs)提示:本地知识库场景下,Ollama + LangChain + Chroma 这套组合是性价比很高的选择,模型和 embedding 都能本地跑,数据不出机器。
3.4 中间件的实现方式
中间件在 LangGraph 里没有官方的一等公民抽象,通常有两种落地方式:包装节点函数和用图的边做拦截。
包装节点函数是最直观的:
def with_logging(node_fn): def wrapped(state): start = time.time() result = node_fn(state) elapsed = time.time() - start logger.info(f"{node_fn.__name__} 耗时 {elapsed:.2f}s") return result return wrapped graph.add_node("executor", with_logging(executor_node))这种方式简单,但只能做前置和后置。如果要在异常时介入,就得用 try/except 包起来。更复杂的场景,比如"工具执行前需要人工确认",更适合用图的边加条件路由来实现——在工具节点前插一个human_approval节点,根据确认结果决定走执行还是走取消。
我个人的经验是:轻量横切逻辑用装饰器,涉及流程分支的用图节点。别硬把所有人都塞进装饰器,那样图会变得很难读。
4. 实操过程与核心环节实现
4.1 环境准备与依赖安装
先把环境搭起来。我用的组合是 Python 3.11 + LangChain + LangGraph + Ollama(本地模型)+ Chroma(向量库)。这套组合的好处是全本地可跑,不依赖外部服务,适合学习和内网部署。
python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install langchain langchain-core langgraph pip install langchain-ollama langchain-chroma pip install chromadb模型侧,本地拉一个能跑 function calling 的模型:
ollama pull qwen2.5:7b ollama pull nomic-embed-text选qwen2.5:7b的原因是它对中文和工具调用支持都不错,7B 参数量在消费级显卡上能跑。如果你机器显存紧张,可以退到 3B 版本,但工具调用的准确率会下降,这点要有心理预期。
4.2 定义状态与图结构
接下来定义状态和图的骨架。这一步是整个项目的"地基",改起来成本最高,所以要想清楚再动手。
from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver class DeepAgentState(TypedDict): messages: Annotated[list, operator.add] plan: list[str] current_step: int artifacts: dict error_count: int builder = StateGraph(DeepAgentState) builder.add_node("planner", planner_node) builder.add_node("executor", executor_node) builder.add_node("reflector", reflector_node) builder.set_entry_point("planner") builder.add_edge("planner", "executor") builder.add_conditional_edges( "executor", should_continue, {"continue": "executor", "reflect": "reflector", "end": END}, ) builder.add_edge("reflector", "executor") checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer)这里我加了reflector节点,作用是每执行若干步后回头检查计划是否需要修订。这是"深度"任务的常见需求——计划不是一次定死的,执行中发现偏差要能调整。MemorySaver是内存版 checkpointer,生产环境要换成数据库版,否则进程重启状态就丢了。
4.3 规划节点的提示词设计
规划节点的提示词直接决定计划质量。我踩过的坑是:一开始提示词写得太随意,模型返回的计划粒度忽大忽小,有的步骤是"分析需求",有的是"打开文件第 3 行读取字段",根本没法统一执行。
后来我改成强制结构化输出,并要求每个步骤满足"单一动作 + 明确输入输出":
PLANNER_PROMPT = """你是一个任务规划器。请把用户任务拆解成有序的执行步骤。 要求: 1. 每个步骤只包含一个明确动作,动词开头。 2. 每个步骤必须说明输入来源和预期输出。 3. 步骤数量控制在 3 到 8 个之间。 4. 以 JSON 数组返回,每个元素包含 step 和 expected_output 两个字段。 用户任务:{task} """配合 Pydantic 做输出解析,基本能保证计划结构稳定。如果模型偶尔返回不合规的 JSON,加一层重试和修复逻辑就行。
4.4 执行节点与工具绑定
执行节点负责把当前步骤落地。核心是把工具绑定到模型上,让模型在步骤内自主选择工具:
from langchain_ollama import ChatOllama llm = ChatOllama(model="qwen2.5:7b", temperature=0) tools = [read_test_case, write_script, validate_syntax] llm_with_tools = llm.bind_tools(tools) def executor_node(state: DeepAgentState): step = state["plan"][state["current_step"]] prompt = f"当前步骤:{step}\n请调用合适的工具完成这一步。" response = llm_with_tools.invoke( state["messages"] + [{"role": "user", "content": prompt}] ) return { "messages": [response], "current_step": state["current_step"] + 1, }temperature=0是为了让工具调用更稳定,规划类任务不需要创造性。bind_tools会把工具的 JSON schema 注入到请求里,模型据此决定调用哪个。
4.5 运行与观测
跑起来之后,观测是必须的。LangGraph 支持流式输出每个节点的状态变化:
config = {"configurable": {"thread_id": "task-001"}} for event in graph.stream( {"messages": [{"role": "user", "content": "读取 test_login.yaml 并生成 Playwright 脚本"}]}, config=config, ): for node_name, node_output in event.items(): print(f"[{node_name}] {node_output}")thread_id是会话标识,同一个 thread 的状态会被 checkpointer 保存,支持中断后恢复。这个机制在做"人工审核"场景时特别有用——审核不通过就中断,人工改完再graph.update_state继续跑。
5. 常见问题与排查技巧实录
5.1 工具调用失败与重试
最常见的报错就是工具调用失败,表现五花八门:模型传了不存在的参数、参数类型不对、工具内部抛异常。我的处理原则是分层拦截:
- 第一层:工具 schema 里用 Pydantic 严格约束参数类型,模型传错直接在校验层拦下。
- 第二层:工具内部对业务规则做校验,比如文件路径合法性。
- 第三层:中间件捕获异常,记录日志并决定是否重试。
重试要区分错误类型。网络超时可以重试,参数错误重试没用,得把错误信息回传给模型让它重新生成。下面是一个简单的重试中间件:
def with_retry(node_fn, max_retry=3): def wrapped(state): for attempt in range(max_retry): try: return node_fn(state) except TransientError as e: if attempt == max_retry - 1: raise time.sleep(2 ** attempt) return wrapped指数退避是标配,别用固定间隔,否则遇到服务端限流会雪上加霜。
5.2 上下文超限的处理
长任务跑到后面,messages 会越来越长,最终顶到模型窗口上限。我试过几种方案,效果排序如下:
| 方案 | 效果 | 适用场景 |
|---|---|---|
| 滑动窗口保留最近 N 轮 | 简单有效,但会丢早期信息 | 短任务 |
| 摘要压缩历史消息 | 保留语义,成本略高 | 中长任务 |
| 状态分区,只回传必要字段 | 最彻底,需要前期设计 | 长任务 |
| 向量检索按需召回 | 灵活,实现复杂 | 超长任务 |
我一般组合使用:状态分区打底,历史消息超过阈值时触发摘要压缩。摘要节点本身也是一次模型调用,要算进成本。
5.3 常见问题速查表
| 现象 | 可能原因 | 排查方向 |
|---|---|---|
| 模型不调用工具,直接回答 | 工具描述不清 / 模型不支持 function calling | 检查 docstring,换支持工具调用的模型 |
| 计划步骤粒度混乱 | 提示词约束不足 | 强制结构化输出,加示例 |
| 图跑到一半卡住 | 条件边没有覆盖所有分支 | 检查should_continue返回值是否都有对应边 |
| 状态丢失 | checkpointer 未配置或 thread_id 变化 | 确认 checkpointer 和 thread_id 一致 |
| 工具参数类型错误 | schema 定义不严 | 用 Pydantic 约束参数 |
| 成本异常高 | 上下文过长 / 重复调用 | 看 token 统计,检查是否有死循环 |
5.4 几个我踩过的坑
坑一:把规划做得太细。一开始我让模型把任务拆到十几步,结果每步都要一次模型调用,成本和延迟都爆炸。后来控制在 3 到 8 步,效果反而更好。规划是给执行指方向的,不是写详细设计文档。
坑二:忽略 checkpointer 的持久化。用MemorySaver开发时一切正常,上线后进程一重启状态全丢。生产环境一定要换成基于数据库的 checkpointer。
坑三:中间件里做重活。有次我在日志中间件里同步写数据库,结果拖慢了整个图。中间件要保持轻量,重活异步做。
坑四:工具之间共享状态。两个工具都读写同一个全局变量,并发时数据错乱。工具要尽量无状态,需要共享的状态走图的状态对象。
6. 关于 DeepAgent 能力边界的一点个人看法
经常有人问 DeepAgent 现在的能力和某些头部产品比差距在哪。我的实际体会是:框架本身不决定上限,模型能力和任务设计才决定上限。DeepAgent 提供的是一套编排骨架,它能让你的 Agent 更稳、更可观测、更容易维护,但它不会让一个 7B 模型突然拥有 70B 模型的推理能力。
所以我的建议是,先把编排做扎实——规划执行分离、状态分区、中间件分层、记忆分层,这四件事做到位,你的 Agent 在复杂任务上的表现会有肉眼可见的提升。至于模型选型,本地能跑就用本地,跑不动再考虑更大的模型,别一上来就追求最强模型而忽略了工程结构。
这套东西我前后迭代了大概两个月,从最开始一个文件几百行的意大利面,到现在图结构清晰、中间件可插拔、状态可恢复,最大的感受就是:Agent 开发的难点从来不在"让模型调用工具",而在"让整个系统在长任务里保持可控"。DeepAgent 和 LangGraph 解决的正是后面这件事。如果你也在做类似的项目,建议从状态设计开始认真对待,这一步偷的懒,后面都会加倍还回来。