在 AI 应用开发领域Agent智能体已经从一个前沿概念迅速演变为工程实践的核心组件。与传统的单任务 AI 模型不同Agent 具备自主规划、工具调用和环境交互的能力能够处理更复杂的多步骤任务。而 Agent Skills技能则是构建这些智能体的基石它们像乐高积木一样通过组合和编排赋予 Agent 解决实际问题的具体能力。无论是处理客户咨询、自动化数据报表还是辅助代码开发一个设计良好的 Skill 集成都直接决定了智能体的实用性和可靠性。然而很多开发者在初次接触 Agent 开发时容易陷入两个误区要么过度关注宏观框架而忽略了具体 Skill 的实现细节导致智能体“想法很多能力很少”要么过早陷入某个特定工具链的配置细节失去了对整体架构和可复用性的把握。真正有价值的 Agent 项目需要在抽象框架和具体技能之间找到平衡点确保每个 Skill 都能独立验证、易于调试并且可以平滑地融入更大的工作流。本文将以一个企业级实战项目为背景带你从零开始构建一套完整的 Agent Skills 体系。你将不仅学会如何编写和调试单个 Skill更重要的是掌握如何将这些技能模块化、标准化并最终集成为一个能够自主处理复杂任务的智能助手。整个过程中我们会重点关注环境隔离、错误处理、日志可观测性和生产就绪的配置管理这些都是教程中容易忽略但实际项目必须面对的工程问题。1. 理解 Agent 和 Skill 的基本工作模型在开始写代码之前必须清晰区分 Agent、Skill 和工具这三个核心概念以及它们之间的协作关系。很多配置错误和运行时问题根源都在于对这些基础组件的职责边界理解模糊。1.1 Agent 作为决策中枢Skill 作为能力单元Agent 的本质是一个决策循环感知输入 - 分析意图 - 规划步骤 - 执行技能 - 观察结果 - 调整策略。这个循环中Agent 本身不直接处理具体任务而是负责判断“现在应该调用哪个 Skill”以及“调用时应该传入什么参数”。Skill 则是封装好的能力单元每个 Skill 专注于完成一类特定任务。例如一个“天气查询 Skill”负责获取天气信息一个“数据统计 Skill”负责生成报表。Skill 内部可以包含简单的 API 调用也可以封装复杂的工作流但对 Agent 来说它只是一个可以通过标准化接口调用的黑盒。工具是 Skill 内部使用的具体实现手段。比如一个“文件处理 Skill”可能会用到读写文件的工具函数一个“网络请求 Skill”会用到 HTTP 客户端库。工具是技术实现细节而 Skill 是业务能力抽象。1.2 主流的 Agent 框架和 Skill 开发范式目前比较成熟的 Agent 框架如 LangChain、AutoGPT、Dify 等虽然具体实现不同但在 Skill 集成方面都遵循类似的模式标准化接口Skill 需要暴露统一的调用方法通常包含execute或run函数接受参数字典并返回标准化结果。能力描述每个 Skill 需要明确声明自己的能力范围、输入参数格式和输出格式以便 Agent 进行正确的路由和参数校验。错误处理Skill 必须妥善处理内部异常并以结构化的方式向上层报告错误类型和恢复建议。下面是一个 Skill 接口的抽象示例展示了这些框架期望的基本契约from abc import ABC, abstractmethod from typing import Dict, Any class BaseSkill(ABC): Skill 基类定义所有具体 Skill 都需要继承并实现这个接口 property abstractmethod def name(self) - str: 返回 Skill 的唯一标识名称 pass property abstractmethod def description(self) - str: 返回 Skill 的功能描述用于 Agent 决策 pass property abstractmethod def parameters(self) - Dict[str, Any]: 返回 Skill 所需的参数定义 pass abstractmethod def execute(self, **kwargs) - Dict[str, Any]: 执行 Skill 的核心逻辑 pass1.3 企业级项目对 Skill 的额外要求在学习环境中我们可能只关注 Skill 的功能正确性。但在企业级项目中Skill 还需要满足可观测性每个 Skill 的执行都需要有详细的日志记录包括输入参数、执行耗时、成功状态和错误信息。资源管理Skill 需要妥善管理数据库连接、HTTP 会话等资源避免泄漏。安全性对输入参数进行严格的验证和清理防止注入攻击。性能约束设置合理的超时时间避免单个 Skill 阻塞整个 Agent 的工作流。理解了这些基础概念后我们就可以开始搭建开发环境准备构建具体的 Skill 实现。2. 准备开发环境与项目结构Agent 开发涉及多个组件协作一个清晰的项目结构能够避免后期大量的配置冲突和导入错误。建议从一开始就采用模块化的布局而不是将所有代码堆在一个文件中。2.1 环境准备与依赖管理首先确保你的 Python 环境是 3.8 或更高版本。强烈建议使用虚拟环境来隔离依赖# 创建并激活虚拟环境 python -m venv agent_skills_env source agent_skills_env/bin/activate # Linux/Mac # agent_skills_env\Scripts\activate # Windows # 安装核心依赖 pip install openai langchain requests python-dotenv对于企业级项目我们还需要一些增强功能的库# 安装增强依赖 pip install pydantic # 数据验证 pip install tenacity # 重试机制 pip install structlog # 结构化日志 pip install prometheus-client # 指标收集可选创建requirements.txt文件固化依赖版本openai1.3.0 langchain0.0.350 requests2.31.0 python-dotenv1.0.0 pydantic2.0.0 tenacity8.2.0 structlog23.0.02.2 项目结构设计采用以下目录结构便于扩展和维护agent_skills_project/ ├── skills/ # 所有 Skill 实现 │ ├── __init__.py │ ├── base.py # Skill 基类定义 │ ├── weather_skill.py # 示例天气查询 Skill │ ├── data_skill.py # 示例数据处理 Skill │ └── calculator_skill.py # 示例计算器 Skill ├── agents/ # Agent 实现 │ ├── __init__.py │ └── main_agent.py # 主 Agent 逻辑 ├── tools/ # 工具函数库 │ ├── __init__.py │ ├── http_client.py # HTTP 客户端封装 │ └── logger.py # 日志配置 ├── config/ # 配置文件 │ ├── __init__.py │ └── settings.py # 应用配置 ├── tests/ # 测试代码 │ ├── __init__.py │ ├── test_skills.py │ └── test_agent.py ├── scripts/ # 辅助脚本 │ └── setup_env.py # 环境设置脚本 ├── .env.example # 环境变量示例 ├── .gitignore ├── requirements.txt └── main.py # 应用入口2.3 配置管理策略企业级项目必须将配置外置化避免将敏感信息硬编码在代码中。使用.env文件管理环境相关配置# config/settings.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件 class Settings: 应用配置类 # API 密钥配置 OPENAI_API_KEY os.getenv(OPENAI_API_KEY, ) WEATHER_API_KEY os.getenv(WEATHER_API_KEY, ) # 服务端点配置 WEATHER_API_URL os.getenv(WEATHER_API_URL, https://api.weatherapi.com/v1) # 运行时配置 LOG_LEVEL os.getenv(LOG_LEVEL, INFO) REQUEST_TIMEOUT int(os.getenv(REQUEST_TIMEOUT, 30)) # 重试配置 MAX_RETRIES int(os.getenv(MAX_RETRIES, 3)) RETRY_DELAY float(os.getenv(RETRY_DELAY, 1.0)) settings Settings()创建.env.example文件作为模板# .env.example OPENAI_API_KEYyour_openai_api_key_here WEATHER_API_KEYyour_weather_api_key_here WEATHER_API_URLhttps://api.weatherapi.com/v1 LOG_LEVELINFO REQUEST_TIMEOUT30 MAX_RETRIES3 RETRY_DELAY1.0这样的环境准备确保了项目的可移植性和安全性为后续的 Skill 开发奠定了坚实基础。3. 实现核心 Skill 模块有了清晰的项目结构后我们可以开始实现具体的 Skill。本节将构建三个具有代表性的 Skill涵盖外部 API 调用、数据处理和计算等常见场景。3.1 设计可复用的 Skill 基类所有具体 Skill 都应该继承自一个统一的基类这样可以确保接口一致性和公共逻辑复用# skills/base.py from abc import ABC, abstractmethod from typing import Dict, Any, Optional import time import structlog from tenacity import retry, stop_after_attempt, wait_exponential logger structlog.get_logger() class BaseSkill(ABC): Skill 基类封装公共逻辑 def __init__(self): self.logger logger.bind(skill_nameself.name) property abstractmethod def name(self) - str: Skill 唯一名称 pass property abstractmethod def description(self) - str: Skill 功能描述 pass property abstractmethod def parameters_schema(self) - Dict[str, Any]: 参数格式定义用于验证和文档 pass def validate_parameters(self, **kwargs) - bool: 验证输入参数是否符合预期格式 try: for param, schema in self.parameters_schema.items(): if param not in kwargs and schema.get(required, True): raise ValueError(f缺少必需参数: {param}) if param in kwargs and not self._validate_type(kwargs[param], schema.get(type)): raise ValueError(f参数 {param} 类型错误期望 {schema.get(type)}) return True except ValueError as e: self.logger.error(参数验证失败, errorstr(e), paramskwargs) return False def _validate_type(self, value, expected_type) - bool: 简单的类型验证 if expected_type is None: return True return isinstance(value, expected_type) retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def execute(self, **kwargs) - Dict[str, Any]: 执行 Skill 的入口方法包含重试和日志 start_time time.time() try: # 参数验证 if not self.validate_parameters(**kwargs): return { success: False, error: 参数验证失败, data: None } self.logger.info(开始执行 Skill, parameterskwargs) # 调用具体实现 result self._execute(**kwargs) execution_time time.time() - start_time self.logger.info(Skill 执行成功, execution_timeexecution_time, result_typetype(result).__name__) return { success: True, data: result, execution_time: execution_time } except Exception as e: execution_time time.time() - start_time self.logger.error(Skill 执行失败, errorstr(e), execution_timeexecution_time) return { success: False, error: str(e), execution_time: execution_time } abstractmethod def _execute(self, **kwargs) - Any: 子类需要实现的具体逻辑 pass3.2 实现天气查询 Skill这是一个典型的外部 API 调用型 Skill演示了如何集成第三方服务# skills/weather_skill.py import requests from typing import Dict, Any from config.settings import settings from skills.base import BaseSkill class WeatherSkill(BaseSkill): 天气查询 Skill支持根据城市名称查询实时天气 property def name(self) - str: return weather_query property def description(self) - str: return 查询指定城市的实时天气信息 property def parameters_schema(self) - Dict[str, Any]: return { city: { type: str, required: True, description: 城市名称如 北京 或 Shanghai } } def _execute(self, **kwargs) - Dict[str, Any]: city kwargs[city] # 构建 API 请求 params { key: settings.WEATHER_API_KEY, q: city, aqi: no } response requests.get( f{settings.WEATHER_API_URL}/current.json, paramsparams, timeoutsettings.REQUEST_TIMEOUT ) # 检查响应状态 response.raise_for_status() data response.json() # 提取关键信息 return { city: data[location][name], country: data[location][country], temperature: data[current][temp_c], condition: data[current][condition][text], humidity: data[current][humidity], wind_speed: data[current][wind_kph] }3.3 实现数据处理 Skill这个 Skill 演示了如何处理结构化数据适合企业内部的数据处理场景# skills/data_skill.py import pandas as pd from typing import Dict, Any, List from skills.base import BaseSkill class DataAnalysisSkill(BaseSkill): 数据分析 Skill支持基本的数据统计和聚合操作 property def name(self) - str: return data_analysis property def description(self) - str: return 对提供的数据集进行统计分析 property def parameters_schema(self) - Dict[str, Any]: return { data: { type: list, required: True, description: 要分析的数据列表每个元素是字典格式 }, operation: { type: str, required: True, description: 分析操作如 summary, average, max, min }, column: { type: str, required: False, description: 要分析的列名仅对某些操作需要 } } def _execute(self, **kwargs) - Dict[str, Any]: data kwargs[data] operation kwargs[operation] column kwargs.get(column) # 转换为 DataFrame 以便分析 df pd.DataFrame(data) if operation summary: result self._get_summary(df) elif operation average and column: result {average: df[column].mean()} elif operation max and column: result {max: df[column].max()} elif operation min and column: result {min: df[column].min()} else: raise ValueError(f不支持的操作: {operation}) return result def _get_summary(self, df: pd.DataFrame) - Dict[str, Any]: 生成数据集的统计摘要 return { row_count: len(df), columns: list(df.columns), numeric_summary: df.describe().to_dict(), missing_values: df.isnull().sum().to_dict() }3.4 实现计算器 Skill这个相对简单的 Skill 演示了纯逻辑运算的实现模式# skills/calculator_skill.py from typing import Dict, Any from skills.base import BaseSkill class CalculatorSkill(BaseSkill): 计算器 Skill支持基本数学运算 property def name(self) - str: return calculator property def description(self) - str: return 执行基本数学运算 property def parameters_schema(self) - Dict[str, Any]: return { operation: { type: str, required: True, description: 运算类型: add, subtract, multiply, divide }, a: { type: (int, float), required: True, description: 第一个操作数 }, b: { type: (int, float), required: True, description: 第二个操作数 } } def _execute(self, **kwargs) - float: operation kwargs[operation] a kwargs[a] b kwargs[b] if operation add: return a b elif operation subtract: return a - b elif operation multiply: return a * b elif operation divide: if b 0: raise ValueError(除数不能为零) return a / b else: raise ValueError(f不支持的运算: {operation})这三个 Skill 覆盖了不同类型的能力实现为后续的 Agent 集成提供了坚实的基础。4. 构建智能体并集成 Skills有了可用的 Skill 模块后我们需要一个智能的 Agent 来协调这些技能的执行。这个 Agent 需要能够理解用户意图选择正确的 Skill并管理执行流程。4.1 设计主 Agent 的决策逻辑主 Agent 的核心职责是路由决策根据用户输入判断应该调用哪个 Skill并提取相应的参数。# agents/main_agent.py import re from typing import Dict, Any, List, Optional import structlog from skills.weather_skill import WeatherSkill from skills.data_skill import DataAnalysisSkill from skills.calculator_skill import CalculatorSkill logger structlog.get_logger() class MainAgent: 主 Agent负责协调多个 Skill 的调用 def __init__(self): self.logger logger.bind(agent_namemain_agent) self.skills self._register_skills() self.intent_patterns self._build_intent_patterns() def _register_skills(self) - Dict[str, Any]: 注册所有可用的 Skill return { weather: WeatherSkill(), data_analysis: DataAnalysisSkill(), calculator: CalculatorSkill() } def _build_intent_patterns(self) - Dict[str, Any]: 构建意图识别模式 return { weather: [ r(查询|查看|获取).*天气, r.*天气.*怎么样, r今天.*天气 ], data_analysis: [ r分析.*数据, r统计.*数据, r计算.*平均, r找出.*最大 ], calculator: [ r计算.*, r.*加.*, r.*减.*, r.*乘.*, r.*除.* ] } def detect_intent(self, user_input: str) - Optional[str]: 检测用户意图返回对应的 Skill 名称 for skill_name, patterns in self.intent_patterns.items(): for pattern in patterns: if re.search(pattern, user_input, re.IGNORECASE): self.logger.info(检测到意图, intentskill_name, patternpattern, inputuser_input) return skill_name return None def extract_parameters(self, skill_name: str, user_input: str) - Dict[str, Any]: 从用户输入中提取 Skill 所需的参数 if skill_name weather: return self._extract_weather_params(user_input) elif skill_name calculator: return self._extract_calculator_params(user_input) elif skill_name data_analysis: return self._extract_data_params(user_input) else: return {} def _extract_weather_params(self, user_input: str) - Dict[str, Any]: 提取天气查询参数 # 简单的城市名称提取逻辑 city_pattern r([北京上海广州深圳杭州成都武汉西安南京天津]|[\u4e00-\u9fa5]{2,5}市?) match re.search(city_pattern, user_input) if match: return {city: match.group(1)} return {city: 北京} # 默认值 def _extract_calculator_params(self, user_input: str) - Dict[str, Any]: 提取计算器参数 # 简化版的参数提取实际项目可以使用更复杂的 NLP 技术 if 加 in user_input: nums re.findall(r\d, user_input) if len(nums) 2: return {operation: add, a: int(nums[0]), b: int(nums[1])} elif 乘 in user_input: nums re.findall(r\d, user_input) if len(nums) 2: return {operation: multiply, a: int(nums[0]), b: int(nums[1])} return {operation: add, a: 0, b: 0} # 默认值 def _extract_data_params(self, user_input: str) - Dict[str, Any]: 提取数据分析参数 # 这里简化处理实际项目需要更复杂的逻辑 return {operation: summary, data: []} def process(self, user_input: str) - Dict[str, Any]: 处理用户输入的主流程 self.logger.info(收到用户输入, inputuser_input) # 1. 意图识别 skill_name self.detect_intent(user_input) if not skill_name: return { success: False, error: 无法识别您的意图请尝试更清晰的表达, suggested_skills: list(self.skills.keys()) } # 2. 参数提取 parameters self.extract_parameters(skill_name, user_input) self.logger.info(提取到参数, skillskill_name, parametersparameters) # 3. 执行 Skill skill self.skills[skill_name] result skill.execute(**parameters) # 4. 格式化响应 return self._format_response(skill_name, result) def _format_response(self, skill_name: str, skill_result: Dict[str, Any]) - Dict[str, Any]: 格式化最终响应 if skill_result[success]: return { success: True, skill_used: skill_name, data: skill_result[data], execution_time: skill_result.get(execution_time, 0) } else: return { success: False, skill_used: skill_name, error: skill_result.get(error, 未知错误), execution_time: skill_result.get(execution_time, 0) }4.2 创建应用入口和交互界面现在我们需要一个简单的方式来测试整个系统# main.py import sys from agents.main_agent import MainAgent def main(): 主函数提供简单的命令行交互界面 agent MainAgent() print(Agent Skills 系统已启动) print(支持的功能: 天气查询、数据分析、数学计算) print(输入 quit 或 exit 退出程序) print(- * 50) while True: try: user_input input(\n请输入您的需求: ).strip() if user_input.lower() in [quit, exit]: print(感谢使用再见) break if not user_input: continue # 处理用户输入 result agent.process(user_input) # 显示结果 if result[success]: print(f✓ 执行成功 (耗时: {result[execution_time]:.2f}s)) print(f使用的技能: {result[skill_used]}) print(结果:, result[data]) else: print(f✗ 执行失败: {result[error]}) if suggested_skills in result: print(建议尝试:, , .join(result[suggested_skills])) except KeyboardInterrupt: print(\n\n程序被用户中断) break except Exception as e: print(f系统错误: {e}) continue if __name__ __main__: main()4.3 运行和测试系统现在可以启动系统进行测试# 确保虚拟环境已激活且依赖已安装 python main.py测试示例交互请输入您的需求: 今天北京天气怎么样 ✓ 执行成功 (耗时: 1.23s) 使用的技能: weather 结果: {city: 北京, country: China, temperature: 15.0, condition: 晴朗, humidity: 45, wind_speed: 12.0} 请输入您的需求: 计算25乘以38 ✓ 执行成功 (耗时: 0.01s) 使用的技能: calculator 结果: 950.0 请输入您的需求: 帮我分析一些数据 ✗ 执行失败: 参数验证失败这个基础版本已经展示了 Agent 和 Skill 协同工作的核心机制。在实际项目中意图识别和参数提取部分通常会使用更先进的 NLP 技术但基本架构是相似的。5. 企业级部署与运维考量将 Agent 系统部署到生产环境需要考虑更多工程因素。本节将讨论监控、安全、性能优化等关键主题。5.1 日志与监控体系建设完善的日志系统是排查生产问题的关键。我们使用结构化日志来确保信息可检索# tools/logger.py import structlog import logging import sys def setup_logging(level: str INFO): 配置结构化日志 # 基础日志配置 logging.basicConfig( format%(message)s, streamsys.stdout, levelgetattr(logging, level.upper()) ) # Structlog 配置 structlog.configure( processors[ structlog.stdlib.filter_by_level, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.processors.TimeStamper(fmtiso), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.UnicodeDecoder(), structlog.processors.JSONRenderer() ], context_classdict, logger_factorystructlog.stdlib.LoggerFactory(), wrapper_classstructlog.stdlib.BoundLogger, cache_logger_on_first_useTrue, )在生产环境中还应该集成指标收集# tools/metrics.py from prometheus_client import Counter, Histogram, Gauge # 定义监控指标 skill_execution_total Counter(skill_executions_total, Total skill executions, [skill_name, status]) skill_execution_duration Histogram(skill_execution_duration_seconds, Skill execution duration, [skill_name]) active_skills Gauge(active_skills, Number of active skills)5.2 安全加固措施企业级 Agent 系统必须考虑安全性# tools/security.py import re from typing import Any def sanitize_input(user_input: str) - str: 清理用户输入防止注入攻击 # 移除潜在的恶意字符 cleaned re.sub(r[;|$], , user_input) # 限制长度 return cleaned[:1000] def validate_parameters(schema: dict, params: dict) - bool: 严格的参数验证 for key, config in schema.items(): if config.get(required, False) and key not in params: return False if key in params: value params[key] expected_type config.get(type) # 类型检查 if expected_type and not isinstance(value, expected_type): return False # 范围检查如果定义了范围 if min in config and value config[min]: return False if max in config and value config[max]: return False # 枚举值检查 if enum in config and value not in config[enum]: return False return True5.3 性能优化策略针对高并发场景的性能优化# tools/cache.py import time from typing import Any, Optional from functools import lru_cache class SkillCache: Skill 结果缓存减少重复计算和 API 调用 def __init__(self, max_size: int 1000, default_ttl: int 300): self._cache {} self.max_size max_size self.default_ttl default_ttl def get(self, key: str) - Optional[Any]: 获取缓存值 if key in self._cache: value, timestamp self._cache[key] if time.time() - timestamp self.default_ttl: return value else: # 缓存过期删除 del self._cache[key] return None def set(self, key: str, value: Any, ttl: Optional[int] None): 设置缓存值 if len(self._cache) self.max_size: # 简单的 LRU 策略删除最旧的一个项目 oldest_key next(iter(self._cache)) del self._cache[oldest_key] actual_ttl ttl or self.default_ttl self._cache[key] (value, time.time() actual_ttl)5.4 配置健康检查端点对于部署在云环境的服务需要提供健康检查# tools/health.py from typing import Dict, Any class HealthChecker: 系统健康检查 def __init__(self, skills: dict): self.skills skills def check_skill_health(self) - Dict[str, Any]: 检查所有 Skill 的健康状态 results {} for name, skill in self.skills.items(): try: # 简单的健康检查尝试执行一个无副作用的操作 if hasattr(skill, health_check): result skill.health_check() else: # 默认健康检查逻辑 result {status: healthy, details: Basic check passed} results[name] result except Exception as e: results[name] {status: unhealthy, error: str(e)} return results def overall_health(self) - Dict[str, Any]: 整体健康状态 skill_health self.check_skill_health() healthy_count sum(1 for result in skill_health.values() if result.get(status) healthy) return { status: healthy if healthy_count len(self.skills) else degraded, skill_count: len(self.skills), healthy_skills: healthy_count, details: skill_health }6. 常见问题排查与调试技巧在实际开发和运维过程中会遇到各种问题。本节提供系统化的排查方法。6.1 Skill 执行失败排查路径当 Skill 执行失败时按照以下顺序排查问题现象可能原因检查方式解决方案Skill 找不到或未注册Skill 名称错误或未正确注册检查_register_skills方法确保 Skill 类已导入并正确注册参数验证失败参数缺失、类型错误或格式不符查看 Skill 的parameters_schema调整参数提取逻辑或用户输入提示API 调用超时网络问题或服务端响应慢检查超时设置和网络连接增加超时时间或添加重试机制外部服务返回错误API 密钥无效或服务不可用查看错误响应和状态码验证 API 配置或检查服务状态内存使用过高数据处理过大或内存泄漏监控内存使用情况优化数据处理逻辑或添加分页6.2 日志分析技巧有效的日志分析可以快速定位问题# 在 Skill 基类中添加详细的调试日志 def execute(self, **kwargs) - Dict[str, Any]: self.logger.debug(开始参数验证, parameterskwargs) # ... 执行逻辑 self.logger.debug(Skill 执行完成, result_previewstr(result)[:200])关键日志搜索模式错误日志grep ERROR app.log特定 Skill 日志grep skill_nameweather app.log慢查询grep execution_time app.log | awk $NF 5.0参数问题grep 参数验证失败 app.log6.3 性能瓶颈识别使用以下方法识别性能瓶颈import cProfile import pstats def profile_skill_execution(): 性能分析工具 agent MainAgent() profiler cProfile.Profile() profiler.enable() # 执行测试用例 for _ in range(100): agent.process(北京天气怎么样) profiler.disable() # 输出分析结果 stats pstats.Stats(profiler) stats.sort_stats(cumulative) stats.print_stats(10) # 显示前10个最耗时的函数常见的性能优化点减少不必要的 API 调用通过缓存优化数据序列化/反序列化使用异步执行提高并发能力批量处理数据减少 I/O 操作6.4 测试策略建议建立完整的测试体系确保系统可靠性# tests/test_skills.py import pytest from skills.weather_skill import WeatherSkill class TestWeatherSkill: def setup_method(self): self.skill WeatherSkill() def test_parameter_validation(self): 测试参数验证逻辑 # 测试必需参数缺失 result self.skill.execute() # 不传参数 assert result[success] is False assert 缺少必需参数 in result[error] def test_skill_description(self): 测试 Skill 元信息 assert self.skill.name weather_query assert 天气 in