1. 从“工具调用”到“技能装配”DeepAgents的范式升级如果你之前跟着这个系列一路走来应该已经对LangChain DeepAgents的Agent有了基本概念一个能思考、能规划、能调用工具Tools来完成任务的智能体。这就像给一个聪明的助手配备了一套瑞士军刀每把刀工具都有明确的功能比如“搜索网络”、“执行代码”、“读写文件”。助手Agent根据你的指令决定何时、按什么顺序使用哪把刀。但“DeepAgents使用Agent Skill”这个标题指向的是一个更高级、更贴近真实工作流的范式。它不再仅仅是“调用工具”而是“装配技能”。这其中的区别我打个比方工具是“螺丝刀”技能则是“组装一台电脑”。后者是一个更复杂的、目标导向的、可能包含多个步骤和决策点的过程。一个“技能”Skill内部可以封装多个工具调用、条件判断、甚至嵌套调用其他技能。它代表了一个完整的、可复用的任务单元。为什么这个升级如此重要在我实际构建复杂Agent系统的经验中单纯罗列工具会遇到几个天花板任务逻辑碎片化、上下文管理困难、错误处理冗余。比如让Agent“分析某公司最新财报并总结其风险”这需要1搜索获取财报PDF2解析PDF文本3调用大模型进行摘要和风险分析4格式化输出。如果只用基础工具Agent的每一步决策都可能出错且整个流程的连贯性很难保证。而“技能”可以将这四步打包内部定义好执行逻辑和异常处理对外只暴露一个干净的接口“执行财报分析技能”。这极大地提升了Agent的可靠性、可维护性和执行复杂任务的能力。接下来我们就深入LangChain DeepAgents的“技能”体系看看如何从“刀匠”进化成“机械师”。2. Skill的核心架构不止于封装理解DeepAgents中的Skill不能停留在“把几个工具包起来”的层面。它的设计包含了一套完整的、用于构建可靠智能工作流的元框架。我们可以从三个层面来剖析它的核心架构。2.1 技能描述与注册让Agent理解“能做什么”一个Skill首先需要被清晰地描述以便Agent的“大脑”通常是LLM能够理解它的用途、输入和输出。这通过Skill基类和相关的装饰器或注册机制来实现。from langchain_deepagents.skills import Skill, skill # 方式一使用装饰器简洁直观 skill( namefinancial_report_analyzer, description分析指定上市公司的最近一期财报PDF总结其核心财务数据和潜在风险。, args_schemaFinancialReportInput, # 一个Pydantic模型定义输入参数 ) async def analyze_financial_report(company_code: str, year: int) - str: 技能的具体实现函数。 :param company_code: 公司股票代码 :param year: 财报年份 :return: 分析报告字符串 # 1. 根据code和year调用搜索工具获取财报PDF URL # 2. 调用PDF解析工具提取文本 # 3. 调用LLM分析文本提取关键指标和风险点 # 4. 格式化并返回结果 analysis_result await _internal_analysis_pipeline(company_code, year) return analysis_result # 方式二继承Skill基类更灵活适合复杂技能 class AdvancedDataFetcherSkill(Skill): name: str advanced_data_fetcher description: str 从多个备用数据源获取信息确保数据可用性。 class InputArgs(BaseModel): query: str priority_source: Literal[api_a, database_b, cache] api_a async def run(self, args: InputArgs) - dict: # 可以实现更复杂的逻辑如重试机制、源切换 for source in [args.priority_source, database_b, cache]: try: data await self._fetch_from_source(source, args.query) return {status: success, source: source, data: data} except Exception as e: self.logger.warning(fSource {source} failed: {e}) continue return {status: error, message: All sources failed}关键点args_schema或Skill类中的InputArgs至关重要。它严格定义了技能的输入契约LLM在规划时会据此生成正确的参数。清晰的description则帮助LLM在众多技能中做出准确选择。2.2 技能的执行流与控制内部的状态与决策技能内部的执行不再是线性的。一个成熟的技能需要处理条件分支、循环、错误重试。DeepAgents的Skill执行环境通常提供了这些控制流原语。假设我们构建一个“竞品调研技能”它需要从不同渠道收集信息并去重skill(namecompetitive_analysis, description对给定产品进行多维度竞品调研。) async def competitive_analysis(product_name: str, dimensions: List[str]): results {} for dimension in dimensions: # 循环控制 # 条件判断根据维度选择不同的信息获取子技能 if dimension pricing: info await get_pricing_analysis(product_name) elif dimension feature: info await get_feature_comparison(product_name) elif dimension user_review: info await aggregate_user_sentiment(product_name) else: info Dimension not supported. # 错误处理某个维度获取失败不影响其他维度 if isinstance(info, Exception): results[dimension] fFailed to fetch: {info} continue # 继续下一个循环 # 去重与合并逻辑技能内部的数据处理 results[dimension] await deduplicate_and_summarize(info) # 最终整合所有维度的结果 final_report await generate_comprehensive_report(results) return final_report在这个例子中技能内部包含了for循环、if-elif-else条件分支、try-except错误处理以及数据清洗逻辑。这些控制流使得技能成为一个智能的、健壮的工作单元而不是简单的工具序列。2.3 技能的组合与嵌套构建技能图谱最强大的地方在于技能可以调用其他技能。这允许你构建层次化的、模块化的技能体系即“技能图谱”。skill(namemarket_entry_feasibility_study, description综合评估新产品进入市场的可行性。) async def market_feasibility_study(product_concept: str, target_region: str): # 嵌套调用其他技能 market_size await competitive_analysis(product_concept, [pricing, feature]) regulatory_check await check_regulatory_compliance(target_region, product_concept) cost_estimation await estimate_operational_costs(target_region) # 基于子技能的结果进行高级推理 feasibility_score await calculate_feasibility_score(market_size, regulatory_check, cost_estimation) return { feasibility_score: feasibility_score, details: { market_analysis: market_size, regulatory_status: regulatory_check, cost_breakdown: cost_estimation } }这里market_feasibility_study这个高级技能通过调用competitive_analysis、check_regulatory_compliance、estimate_operational_costs三个子技能完成了更复杂的综合分析。这种组合能力让你能够像搭积木一样用简单的技能构建出解决极其复杂问题的超级智能体。注意技能嵌套虽然强大但需注意循环调用和权限问题。确保技能图谱是无环的并为不同层级的技能设置适当的执行权限和资源访问限制。3. 实战构建一个“智能内容创作”技能链理论说得再多不如动手实践。我们以“为一个科技博客创作一篇关于AI代理的短文”为例构建一个从主题分析到初稿生成的技能链。这个任务单靠一个提示词很难做好但拆分成技能后质量会大幅提升。3.1 技能分解与设计我们将整个过程分解为四个技能Topic Brainstormer主题头脑风暴根据核心关键词生成文章的几个潜在角度和标题。Outline Generator大纲生成器选定一个角度后生成详细的文章大纲。Section Writer段落写手根据大纲中的某一节标题和要点撰写该节的具体内容。Editor Polisher编辑润色器对完成的初稿进行语言润色、逻辑检查和格式统一。3.2 技能实现代码详解首先定义技能所需的输入输出模型使用Pydantic。from pydantic import BaseModel, Field from typing import List, Optional from langchain_deepagents.skills import skill from langchain_deepagents.agents import AgentExecutor from langchain_deepagents.tools import Tool # 假设我们有一个配置好的LLM from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4, temperature0.7) # 1. Topic Brainstormer Skill class BrainstormInput(BaseModel): core_keyword: str Field(description文章的核心关键词) num_ideas: int Field(default3, description需要生成的主题想法数量) skill(nametopic_brainstormer, description为给定的关键词生成多个文章写作角度和标题。, args_schemaBrainstormInput) async def brainstorm_topics(core_keyword: str, num_ideas: int 3) - List[dict]: prompt f 你是一位资深的科技博客编辑。请针对关键词“{core_keyword}”构思 {num_ideas} 个不同的文章写作角度。 每个角度需要包含 1. 一个吸引人的文章标题。 2. 一句话阐述的核心观点。 3. 目标读者是谁。 请以JSON列表格式返回每个元素是一个包含title, core_idea, target_audience键的字典。 # 这里简单演示实际应通过Agent或直接调用LLM response await llm.ainvoke(prompt) # 解析response.content为JSON import json try: ideas json.loads(response.content) return ideas[:num_ideas] # 确保返回数量正确 except json.JSONDecodeError: # 优雅降级如果LLM没返回JSON尝试提取 return [{title: fIdea about {core_keyword}, core_idea: Parsing failed, target_audience: General}] # 2. Outline Generator Skill class OutlineInput(BaseModel): chosen_topic: dict Field(description从头脑风暴中选择的主题字典) depth: str Field(defaultstandard, description大纲深度brief仅H2 standardH2H3 detailed到H4) skill(nameoutline_generator, description根据选定的文章主题生成详细的结构化大纲。, args_schemaOutlineInput) async def generate_outline(chosen_topic: dict, depth: str) - dict: title chosen_topic.get(title, Unknown Title) prompt f 基于文章标题《{title}》和核心观点“{chosen_topic.get(core_idea)}”生成一份文章大纲。 大纲深度要求{depth}。 请以JSON格式返回结构示例 {{ title: ..., introduction: [要点1, 要点2], sections: [ {{heading: H2标题1, subpoints: [H3要点1.1, H3要点1.2]}}, {{heading: H2标题2, subpoints: [...]}} ], conclusion: [总结要点1, 总结要点2] }} response await llm.ainvoke(prompt) # ... JSON解析逻辑 return parsed_outline # 3. Section Writer Skill class SectionInput(BaseModel): section_heading: str Field(description当前要撰写的章节标题) key_points: List[str] Field(description该章节需要涵盖的要点列表) writing_style: str Field(defaultprofessional, description写作风格如 professional, conversational, technical) word_limit: Optional[int] Field(default300, description字数限制) skill(namesection_writer, description根据章节标题和要点撰写完整的章节内容。, args_schemaSectionInput) async def write_section(section_heading: str, key_points: List[str], writing_style: str, word_limit: int 300) - str: points_str \n.join([f- {p} for p in key_points]) prompt f 请以“{writing_style}”的写作风格撰写文章章节“{section_heading}”的内容。 本章节必须涵盖以下要点 {points_str} 要求逻辑清晰论述有力字数控制在{word_limit}字左右。 直接输出章节正文无需再次说明标题。 response await llm.ainvoke(prompt) return response.content # 4. Editor Polisher Skill class EditorInput(BaseModel): full_draft: str Field(description需要润色的完整文章草稿) focus_areas: List[str] Field(default_factorylambda: [grammar, coherence], description重点润色领域如 grammar, clarity, conciseness, tone) skill(nameeditor_polisher, description对文章草稿进行编辑和润色提升语言质量和逻辑连贯性。, args_schemaEditorInput) async def edit_and_polish(full_draft: str, focus_areas: List[str]) - str: focus_str , .join(focus_areas) prompt f 你是一位专业的文本编辑。请对以下文章草稿进行润色重点关注{focus_str}。 要求 1. 修正语法和拼写错误。 2. 优化句子结构使其更流畅易读。 3. 确保段落间过渡自然逻辑连贯。 4. 保持原文核心内容和风格不变。 请直接输出润色后的完整文章。 ---文章草稿开始--- {full_draft} ---文章草稿结束--- response await llm.ainvoke(prompt) return response.content3.3 组装与执行让Agent驾驭技能链有了这些技能我们需要一个主Agent来协调它们。这个Agent本身可以看作一个“项目管理”技能。from langchain_deepagents.agents import create_react_agent from langchain_deepagents.tools import SkillTool # 将技能包装成Agent可以使用的Tool brainstorm_tool SkillTool(skillbrainstorm_topics, namebrainstorm_topics, descriptionbrainstorm_topics.description) outline_tool SkillTool(skillgenerate_outline, namegenerate_outline, descriptiongenerate_outline.description) section_writer_tool SkillTool(skillwrite_section, namewrite_section, descriptionwrite_section.description) editor_tool SkillTool(skilledit_and_polish, nameedit_and_polish, descriptionedit_and_polish.description) # 创建主Agent它可以使用上述所有技能工具 agent_executor create_react_agent( llmllm, tools[brainstorm_tool, outline_tool, section_writer_tool, editor_tool], verboseTrue ) # 定义任务 task 请创作一篇关于“AI Agent在自动化工作流中的应用”的科技博客短文。 请先进行头脑风暴给出3个不同的写作角度供我选择。 # 执行任务 async def run_content_creation(): result await agent_executor.ainvoke({input: task}) print(result[output]) # 在实际运行中Agent会自主决定调用流程 # 1. 调用 brainstorm_topics获得3个主题。 # 2. 可能需要与用户交互在模拟中我们可以预设选择第一个或者自行根据某种规则选择。 # 3. 调用 generate_outline为选定的主题生成大纲。 # 4. 遍历大纲的 sections依次调用 write_section 撰写每个章节。 # 5. 将所有章节组合成初稿后调用 edit_and_polish 进行润色。 # 6. 输出最终文章。实操心得在这个流程中最大的挑战不是技能的实现而是主Agent的提示工程。你需要精心设计给主Agent的系统提示System Prompt明确指导它如何按步骤使用这些技能并在适当的时候做出决策例如如何从头脑风暴的结果中选择一个主题。一个简单的提示可能不够可能需要让Agent在关键步骤上输出中间结果以供“人工”确认或者在技能链中设计“检查点”技能来评估当前产出质量决定是继续、重做还是调整方向。4. 高级技巧技能的管理、评估与调试当技能体系变得庞大时管理、评估和调试就成为关键。这部分是很多教程不会涉及的“脏活累活”但决定了你的Agent系统能否在生产环境稳定运行。4.1 技能的版本管理与依赖像管理代码库一样管理你的技能。为每个技能定义清晰的版本号并记录其输入输出模式Schema的变更。# skill_registry.py skill_registry { financial_report_analyzer: { skill: analyze_financial_report, version: 1.2.0, dependencies: [pdf_parser_tool^2.0, financial_llm_tool^1.5], changelog: { 1.2.0: Added support for quarterly reports., 1.1.0: Improved error handling for network timeouts., 1.0.0: Initial release. } }, # ... 其他技能 }为什么需要版本管理当某个底层工具如PDF解析器升级导致API变化时依赖它的所有技能都可能受影响。通过版本和依赖声明你可以快速进行影响性分析并决定是升级技能还是回滚工具。4.2 技能的性能监控与评估为关键技能添加监控指标这对于优化和排错至关重要。import time import asyncio from functools import wraps from prometheus_client import Counter, Histogram SKILL_EXECUTION_COUNT Counter(skill_execution_total, Total skill executions, [skill_name, status]) SKILL_EXECUTION_DURATION Histogram(skill_execution_duration_seconds, Skill execution duration, [skill_name]) def monitor_skill(skill_func): wraps(skill_func) async def wrapper(*args, **kwargs): skill_name skill_func.__name__ start_time time.time() status success try: result await skill_func(*args, **kwargs) return result except Exception as e: status ferror_{type(e).__name__} raise e finally: duration time.time() - start_time SKILL_EXECUTION_COUNT.labels(skill_nameskill_name, statusstatus).inc() SKILL_EXECUTION_DURATION.labels(skill_nameskill_name).observe(duration) return wrapper # 使用装饰器包装技能 skill(...) monitor_skill async def my_critical_skill(...): ...通过监控面板你可以一目了然地看到哪个技能被调用最频繁平均耗时是多少失败率如何哪些错误最常见这些数据是优化技能逻辑如增加缓存、重试或进行资源扩容的直接依据。4.3 技能的调试与“可观测性”当Agent执行出错时定位问题可能很困难。是技能本身的bug是LLM错误地调用了技能还是输入数据有问题你需要为技能执行增加详细的日志和追踪。import logging from contextvars import ContextVar request_id_var ContextVar(request_id, defaultunknown) class TracedSkill(Skill): async def run(self, args): request_id request_id_var.get() self.logger.info(f[{request_id}] Starting skill {self.name} with args: {args}) # 记录中间状态 intermediate_states [] try: # 假设技能有多个步骤 step1_result await self._step1(args) intermediate_states.append({step: 1, result: str(step1_result)[:100]}) # 记录摘要 self.logger.debug(f[{request_id}] Step1 completed: {step1_result}) step2_result await self._step2(step1_result) intermediate_states.append({step: 2, result: str(step2_result)[:100]}) final_result await self._finalize(step2_result) self.logger.info(f[{request_id}] Skill {self.name} completed successfully.) return final_result except Exception as e: self.logger.error(f[{request_id}] Skill {self.name} failed at intermediate states {intermediate_states}. Error: {e}, exc_infoTrue) # 可以将错误和中间状态上报到错误追踪系统如Sentry raise调试技巧除了日志还可以为技能开发一个“沙盒测试界面”。这个界面允许你手动输入参数直接运行某个技能并查看其每一步的详细输出和内部状态。这对于在复杂技能链中隔离和复现问题非常有效。本质上你是为技能创建了单元测试和集成测试的环境。5. 避坑指南从技能设计到生产部署的常见陷阱基于我过去在多个项目中构建Agent技能系统的经验以下是一些最容易踩坑的地方及其解决方案。5.1 技能粒度过细或过粗问题把每个简单的API调用都包装成一个技能导致技能数量爆炸Agent规划负担加重决策缓慢。反之将一个包含数十个步骤的复杂流程做成一个技能内部逻辑混乱难以调试和复用。解决方案遵循“单一职责”和“高内聚”原则。一个理想的技能应该对应一个明确的、有价值的用户意图或任务阶段。例如“验证用户邮箱”是一个好技能单一职责“处理用户注册流程”则可能包含“验证邮箱”、“创建用户记录”、“发送欢迎邮件”等多个步骤更适合拆分成多个技能或作为一个高级协调技能。5.2 输入输出Schema设计不当问题Schema定义过于宽松如args: dict导致LLM容易传错参数或过于复杂嵌套LLM难以正确生成。解决方案使用强类型的Pydantic模型为每个字段提供清晰的description和示例examples。保持Schema扁平化尽量避免多层嵌套。如果参数确实复杂考虑将其拆分为多个技能调用。为枚举类型提供明确选项。例如writing_style: Literal[“professional“, ”conversational“, ”technical“]比writing_style: str好得多。5.3 技能间的状态污染与依赖问题技能A修改了某个全局状态如一个缓存字典技能B在不知情的情况下依赖了这个状态导致非预期的行为或难以追踪的bug。解决方案原则技能应尽可能无状态Stateless。所有需要的信息都通过输入参数传递。如果必须共享状态如缓存、数据库连接池应通过明确的、受管理的上下文Context对象来传递而不是全局变量。在技能文档中清晰说明其依赖的外部状态和可能产生的副作用。5.4 错误处理与技能韧性不足问题技能内部遇到网络超时、API限流、数据格式异常时直接崩溃导致整个Agent任务失败。解决方案在技能内部实现分层的错误处理策略。重试对于暂时的网络故障HTTP 5xx错误、超时使用指数退避策略进行自动重试。降级当主数据源失败时尝试从备用源获取数据。例如新闻抓取技能在主新闻API失败时可以降级到使用搜索引擎的摘要。优雅失败如果无法完成核心任务应返回一个结构化的错误信息而不是抛出异常。例如{“status“: ”partial_success“, ”data“: {...}, ”errors“: [”Failed to fetch stock price from source A“]}让上游Agent或技能决定如何处理部分成功的结果。5.5 忽视技能的安全性与权限问题一个用于“读取用户数据”的技能可能被恶意提示诱导去执行“删除用户数据”的操作如果底层工具权限控制不严。解决方案最小权限原则为每个技能配置执行所需的最小权限。在技能执行前进行权限校验。输入验证与净化对所有来自不可信源如用户输入、LLM生成的参数进行严格的验证和净化防止注入攻击。敏感操作确认对于删除、修改、支付等高风险操作技能内部应设计二次确认机制或者仅允许在特定的、受信任的Agent工作流中被调用。掌握DeepAgents的Agent Skill意味着你从编写简单的工具调用脚本进阶到了设计和编排智能工作流系统的阶段。这要求你不仅是一个程序员更要成为一个系统架构师思考如何将复杂任务分解为可管理、可复用、可观测的智能模块。这个过程充满挑战但当你看到一个个技能像齿轮一样精密咬合驱动Agent完成那些曾经觉得不可思议的复杂任务时所有的努力都是值得的。