在AI技术浪潮席卷全球的今天智能体Agent正从概念走向落地成为连接大模型与具体业务场景的关键桥梁。然而许多开发者在尝试构建自己的Agent时常常陷入“理论懂落地难”的困境如何设计一个稳定、可扩展的架构如何让Agent具备执行复杂任务的能力Skills如何整合外部工具与数据本文将围绕“Agent Skills架构”这一核心主题为你带来一份从零到一的完整实战教程。我们将手把手带你搭建一个具备基础技能如网络搜索、文件读写的智能体项目并深入剖析其背后的架构设计思想让你不仅能跑通代码更能理解其运作机理为构建更复杂的AI应用打下坚实基础。1. 智能体Agent与技能Skills核心概念解析在深入代码之前我们必须厘清几个核心概念这是理解后续所有架构和实现的基础。1.1 什么是智能体Agent在AI语境下智能体Agent并非一个全新的概念。简单来说它是一个能够感知环境、进行决策并执行动作以实现特定目标的软件实体。在大模型时代Agent通常指一个以大语言模型LLM为“大脑”的系统它能够理解用户意图、规划任务步骤、调用工具技能并处理结果。一个典型的AI Agent工作流程可以概括为感知Perception - 规划Planning - 执行Action - 观察Observation的循环直到任务完成。例如当用户请求“帮我查一下今天北京的天气然后总结成一份简报”Agent需要理解这个复杂请求规划出“搜索天气”和“总结简报”两个子任务依次调用对应的工具执行最后将整合的结果返回给用户。1.2 什么是技能Skills技能Skills是Agent能力的具象化体现是Agent能够执行的具体操作单元。你可以将其理解为Agent的“手脚”或“工具箱”里的工具。一个Skill通常封装了一个特定的功能例如网络搜索Skill调用搜索引擎API获取实时信息。计算器Skill执行数学运算。文件读写Skill读取本地文件内容或写入数据。数据库查询Skill连接数据库并执行SQL。Skills与大型语言模型本身的知识库形成互补。LLM擅长理解和生成语言但缺乏获取实时信息、执行精确计算或操作外部系统的能力。Skills正是为了弥补这一短板而存在。1.3 Agent Skills 架构的核心价值那么为什么要专门设计一个“Agent Skills架构”其核心价值在于解耦、复用与扩展。解耦将Agent的核心决策逻辑大脑与具体的执行能力手脚分离。大脑LLM只负责“想”手脚Skills负责“做”。这使得两者可以独立演进和优化。复用设计良好的Skill可以被不同的Agent重复使用。一个写好的“天气查询Skill”既可以用于个人助手Agent也可以用于旅行规划Agent。扩展当需要为Agent增加新能力时无需改动核心架构只需开发并注册一个新的Skill即可。这种插件化的方式极大地提升了系统的可扩展性。1.4 与相关概念的区分MCP, Plugin等在社区中你可能会遇到类似的概念如MCPModel Context Protocol、Plugin插件等。这里做一个简要区分Skills本文焦点通常指在Agent框架内部定义的、可被Agent直接调用的功能模块。它们与Agent框架耦合较紧密实现相对轻量。MCP是一个由Anthropic提出的开放协议旨在为大模型提供标准化的方式来连接数据源、工具和应用程序。它更侧重于定义一套通用的通信标准使得任何兼容MCP的服务都能被支持该协议的模型或客户端使用。你可以把MCP Server看作是一种标准化、协议化的Skill提供方。Plugin概念更泛化在某些框架如LangChain中Plugin可能等同于Tool或Skill在ChatGPT等产品中特指通过官方平台审核上架的三方扩展。对于初学者和自定义项目从实现自有的Skills架构入手是理解其原理的最佳途径。2. 环境准备与项目初始化我们将使用Python作为开发语言因为它拥有最丰富的AI生态库。本教程将构建一个控制台应用的Agent重点在于架构理解。2.1 基础环境要求操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)Python版本3.8 - 3.11推荐3.9或3.10兼容性最好包管理工具pip (建议版本21.0)代码编辑器VS Code, PyCharm 或任何你熟悉的IDE。2.2 创建项目与虚拟环境首先为项目创建一个独立的目录和Python虚拟环境这是管理依赖的最佳实践。# 1. 创建项目目录并进入 mkdir agent-skills-tutorial cd agent-skills-tutorial # 2. 创建虚拟环境 (以venv为例) python -m venv venv # 3. 激活虚拟环境 # Windows (PowerShell) .\venv\Scripts\Activate.ps1 # Windows (CMD) .\venv\Scripts\activate.bat # macOS / Linux source venv/bin/activate # 激活后命令行提示符前应显示 (venv)2.3 安装核心依赖我们将使用openai库作为与大模型交互的客户端并使用requests进行网络请求。pydantic用于数据验证和设置管理python-dotenv用于管理环境变量如API密钥。# 在激活的虚拟环境中执行 pip install openai requests pydantic python-dotenv为了后续示例我们还需要一个能进行简单网页搜索的工具。这里我们使用duckduckgo-search作为免费示例。请注意对于生产环境建议使用更稳定、功能更强的搜索引擎API如Serper、Google Custom Search等。pip install duckduckgo-search2.4 项目结构设计一个清晰的项目结构是良好架构的开始。我们的项目结构如下agent-skills-tutorial/ ├── .env # 存储敏感信息如API密钥切勿提交到Git ├── .gitignore # Git忽略文件 ├── requirements.txt # 项目依赖清单 ├── main.py # 应用主入口 ├── core/ # 核心架构模块 │ ├── __init__.py │ ├── agent.py # Agent核心类 │ ├── skill.py # Skill基类与注册器 │ └── models.py # 数据模型消息、结果等 ├── skills/ # 技能实现目录 │ ├── __init__.py │ ├── web_search.py # 网络搜索技能 │ ├── calculator.py # 计算器技能 │ └── file_io.py # 文件读写技能 └── utils/ # 工具函数 ├── __init__.py └── config.py # 配置加载你可以使用以下命令快速创建这个结构# 在项目根目录下执行 mkdir core skills utils touch .env .gitignore requirements.txt main.py touch core/__init__.py core/agent.py core/skill.py core/models.py touch skills/__init__.py skills/web_search.py skills/calculator.py skills/file_io.py touch utils/__init__.py utils/config.py在.gitignore文件中至少添加以下内容venv/ .env __pycache__/ *.pyc3. 核心架构设计与模块拆解现在我们来逐一实现这个架构的核心模块。3.1 数据模型定义 (core/models.py)我们首先定义在整个系统中流转的数据结构使用pydantic可以方便地进行数据验证和序列化。# core/models.py from pydantic import BaseModel from typing import Dict, Any, Optional, List from enum import Enum class MessageRole(str, Enum): 消息角色枚举 USER user ASSISTANT assistant SYSTEM system class Message(BaseModel): 对话消息 role: MessageRole content: str class SkillInput(BaseModel): Skill执行的输入参数 args: Dict[str, Any] # 参数字典 class SkillResult(BaseModel): Skill执行的结果 success: bool output: str # 执行输出的文本 error: Optional[str] None # 如果失败错误信息 data: Optional[Dict[str, Any]] None # 额外的结构化数据 class AgentResponse(BaseModel): Agent对用户的最终响应 message: str # 回复内容 used_skills: List[str] [] # 本次响应使用了哪些技能 raw_data: Optional[Dict[str, Any]] None # 原始数据如搜索结果的JSON3.2 Skill基类与注册机制 (core/skill.py)这是架构中最关键的部分之一。我们定义一个所有Skill都必须继承的基类并实现一个简单的注册中心来管理它们。# core/skill.py from abc import ABC, abstractmethod from typing import Dict, Any from .models import SkillInput, SkillResult class BaseSkill(ABC): Skill抽象基类 def __init__(self, name: str, description: str): self.name name # Skill的唯一标识如 “web_search” self.description description # 对LLM的描述用于让LLM知道何时调用此Skill abstractmethod def execute(self, input_data: SkillInput) - SkillResult: 执行技能的核心方法子类必须实现 pass def get_schema(self) - Dict[str, Any]: 返回Skill的调用模式用于告知LLM如何调用 # 这是一个简化示例更复杂的实现可以动态生成JSON Schema return { name: self.name, description: self.description, parameters: { type: object, properties: self._get_parameter_schema(), required: self._get_required_parameters() } } def _get_parameter_schema(self) - Dict[str, Any]: 子类可重写此方法来定义参数结构 return {} def _get_required_parameters(self) - List[str]: 子类可重写此方法来定义必填参数 return [] class SkillRegistry: Skill注册中心单例模式 _instance None _skills: Dict[str, BaseSkill] {} def __new__(cls): if cls._instance is None: cls._instance super(SkillRegistry, cls).__new__(cls) return cls._instance def register(self, skill: BaseSkill): 注册一个Skill if skill.name in self._skills: raise ValueError(fSkill with name {skill.name} is already registered.) self._skills[skill.name] skill print(f[SkillRegistry] Registered skill: {skill.name}) def get_skill(self, name: str) - BaseSkill: 根据名称获取Skill skill self._skills.get(name) if not skill: raise KeyError(fSkill {name} not found in registry.) return skill def list_skills(self) - Dict[str, str]: 列出所有已注册Skill的名称和描述 return {name: skill.description for name, skill in self._skills.items()} def clear(self): 清空注册表主要用于测试 self._skills.clear() # 全局注册中心实例 registry SkillRegistry()3.3 Agent核心类 (core/agent.py)Agent类负责协调整个工作流程接收用户输入与LLM交互决定是否调用及调用哪个Skill执行Skill并将结果整合返回。# core/agent.py import json import openai from typing import List, Optional from .models import Message, MessageRole, AgentResponse, SkillInput, SkillResult from .skill import registry from utils.config import settings class Agent: 智能体核心类 def __init__(self, system_prompt: Optional[str] None): 初始化Agent。 :param system_prompt: 系统提示词用于设定Agent的角色和行为。 self.client openai.OpenAI(api_keysettings.OPENAI_API_KEY, base_urlsettings.OPENAI_BASE_URL) self.system_prompt system_prompt or self._default_system_prompt() self.conversation_history: List[Message] [] if self.system_prompt: self.conversation_history.append(Message(roleMessageRole.SYSTEM, contentself.system_prompt)) def _default_system_prompt(self) - str: 默认系统提示词描述了Agent的能力和行为规范 skills_desc \n.join([f- {name}: {desc} for name, desc in registry.list_skills().items()]) return f你是一个有帮助的AI助手可以调用工具Skills来帮助用户解决问题。 你可以使用的工具Skills有 {skills_desc} 当用户的问题需要用到上述工具时请严格按照以下JSON格式响应且只输出这个JSON {{ thought: 你的思考过程分析是否需要以及使用哪个工具, skill_to_use: 要使用的工具名称如果不需要则为null, skill_input: {{}} // 工具的输入参数如果不需要工具则为空对象 }} 如果不需要使用工具请直接以自然语言回复用户。 def chat(self, user_input: str) - AgentResponse: 主要聊天接口。 :param user_input: 用户输入 :return: Agent的响应 # 1. 将用户输入加入历史 self.conversation_history.append(Message(roleMessageRole.USER, contentuser_input)) # 2. 调用LLM获取响应 llm_response self._call_llm() # 3. 解析LLM响应判断是否需要调用Skill used_skills [] final_response_text raw_data None try: # 尝试解析JSON如果解析成功说明LLM决定调用Skill action_data json.loads(llm_response) skill_name action_data.get(skill_to_use) if skill_name and skill_name ! null: # 4. 调用Skill skill_input SkillInput(argsaction_data.get(skill_input, {})) skill_result self._execute_skill(skill_name, skill_input) used_skills.append(skill_name) raw_data skill_result.data # 5. 将Skill执行结果作为新的上下文再次调用LLM进行总结或回答 self.conversation_history.append( Message(roleMessageRole.ASSISTANT, contentf[调用工具 {skill_name}] 结果: {skill_result.output}) ) # 获取LLM基于工具结果的最终回复 final_llm_response self._call_llm() final_response_text final_llm_response else: # LLM直接回复无需调用工具 final_response_text action_data.get(thought, llm_response) # 如果JSON里有thought用它 except json.JSONDecodeError: # LLM的响应不是JSON说明它直接进行了自然语言回复 final_response_text llm_response # 6. 将Assistant的最终回复加入历史 self.conversation_history.append(Message(roleMessageRole.ASSISTANT, contentfinal_response_text)) # 7. 返回结构化响应 return AgentResponse( messagefinal_response_text, used_skillsused_skills, raw_dataraw_data ) def _call_llm(self) - str: 调用OpenAI API或其他兼容API try: response self.client.chat.completions.create( modelsettings.LLM_MODEL, # 例如 gpt-3.5-turbo messages[msg.dict() for msg in self.conversation_history], temperature0.1, # 低温度使输出更确定更适合工具调用 max_tokens500 ) return response.choices[0].message.content.strip() except Exception as e: return f调用语言模型时出错: {str(e)} def _execute_skill(self, skill_name: str, skill_input: SkillInput) - SkillResult: 执行指定的Skill try: skill registry.get_skill(skill_name) return skill.execute(skill_input) except Exception as e: return SkillResult(successFalse, output, errorf执行技能 {skill_name} 时出错: {str(e)}) def clear_history(self): 清空对话历史保留系统提示 self.conversation_history [msg for msg in self.conversation_history if msg.role MessageRole.SYSTEM]3.4 配置管理 (utils/config.py)使用环境变量来管理配置避免将API密钥等敏感信息硬编码在代码中。# utils/config.py import os from pydantic_settings import BaseSettings from pydantic import Field class Settings(BaseSettings): 应用配置 OPENAI_API_KEY: str Field(default, descriptionOpenAI API Key) OPENAI_BASE_URL: str Field(defaulthttps://api.openai.com/v1, descriptionOpenAI API Base URL) LLM_MODEL: str Field(defaultgpt-3.5-turbo, description使用的LLM模型) class Config: env_file .env # 从 .env 文件加载配置 env_file_encoding utf-8 # 创建全局配置实例 settings Settings()在项目根目录创建.env文件并填入你的配置请勿提交此文件到版本控制# .env OPENAI_API_KEYsk-your-openai-api-key-here # 如果你使用其他兼容OpenAI API的服务可以修改以下两项 # OPENAI_BASE_URLhttps://api.xxx.com/v1 # LLM_MODELgpt-3.5-turbo4. 实战实现并注册具体Skills架构搭好了现在我们来为Agent添砖加瓦实现几个实用的Skill。4.1 网络搜索Skill (skills/web_search.py)这个Skill使用duckduckgo_search来获取网络信息。# skills/web_search.py from duckduckgo_search import DDGS from core.skill import BaseSkill from core.models import SkillInput, SkillResult from typing import Dict, Any, List class WebSearchSkill(BaseSkill): 网络搜索技能 def __init__(self): super().__init__( nameweb_search, description使用搜索引擎在互联网上搜索信息。当用户询问实时信息、新闻、最新事件或未知知识时使用。 ) def execute(self, input_data: SkillInput) - SkillResult: query input_data.args.get(query) if not query: return SkillResult(successFalse, output, error搜索技能需要 query 参数。) try: with DDGS() as ddgs: # 获取最相关的5条结果 results list(ddgs.text(query, max_results5)) if not results: return SkillResult(successTrue, outputf未找到关于 {query} 的搜索结果。, data{results: []}) # 格式化输出 formatted_results [] output_lines [f关于 {query} 的搜索结果] for i, r in enumerate(results, 1): output_lines.append(f{i}. {r[title]}) output_lines.append(f 链接: {r[href]}) output_lines.append(f 摘要: {r[body][:150]}...) output_lines.append() formatted_results.append({title: r[title], href: r[href], snippet: r[body]}) output_text \n.join(output_lines) return SkillResult( successTrue, outputoutput_text, data{query: query, results: formatted_results} ) except Exception as e: return SkillResult(successFalse, output, errorf搜索过程中发生错误: {str(e)}) def _get_parameter_schema(self) - Dict[str, Any]: return { query: { type: string, description: 要搜索的关键词或问题 } } def _get_required_parameters(self) - List[str]: return [query]4.2 计算器Skill (skills/calculator.py)这个Skill使用Python的eval函数进行数学计算。注意在生产环境中使用eval是危险的这里仅作演示。安全做法是使用ast.literal_eval或专门的数学表达式解析库。# skills/calculator.py import ast import operator as op from core.skill import BaseSkill from core.models import SkillInput, SkillResult # 安全地支持的操作符 (简化版生产环境建议使用更安全的库如 numexpr) _supported_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, } def _safe_eval(node): 安全地评估一个AST数字表达式节点 if isinstance(node, ast.Num): # Python 3.8 return node.n elif isinstance(node, ast.Constant): # Python 3.8 return node.value elif isinstance(node, ast.BinOp): left _safe_eval(node.left) right _safe_eval(node.right) operator_func _supported_operators.get(type(node.op)) if operator_func is None: raise TypeError(f不支持的运算符: {node.op}) return operator_func(left, right) elif isinstance(node, ast.UnaryOp): operand _safe_eval(node.operand) operator_func _supported_operators.get(type(node.op)) if operator_func is None: raise TypeError(f不支持的运算符: {node.op}) return operator_func(operand) else: raise TypeError(f不支持的表达式类型: {node}) class CalculatorSkill(BaseSkill): 计算器技能安全版 def __init__(self): super().__init__( namecalculator, description执行数学计算。当用户需要计算数学表达式时使用例如 23*4 或 sin(30)。注意本技能仅支持基本算术。 ) def execute(self, input_data: SkillInput) - SkillResult: expression input_data.args.get(expression) if not expression: return SkillResult(successFalse, output, error计算器技能需要 expression 参数。) # 清理表达式移除可能的安全风险字符非常基础的过滤 expression expression.strip().replace( , ) # 禁止导入、函数调用等 for forbidden in [import, exec, eval, __, open, file, sys, os]: if forbidden in expression.lower(): return SkillResult(successFalse, output, error表达式包含不安全内容。) try: # 使用ast解析确保是字面量表达式 tree ast.parse(expression, modeeval) result _safe_eval(tree.body) return SkillResult(successTrue, outputf{expression} {result}, data{expression: expression, result: result}) except SyntaxError: return SkillResult(successFalse, output, error数学表达式语法错误。) except TypeError as e: return SkillResult(successFalse, output, errorf计算错误: {str(e)}) except Exception as e: return SkillResult(successFalse, output, errorf计算过程中发生未知错误: {str(e)}) def _get_parameter_schema(self) - Dict[str, Any]: return { expression: { type: string, description: 要计算的数学表达式例如 3 5 * 2 或 (10 - 4) / 2 } } def _get_required_parameters(self) - List[str]: return [expression]4.3 文件读写Skill (skills/file_io.py)这个Skill允许Agent读取和写入文本文件但必须严格限制路径防止任意文件访问。# skills/file_io.py import os from pathlib import Path from core.skill import BaseSkill from core.models import SkillInput, SkillResult from typing import Dict, Any, List class FileIOSkill(BaseSkill): 文件读写技能限制在项目目录内 def __init__(self, base_dir: str ./workspace): :param base_dir: 允许文件操作的基础目录默认为项目下的workspace文件夹 super().__init__( namefile_io, description读取或写入文本文件。可以用于记录信息、读取配置等。操作被限制在指定目录内。 ) self.base_dir Path(base_dir).resolve() # 确保基础目录存在 self.base_dir.mkdir(parentsTrue, exist_okTrue) def execute(self, input_data: SkillInput) - SkillResult: action input_data.args.get(action) # read 或 write filename input_data.args.get(filename) content input_data.args.get(content, ) # 写操作时需要 if not action or action not in [read, write]: return SkillResult(successFalse, output, error参数 action 必须为 read 或 write。) if not filename: return SkillResult(successFalse, output, error必须提供 filename 参数。) # 1. 构造绝对路径并确保它在允许的基础目录内防止路径遍历攻击 file_path (self.base_dir / filename).resolve() try: file_path.relative_to(self.base_dir) except ValueError: return SkillResult(successFalse, output, errorf文件路径 {filename} 不在允许的目录 {self.base_dir} 内。) try: if action read: if not file_path.is_file(): return SkillResult(successFalse, output, errorf文件 {filename} 不存在。) with open(file_path, r, encodingutf-8) as f: file_content f.read() return SkillResult( successTrue, outputf文件 {filename} 的内容\n\n{file_content}\n, data{action: read, filename: filename, content: file_content} ) else: # write # 确保目录存在 file_path.parent.mkdir(parentsTrue, exist_okTrue) with open(file_path, w, encodingutf-8) as f: f.write(content) return SkillResult( successTrue, outputf已成功将内容写入文件 {filename}。, data{action: write, filename: filename, content: content} ) except IOError as e: return SkillResult(successFalse, output, errorf文件操作失败: {str(e)}) except Exception as e: return SkillResult(successFalse, output, errorf发生未知错误: {str(e)}) def _get_parameter_schema(self) - Dict[str, Any]: return { action: { type: string, description: 要执行的操作read 或 write, enum: [read, write] }, filename: { type: string, description: 相对于基础工作目录的文件名例如 notes.txt 或 data/log.md }, content: { type: string, description: 当 action 为 write 时要写入文件的内容 } } def _get_required_parameters(self) - List[str]: return [action, filename]4.4 技能注册与整合我们需要一个地方来初始化并注册所有Skill。通常在应用启动时完成。# skills/__init__.py from .web_search import WebSearchSkill from .calculator import CalculatorSkill from .file_io import FileIOSkill from core.skill import registry def register_all_skills(): 注册所有技能到全局注册中心 registry.register(WebSearchSkill()) registry.register(CalculatorSkill()) registry.register(FileIOSkill()) print(所有技能注册完成。)5. 运行与测试构建完整的Agent应用现在我们将所有模块组合起来创建一个可运行的Agent应用。5.1 应用主入口 (main.py)# main.py import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from skills import register_all_skills from core.agent import Agent from utils.config import settings def main(): # 0. 检查配置 if not settings.OPENAI_API_KEY: print(错误未设置 OPENAI_API_KEY。请在 .env 文件中配置。) print(示例OPENAI_API_KEYsk-...) return # 1. 注册所有技能 print(正在初始化技能...) register_all_skills() # 2. 创建Agent实例 print(正在启动智能体...) agent Agent() # 3. 简单的命令行交互循环 print(\n *50) print(Agent Skills 演示系统已启动) print(输入 quit 或 exit 退出程序。) print(*50 \n) while True: try: user_input input(\n你: ).strip() if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input: continue # 4. 调用Agent处理用户输入 print(Agent: 思考中...) response agent.chat(user_input) # 5. 打印结果 print(f\nAgent: {response.message}) if response.used_skills: print(f[本次使用了技能: {, .join(response.used_skills)}]) except KeyboardInterrupt: print(\n\n程序被中断。) break except Exception as e: print(f\n系统错误: {e}) if __name__ __main__: main()5.2 运行与交互示例确保你的.env文件已正确配置API密钥然后在项目根目录运行python main.py你将看到类似以下的输出和交互过程正在初始化技能... [SkillRegistry] Registered skill: web_search [SkillRegistry] Registered skill: calculator [SkillRegistry] Registered skill: file_io 所有技能注册完成。 正在启动智能体... Agent Skills 演示系统已启动 输入 quit 或 exit 退出程序。 你: 计算一下 (15 7) * 3 等于多少 Agent: 思考中... Agent: (15 7) * 3 66 [本次使用了技能: calculator] 你: 搜索一下今天关于人工智能的最新新闻 Agent: 思考中... Agent: 根据搜索结果今天关于人工智能的最新新闻主要集中在以下几个方面 1. OpenAI发布了新的多模态模型... 2. 某科技公司宣布开源其大语言模型... ... (具体摘要) [本次使用了技能: web_search] 你: 帮我把“今天学习了Agent架构”这句话保存到 notes.txt 文件里 Agent: 思考中... Agent: 已成功将内容写入文件 notes.txt。 [本次使用了技能: file_io] 你: 退出 再见6. 常见问题与排查思路在实践过程中你可能会遇到一些问题。以下是一些常见问题的排查思路。问题现象可能原因解决思路导入错误ModuleNotFoundError1. 未安装依赖包。2. 虚拟环境未激活。3.PYTHONPATH设置问题。1. 运行pip install -r requirements.txt。2. 确认命令行提示符前有(venv)。3. 在main.py开头已添加sys.path确保从根目录运行。OpenAI API 调用失败1. API Key 错误或未设置。2. 网络连接问题。3. 额度不足或账号被封。1. 检查.env文件中的OPENAI_API_KEY。2. 检查网络或尝试设置代理注意此处不讨论具体代理工具。3. 登录OpenAI平台检查账号状态和额度。Agent 不调用 Skill直接回复1. 系统提示词Prompt未正确描述 Skills。2. LLM 温度temperature设置过高输出不稳定。3. Skill 描述不够清晰LLM 无法判断何时调用。1. 检查_default_system_prompt方法确保Skill列表被正确格式化并包含在提示词中。2. 在_call_llm方法中将temperature调低如0.1。3. 优化Skill的description使其更具体例如“当用户需要计算数学表达式时使用”。Skill 执行出错1. Skill 的execute方法有bug。2. LLM 生成的参数格式不符合 Skill 预期。3. 外部API或资源不可用如搜索失败。1. 在Skill的execute方法中添加更详细的异常捕获和日志。2. 在Agent的_execute_skill方法中打印skill_input进行调试。3. 为依赖外部网络的Skill添加重试机制和超时设置。路径遍历安全风险FileIOSkill未对文件路径进行严格限制可能导致读取系统文件。已通过pathlib.Path.resolve()和relative_to()方法进行了防护。确保base_dir被设置为一个安全的、专用的目录。7. 架构演进与最佳实践我们实现了一个基础但完整的Agent Skills架构。要将其用于更严肃的项目还需要考虑以下进阶方向和最佳实践。7.1 架构扩展方向技能发现与动态加载当前技能是硬编码注册的。可以改为从指定目录自动扫描并加载继承自BaseSkill的类实现真正的插件化。技能编排与工作流当前Agent一次只能调用一个Skill。复杂的任务需要多个Skill按顺序或条件执行。可以引入“工作流引擎”或“规划器”模块让LLM生成一个技能调用序列Plan。技能结果验证与重试为Skill执行结果添加验证逻辑如果结果不满足要求如搜索无结果可以触发重试或尝试其他技能。长期记忆与上下文管理当前的对话历史是临时的。可以集成向量数据库如Chroma, Pinecone来存储和检索长期记忆使Agent能记住跨会话的信息。多模态技能扩展Skill基类支持处理图像、音频等多模态输入输出。7.2 工程化最佳实践配置中心化使用pydantic-settings管理所有配置支持不同环境开发、测试、生产。全面的日志记录为Agent的决策过程、Skill的调用和结果添加结构化日志如使用logging模块便于监控和调试。异步化改造LLM API调用和某些Skill如网络请求是I/O密集型操作。使用asyncio和aiohttp进行异步改造可以大幅提升Agent的响应速度和吞吐量。技能权限与安全为Skill定义权限等级如“读取文件”、“执行命令”、“访问网络”并在Agent调用前进行权限校验。对用户输入和Skill参数进行严格的清洗和验证。单元测试与集成测试为每个Skill编写单元测试模拟各种输入。为Agent编写集成测试验证其端到端的决策和调用流程。使用更成熟的框架对于生产级应用建议基于成熟的开源框架进行开发如LangChain、LlamaIndex、AutoGen等。它们提供了更健壮的工具调用、记忆、流程控制等组件。本教程的自实现旨在帮助你理解底层原理。7.3 提示词Prompt工程优化Agent的智能程度很大程度上取决于给LLM的提示词。优化你的系统提示词明确指令清晰告诉LLM必须使用JSON格式回复以调用工具。提供示例在提示词中加入1-2个用户请求和正确调用Skill的JSON示例Few-Shot Learning。限制幻觉明确告知LLM“如果你不知道或工具无法处理请直接说不知道不要编造信息”。结构化输出要求LLM在“thought”字段中先进行思考链Chain-of-Thought这能提高决策的可靠性。通过本教程你不仅完成了一个可运行的Agent Skills项目更重要的是理解了其核心架构思想通过注册中心解耦能力利用LLM进行规划与调度通过标准化接口执行具体操作。这套模式是构建复杂AI应用的基础。接下来你可以尝试添加更多有趣的Skill如发送邮件、查询数据库、控制智能家居等或者尝试集成不同的LLM如通义千问、DeepSeek等兼容OpenAI API的模型探索Agent技术的无限可能。
在AI技术浪潮席卷全球的今天智能体Agent正从概念走向落地成为连接大模型与具体业务场景的关键桥梁。然而许多开发者在尝试构建自己的Agent时常常陷入“理论懂落地难”的困境如何设计一个稳定、可扩展的架构如何让Agent具备执行复杂任务的能力Skills如何整合外部工具与数据本文将围绕“Agent Skills架构”这一核心主题为你带来一份从零到一的完整实战教程。我们将手把手带你搭建一个具备基础技能如网络搜索、文件读写的智能体项目并深入剖析其背后的架构设计思想让你不仅能跑通代码更能理解其运作机理为构建更复杂的AI应用打下坚实基础。1. 智能体Agent与技能Skills核心概念解析在深入代码之前我们必须厘清几个核心概念这是理解后续所有架构和实现的基础。1.1 什么是智能体Agent在AI语境下智能体Agent并非一个全新的概念。简单来说它是一个能够感知环境、进行决策并执行动作以实现特定目标的软件实体。在大模型时代Agent通常指一个以大语言模型LLM为“大脑”的系统它能够理解用户意图、规划任务步骤、调用工具技能并处理结果。一个典型的AI Agent工作流程可以概括为感知Perception - 规划Planning - 执行Action - 观察Observation的循环直到任务完成。例如当用户请求“帮我查一下今天北京的天气然后总结成一份简报”Agent需要理解这个复杂请求规划出“搜索天气”和“总结简报”两个子任务依次调用对应的工具执行最后将整合的结果返回给用户。1.2 什么是技能Skills技能Skills是Agent能力的具象化体现是Agent能够执行的具体操作单元。你可以将其理解为Agent的“手脚”或“工具箱”里的工具。一个Skill通常封装了一个特定的功能例如网络搜索Skill调用搜索引擎API获取实时信息。计算器Skill执行数学运算。文件读写Skill读取本地文件内容或写入数据。数据库查询Skill连接数据库并执行SQL。Skills与大型语言模型本身的知识库形成互补。LLM擅长理解和生成语言但缺乏获取实时信息、执行精确计算或操作外部系统的能力。Skills正是为了弥补这一短板而存在。1.3 Agent Skills 架构的核心价值那么为什么要专门设计一个“Agent Skills架构”其核心价值在于解耦、复用与扩展。解耦将Agent的核心决策逻辑大脑与具体的执行能力手脚分离。大脑LLM只负责“想”手脚Skills负责“做”。这使得两者可以独立演进和优化。复用设计良好的Skill可以被不同的Agent重复使用。一个写好的“天气查询Skill”既可以用于个人助手Agent也可以用于旅行规划Agent。扩展当需要为Agent增加新能力时无需改动核心架构只需开发并注册一个新的Skill即可。这种插件化的方式极大地提升了系统的可扩展性。1.4 与相关概念的区分MCP, Plugin等在社区中你可能会遇到类似的概念如MCPModel Context Protocol、Plugin插件等。这里做一个简要区分Skills本文焦点通常指在Agent框架内部定义的、可被Agent直接调用的功能模块。它们与Agent框架耦合较紧密实现相对轻量。MCP是一个由Anthropic提出的开放协议旨在为大模型提供标准化的方式来连接数据源、工具和应用程序。它更侧重于定义一套通用的通信标准使得任何兼容MCP的服务都能被支持该协议的模型或客户端使用。你可以把MCP Server看作是一种标准化、协议化的Skill提供方。Plugin概念更泛化在某些框架如LangChain中Plugin可能等同于Tool或Skill在ChatGPT等产品中特指通过官方平台审核上架的三方扩展。对于初学者和自定义项目从实现自有的Skills架构入手是理解其原理的最佳途径。2. 环境准备与项目初始化我们将使用Python作为开发语言因为它拥有最丰富的AI生态库。本教程将构建一个控制台应用的Agent重点在于架构理解。2.1 基础环境要求操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)Python版本3.8 - 3.11推荐3.9或3.10兼容性最好包管理工具pip (建议版本21.0)代码编辑器VS Code, PyCharm 或任何你熟悉的IDE。2.2 创建项目与虚拟环境首先为项目创建一个独立的目录和Python虚拟环境这是管理依赖的最佳实践。# 1. 创建项目目录并进入 mkdir agent-skills-tutorial cd agent-skills-tutorial # 2. 创建虚拟环境 (以venv为例) python -m venv venv # 3. 激活虚拟环境 # Windows (PowerShell) .\venv\Scripts\Activate.ps1 # Windows (CMD) .\venv\Scripts\activate.bat # macOS / Linux source venv/bin/activate # 激活后命令行提示符前应显示 (venv)2.3 安装核心依赖我们将使用openai库作为与大模型交互的客户端并使用requests进行网络请求。pydantic用于数据验证和设置管理python-dotenv用于管理环境变量如API密钥。# 在激活的虚拟环境中执行 pip install openai requests pydantic python-dotenv为了后续示例我们还需要一个能进行简单网页搜索的工具。这里我们使用duckduckgo-search作为免费示例。请注意对于生产环境建议使用更稳定、功能更强的搜索引擎API如Serper、Google Custom Search等。pip install duckduckgo-search2.4 项目结构设计一个清晰的项目结构是良好架构的开始。我们的项目结构如下agent-skills-tutorial/ ├── .env # 存储敏感信息如API密钥切勿提交到Git ├── .gitignore # Git忽略文件 ├── requirements.txt # 项目依赖清单 ├── main.py # 应用主入口 ├── core/ # 核心架构模块 │ ├── __init__.py │ ├── agent.py # Agent核心类 │ ├── skill.py # Skill基类与注册器 │ └── models.py # 数据模型消息、结果等 ├── skills/ # 技能实现目录 │ ├── __init__.py │ ├── web_search.py # 网络搜索技能 │ ├── calculator.py # 计算器技能 │ └── file_io.py # 文件读写技能 └── utils/ # 工具函数 ├── __init__.py └── config.py # 配置加载你可以使用以下命令快速创建这个结构# 在项目根目录下执行 mkdir core skills utils touch .env .gitignore requirements.txt main.py touch core/__init__.py core/agent.py core/skill.py core/models.py touch skills/__init__.py skills/web_search.py skills/calculator.py skills/file_io.py touch utils/__init__.py utils/config.py在.gitignore文件中至少添加以下内容venv/ .env __pycache__/ *.pyc3. 核心架构设计与模块拆解现在我们来逐一实现这个架构的核心模块。3.1 数据模型定义 (core/models.py)我们首先定义在整个系统中流转的数据结构使用pydantic可以方便地进行数据验证和序列化。# core/models.py from pydantic import BaseModel from typing import Dict, Any, Optional, List from enum import Enum class MessageRole(str, Enum): 消息角色枚举 USER user ASSISTANT assistant SYSTEM system class Message(BaseModel): 对话消息 role: MessageRole content: str class SkillInput(BaseModel): Skill执行的输入参数 args: Dict[str, Any] # 参数字典 class SkillResult(BaseModel): Skill执行的结果 success: bool output: str # 执行输出的文本 error: Optional[str] None # 如果失败错误信息 data: Optional[Dict[str, Any]] None # 额外的结构化数据 class AgentResponse(BaseModel): Agent对用户的最终响应 message: str # 回复内容 used_skills: List[str] [] # 本次响应使用了哪些技能 raw_data: Optional[Dict[str, Any]] None # 原始数据如搜索结果的JSON3.2 Skill基类与注册机制 (core/skill.py)这是架构中最关键的部分之一。我们定义一个所有Skill都必须继承的基类并实现一个简单的注册中心来管理它们。# core/skill.py from abc import ABC, abstractmethod from typing import Dict, Any from .models import SkillInput, SkillResult class BaseSkill(ABC): Skill抽象基类 def __init__(self, name: str, description: str): self.name name # Skill的唯一标识如 “web_search” self.description description # 对LLM的描述用于让LLM知道何时调用此Skill abstractmethod def execute(self, input_data: SkillInput) - SkillResult: 执行技能的核心方法子类必须实现 pass def get_schema(self) - Dict[str, Any]: 返回Skill的调用模式用于告知LLM如何调用 # 这是一个简化示例更复杂的实现可以动态生成JSON Schema return { name: self.name, description: self.description, parameters: { type: object, properties: self._get_parameter_schema(), required: self._get_required_parameters() } } def _get_parameter_schema(self) - Dict[str, Any]: 子类可重写此方法来定义参数结构 return {} def _get_required_parameters(self) - List[str]: 子类可重写此方法来定义必填参数 return [] class SkillRegistry: Skill注册中心单例模式 _instance None _skills: Dict[str, BaseSkill] {} def __new__(cls): if cls._instance is None: cls._instance super(SkillRegistry, cls).__new__(cls) return cls._instance def register(self, skill: BaseSkill): 注册一个Skill if skill.name in self._skills: raise ValueError(fSkill with name {skill.name} is already registered.) self._skills[skill.name] skill print(f[SkillRegistry] Registered skill: {skill.name}) def get_skill(self, name: str) - BaseSkill: 根据名称获取Skill skill self._skills.get(name) if not skill: raise KeyError(fSkill {name} not found in registry.) return skill def list_skills(self) - Dict[str, str]: 列出所有已注册Skill的名称和描述 return {name: skill.description for name, skill in self._skills.items()} def clear(self): 清空注册表主要用于测试 self._skills.clear() # 全局注册中心实例 registry SkillRegistry()3.3 Agent核心类 (core/agent.py)Agent类负责协调整个工作流程接收用户输入与LLM交互决定是否调用及调用哪个Skill执行Skill并将结果整合返回。# core/agent.py import json import openai from typing import List, Optional from .models import Message, MessageRole, AgentResponse, SkillInput, SkillResult from .skill import registry from utils.config import settings class Agent: 智能体核心类 def __init__(self, system_prompt: Optional[str] None): 初始化Agent。 :param system_prompt: 系统提示词用于设定Agent的角色和行为。 self.client openai.OpenAI(api_keysettings.OPENAI_API_KEY, base_urlsettings.OPENAI_BASE_URL) self.system_prompt system_prompt or self._default_system_prompt() self.conversation_history: List[Message] [] if self.system_prompt: self.conversation_history.append(Message(roleMessageRole.SYSTEM, contentself.system_prompt)) def _default_system_prompt(self) - str: 默认系统提示词描述了Agent的能力和行为规范 skills_desc \n.join([f- {name}: {desc} for name, desc in registry.list_skills().items()]) return f你是一个有帮助的AI助手可以调用工具Skills来帮助用户解决问题。 你可以使用的工具Skills有 {skills_desc} 当用户的问题需要用到上述工具时请严格按照以下JSON格式响应且只输出这个JSON {{ thought: 你的思考过程分析是否需要以及使用哪个工具, skill_to_use: 要使用的工具名称如果不需要则为null, skill_input: {{}} // 工具的输入参数如果不需要工具则为空对象 }} 如果不需要使用工具请直接以自然语言回复用户。 def chat(self, user_input: str) - AgentResponse: 主要聊天接口。 :param user_input: 用户输入 :return: Agent的响应 # 1. 将用户输入加入历史 self.conversation_history.append(Message(roleMessageRole.USER, contentuser_input)) # 2. 调用LLM获取响应 llm_response self._call_llm() # 3. 解析LLM响应判断是否需要调用Skill used_skills [] final_response_text raw_data None try: # 尝试解析JSON如果解析成功说明LLM决定调用Skill action_data json.loads(llm_response) skill_name action_data.get(skill_to_use) if skill_name and skill_name ! null: # 4. 调用Skill skill_input SkillInput(argsaction_data.get(skill_input, {})) skill_result self._execute_skill(skill_name, skill_input) used_skills.append(skill_name) raw_data skill_result.data # 5. 将Skill执行结果作为新的上下文再次调用LLM进行总结或回答 self.conversation_history.append( Message(roleMessageRole.ASSISTANT, contentf[调用工具 {skill_name}] 结果: {skill_result.output}) ) # 获取LLM基于工具结果的最终回复 final_llm_response self._call_llm() final_response_text final_llm_response else: # LLM直接回复无需调用工具 final_response_text action_data.get(thought, llm_response) # 如果JSON里有thought用它 except json.JSONDecodeError: # LLM的响应不是JSON说明它直接进行了自然语言回复 final_response_text llm_response # 6. 将Assistant的最终回复加入历史 self.conversation_history.append(Message(roleMessageRole.ASSISTANT, contentfinal_response_text)) # 7. 返回结构化响应 return AgentResponse( messagefinal_response_text, used_skillsused_skills, raw_dataraw_data ) def _call_llm(self) - str: 调用OpenAI API或其他兼容API try: response self.client.chat.completions.create( modelsettings.LLM_MODEL, # 例如 gpt-3.5-turbo messages[msg.dict() for msg in self.conversation_history], temperature0.1, # 低温度使输出更确定更适合工具调用 max_tokens500 ) return response.choices[0].message.content.strip() except Exception as e: return f调用语言模型时出错: {str(e)} def _execute_skill(self, skill_name: str, skill_input: SkillInput) - SkillResult: 执行指定的Skill try: skill registry.get_skill(skill_name) return skill.execute(skill_input) except Exception as e: return SkillResult(successFalse, output, errorf执行技能 {skill_name} 时出错: {str(e)}) def clear_history(self): 清空对话历史保留系统提示 self.conversation_history [msg for msg in self.conversation_history if msg.role MessageRole.SYSTEM]3.4 配置管理 (utils/config.py)使用环境变量来管理配置避免将API密钥等敏感信息硬编码在代码中。# utils/config.py import os from pydantic_settings import BaseSettings from pydantic import Field class Settings(BaseSettings): 应用配置 OPENAI_API_KEY: str Field(default, descriptionOpenAI API Key) OPENAI_BASE_URL: str Field(defaulthttps://api.openai.com/v1, descriptionOpenAI API Base URL) LLM_MODEL: str Field(defaultgpt-3.5-turbo, description使用的LLM模型) class Config: env_file .env # 从 .env 文件加载配置 env_file_encoding utf-8 # 创建全局配置实例 settings Settings()在项目根目录创建.env文件并填入你的配置请勿提交此文件到版本控制# .env OPENAI_API_KEYsk-your-openai-api-key-here # 如果你使用其他兼容OpenAI API的服务可以修改以下两项 # OPENAI_BASE_URLhttps://api.xxx.com/v1 # LLM_MODELgpt-3.5-turbo4. 实战实现并注册具体Skills架构搭好了现在我们来为Agent添砖加瓦实现几个实用的Skill。4.1 网络搜索Skill (skills/web_search.py)这个Skill使用duckduckgo_search来获取网络信息。# skills/web_search.py from duckduckgo_search import DDGS from core.skill import BaseSkill from core.models import SkillInput, SkillResult from typing import Dict, Any, List class WebSearchSkill(BaseSkill): 网络搜索技能 def __init__(self): super().__init__( nameweb_search, description使用搜索引擎在互联网上搜索信息。当用户询问实时信息、新闻、最新事件或未知知识时使用。 ) def execute(self, input_data: SkillInput) - SkillResult: query input_data.args.get(query) if not query: return SkillResult(successFalse, output, error搜索技能需要 query 参数。) try: with DDGS() as ddgs: # 获取最相关的5条结果 results list(ddgs.text(query, max_results5)) if not results: return SkillResult(successTrue, outputf未找到关于 {query} 的搜索结果。, data{results: []}) # 格式化输出 formatted_results [] output_lines [f关于 {query} 的搜索结果] for i, r in enumerate(results, 1): output_lines.append(f{i}. {r[title]}) output_lines.append(f 链接: {r[href]}) output_lines.append(f 摘要: {r[body][:150]}...) output_lines.append() formatted_results.append({title: r[title], href: r[href], snippet: r[body]}) output_text \n.join(output_lines) return SkillResult( successTrue, outputoutput_text, data{query: query, results: formatted_results} ) except Exception as e: return SkillResult(successFalse, output, errorf搜索过程中发生错误: {str(e)}) def _get_parameter_schema(self) - Dict[str, Any]: return { query: { type: string, description: 要搜索的关键词或问题 } } def _get_required_parameters(self) - List[str]: return [query]4.2 计算器Skill (skills/calculator.py)这个Skill使用Python的eval函数进行数学计算。注意在生产环境中使用eval是危险的这里仅作演示。安全做法是使用ast.literal_eval或专门的数学表达式解析库。# skills/calculator.py import ast import operator as op from core.skill import BaseSkill from core.models import SkillInput, SkillResult # 安全地支持的操作符 (简化版生产环境建议使用更安全的库如 numexpr) _supported_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, } def _safe_eval(node): 安全地评估一个AST数字表达式节点 if isinstance(node, ast.Num): # Python 3.8 return node.n elif isinstance(node, ast.Constant): # Python 3.8 return node.value elif isinstance(node, ast.BinOp): left _safe_eval(node.left) right _safe_eval(node.right) operator_func _supported_operators.get(type(node.op)) if operator_func is None: raise TypeError(f不支持的运算符: {node.op}) return operator_func(left, right) elif isinstance(node, ast.UnaryOp): operand _safe_eval(node.operand) operator_func _supported_operators.get(type(node.op)) if operator_func is None: raise TypeError(f不支持的运算符: {node.op}) return operator_func(operand) else: raise TypeError(f不支持的表达式类型: {node}) class CalculatorSkill(BaseSkill): 计算器技能安全版 def __init__(self): super().__init__( namecalculator, description执行数学计算。当用户需要计算数学表达式时使用例如 23*4 或 sin(30)。注意本技能仅支持基本算术。 ) def execute(self, input_data: SkillInput) - SkillResult: expression input_data.args.get(expression) if not expression: return SkillResult(successFalse, output, error计算器技能需要 expression 参数。) # 清理表达式移除可能的安全风险字符非常基础的过滤 expression expression.strip().replace( , ) # 禁止导入、函数调用等 for forbidden in [import, exec, eval, __, open, file, sys, os]: if forbidden in expression.lower(): return SkillResult(successFalse, output, error表达式包含不安全内容。) try: # 使用ast解析确保是字面量表达式 tree ast.parse(expression, modeeval) result _safe_eval(tree.body) return SkillResult(successTrue, outputf{expression} {result}, data{expression: expression, result: result}) except SyntaxError: return SkillResult(successFalse, output, error数学表达式语法错误。) except TypeError as e: return SkillResult(successFalse, output, errorf计算错误: {str(e)}) except Exception as e: return SkillResult(successFalse, output, errorf计算过程中发生未知错误: {str(e)}) def _get_parameter_schema(self) - Dict[str, Any]: return { expression: { type: string, description: 要计算的数学表达式例如 3 5 * 2 或 (10 - 4) / 2 } } def _get_required_parameters(self) - List[str]: return [expression]4.3 文件读写Skill (skills/file_io.py)这个Skill允许Agent读取和写入文本文件但必须严格限制路径防止任意文件访问。# skills/file_io.py import os from pathlib import Path from core.skill import BaseSkill from core.models import SkillInput, SkillResult from typing import Dict, Any, List class FileIOSkill(BaseSkill): 文件读写技能限制在项目目录内 def __init__(self, base_dir: str ./workspace): :param base_dir: 允许文件操作的基础目录默认为项目下的workspace文件夹 super().__init__( namefile_io, description读取或写入文本文件。可以用于记录信息、读取配置等。操作被限制在指定目录内。 ) self.base_dir Path(base_dir).resolve() # 确保基础目录存在 self.base_dir.mkdir(parentsTrue, exist_okTrue) def execute(self, input_data: SkillInput) - SkillResult: action input_data.args.get(action) # read 或 write filename input_data.args.get(filename) content input_data.args.get(content, ) # 写操作时需要 if not action or action not in [read, write]: return SkillResult(successFalse, output, error参数 action 必须为 read 或 write。) if not filename: return SkillResult(successFalse, output, error必须提供 filename 参数。) # 1. 构造绝对路径并确保它在允许的基础目录内防止路径遍历攻击 file_path (self.base_dir / filename).resolve() try: file_path.relative_to(self.base_dir) except ValueError: return SkillResult(successFalse, output, errorf文件路径 {filename} 不在允许的目录 {self.base_dir} 内。) try: if action read: if not file_path.is_file(): return SkillResult(successFalse, output, errorf文件 {filename} 不存在。) with open(file_path, r, encodingutf-8) as f: file_content f.read() return SkillResult( successTrue, outputf文件 {filename} 的内容\n\n{file_content}\n, data{action: read, filename: filename, content: file_content} ) else: # write # 确保目录存在 file_path.parent.mkdir(parentsTrue, exist_okTrue) with open(file_path, w, encodingutf-8) as f: f.write(content) return SkillResult( successTrue, outputf已成功将内容写入文件 {filename}。, data{action: write, filename: filename, content: content} ) except IOError as e: return SkillResult(successFalse, output, errorf文件操作失败: {str(e)}) except Exception as e: return SkillResult(successFalse, output, errorf发生未知错误: {str(e)}) def _get_parameter_schema(self) - Dict[str, Any]: return { action: { type: string, description: 要执行的操作read 或 write, enum: [read, write] }, filename: { type: string, description: 相对于基础工作目录的文件名例如 notes.txt 或 data/log.md }, content: { type: string, description: 当 action 为 write 时要写入文件的内容 } } def _get_required_parameters(self) - List[str]: return [action, filename]4.4 技能注册与整合我们需要一个地方来初始化并注册所有Skill。通常在应用启动时完成。# skills/__init__.py from .web_search import WebSearchSkill from .calculator import CalculatorSkill from .file_io import FileIOSkill from core.skill import registry def register_all_skills(): 注册所有技能到全局注册中心 registry.register(WebSearchSkill()) registry.register(CalculatorSkill()) registry.register(FileIOSkill()) print(所有技能注册完成。)5. 运行与测试构建完整的Agent应用现在我们将所有模块组合起来创建一个可运行的Agent应用。5.1 应用主入口 (main.py)# main.py import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from skills import register_all_skills from core.agent import Agent from utils.config import settings def main(): # 0. 检查配置 if not settings.OPENAI_API_KEY: print(错误未设置 OPENAI_API_KEY。请在 .env 文件中配置。) print(示例OPENAI_API_KEYsk-...) return # 1. 注册所有技能 print(正在初始化技能...) register_all_skills() # 2. 创建Agent实例 print(正在启动智能体...) agent Agent() # 3. 简单的命令行交互循环 print(\n *50) print(Agent Skills 演示系统已启动) print(输入 quit 或 exit 退出程序。) print(*50 \n) while True: try: user_input input(\n你: ).strip() if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input: continue # 4. 调用Agent处理用户输入 print(Agent: 思考中...) response agent.chat(user_input) # 5. 打印结果 print(f\nAgent: {response.message}) if response.used_skills: print(f[本次使用了技能: {, .join(response.used_skills)}]) except KeyboardInterrupt: print(\n\n程序被中断。) break except Exception as e: print(f\n系统错误: {e}) if __name__ __main__: main()5.2 运行与交互示例确保你的.env文件已正确配置API密钥然后在项目根目录运行python main.py你将看到类似以下的输出和交互过程正在初始化技能... [SkillRegistry] Registered skill: web_search [SkillRegistry] Registered skill: calculator [SkillRegistry] Registered skill: file_io 所有技能注册完成。 正在启动智能体... Agent Skills 演示系统已启动 输入 quit 或 exit 退出程序。 你: 计算一下 (15 7) * 3 等于多少 Agent: 思考中... Agent: (15 7) * 3 66 [本次使用了技能: calculator] 你: 搜索一下今天关于人工智能的最新新闻 Agent: 思考中... Agent: 根据搜索结果今天关于人工智能的最新新闻主要集中在以下几个方面 1. OpenAI发布了新的多模态模型... 2. 某科技公司宣布开源其大语言模型... ... (具体摘要) [本次使用了技能: web_search] 你: 帮我把“今天学习了Agent架构”这句话保存到 notes.txt 文件里 Agent: 思考中... Agent: 已成功将内容写入文件 notes.txt。 [本次使用了技能: file_io] 你: 退出 再见6. 常见问题与排查思路在实践过程中你可能会遇到一些问题。以下是一些常见问题的排查思路。问题现象可能原因解决思路导入错误ModuleNotFoundError1. 未安装依赖包。2. 虚拟环境未激活。3.PYTHONPATH设置问题。1. 运行pip install -r requirements.txt。2. 确认命令行提示符前有(venv)。3. 在main.py开头已添加sys.path确保从根目录运行。OpenAI API 调用失败1. API Key 错误或未设置。2. 网络连接问题。3. 额度不足或账号被封。1. 检查.env文件中的OPENAI_API_KEY。2. 检查网络或尝试设置代理注意此处不讨论具体代理工具。3. 登录OpenAI平台检查账号状态和额度。Agent 不调用 Skill直接回复1. 系统提示词Prompt未正确描述 Skills。2. LLM 温度temperature设置过高输出不稳定。3. Skill 描述不够清晰LLM 无法判断何时调用。1. 检查_default_system_prompt方法确保Skill列表被正确格式化并包含在提示词中。2. 在_call_llm方法中将temperature调低如0.1。3. 优化Skill的description使其更具体例如“当用户需要计算数学表达式时使用”。Skill 执行出错1. Skill 的execute方法有bug。2. LLM 生成的参数格式不符合 Skill 预期。3. 外部API或资源不可用如搜索失败。1. 在Skill的execute方法中添加更详细的异常捕获和日志。2. 在Agent的_execute_skill方法中打印skill_input进行调试。3. 为依赖外部网络的Skill添加重试机制和超时设置。路径遍历安全风险FileIOSkill未对文件路径进行严格限制可能导致读取系统文件。已通过pathlib.Path.resolve()和relative_to()方法进行了防护。确保base_dir被设置为一个安全的、专用的目录。7. 架构演进与最佳实践我们实现了一个基础但完整的Agent Skills架构。要将其用于更严肃的项目还需要考虑以下进阶方向和最佳实践。7.1 架构扩展方向技能发现与动态加载当前技能是硬编码注册的。可以改为从指定目录自动扫描并加载继承自BaseSkill的类实现真正的插件化。技能编排与工作流当前Agent一次只能调用一个Skill。复杂的任务需要多个Skill按顺序或条件执行。可以引入“工作流引擎”或“规划器”模块让LLM生成一个技能调用序列Plan。技能结果验证与重试为Skill执行结果添加验证逻辑如果结果不满足要求如搜索无结果可以触发重试或尝试其他技能。长期记忆与上下文管理当前的对话历史是临时的。可以集成向量数据库如Chroma, Pinecone来存储和检索长期记忆使Agent能记住跨会话的信息。多模态技能扩展Skill基类支持处理图像、音频等多模态输入输出。7.2 工程化最佳实践配置中心化使用pydantic-settings管理所有配置支持不同环境开发、测试、生产。全面的日志记录为Agent的决策过程、Skill的调用和结果添加结构化日志如使用logging模块便于监控和调试。异步化改造LLM API调用和某些Skill如网络请求是I/O密集型操作。使用asyncio和aiohttp进行异步改造可以大幅提升Agent的响应速度和吞吐量。技能权限与安全为Skill定义权限等级如“读取文件”、“执行命令”、“访问网络”并在Agent调用前进行权限校验。对用户输入和Skill参数进行严格的清洗和验证。单元测试与集成测试为每个Skill编写单元测试模拟各种输入。为Agent编写集成测试验证其端到端的决策和调用流程。使用更成熟的框架对于生产级应用建议基于成熟的开源框架进行开发如LangChain、LlamaIndex、AutoGen等。它们提供了更健壮的工具调用、记忆、流程控制等组件。本教程的自实现旨在帮助你理解底层原理。7.3 提示词Prompt工程优化Agent的智能程度很大程度上取决于给LLM的提示词。优化你的系统提示词明确指令清晰告诉LLM必须使用JSON格式回复以调用工具。提供示例在提示词中加入1-2个用户请求和正确调用Skill的JSON示例Few-Shot Learning。限制幻觉明确告知LLM“如果你不知道或工具无法处理请直接说不知道不要编造信息”。结构化输出要求LLM在“thought”字段中先进行思考链Chain-of-Thought这能提高决策的可靠性。通过本教程你不仅完成了一个可运行的Agent Skills项目更重要的是理解了其核心架构思想通过注册中心解耦能力利用LLM进行规划与调度通过标准化接口执行具体操作。这套模式是构建复杂AI应用的基础。接下来你可以尝试添加更多有趣的Skill如发送邮件、查询数据库、控制智能家居等或者尝试集成不同的LLM如通义千问、DeepSeek等兼容OpenAI API的模型探索Agent技术的无限可能。
函数详解：从基础概念到实战应用与性能优化)
