news 2026/11/3 5:36:51

DeepAgent 深度解析:基于 LangGraph 的 Agent 编排与中间件实践

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
DeepAgent 深度解析:基于 LangGraph 的 Agent 编排与中间件实践

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 提供的是针对复杂任务的装配工艺。

层级代表组件职责类比
LangChainChatModel、Tool、Retriever、Prompt提供模型调用、工具封装、检索等原子能力汽车零件
LangGraphStateGraph、Node、Edge、Checkpointer把原子能力编排成有状态、可循环、可中断的图装配流水线
DeepAgent中间件、规划节点、执行节点、记忆层针对深度任务定义编排模式与扩展点装配工艺标准

这个分层带来的直接好处是:你换模型、换工具、换检索库,都不需要动编排逻辑;你改编排逻辑,也不需要重写工具。这就是"中间件"这个词在这里的意义——它把横切关注点(日志、重试、权限、人工审核、记忆写入)从业务节点里抽出来,做成可插拔的一层。

2.3 中间件机制为什么是核心

中间件(middleware)在 DeepAgent 里不是可有可无的装饰,而是整个架构的骨架。它的工作方式很像 Web 框架里的请求中间件:一个请求(这里是一次 Agent 步骤)从进入节点到离开节点,会依次穿过若干中间件,每个中间件可以在前置、后置、异常三个时机介入。

我实际用下来,中间件主要解决四类问题:

  1. 可观测性:记录每一步的输入输出、耗时、token 消耗,方便排查和成本核算。
  2. 可靠性:工具调用失败时自动重试、降级、熔断,避免整个图因为一次网络抖动崩掉。
  3. 安全与合规:在工具真正执行前做参数校验、权限判断,拦截危险操作。
  4. 人机协同:在关键节点插入人工确认,确认通过才继续往下走。

这四类需求几乎每个生产级 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 解决的正是后面这件事。如果你也在做类似的项目,建议从状态设计开始认真对待,这一步偷的懒,后面都会加倍还回来。

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

模型中立:用baseUrl和OpenAI兼容协议解耦大模型与业务代码

1. 为什么你的业务代码不该跟某一家大模型“领证”我见过太多项目,第一版跑通的时候欢天喜地,三个月后想换个模型,结果发现整个业务层里散落着几十处openai.ChatCompletion.create,连错误处理都是照着某一家 SDK 的异常类型写的。…

作者头像 李华
网站建设 2026/11/3 5:36:29

C# 解析 DXF 并计算 CAD 图纸面积:组码、凸度与块参照实践

先说结论:C# 读取 CAD 图纸这件事,难点主要不在语言,而在格式。DXF 是明文的组码-值对,DWG 是压缩过的二进制私有结构,两者的解析成本差着数量级。我这些年做过不少和图纸打交道的活——批量统计厂房构件面积、从图纸里…

作者头像 李华
网站建设 2026/11/3 5:36:28

大模型集成架构:实现模型中立的业务代码设计

1. 为什么你的业务代码不该跟任何一家大模型“领证”我见过太多项目,第一版跑通的时候欢天喜地,三个月后想换个模型,发现代码里到处都是openai.ChatCompletion.create的调用,API Key 硬编码在配置文件里,连 baseUrl 都…

作者头像 李华
网站建设 2026/11/3 5:36:28

OPC模型:技术创意商业化的解码系统

1. 从创意到商业化的关键跃迁上周参加完Mixlab Launchpad闭门交流会,有个问题一直萦绕在我脑海:为什么90%的创意项目最终都死在了从"我想做"到"有人付钱"的转化路上?这个问题困扰了我三年,直到自己真正跑通从…

作者头像 李华
网站建设 2026/11/3 5:35:11

ERP实施是管理改造工程而非技术项目

简介:本资源是一份面向ERP实施顾问、企业信息化负责人及用友生态从业者的专业培训课件,系统讲解用友ERP项目实施的方法论体系与全流程管理要点。内容覆盖系统实施本质认知、标准实施路线图(售前咨询→项目准备→项目建设→项目交付→运行支持…

作者头像 李华
网站建设 2026/11/3 5:34:27

汽车零部件行业ERP选型指南与实施策略

1. 汽车零部件行业ERP选型核心痛点汽车零部件制造是典型的离散型生产模式,具有多品种小批量、工艺路线复杂、质量追溯严格等行业特性。我在为三家 Tier1 供应商实施ERP系统的过程中发现,这个行业对ERP的特殊需求主要集中在四个方面:追溯性要求…

作者头像 李华