news 2026/8/18 22:49:48

AI智能体技能架构实战:从零构建可扩展的Agent系统

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
AI智能体技能架构实战:从零构建可扩展的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架构”?其核心价值在于解耦、复用与扩展

  1. 解耦:将Agent的核心决策逻辑(大脑)与具体的执行能力(手脚)分离。大脑(LLM)只负责“想”,手脚(Skills)负责“做”。这使得两者可以独立演进和优化。
  2. 复用:设计良好的Skill可以被不同的Agent重复使用。一个写好的“天气查询Skill”,既可以用于个人助手Agent,也可以用于旅行规划Agent。
  3. 扩展:当需要为Agent增加新能力时,无需改动核心架构,只需开发并注册一个新的Skill即可。这种插件化的方式极大地提升了系统的可扩展性。

1.4 与相关概念的区分(MCP, Plugin等)

在社区中,你可能会遇到类似的概念,如MCP(Model 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-search

2.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__/ *.pyc

3. 核心架构设计与模块拆解

现在,我们来逐一实现这个架构的核心模块。

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 # 原始数据(如搜索结果的JSON)

3.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(f"Skill 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(f"Skill '{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_key=settings.OPENAI_API_KEY, base_url=settings.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(role=MessageRole.SYSTEM, content=self.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(role=MessageRole.USER, content=user_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(args=action_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(role=MessageRole.ASSISTANT, content=f"[调用工具 {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(role=MessageRole.ASSISTANT, content=final_response_text)) # 7. 返回结构化响应 return AgentResponse( message=final_response_text, used_skills=used_skills, raw_data=raw_data ) def _call_llm(self) -> str: """调用OpenAI API(或其他兼容API)""" try: response = self.client.chat.completions.create( model=settings.LLM_MODEL, # 例如 "gpt-3.5-turbo" messages=[msg.dict() for msg in self.conversation_history], temperature=0.1, # 低温度使输出更确定,更适合工具调用 max_tokens=500 ) 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(success=False, output="", error=f"执行技能 '{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="", description="OpenAI API Key") OPENAI_BASE_URL: str = Field(default="https://api.openai.com/v1", description="OpenAI API Base URL") LLM_MODEL: str = Field(default="gpt-3.5-turbo", description="使用的LLM模型") class Config: env_file = ".env" # 从 .env 文件加载配置 env_file_encoding = 'utf-8' # 创建全局配置实例 settings = Settings()

在项目根目录创建.env文件,并填入你的配置(请勿提交此文件到版本控制):

# .env OPENAI_API_KEY=sk-your-openai-api-key-here # 如果你使用其他兼容OpenAI API的服务,可以修改以下两项 # OPENAI_BASE_URL=https://api.xxx.com/v1 # LLM_MODEL=gpt-3.5-turbo

4. 实战:实现并注册具体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__( name="web_search", description="使用搜索引擎在互联网上搜索信息。当用户询问实时信息、新闻、最新事件或未知知识时使用。" ) def execute(self, input_data: SkillInput) -> SkillResult: query = input_data.args.get("query") if not query: return SkillResult(success=False, output="", error="搜索技能需要 'query' 参数。") try: with DDGS() as ddgs: # 获取最相关的5条结果 results = list(ddgs.text(query, max_results=5)) if not results: return SkillResult(success=True, output=f"未找到关于 '{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( success=True, output=output_text, data={"query": query, "results": formatted_results} ) except Exception as e: return SkillResult(success=False, output="", error=f"搜索过程中发生错误: {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__( name="calculator", description="执行数学计算。当用户需要计算数学表达式时使用,例如 '2+3*4' 或 'sin(30)'。注意:本技能仅支持基本算术。" ) def execute(self, input_data: SkillInput) -> SkillResult: expression = input_data.args.get("expression") if not expression: return SkillResult(success=False, output="", error="计算器技能需要 'expression' 参数。") # 清理表达式,移除可能的安全风险字符(非常基础的过滤) expression = expression.strip().replace(" ", "") # 禁止导入、函数调用等 for forbidden in ["import", "exec", "eval", "__", "open", "file", "sys", "os"]: if forbidden in expression.lower(): return SkillResult(success=False, output="", error="表达式包含不安全内容。") try: # 使用ast解析,确保是字面量表达式 tree = ast.parse(expression, mode='eval') result = _safe_eval(tree.body) return SkillResult(success=True, output=f"{expression} = {result}", data={"expression": expression, "result": result}) except SyntaxError: return SkillResult(success=False, output="", error="数学表达式语法错误。") except TypeError as e: return SkillResult(success=False, output="", error=f"计算错误: {str(e)}") except Exception as e: return SkillResult(success=False, output="", error=f"计算过程中发生未知错误: {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__( name="file_io", description="读取或写入文本文件。可以用于记录信息、读取配置等。操作被限制在指定目录内。" ) self.base_dir = Path(base_dir).resolve() # 确保基础目录存在 self.base_dir.mkdir(parents=True, exist_ok=True) 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(success=False, output="", error="参数 'action' 必须为 'read' 或 'write'。") if not filename: return SkillResult(success=False, output="", error="必须提供 'filename' 参数。") # 1. 构造绝对路径,并确保它在允许的基础目录内(防止路径遍历攻击) file_path = (self.base_dir / filename).resolve() try: file_path.relative_to(self.base_dir) except ValueError: return SkillResult(success=False, output="", error=f"文件路径 '{filename}' 不在允许的目录 '{self.base_dir}' 内。") try: if action == "read": if not file_path.is_file(): return SkillResult(success=False, output="", error=f"文件 '{filename}' 不存在。") with open(file_path, 'r', encoding='utf-8') as f: file_content = f.read() return SkillResult( success=True, output=f"文件 '{filename}' 的内容:\n```\n{file_content}\n```", data={"action": "read", "filename": filename, "content": file_content} ) else: # write # 确保目录存在 file_path.parent.mkdir(parents=True, exist_ok=True) with open(file_path, 'w', encoding='utf-8') as f: f.write(content) return SkillResult( success=True, output=f"已成功将内容写入文件 '{filename}'。", data={"action": "write", "filename": filename, "content": content} ) except IOError as e: return SkillResult(success=False, output="", error=f"文件操作失败: {str(e)}") except Exception as e: return SkillResult(success=False, output="", error=f"发生未知错误: {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_KEY=sk-...") 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 架构扩展方向

  1. 技能发现与动态加载:当前技能是硬编码注册的。可以改为从指定目录自动扫描并加载继承自BaseSkill的类,实现真正的插件化。
  2. 技能编排与工作流:当前Agent一次只能调用一个Skill。复杂的任务需要多个Skill按顺序或条件执行。可以引入“工作流引擎”或“规划器”模块,让LLM生成一个技能调用序列(Plan)。
  3. 技能结果验证与重试:为Skill执行结果添加验证逻辑,如果结果不满足要求(如搜索无结果),可以触发重试或尝试其他技能。
  4. 长期记忆与上下文管理:当前的对话历史是临时的。可以集成向量数据库(如Chroma, Pinecone)来存储和检索长期记忆,使Agent能记住跨会话的信息。
  5. 多模态技能:扩展Skill基类,支持处理图像、音频等多模态输入输出。

7.2 工程化最佳实践

  1. 配置中心化:使用pydantic-settings管理所有配置,支持不同环境(开发、测试、生产)。
  2. 全面的日志记录:为Agent的决策过程、Skill的调用和结果添加结构化日志(如使用logging模块),便于监控和调试。
  3. 异步化改造:LLM API调用和某些Skill(如网络请求)是I/O密集型操作。使用asyncioaiohttp进行异步改造可以大幅提升Agent的响应速度和吞吐量。
  4. 技能权限与安全:为Skill定义权限等级(如“读取文件”、“执行命令”、“访问网络”),并在Agent调用前进行权限校验。对用户输入和Skill参数进行严格的清洗和验证。
  5. 单元测试与集成测试:为每个Skill编写单元测试,模拟各种输入。为Agent编写集成测试,验证其端到端的决策和调用流程。
  6. 使用更成熟的框架:对于生产级应用,建议基于成熟的开源框架进行开发,如LangChainLlamaIndexAutoGen等。它们提供了更健壮的工具调用、记忆、流程控制等组件。本教程的自实现旨在帮助你理解底层原理。

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技术的无限可能。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/8/18 22:48:51

PPPoE协议深度解析:从拨号原理到家庭网络优化实践

你有没有想过&#xff0c;为什么家里的宽带&#xff0c;明明已经插上了光猫和路由器&#xff0c;有时候还需要在电脑上点一下那个“宽带连接”&#xff0c;输入账号密码才能上网&#xff1f;这个看似“古老”的操作&#xff0c;在光纤入户、千兆宽带普及的今天&#xff0c;依然…

作者头像 李华
网站建设 2026/8/18 22:48:47

基于开源代码挖掘的智能体技能自动化提取框架设计与实现

1. 项目概述&#xff1a;从海量开源智能体仓库中“挖矿”学技能最近在搞多智能体系统开发的朋友&#xff0c;估计都遇到过同一个头疼的问题&#xff1a;想让智能体学会一个新技能&#xff0c;比如“如何用Python的requests库处理OAuth 2.0授权流程”&#xff0c;或者“如何用Do…

作者头像 李华
网站建设 2026/8/18 22:48:44

海量智能体轨迹安全违规检测:工程架构与实战解析

1. 从海量智能体轨迹中嗅探安全违规&#xff1a;一个被低估的工程挑战最近和几个做多智能体系统&#xff08;Multi-Agent System, MAS&#xff09;和机器人流程自动化&#xff08;RPA&#xff09;的朋友聊天&#xff0c;大家不约而同地提到了同一个痛点&#xff1a;系统跑起来了…

作者头像 李华
网站建设 2026/8/18 22:48:08

SystemVerilog覆盖率:芯片验证的量化指标与实战建模指南

1. 项目概述&#xff1a;为什么覆盖率是芯片验证的“体检报告” 做芯片验证的&#xff0c;最怕听到的一句话可能就是“流片回来发现功能有问题”。那感觉&#xff0c;就像你花了几个月盖了一栋大楼&#xff0c;最后验收时发现承重墙没放钢筋&#xff0c;推倒重来的成本高到让人…

作者头像 李华
网站建设 2026/8/18 22:44:25

Simulink频域分析实战:线性化原理与稳定性评估指南

1. 项目概述&#xff1a;从“感觉”到“数据”的跨越 做控制、做信号处理&#xff0c;或者搞机电系统仿真的朋友&#xff0c;肯定都遇到过这样的场景&#xff1a;你辛辛苦苦搭好了一个Simulink模型&#xff0c;参数调来调去&#xff0c;阶跃响应看起来也“差不多”了&#xff0…

作者头像 李华
网站建设 2026/8/18 22:42:34

基于Docker的SoNovel本地小说库部署与自动化管理指南

这次我们来看一个能让你彻底摆脱小说平台限制的开源神器——SoNovel。它是一个基于Docker的本地小说库解决方案&#xff0c;核心目标就是让你能自由地下载、管理和阅读网络小说&#xff0c;构建一个完全私有的、不受任何平台规则约束的个人图书馆。对于经常追更、又苦于平台广告…

作者头像 李华