这是我的第468篇原创文章。一、引言LangChain 的 Deep Agents 框架引入了一个基于 LangGraph 构建的智能体框架agent harness赋予大语言模型显式规划、持久文件系统记忆、子代理派生和极端上下文工程的能力。ReAct 智能体浅层智能体的局限上下文窗口溢出每次工具调用都会将其结果添加到对话历史中。经过二三十次工具调用后上下文窗口中会塞满数千个 token 的中间结果。LLM 开始无法区分哪些信息是关键的、哪些是噪音。早期的重要指令被推离了模型的注意力窗口。幻觉漂移随着上下文的增长模型对原始目标的把握逐渐松动。它开始在无关的工具结果之间幻觉出不存在的关联混淆来自不同步骤的信息并逐渐偏离你实际要求它做的事情。智能体不会崩溃——它只是悄悄地开始解决错误的问题。缺乏持久记忆ReAct 智能体的唯一记忆就是它的对话历史。没有办法保存中间发现、组织研究笔记或在步骤之间维护结构化状态。一切都存在于一条线性的消息流中模型在每次迭代中都必须重新阅读全部内容。单体上下文污染当 ReAct 智能体需要执行一个专门的子任务时该子任务的整个上下文工具调用、观察结果、推理过程会混入主对话中。研究子任务会污染后续写作子任务的上下文。没有隔离也没有关注点分离。Deep Agents 是 LangChain 对上述局限的回应。deepagents 包于 2025 年 9 月首次发布目前在 PyPI 上已更新至0.4.10版本提供了一个基于 LangGraph 构建的智能体框架。该框架为智能体配备了内置能力使其适合处理长周期、复杂的任务。ReAct 智能体并未过时。对于少于十次工具调用的简单、定义明确的任务ReAct 智能体更快、更易于调试且消耗更少的 token。当任务能舒适地容纳在单个上下文窗口内时规划、文件系统管理和子代理协调的开销是不必要的。关键在于将智能体架构与任务复杂度相匹配。维度ReActAgent 1.0Deep AgentAgent 2.0设置复杂度极低中等最佳适用场景短期、聚焦型任务长周期、多步骤项目上下文管理仅依赖对话历史文件系统 子代理简单任务 Token 效率更高较低规划开销复杂任务 Token 效率较低上下文溢出更高上下文卸载调试难度直接需要跨代理追踪目标持久性随时间退化通过规划工具维持二、实现过程接下来通过一个具体的应用实践来说明deep agents的使用搭一个城市智慧大脑里面有三个专家行政文秘专家secretary_expert专门写公文、报告政策咨询专家policy_expert专门查最新的政策带搜索工具行政审批专家approval_expert专门处理审批流程有自己的固定工作流主智能体自动帮你做任务调度它会看每个子智能体的描述自动判断用户的问题该分给谁甚至复杂的问题它会自动协调多个子智能体依次处理。2.1 第一步先定义基础工具和状态定义网络搜索工具给政策专家用用来查最新的政策tool def internet_search(query: str) - str: 使用 DuckDuckGo 进行网络搜索查询最新的政策、新闻等信息 print(f\n[工具调用] internet_search(query{query})) try: results [] with DDGS() as ddgs: for r in ddgs.text(query, max_results5): results.append(f- [{r[title]}]({r[href]}): {r[body][:200]}) if not results: return 未找到相关结果。 return \n.join(results) except Exception as e: return f搜索失败: {str(e)}定义行政审批专家的状态结构class ExpertState(TypedDict): messages: Annotated[list[BaseMessage], lambda x, y: x y] expert_output: NotRequired[str]2.2 第二步定义三个专家子智能体子智能体 1: 行政文秘专家 (使用字典式SubAgent最简单的那种方式)secretary_expert { name: 行政文秘专家, description: 精通公文写作和文档处理负责各类行政文书、方案报告和公函的撰写, system_prompt: ( 你是一位行政文秘专家。请在回答开头加上[子智能体: 行政文秘专家]。\n 你具备强大的文字写作和文档编辑能力可以\n 1. 撰写各类行政公文、通知、报告\n 2. 制定工作方案、计划总结\n 3. 编写会议纪要、调研报告\n 专注于用专业的文笔完成各类文书撰写任务。 ), }子智能体 2: 政策咨询专家 (使用CompiledSubAgent带搜索工具的自定义Agent) def create_policy_advisor_subagent(model): from langchain.agents import create_agent # 我们自己创建一个带搜索工具的Agent policy_graph create_agent( model, tools[internet_search], system_prompt( 你是一位城市政策咨询专家。\n 你必须为每个政策查询调用 internet_search 工具搜索相关政策文件。\n 绝不能在没有搜索的情况下编造政策内容。\n 当用户询问政策法规、补贴申请、办事流程等问题时\n 1. 立即使用 internet_search 工具搜索最新政策\n 2. 基于搜索结果提供准确的政策信息和办事指南\n 3. 回答开头必须加上[子智能体: 政策咨询专家]\n 4. 如果搜索结果不确定要明确告知用户以官方最新发布为准 ) ) # 把我们自定义的Agent包装成CompiledSubAgent主智能体就能调度它了 return CompiledSubAgent( name政策咨询专家, description负责城市政策法规咨询、补贴申请指引和办事流程指导, runnablepolicy_graph )子智能体 3: 行政审批专家 (使用CompiledSubAgent自定义LangGraph工作流)def create_approval_subagent(model): # 我们自己定义一个审批的工作流接收申请 - 审核材料 - 审批完成 def receive_node(state: ExpertState): 第一个节点接收用户的申请 query state[messages][-1].content print(f\n[子智能体: 行政审批专家] 受理窗口 - 接收申请...) return {expert_output: f申请事项: {query}} def review_node(state: ExpertState): 第二个节点审核用户的材料 output state.get(expert_output, ) print(f[子智能体: 行政审批专家] 审核窗口 - 审查材料...) return {expert_output: output → 材料审核中} def approve_node(state: ExpertState): 第三个节点完成审批返回结果 output state.get(expert_output, ) print(f[子智能体: 行政审批专家] 审批窗口 - 办理审批...) return {messages: [AIMessage(contentf[子智能体: 行政审批专家] 办事流程指导{output} → 审批完成。请携带身份证到市民中心窗口办理。)]} # 把节点串成LangGraph的图 builder StateGraph(ExpertState) builder.add_node(receive, receive_node) builder.add_node(review, review_node) builder.add_node(approve, approve_node) builder.add_edge(START, receive) builder.add_edge(receive, review) builder.add_edge(review, approve) builder.add_edge(approve, END) # 把我们自定义的工作流图包装成CompiledSubAgent return CompiledSubAgent( name行政审批专家, description负责行政审批事项咨询、SOP流程指导和办事指南服务, runnablebuilder.compile() )2.3 第三步搭主智能体把专家都加进去把三个专家加到主智能体里写几个测试用例看看它是不是自动把任务分给对应的专家def run_city_brain_tests(): # 初始化大模型 model ChatOpenAI( modeldeepseek-chat, api_keysk-xxxxxxxxxxxxxxxx, base_urlhttps://api.deepseek.com/v1, ) # 初始化两个自定义的子智能体 policy_expert create_policy_advisor_subagent(model) approval_expert create_approval_subagent(model) # 创建主智能体城市智慧大脑把三个子智能体都传进去 main_agent create_deep_agent( modelmodel, name城市智慧大脑, system_prompt( 你是城市智慧大脑智能体的核心调度中枢。\n 根据用户问题类型将任务分配给最合适的专家子智能体\n - 行政文秘专家处理公文写作、方案报告、文档撰写等文书相关问题\n - 政策咨询专家处理政策法规、补贴申请、办事流程相关问题\n - 行政审批专家处理行政审批事项、SOP流程和办事指南相关问题 ), subagents[secretary_expert, policy_expert, approval_expert] ) # 测试用例覆盖不同的场景 test_cases [ (帮我写一份关于城市环境整治的工作报告, 应路由到 行政文秘专家), (个人创业有什么补贴政策, 应路由到 政策咨询专家), (我想注册一家公司需要准备什么材料, 应路由到 行政审批专家), (你好今天天气不错, 闲聊 - 无需路由到子代理), (我需要开办一家餐馆帮我写一份申请报告再查下有什么补贴政策最后告诉我办理流程, 需要多子代理协调文秘(报告) - 政策(补贴) - 审批(流程)), ] # 跑测试看结果 active_subagents {} for i, (query, expected) in enumerate(test_cases, 1): print(f\n{*70}) print(f测试用例 {i}: {expected}) print(f用户问题: {query}) print(f{*70}) # 流式调用主智能体它会自动调度子智能体 for chunk in main_agent.stream( {messages: [HumanMessage(contentquery)]}, stream_mode[updates, messages], subgraphsTrue, versionv2 ): # 这里我们可以监控子智能体的生命周期 is_subagent any(s.startswith(tools:) for s in chunk[ns]) if chunk[type] updates: for node_name, data in chunk[data].items(): if not chunk[ns] and node_name model_request: for msg in data.get(messages, []): for tc in getattr(msg, tool_calls, []): if tc.get(name) task: sub_id tc.get(id, ) sub_type tc.get(args, {}).get(subagent_type, unknown) active_subagents[sub_id] { type: sub_type, description: tc.get(args, {}).get(description, )[:50], status: pending } print(f\n[lifecycle] PENDING → 子智能体 {sub_type})结果整个流程主智能体自动帮你搞定用户提问主智能体判断该分给哪个专家然后调用子智能体子智能体自己处理中间过程全在自己的上下文里处理完了把结果返回给主智能体主智能体再把结果整合了返回给用户。作者简介读研期间发表6篇SCI数据挖掘相关论文现在某研究院从事数据算法相关科研工作结合自身科研实践经历不定期分享关于Python、机器学习、深度学习、人工智能系列基础知识与应用案例。致力于只做原创以最简单的方式理解和学习关注我一起交流成长。需要数据集和源码的小伙伴可以关注底部公众号添加作者微信。