1. 项目缘起从LangChain Agent到DeepAgents的实践跨越最近在折腾AI应用开发尤其是想搞点能自主处理复杂任务的智能体Agent。相信不少朋友和我一样最开始都是从LangChain入门的。LangChain的Agent框架确实经典它把工具调用、记忆、推理链这些概念带到了大众视野让我们能相对容易地搭出一个能“思考”和“行动”的AI程序。但用久了尤其是在处理一些需要多步骤协作、状态管理或者长周期任务的场景时总会感觉有点“力不从心”。比如一个需要先查资料、再分析、最后生成报告的任务用传统的LangChain Agent来编排状态流转和错误处理写起来就挺啰嗦的。就在这时DeepAgents进入了我的视线。它并不是一个完全独立于LangChain的新框架你可以把它理解为在LangChain特别是LangGraph思想基础上一个更聚焦于生产级、复杂、可协作智能体的实现方案和最佳实践集合。如果说LangChain提供的是乐高积木那么DeepAgents更像是一套已经设计好的、针对特定建筑类型比如智能体协作系统的预制构件和搭建手册。它特别强调“Skills”技能的模块化封装和智能体之间的协同工作流。看到网上很多人在讨论LangGraph和LangChain的区别或者寻找更强大的Agent实战方案我觉得是时候把我基于DeepAgents进行Skills应用实践的这些踩坑和收获系统地整理出来了。这次我们不只停留在“Hello World”而是深入一个具备多个技能、能协同工作的智能体系统内部看看怎么把它真正用起来。2. DeepAgents核心概念拆解不只是另一个Agent框架在开始动手之前有必要先厘清几个关键概念。很多人容易把DeepAgents和LangChain、LangGraph混为一谈或者觉得它只是一个高级封装。其实不然它的设计哲学有自己独特的侧重点。2.1 DeepAgents与LangChain/LangGraph的定位关系首先明确一点DeepAgents通常构建在LangChain生态之上尤其是利用LangGraph来定义和运行智能体工作流。你可以认为LangChain是基础工具库LangGraph是负责编排工作流有向图的引擎而DeepAgents则是在此之上定义了一套如何构建复杂、可复用、可协作智能体系统的架构模式和实现库。它们的关系有点像Spring Boot、Spring Framework和Java EE。LangChain提供了各种组件LLM调用、工具、记忆LangGraph提供了编排这些组件的“流程图”能力而DeepAgents则约定了一套更上层的“应用架构”比如如何组织代码、如何定义技能、智能体之间如何通信和共享状态。它解决的是“用LangGraph搭建大型应用时项目结构容易混乱”的问题。因此当你搜索“langgraph和langchain的区别”时可以这样理解LangGraph是LangChain中用于构建有状态、多步骤应用特别是Agent的一个子库而DeepAgents是应用了LangGraph的一种更高级的架构实践。2.2 Skills智能体的“超能力”模块“Skills”是DeepAgents里最核心的概念之一也是本文实践的重点。什么是Skill你可以把它理解为一个智能体所具备的、可独立执行且功能明确的“超能力”单元。它比LangChain里的“Tool”工具概念更丰富、更内聚。一个LangChain Tool通常只做一件事比如“搜索网络”、“执行计算”。而一个Skill可以封装一个完整的子任务流程内部可能包含多个工具的调用、条件判断甚至内部状态。例如一个“数据获取与清洗”Skill内部可能依次调用“网络搜索API”、“数据解析器”、“格式校验工具”等多个步骤。在DeepAgents的范式里我们会把这样的复杂能力打包成一个Skill对外提供统一的接口。这样做的好处是高内聚、低耦合每个Skill负责一个明确的业务领域内部逻辑变更不影响其他部分。可复用性一个写好的“邮件总结”Skill可以被不同的智能体客服机器人、个人助理调用。易于测试和维护可以针对单个Skill进行单元测试和调试。网上热词“superpower skills”很形象地表达了这一点。我们的目标就是为智能体装配上各种强大的、模块化的超能力技能。2.3 智能体协作与工作流DeepAgents的另一个强项是智能体Agent间的协作。在一个系统中你可以有多个各司其职的智能体比如一个负责信息检索的“研究员”Agent一个负责内容创作的“作家”Agent一个负责代码生成的“程序员”Agent。这些Agent通过预定义的工作流进行协作共同完成一个宏大的任务。这比让一个“全能”Agent去处理所有事情要更清晰、更健壮也更容易扩展。LangGraph的工作流定义能力在这里得到了充分发挥。我们可以用图Graph来清晰地描述这些Agent和Skill之间的调用关系、状态流转和条件分支。DeepAgents则提供了更友好的抽象让我们定义这种多智能体协作系统时代码结构更清晰。3. 实战构建一个多技能内容创作智能体理论说了这么多我们直接进入实战。假设我们要构建一个“内容创作助理”智能体。它的任务是给定一个主题能自动完成资料搜集、大纲生成、内容撰写和格式检查。我们将把这个任务拆解成多个Skill并由一个主控Agent来协调。3.1 环境搭建与基础依赖首先确保你的Python环境建议3.9以上并安装核心库。这里不只用langchain还需要langgraph和deepagents或其理念下的实现库。请注意截至我实践时“DeepAgents”可能更多指的是一种架构模式或者某个特定实现。为了演示我会使用LangGraph和LangChain来模拟实现DeepAgents的核心思想因为这是最通用和可控的方式。pip install langchain langgraph langchain-openai我们使用OpenAI的模型作为LLM引擎你也可以替换为Anthropic Claude、DeepSeek等LangChain支持的其他模型。import os from langchain_openai import ChatOpenAI # 设置你的OpenAI API Key os.environ[OPENAI_API_KEY] your-api-key-here # 初始化一个功能较强的LLM比如GPT-4用于复杂的规划和推理 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0.1) # 初始化一个用于一般文本生成的LLM text_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7)注意在实际项目中API密钥应通过环境变量或安全的配置管理系统读取切勿硬编码在代码中。3.2 定义核心Skills模块化超能力我们将创建三个核心SkillResearchSkill研究技能、OutlineSkill大纲技能和WritingSkill写作技能。每个Skill都是一个类继承自一个基础的Skill类这里我们先抽象一个。它需要有一个清晰的description描述智能体何时使用它和一个execute方法执行具体逻辑。3.2.1 ResearchSkill网络信息搜集与摘要这个Skill负责根据主题进行网络搜索这里我们用模拟的搜索函数代替真实API避免依赖并返回关键信息的摘要。from typing import Dict, Any from langchain_core.tools import tool # 模拟一个网络搜索工具 tool def web_search_tool(query: str) - str: 根据查询词进行网络搜索返回摘要信息。 在实际应用中这里应接入Serper API、Google Search API或爬虫。 # 这里是模拟数据 mock_data { 机器学习入门: 机器学习是人工智能的核心分支主要分为监督学习、无监督学习和强化学习。常用库包括Scikit-learn、TensorFlow和PyTorch。, Python异步编程: Python的asyncio库用于编写并发代码核心概念是协程Coroutine、事件循环Event Loop和任务Task。, 健康饮食: 均衡膳食应包含充足的蔬菜水果、优质蛋白质、全谷物和健康脂肪。建议减少加工食品和含糖饮料的摄入。 } return mock_data.get(query, f未找到关于{query}的详细信息。) class ResearchSkill: 研究技能负责搜集和总结给定主题的信息。 property def description(self) - str: return 当需要获取某个主题的最新、最全的客观事实和背景信息时使用此技能。输入应为一个明确的主题或问题。 def execute(self, topic: str) - Dict[str, Any]: print(f[ResearchSkill] 正在研究主题: {topic}) # 1. 调用搜索工具 search_result web_search_tool.invoke(topic) # 2. 使用LLM对搜索结果进行提炼和摘要 from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的研究助理。请将以下原始搜索信息提炼成3-5个最关键、最相关的要点用简洁明了的语言列出。), (user, 主题{topic}\n原始信息{raw_info}) ]) chain prompt | llm summarized_points chain.invoke({topic: topic, raw_info: search_result}) print(f[ResearchSkill] 研究完成生成{len(summarized_points.content)}个要点。) return { topic: topic, raw_search_result: search_result, summarized_points: summarized_points.content }3.2.2 OutlineSkill基于资料生成内容大纲这个Skill接收研究结果并生成一份内容大纲。class OutlineSkill: 大纲技能基于研究资料生成一份结构化的内容大纲。 property def description(self) - str: return 当已经拥有关于某个主题的研究资料并需要规划内容的整体结构和逻辑流程时使用此技能。输入应为研究结果。 def execute(self, research_result: Dict[str, Any]) - Dict[str, Any]: print(f[OutlineSkill] 正在为主题{research_result[topic]}生成大纲。) from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一位经验丰富的编辑。请根据以下关于某个主题的研究要点生成一份详细的内容大纲。大纲应包含引言、主体至少3个主要部分每部分下有2-3个子点和结论。请使用清晰的标题格式。), (user, 主题{topic}\n研究要点{points}) ]) chain prompt | llm outline chain.invoke({ topic: research_result[topic], points: research_result[summarized_points] }) print(f[OutlineSkill] 大纲生成完成。) return { topic: research_result[topic], research_summary: research_result[summarized_points], content_outline: outline.content }3.2.3 WritingSkill根据大纲撰写完整内容这个Skill是最终的生产者它利用大纲和研究摘要撰写完整的文章或报告。class WritingSkill: 写作技能根据内容大纲和研究摘要撰写完整的、高质量的文章。 property def description(self) - str: return 当已经拥有明确的内容大纲和支撑材料需要产出最终的、连贯的文本内容时使用此技能。 def execute(self, outline_result: Dict[str, Any]) - Dict[str, Any]: print(f[WritingSkill] 正在根据大纲撰写完整内容。) from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一位专业的作家。请严格按照提供的内容大纲进行写作并巧妙融入研究要点中的事实信息。文章应流畅、专业、信息量大。字数在800-1000字左右。), (user, 主题{topic}\n大纲{outline}\n参考研究要点{research_points}) ]) chain prompt | text_llm # 使用text_llm温度稍高创造性更强 full_content chain.invoke({ topic: outline_result[topic], outline: outline_result[content_outline], research_points: outline_result[research_summary] }) print(f[WritingSkill] 内容撰写完成。) return { topic: outline_result[topic], final_content: full_content.content }实操心得在定义Skill时description属性至关重要。它不仅是给人看的注释未来如果想让一个“规划Agent”自动选择使用哪个Skill这个描述就是LLM做决策的关键依据。描述要具体、清晰说明技能的适用场景和输入要求。3.3 构建智能体工作流用LangGraph编排Skills现在我们有三个独立的Skill我们需要一个“大脑”来按顺序调用它们。这就是主控Agent我们用LangGraph来定义它的工作流。这个工作流是一个简单的顺序链研究 - 生成大纲 - 写作。from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END # 首先定义整个工作流的状态结构 class ContentCreationState(TypedDict): 内容创作工作流的状态定义。 topic: str # 用户输入的主题 research_result: Annotated[dict, operator.add] # 研究技能的输出 outline_result: Annotated[dict, operator.add] # 大纲技能的输出 final_result: Annotated[dict, operator.add] # 写作技能的输出最终文章 # Annotated 和 operator.add 用于LangGraph的状态合并这里简化处理 # 初始化技能实例 research_skill ResearchSkill() outline_skill OutlineSkill() writing_skill WritingSkill() # 定义各个节点Node函数每个节点对应一个Skill的执行 def research_node(state: ContentCreationState) - ContentCreationState: 执行研究节点。 result research_skill.execute(state[topic]) return {research_result: result} def outline_node(state: ContentCreationState) - ContentCreationState: 执行大纲节点依赖研究结果。 result outline_skill.execute(state[research_result]) return {outline_result: result} def writing_node(state: ContentCreationState) - ContentCreationState: 执行写作节点依赖大纲结果。 result writing_skill.execute(state[outline_result]) return {final_result: result} # 构建工作流图 workflow StateGraph(ContentCreationState) # 添加节点 workflow.add_node(research, research_node) workflow.add_node(outline, outline_node) workflow.add_node(write, writing_node) # 设置边的连接关系research - outline - write - END workflow.add_edge(research, outline) workflow.add_edge(outline, write) workflow.add_edge(write, END) # 设置入口点 workflow.set_entry_point(research) # 编译图得到可执行的应用 app workflow.compile()现在我们的多技能智能体系统就构建完成了。你可以通过以下方式运行它# 初始化输入状态 initial_state ContentCreationState(topic机器学习入门) # 运行工作流 final_state app.invoke(initial_state) # 查看最终结果 print(最终生成的文章) print(final_state[final_result][final_content])这个工作流会依次触发三个Skill并将中间结果传递给下一个节点最终产出文章。4. 进阶实现动态Skill路由与智能体协作上面的例子是一个固定的线性流程。但在更复杂的场景中主控Agent可能需要根据当前情况动态决定使用哪个Skill。这就涉及到Skill的路由Routing问题。同时我们也可以引入多个专门的智能体进行协作。4.1 构建一个动态路由控制器我们创建一个RouterAgent它的职责是根据用户请求和当前上下文从技能库中选择最合适的一个或多个技能来执行。这需要LLM具备一定的规划和工具调用能力。from langchain.agents import create_openai_tools_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder # 首先将我们的Skill包装成LangChain Tool以便被Agent调用 from langchain_core.tools import BaseTool from pydantic import BaseModel, Field class ResearchInput(BaseModel): topic: str Field(description需要研究的主题) class ResearchTool(BaseTool): name research_skill description research_skill.description # 使用Skill自身的描述 args_schema ResearchInput def _run(self, topic: str) - str: result research_skill.execute(topic) # 将结果格式化为字符串方便Agent阅读 return f研究主题{result[topic]}\n研究摘要{result[summarized_points]} # 类似地为OutlineSkill和WritingSkill创建Tool此处省略... # outline_tool OutlineTool(...) # writing_tool WritingTool(...) # 假设我们有一个技能工具列表 skill_tools [ResearchTool(), OutlineTool(), WritingTool()] # OutlineTool和WritingTool需要类似定义 # 创建路由Agent router_prompt ChatPromptTemplate.from_messages([ (system, 你是一个智能任务路由控制器。你的职责是分析用户请求并决定调用哪一个技能工具来完成任务。 可用的技能工具有 {tools} 请仔细分析用户请求。如果请求是开放性的或需要多步骤完成例如“帮我写一篇关于X的文章”你应该先调用research_skill获取信息然后根据需要调用后续技能。 一次只调用一个工具。根据工具的返回结果和用户需求决定下一步动作。), MessagesPlaceholder(variable_namechat_history), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) router_agent create_openai_tools_agent(llm, skill_tools, router_prompt) agent_executor AgentExecutor(agentrouter_agent, toolsskill_tools, verboseTrue, handle_parsing_errorsTrue) # 使用路由Agent处理复杂请求 result agent_executor.invoke({ input: 请帮我撰写一篇关于Python异步编程的科普文章要求通俗易懂。, chat_history: [] }) print(result[output])这个RouterAgent内部集成了LLM的推理能力它会根据你的请求“写一篇科普文章”自动规划步骤先调用research_skill根据返回的研究摘要再决定调用outline_skill和writing_skill。这就实现了动态的技能路由。4.2 多智能体协作系统设计更进一步我们可以将每个Skill升级为一个独立的、拥有自己LLM的Agent。例如ResearcherAgent、OutlineAgent、WriterAgent。它们通过一个协调者AgentOrchestrator来管理。协调者负责分解总任务并将子任务分配给最合适的专家Agent。这种架构的优点是职责更清晰每个Agent可以针对其专业领域进行深度优化和提示词工程。容错性更强一个Agent失败不影响其他Agent。易于扩展新增一个“校对Agent”或“配图Agent”非常容易。用LangGraph实现这种模式非常直观。你可以定义一个State其中包含任务描述、当前阶段、各Agent的产出等。图中包含多个节点每个节点是一个Agent的执行函数。协调逻辑可以写在节点之间的条件边Conditional Edge上根据上一个Agent的输出来决定下一步走向哪个Agent。# 伪代码示意多智能体协作图 graph_builder StateGraph(State) graph_builder.add_node(“orchestrator”, orchestrator_agent_node) # 协调者节点负责任务分解和分配 graph_builder.add_node(“researcher”, researcher_agent_node) graph_builder.add_node(“outliner”, outliner_agent_node) graph_builder.add_node(“writer”, writer_agent_node) # 设置条件路由协调者根据任务类型决定下一个执行者是谁 def route_after_orchestrator(state): if state[“subtask_type”] “research”: return “researcher” elif state[“subtask_type”] “outline”: return “outliner” else: return “writer” graph_builder.add_conditional_edges(“orchestrator”, route_after_orchestrator) graph_builder.add_edge(“researcher”, “orchestrator”) # 研究员完成后交还给协调者评估 graph_builder.add_edge(“outliner”, “orchestrator”) graph_builder.add_edge(“writer”, END) graph_builder.set_entry_point(“orchestrator”) collab_app graph_builder.compile()这种模式非常强大能够处理极其复杂的任务也是DeepAgents理念中“深度”的体现——智能体之间可以进行深度的、有状态的协作。5. 生产环境考量与避坑指南将这样一个基于Skills和LangGraph的智能体系统投入生产会面临许多在Demo中遇不到的问题。下面分享几个关键的实践经验和踩过的坑。5.1 状态管理与持久化在长时间运行或异步任务中工作流的状态State需要持久化。LangGraph的状态默认在内存中进程重启就丢失。对于生产环境你需要集成一个外部状态存储。解决方案使用CheckpointerLangGraph提供了MemorySaver内存和SqliteSaverSQLite数据库等内置检查点实现。对于分布式系统你需要自定义Checkpointer将状态存入Redis、PostgreSQL或云数据库。关键点状态对象必须是可序列化的JSON兼容。避免在State中存储复杂的Python对象如数据库连接、LLM实例只存储ID或配置参数。from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph memory SqliteSaver.from_conn_string(“:memory:”) # 生产环境换成文件或服务器路径 workflow StateGraph(..., checkpointermemory)5.2 错误处理与重试机制网络调用、API限流、模型输出格式错误等都可能导致单个Skill或Agent失败。必须有健壮的错误处理。实践经验Skill内部容错在每个Skill的execute方法内部对可能失败的操作如API调用进行try-catch并返回结构化的错误信息而不是抛出异常导致整个工作流崩溃。工作流层面的重试利用LangGraph的prebuilt工具或自定义节点包装器为关键节点添加重试逻辑。例如使用tenacity库。超时控制为每个Skill或Agent的执行设置超时时间防止因某个环节卡死而阻塞整个系统。import tenacity from tenacity import retry, stop_after_attempt, wait_exponential class RobustResearchSkill(ResearchSkill): retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def execute_with_retry(self, topic: str) - Dict[str, Any]: # 原有的execute逻辑 return self.execute(topic) def execute(self, topic: str) - Dict[str, Any]: try: # ... 原有逻辑 return result except SomeSpecificAPIError as e: return {“error”: f“API调用失败: {e}”, “topic”: topic} except Exception as e: # 记录日志 logger.error(f“ResearchSkill执行异常: {e}”) return {“error”: “技能执行内部错误”, “topic”: topic}5.3 技能的可发现性与注册表当系统中有几十上百个Skill时如何让路由Agent或协调者知道所有可用的Skill你需要一个技能注册表。实现模式创建一个全局的SkillRegistry单例或类。每个Skill在定义时向注册表注册自己提供name,description,input_schema,execute_function。路由Agent或工具调用框架从注册表中动态加载可用的工具列表。class SkillRegistry: _skills {} classmethod def register(cls, name: str, description: str, func: callable): cls._skills[name] {“description”: description, “func”: func} classmethod def get_tool_definitions_for_agent(cls): # 将注册的技能转化为LangChain Tool列表 tools [] for name, info in cls._skills.items(): # ... 动态创建Tool类 tools.append(tool) return tools # 技能定义时自动注册 def skill(name, description): def decorator(func): SkillRegistry.register(name, description, func) return func return decorator skill(name“web_research”, description“搜索网络获取信息”) def web_research_skill(topic: str) - str: # ... 技能实现5.4 监控、日志与可观测性生产系统必须有完善的监控。你需要记录工作流执行轨迹每个节点的开始/结束时间、输入/输出、状态变化。这有助于调试和复现问题。LLM调用成本与延迟记录每次模型调用的token消耗、模型名称、耗时用于成本核算和性能优化。技能执行成功率统计每个Skill的成功/失败次数定位薄弱环节。可以考虑集成像LangSmith这样的LLM应用监控平台它能无缝对接LangChain/LangGraph提供强大的追踪和评估功能。至少你应该在关键节点添加结构化的日志输出。6. 性能优化与扩展思考当你的智能体系统开始处理真实流量时性能会成为瓶颈。以下是一些优化方向LLM调用优化缓存对频繁出现的、结果确定的查询如“什么是机器学习”的LLM响应进行缓存。可以使用langchain.cache如SQLiteCache,RedisCache或简单的functools.lru_cache适用于单进程。批处理如果多个用户请求类似任务可以考虑将请求合并后批量调用LLM API某些提供商如OpenAI的批处理API更经济。模型分级将任务分级。重度的规划、推理用GPT-4简单的文本生成用GPT-3.5 Turbo或更便宜的模型甚至本地模型。工作流异步化LangGraph支持异步节点。将IO密集型的Skill如网络请求、数据库查询定义为async函数并使用异步LLM客户端可以大幅提升系统的并发吞吐量。技能的粒度与组合不要过度设计细粒度Skill。如果一个Skill很少被独立使用或者总是和另一个Skill一起被调用考虑将它们合并。反之如果一个Skill过于复杂内部逻辑太多应考虑拆分成更小的、可复用的子Skill。这需要在“复用性”和“复杂性”之间找到平衡。向更复杂的架构演进当单个协调者成为瓶颈时可以考虑分层协调或去中心化的智能体网络。例如引入“元协调者”来管理多个领域的协调者Agent。探索智能体间的通信协议如通过消息队列RabbitMQ, Kafka传递结构化消息实现完全解耦和水平扩展。基于LangChain和DeepAgents理念构建Skills应用是一个从“玩具”到“工具”再到“系统”的演进过程。起点可以很简单一个线性的工作流加上几个Skill就能解决实际问题。随着需求复杂再逐步引入动态路由、多智能体协作、状态持久化和生产级监控。关键在于始终把握“模块化”和“可组合”这两个核心思想让每个Skill成为一块坚实的积木最终搭建出能应对复杂现实挑战的智能体大厦。
)
