最近在探索AI智能体Agent的落地应用时发现很多开发者对如何构建一个具备复杂技能Skills的智能体系统感到困惑。网上资料要么过于理论化要么代码片段零散难以整合成一个可运行、可扩展的项目。本文将基于一个名为“Agent Skills”的架构项目手把手带你从零开始构建一个具备多技能协同与深度研究DeepResearch能力的智能体系统。无论你是想入门AI Agent开发还是希望将智能体技术应用到实际业务中这篇实战教程都能提供一条清晰的路径和一套完整的代码。1. 背景与核心概念什么是Agent Skills架构在深入代码之前我们有必要厘清几个核心概念。这能帮助我们在后续开发中做出正确的设计决策。1.1 智能体Agent与技能Skills智能体Agent在AI语境下通常指一个能够感知环境、进行决策并执行行动以达成目标的自治软件实体。它不仅仅是调用一次大模型API而是一个具备记忆、规划和工具使用能力的持续运行系统。技能Skills是智能体能力的具象化单元。一个复杂的任务可以被拆解为多个基础技能的组合。例如一个“市场分析Agent”可能由“网络搜索Skill”、“数据提取Skill”、“报告生成Skill”等构成。将功能模块化为Skill极大地提升了系统的可维护性和可扩展性。1.2 Agent Skills 架构的核心思想Agent Skills架构是一种设计模式它强调模块化每个Skill是独立的、功能单一的模块通过清晰的接口与Agent核心交互。可编排Agent核心或称“大脑”负责根据目标动态地选择、组合和调用不同的Skills。状态管理Agent需要维护对话历史、任务上下文、工具调用结果等状态供后续决策使用。工具集成Skills的本质是让Agent能够安全、有效地使用外部工具如搜索引擎、数据库、API。1.3 DeepResearch深度研究能力DeepResearch深度研究是本项目要实现的一个高级技能示例。它模拟了一个研究员的完整工作流理解复杂问题 - 拆解子问题 - 并行搜索与信息收集 - 综合分析与验证 - 产出结构化报告。这远非一次简单的网络搜索而是体现了智能体在复杂任务上的规划、执行与反思能力。1.4 与相关概念的区分Skills vs. MCP (Model Context Protocol)MCP是Anthropic提出的一种协议用于让大模型安全、标准化地使用服务器端工具和资源。你可以把Skills看作是应用层的功能实现而MCP是底层通信协议的一种选择。我们的项目侧重于应用层Skill的构建与编排可以兼容不同的底层协议如OpenAI的Function Calling或MCP。Agent vs. 简单提示工程传统的提示工程是静态的、一次性的交互。Agent是动态的、有状态的它可以根据历史交互调整策略主动规划步骤形成一个闭环。2. 环境准备与版本说明我们将使用Python作为开发语言并依托于LangChain这一流行的AI应用开发框架来构建我们的Agent系统。LangChain提供了丰富的组件来简化Agent、Tools、Memory的开发。核心环境要求操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文命令以macOS/Linux的bash为例Windows用户可在PowerShell或WSL中运行。Python版本 3.9 或 3.10。推荐使用3.10以获得最佳兼容性。python --version # 检查版本包管理工具pip或poetry。本文使用pip。大模型API你需要一个可用的大模型API密钥例如OpenAI GPT-3.5/4Anthropic Claude或国内可用的智谱、月之暗面等平台的API。注意我们将使用OpenAI格式的接口进行演示其他兼容API如OpenRouter、Ollama本地模型只需调整基础URL和API Key即可。项目初始化创建项目目录并进入mkdir agent-skills-tutorial cd agent-skills-tutorial创建虚拟环境强烈推荐python -m venv venv # 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows: # venv\Scripts\activate创建核心依赖文件requirements.txtlangchain0.1.0 langchain-openai0.0.5 # 用于集成OpenAI模型 langchain-community0.0.10 # 包含许多社区工具和集成 python-dotenv1.0.0 # 用于管理环境变量 duckduckgo-search4.1.0 # 用于实现搜索技能 beautifulsoup44.12.0 # 用于网页内容解析 requests2.31.0 # 用于HTTP请求 pydantic2.0.0 # 用于数据验证和设置管理安装依赖pip install -r requirements.txt创建环境变量文件.env用于安全存储API密钥# .env OPENAI_API_KEYyour-openai-api-key-here # 如果你使用其他模型例如Ollama # OLLAMA_BASE_URLhttp://localhost:11434 # OLLAMA_MODELllama3.2重要请将.env添加到.gitignore文件中避免密钥泄露。3. 项目架构设计与核心模块拆解在写代码前我们先设计整个项目的结构。一个清晰的架构是项目可维护性的基石。agent-skills-tutorial/ ├── .env # 环境变量密钥等 ├── .gitignore ├── requirements.txt # Python依赖 ├── main.py # 主程序入口 ├── core/ # 核心Agent逻辑 │ ├── __init__.py │ ├── agent.py # Agent核心类定义 │ └── memory.py # 记忆管理对话历史 ├── skills/ # 技能模块目录 │ ├── __init__.py │ ├── base_skill.py # 技能基类 │ ├── web_search.py # 网络搜索技能 │ ├── calculator.py # 计算器技能 │ └── deep_research.py # 深度研究技能复合技能 ├── tools/ # 底层工具可被技能调用 │ ├── __init__.py │ └── web_scraper.py # 网页抓取工具 └── utils/ # 工具函数 ├── __init__.py └── config.py # 配置加载各模块职责说明core/agent.py定义Agent类。它是系统的大脑负责加载技能、解析用户意图、维护执行状态、并调用相应的技能。core/memory.py管理Agent的对话历史和工作记忆通常使用ConversationBufferMemory。skills/base_skill.py定义所有技能的抽象基类BaseSkill规定每个技能必须实现的接口如execute方法。skills/*.py具体的技能实现。每个技能文件都是一个独立的、可插拔的功能模块。tools/存放更底层的、单一功能的工具例如一个纯粹的网页抓取函数。技能可以组合调用多个工具。utils/config.py统一加载环境配置和模型设置。4. 核心代码实战一步步构建Agent系统接下来我们将从底层到上层逐一实现上述模块。4.1 实现技能基类与配置加载首先定义技能的契约。创建skills/base_skill.py# skills/base_skill.py from abc import ABC, abstractmethod from typing import Any, Dict from pydantic import BaseModel, Field class SkillInput(BaseModel): 技能的输入参数模型 input_text: str Field(description用户输入或上级指令) context: Dict[str, Any] Field(default_factorydict, description执行上下文信息) class BaseSkill(ABC): 所有技能的抽象基类 name: str base_skill description: str 一个基础的技能 def __init__(self, config: Dict[str, Any] None): self.config config or {} abstractmethod async def execute(self, skill_input: SkillInput) - Dict[str, Any]: 执行技能的核心方法。 返回一个字典至少包含 output (技能输出) 和 status (执行状态)。 pass def get_description(self) - str: 获取技能的描述用于Agent决策 return f{self.name}: {self.description}接着创建配置加载工具utils/config.py# utils/config.py import os from dotenv import load_dotenv from typing import Optional load_dotenv() # 加载 .env 文件中的环境变量 class Config: 全局配置类 OPENAI_API_KEY: Optional[str] os.getenv(OPENAI_API_KEY) # 可以在这里添加其他配置如模型名称、温度等 MODEL_NAME: str gpt-3.5-turbo MODEL_TEMPERATURE: float 0.1 # 低温度使输出更确定 classmethod def validate(cls): 验证必要配置是否存在 if not cls.OPENAI_API_KEY: raise ValueError(OPENAI_API_KEY 未在环境变量中设置。请检查 .env 文件。) config Config() config.validate()4.2 实现具体技能Web搜索与计算器让我们实现两个基础技能感受一下技能模块的独立性。技能一网络搜索 (skills/web_search.py)这个技能利用duckduckgo-search库进行实时网络搜索。# skills/web_search.py from duckduckgo_search import DDGS from .base_skill import BaseSkill, SkillInput from typing import Dict, Any import asyncio class WebSearchSkill(BaseSkill): name web_search description 使用DuckDuckGo进行网络搜索获取最新信息。 def __init__(self, config: Dict[str, Any] None): super().__init__(config) self.max_results self.config.get(max_results, 3) async def execute(self, skill_input: SkillInput) - Dict[str, Any]: query skill_input.input_text if not query: return {output: 搜索查询不能为空。, status: error} try: # 注意DDGS() 同步调用我们用线程池包装以避免阻塞事件循环 loop asyncio.get_event_loop() results await loop.run_in_executor( None, self._sync_search, query ) formatted_results [] for r in results[:self.max_results]: formatted_results.append(f标题: {r[title]}\n链接: {r[href]}\n摘要: {r[body]}\n) output f关于 {query} 的搜索结果共{len(formatted_results)}条\n -*50 \n output \n.join(formatted_results) return { output: output, status: success, raw_results: results[:self.max_results] # 保留原始数据供其他技能使用 } except Exception as e: return {output: f搜索过程中发生错误{str(e)}, status: error} def _sync_search(self, query: str): 同步执行搜索的函数 with DDGS() as ddgs: return list(ddgs.text(query, max_resultsself.max_results2))技能二计算器 (skills/calculator.py)这是一个安全的、受限的计算技能使用eval但进行了严格限制。# skills/calculator.py import ast import operator as op from .base_skill import BaseSkill, SkillInput from typing import Dict, Any # 定义允许的运算符 allowed_operators { ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv, ast.Pow: op.pow, ast.USub: op.neg, } class CalculatorSkill(BaseSkill): name calculator description 执行安全的数学表达式计算支持 , -, *, /, **, 括号。 async def execute(self, skill_input: SkillInput) - Dict[str, Any]: expression skill_input.input_text.strip() if not expression: return {output: 计算表达式不能为空。, status: error} try: # 使用抽象语法树进行安全评估 node ast.parse(expression, modeeval).body result self._eval_expr(node) return {output: f{expression} {result}, status: success, result: result} except (SyntaxError, TypeError, KeyError, ZeroDivisionError) as e: return {output: f计算错误或表达式不安全{str(e)}, status: error} def _eval_expr(self, node): 递归安全地评估AST节点 if isinstance(node, ast.Num): # 数字 return node.n elif isinstance(node, ast.BinOp): # 二元操作 left_val self._eval_expr(node.left) right_val self._eval_expr(node.right) operator_func allowed_operators.get(type(node.op)) if operator_func is None: raise TypeError(f不允许的操作符: {type(node.op)}) return operator_func(left_val, right_val) elif isinstance(node, ast.UnaryOp): # 一元操作如负数 operand_val self._eval_expr(node.operand) operator_func allowed_operators.get(type(node.op)) if operator_func is None: raise TypeError(f不允许的操作符: {type(node.op)}) return operator_func(operand_val) else: raise TypeError(f不支持的表达式类型: {type(node)})4.3 实现复合技能DeepResearch深度研究这是本项目的亮点。DeepResearchSkill本身不直接做具体工作而是扮演一个“子项目经理”的角色它利用已有的WebSearchSkill和规划能力完成一个多步骤的研究任务。创建skills/deep_research.py# skills/deep_research.py import asyncio from typing import Dict, Any, List from .base_skill import BaseSkill, SkillInput from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema import SystemMessage, HumanMessage from utils.config import config class DeepResearchSkill(BaseSkill): name deep_research description 对一个复杂主题进行深度研究分解问题、并行搜索、综合分析并生成报告。 def __init__(self, config: Dict[str, Any] None): super().__init__(config) # 初始化一个大语言模型用于规划和分析 self.llm ChatOpenAI( api_keyconfig.OPENAI_API_KEY, modelconfig.MODEL_NAME, temperature0.2, ) # 这里假设我们已经有了一个搜索技能实例实际中可通过依赖注入传入 from .web_search import WebSearchSkill self.search_skill WebSearchSkill(config{max_results: 4}) async def execute(self, skill_input: SkillInput) - Dict[str, Any]: research_topic skill_input.input_text print(f[DeepResearch] 开始研究主题: {research_topic}) # 步骤1规划研究子问题 sub_questions await self._plan_research(research_topic) print(f[DeepResearch] 生成的子问题: {sub_questions}) # 步骤2并行搜索所有子问题 search_tasks [] for q in sub_questions: task self.search_skill.execute(SkillInput(input_textq)) search_tasks.append(task) search_results await asyncio.gather(*search_tasks) print(f[DeepResearch] 并行搜索完成共 {len(search_results)} 组结果。) # 步骤3综合分析与报告生成 final_report await self._synthesize_report(research_topic, sub_questions, search_results) return { output: final_report, status: success, metadata: { topic: research_topic, sub_questions: sub_questions, search_summary: [r.get(status) for r in search_results] } } async def _plan_research(self, topic: str) - List[str]: 使用LLM将宽泛的研究主题分解为3-5个具体的子问题 prompt ChatPromptTemplate.from_messages([ SystemMessage(content你是一个资深研究助理。请将用户给出的宽泛研究主题分解为3到5个具体、可搜索的关键子问题。这些子问题应覆盖主题的不同方面并有助于形成全面的理解。请直接输出问题列表每行一个不要编号以外的额外解释。), HumanMessage(contentf研究主题{topic}) ]) messages prompt.format_messages() response await self.llm.ainvoke(messages) # 解析LLM返回的文本拆分成问题列表 questions_text response.content.strip() questions [q.strip() for q in questions_text.split(\n) if q.strip()] # 清理可能存在的编号前缀如“1. ”“- ” cleaned_questions [] for q in questions: # 移除常见的列表标记 for prefix in [1., 2., 3., 4., 5., - , * ]: if q.startswith(prefix): q q[len(prefix):].strip() break cleaned_questions.append(q) return cleaned_questions[:5] # 最多返回5个问题 async def _synthesize_report(self, topic: str, sub_questions: List[str], search_results: List[Dict]) - str: 综合搜索结果为一份结构化报告 # 准备给LLM的上下文 context_parts [] for i, (q, result) in enumerate(zip(sub_questions, search_results)): if result[status] success: context_parts.append(f## 子问题 {i1}: {q}\n搜索到的相关信息\n{result[output][:1500]}...\n) # 截取部分内容避免token超限 else: context_parts.append(f## 子问题 {i1}: {q}\n[信息获取失败: {result[output]}]\n) research_context \n.join(context_parts) synthesis_prompt ChatPromptTemplate.from_messages([ SystemMessage(content你是一个分析专家。请基于以下关于某个主题的多个子问题的搜索信息撰写一份简洁、清晰、结构化的研究报告。报告应包含1) 概述2) 针对每个子问题的关键发现3) 综合结论与见解。确保信息准确如果某些部分信息不足请注明。), HumanMessage(contentf研究主题{topic}\n\n以下是各子问题的搜索信息\n{research_context}\n\n请生成研究报告) ]) messages synthesis_prompt.format_messages() response await self.llm.ainvoke(messages) return response.content4.4 实现Agent核心与记忆管理现在我们来创建Agent的大脑。创建core/agent.py# core/agent.py from typing import Dict, Any, List, Optional from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema import SystemMessage, HumanMessage, AIMessage from skills.base_skill import BaseSkill, SkillInput from utils.config import config import json class Agent: 智能体核心负责技能管理和任务调度 def __init__(self, skills: List[BaseSkill], model_name: str None): self.skills {skill.name: skill for skill in skills} self.llm ChatOpenAI( api_keyconfig.OPENAI_API_KEY, modelmodel_name or config.MODEL_NAME, temperatureconfig.MODEL_TEMPERATURE, ) # 简单的对话历史记忆 self.conversation_history: List[Dict[str, str]] [] def _build_system_prompt(self) - str: 构建系统提示词描述Agent的能力和规则 skill_descriptions \n.join([skill.get_description() for skill in self.skills.values()]) return f你是一个多功能AI助手可以调用以下技能来帮助用户 {skill_descriptions} 请遵循以下规则 1. 仔细分析用户的请求。 2. 如果请求可以通过直接对话回答请直接回答。 3. 如果需要使用上述技能请严格按照以下JSON格式回复 json {{action: use_skill, skill_name: 技能名称, input: 传递给技能的输入文本}}不要解释你将做什么直接输出JSON。 async def process_query(self, user_input: str) - str: 处理用户输入的主要方法 # 将用户输入加入历史 self.conversation_history.append({role: user, content: user_input})# 准备包含历史的对话上下文 messages [SystemMessage(contentself._build_system_prompt())] for msg in self.conversation_history[-6:]: # 保留最近6轮对话作为上下文 if msg[role] user: messages.append(HumanMessage(contentmsg[content])) else: messages.append(AIMessage(contentmsg[content])) # 调用LLM进行决策 llm_response await self.llm.ainvoke(messages) response_text llm_response.content.strip() # 尝试解析LLM的响应看是否是技能调用指令 skill_result None try: # 尝试从响应中提取JSON代码块 if json in response_text: json_str response_text.split(json)[1].split()[0].strip() elif response_text.startswith({) and response_text.endswith(}): json_str response_text else: json_str None if json_str: action_data json.loads(json_str) if action_data.get(action) use_skill: skill_name action_data[skill_name] skill_input_text action_data[input] if skill_name in self.skills: print(f[Agent] 决定调用技能: {skill_name}, 输入: {skill_input_text}) skill self.skills[skill_name] skill_result await skill.execute(SkillInput(input_textskill_input_text, context{user_query: user_input})) # 将技能执行结果格式化后作为Agent的最终回复 if skill_result[status] success: final_response f【{skill.name} 执行结果】\n{skill_result[output]} else: final_response f【技能执行出错】\n{skill_result[output]} else: final_response f未知的技能名称: {skill_name}。可用技能: {list(self.skills.keys())} else: final_response response_text # LLM直接回复了文本 else: final_response response_text # LLM直接回复了文本 except json.JSONDecodeError: # 如果响应不是JSON则当作普通对话回复 final_response response_text except Exception as e: final_response f处理技能调用时发生错误: {str(e)} # 将Agent的回复加入历史 self.conversation_history.append({role: assistant, content: final_response}) return final_response记忆管理模块 core/memory.py 可以更复杂但为了简化我们暂时使用Agent内部的列表。在实际项目中你可能需要集成 ConversationBufferMemory 或向量数据库。 ### 4.5 主程序入口与运行测试 最后创建 main.py 来组装一切并提供一个简单的交互界面。 python # main.py import asyncio from core.agent import Agent from skills.web_search import WebSearchSkill from skills.calculator import CalculatorSkill from skills.deep_research import DeepResearchSkill from utils.config import config async def main(): print(初始化 Agent Skills 系统...) # 1. 实例化所有技能 skills [ WebSearchSkill(config{max_results: 3}), CalculatorSkill(), DeepResearchSkill(config), # 传入配置 ] # 2. 创建智能体 agent Agent(skillsskills, model_namegpt-3.5-turbo) print(系统初始化完成) print(可用技能:, list(agent.skills.keys())) print(输入 quit 或 exit 退出程序。) print(- * 50) # 3. 交互循环 while True: try: user_input input(\n您: ).strip() if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input: continue print(Agent 正在思考...) response await agent.process_query(user_input) print(f\n助手: {response}) except KeyboardInterrupt: print(\n\n程序被中断。) break except Exception as e: print(f\n发生未预期错误: {e}) if __name__ __main__: asyncio.run(main())运行与测试确保你的.env文件已正确配置OPENAI_API_KEY。在终端运行python main.py尝试以下交互“计算一下 (15 27) * 3 等于多少”- 应触发计算器技能。“搜索一下今天 OpenAI 有什么新闻。”- 应触发网络搜索技能。“请深度研究一下‘人工智能在医疗诊断中的最新应用’。”- 应触发深度研究技能。你会看到它先规划子问题然后并行搜索最后生成报告。5. 常见问题与排查思路在构建和运行此类Agent系统时你可能会遇到以下典型问题问题现象可能原因解决思路导入错误ModuleNotFoundError1. 未安装依赖。2. 虚拟环境未激活。3. PYTHONPATH 不正确。1. 运行pip install -r requirements.txt。2. 确认终端提示符前有(venv)。3. 在项目根目录下运行脚本。API密钥错误或超时1..env文件未创建或路径不对。2. API Key 无效或余额不足。3. 网络连接问题。1. 确认.env文件在项目根目录且内容正确。2. 登录对应平台检查API Key状态和余额。3. 检查网络或尝试使用curl测试API端点。Agent不调用技能总是直接回复1. 系统提示词_build_system_prompt不够清晰。2. 模型温度temperature过高导致输出随机。3. LLM的响应格式解析失败。1. 强化提示词明确要求输出JSON。2. 将temperature调低如0.1。3. 在agent.py的process_query方法中添加更健壮的JSON解析和日志打印出LLM的原始响应进行调试。深度研究技能运行非常慢1. 并行搜索的异步任务管理不当。2. 网络搜索本身耗时。3. 生成的报告过长LLM生成慢。1. 确保正确使用asyncio.gather。2. 为搜索设置超时 (asyncio.wait_for)。3. 限制搜索结果的长度和最终报告的token数。技能执行出错如搜索失败1. 第三方库如duckduckgo-search的API变化或网络封锁。2. 技能内部代码逻辑错误。1. 在技能代码中加入更详细的异常捕获和日志。2. 考虑使用备用搜索源如SerpAPI、Google Search API。3. 对用户输入进行预处理和清洗。对话历史混乱或丢失1.conversation_history列表管理不当。2. 长时间运行后内存累积。1. 实现一个真正的记忆管理类如使用ConversationBufferWindowMemory限制轮数。2. 定期清理或持久化历史记录。6. 最佳实践与工程建议将原型转化为可投入生产环境的系统需要考虑更多工程化细节。6.1 技能设计规范单一职责每个技能只做一件事并做好。避免创建“万能技能”。明确接口输入输出使用Pydantic模型严格定义便于验证和文档化。错误处理技能内部必须进行细致的异常处理并返回统一的错误格式避免整个Agent崩溃。可配置化像max_results这样的参数应从外部传入提高灵活性。6.2 Agent核心优化技能路由优化当前的基于LLM的“思考-决策”循环成本较高且可能不稳定。对于简单、明确的意图可以优先使用意图识别分类器例如微调一个小模型或使用规则来路由降低成本和提高速度。流式输出对于耗时的技能如DeepResearch实现流式输出Streaming让用户能看到“思考中...”、“正在搜索...”、“正在分析...”等中间状态提升体验。持久化记忆集成向量数据库如Chroma, Pinecone来存储和检索长期记忆使Agent能记住跨会话的关键信息。6.3 系统架构扩展技能市场与动态加载设计一个技能注册中心支持从配置文件或数据库动态加载技能无需重启服务。技能编排与工作流引入工作流引擎如Prefect、Airflow来管理复杂技能的执行顺序和条件分支实现更强大的自动化流程。监控与可观测性为每个技能调用和Agent决策记录日志、指标如耗时、成功率便于监控和调试。使用像LangSmith这样的工具可以极大提升开发效率。6.4 安全与合规输入净化对所有用户输入和技能输入进行严格的验证和净化防止注入攻击。工具权限控制为技能分级如“读取网络信息”、“执行计算”、“写入数据库”并根据用户或会话上下文进行权限校验。内容审核在技能输出最终给用户前可增加一层内容安全审核过滤不当信息。数据隐私如果技能处理用户个人数据需确保符合数据隐私法规避免在日志或记忆中泄露敏感信息。通过本教程你不仅实现了一个具备基础技能和深度研究能力的AI智能体更重要的是掌握了一套构建可扩展、模块化Agent系统的设计方法论和实战代码。从技能抽象、Agent决策循环到复合技能编排这些模式是开发更复杂AI应用的基础。你可以在此基础上继续集成更多技能如数据库查询、发送邮件、生成图表优化决策逻辑或将其封装为API服务从而打造出真正解决实际业务问题的智能助手。