这次我们来深入探讨AI Agent开发中的两个核心概念Skill技能和Token令牌。如果你正在学习Agent开发或者想要理解如何让AI Agent具备特定能力这篇文章将带你从零开始掌握Skill的编写方法和Token的工作原理。在AI Agent体系中Skill是Agent能够执行的具体任务能力而Token则是控制访问和计算资源的关键凭证。理解这两者的关系是构建实用Agent系统的基石。本文将通过实际案例演示如何编写Skill、如何管理Token以及如何让Agent真正理解你的指令。1. 核心能力速览能力项说明Skill类型文本处理、数据分析、API调用、文件操作、自动化任务等Token作用身份验证、API调用配额、资源访问控制、计费单位开发门槛需要基础编程知识熟悉Python或JavaScript更佳测试环境本地开发环境或云平台测试环境适用场景智能助手、自动化流程、数据分析、内容生成等2. Agent开发基础Skill与Token的关系在AI Agent生态中Skill和Token是相辅相成的两个核心组件。Skill定义了Agent能做什么而Token则决定了Agent被允许做什么以及能做多少。2.1 什么是SkillSkill是Agent执行特定任务的能力单元。每个Skill都包含三个基本要素意图识别理解用户想要执行什么操作参数提取从用户输入中提取执行任务所需的信息动作执行调用相应的API或执行具体操作例如一个天气查询Skill需要识别用户询问天气的意图提取地点参数然后调用天气API返回结果。2.2 什么是TokenToken在Agent开发中有多重含义API访问令牌用于身份验证和授权计算资源单位衡量AI模型处理文本的复杂度会话标识维持对话上下文的一致性理解Token的不同含义有助于避免开发过程中的常见错误。3. 环境准备与开发工具开始编写Skill前需要准备合适的开发环境。以下是推荐的配置方案3.1 基础开发环境# 创建Python虚拟环境 python -m venv agent_env source agent_env/bin/activate # Linux/Mac # 或 agent_env\Scripts\activate # Windows # 安装核心依赖 pip install openai pip install requests pip install python-dotenv3.2 常用Agent开发框架根据项目需求选择合适的框架LangChain功能全面的Agent开发框架AutoGPT专注于自主任务执行的Agent自定义框架针对特定需求的轻量级解决方案# 基础环境检查脚本 import sys import requests def check_environment(): print(fPython版本: {sys.version}) try: response requests.get(https://httpbin.org/get, timeout5) print(网络连接: 正常) except: print(网络连接: 异常) if __name__ __main__: check_environment()4. 编写你的第一个Skill让我们通过一个实际的例子来学习Skill的编写方法。我们将创建一个时间查询Skill它能够回答当前时间或指定时区的时间。4.1 Skill基本结构import datetime import pytz from typing import Dict, Any class TimeSkill: def __init__(self): self.skill_name 时间查询 self.description 查询当前时间或指定时区的时间 def recognize_intent(self, user_input: str) - bool: 识别用户是否想要查询时间 time_keywords [时间, 几点, 钟点, 现在几点] return any(keyword in user_input for keyword in time_keywords) def extract_parameters(self, user_input: str) - Dict[str, Any]: 从用户输入中提取参数 parameters {} # 提取时区信息 if 北京时间 in user_input: parameters[timezone] Asia/Shanghai elif 纽约时间 in user_input: parameters[timezone] America/New_York else: parameters[timezone] local # 默认本地时间 return parameters def execute(self, parameters: Dict[str, Any]) - str: 执行时间查询操作 timezone parameters.get(timezone, local) if timezone local: current_time datetime.datetime.now() return f当前时间是: {current_time.strftime(%Y-%m-%d %H:%M:%S)} else: try: tz pytz.timezone(timezone) current_time datetime.datetime.now(tz) return f{timezone} 当前时间是: {current_time.strftime(%Y-%m-%d %H:%M:%S)} except pytz.UnknownTimeZoneError: return 抱歉我不认识这个时区 def process(self, user_input: str) - str: 完整的Skill处理流程 if not self.recognize_intent(user_input): return 这不是时间查询请求 parameters self.extract_parameters(user_input) result self.execute(parameters) return result # 测试Skill if __name__ __main__: time_skill TimeSkill() test_inputs [ 现在几点了, 查询北京时间, 纽约现在几点钟 ] for test_input in test_inputs: print(f输入: {test_input}) print(f输出: {time_skill.process(test_input)}) print(- * 50)4.2 Skill的进阶特性一个成熟的Skill还应该包含以下特性class AdvancedTimeSkill(TimeSkill): def __init__(self): super().__init__() self.supported_timezones [ Asia/Shanghai, America/New_York, Europe/London, Asia/Tokyo ] def validate_parameters(self, parameters: Dict[str, Any]) - bool: 验证参数有效性 timezone parameters.get(timezone) if timezone ! local and timezone not in self.supported_timezones: return False return True def get_skill_info(self) - Dict[str, Any]: 返回Skill的元信息 return { name: self.skill_name, description: self.description, version: 1.0, author: Your Name }5. Token的管理与使用Token管理是Agent开发中的关键环节。不当的Token管理会导致API调用失败、资源浪费或安全风险。5.1 API Token的配置与管理import os from dotenv import load_dotenv import hashlib class TokenManager: def __init__(self): load_dotenv() # 加载环境变量 self.tokens {} def load_tokens(self): 从环境变量加载Token self.tokens[openai] os.getenv(OPENAI_API_KEY) self.tokens[weather] os.getenv(WEATHER_API_KEY) # 添加更多服务的Token def validate_token(self, service_name: str) - bool: 验证Token有效性 token self.tokens.get(service_name) if not token: print(f未找到 {service_name} 的Token) return False # 基础格式验证 if len(token) 10: # 假设Token至少10个字符 print(f{service_name} Token格式异常) return False return True def get_token_hash(self, service_name: str) - str: 获取Token的哈希值用于日志记录不暴露真实Token token self.tokens.get(service_name, ) return hashlib.md5(token.encode()).hexdigest()[:8] # 使用示例 token_manager TokenManager() token_manager.load_tokens() if token_manager.validate_token(openai): print(OpenAI Token有效) else: print(OpenAI Token无效或未配置)5.2 Token的安全最佳实践import keyring # 用于安全存储密码 class SecureTokenManager: def __init__(self, service_name: str): self.service_name service_name def store_token(self, token: str): 安全存储Token keyring.set_password(self.service_name, api_token, token) def retrieve_token(self) - str: 安全获取Token return keyring.get_password(self.service_name, api_token) def clear_token(self): 清除存储的Token keyring.delete_password(self.service_name, api_token) # 环境变量配置示例 (.env文件) # API Tokens OPENAI_API_KEYsk-your-openai-token-here WEATHER_API_KEYyour-weather-api-key CUSTOM_SERVICE_TOKENyour-custom-token # 配置参数 API_TIMEOUT30 MAX_RETRIES3 6. Skill与Token的集成实战现在我们将Skill和Token结合起来创建一个实际可用的天气查询Skill。6.1 天气查询Skill实现import requests import json from typing import Dict, Any class WeatherSkill: def __init__(self, token_manager): self.skill_name 天气查询 self.token_manager token_manager self.base_url https://api.weatherapi.com/v1 def recognize_intent(self, user_input: str) - bool: weather_keywords [天气, 气温, 温度, 天气预报] return any(keyword in user_input for keyword in weather_keywords) def extract_parameters(self, user_input: str) - Dict[str, Any]: parameters {} # 简单的地点提取逻辑实际项目可以使用NLP模型 locations [北京, 上海, 广州, 深圳, 纽约, 伦敦] for location in locations: if location in user_input: parameters[location] location break if location not in parameters: parameters[location] 北京 # 默认地点 return parameters def execute(self, parameters: Dict[str, Any]) - str: 调用天气API查询天气 location parameters.get(location, 北京) api_key self.token_manager.tokens.get(weather) if not api_key: return 天气服务暂不可用 try: url f{self.base_url}/current.json params { key: api_key, q: location, lang: zh } response requests.get(url, paramsparams, timeout10) if response.status_code 200: data response.json() current data[current] result f{location}天气{current[condition][text]}\n result f温度{current[temp_c]}°C\n result f湿度{current[humidity]}%\n result f风速{current[wind_kph]} km/h return result else: return f天气查询失败错误代码{response.status_code} except requests.exceptions.Timeout: return 天气查询超时请稍后重试 except Exception as e: return f天气查询出错{str(e)} def process(self, user_input: str) - str: if not self.recognize_intent(user_input): return 这不是天气查询请求 parameters self.extract_parameters(user_input) return self.execute(parameters) # 集成测试 token_manager TokenManager() token_manager.load_tokens() weather_skill WeatherSkill(token_manager) test_queries [ 北京天气怎么样, 查询上海气温, 今天纽约的天气 ] for query in test_queries: print(f用户: {query}) print(fAgent: {weather_skill.process(query)}) print()7. Token限额与用量监控在实际应用中需要监控Token的使用情况避免超出限额。7.1 Token用量监控器import time from datetime import datetime, timedelta class TokenUsageMonitor: def __init__(self, limits: Dict[str, int]): limits: 服务名称到每分钟限额的映射 {openai: 100, weather: 50} self.limits limits self.usage {service: [] for service in limits.keys()} def record_usage(self, service_name: str, tokens_used: int 1): 记录Token使用情况 current_time time.time() if service_name in self.usage: self.usage[service_name].append((current_time, tokens_used)) # 清理过期的使用记录保留最近1小时 one_hour_ago current_time - 3600 self.usage[service_name] [ record for record in self.usage[service_name] if record[0] one_hour_ago ] def check_limit(self, service_name: str) - bool: 检查是否超过限额 if service_name not in self.limits: return True current_time time.time() one_minute_ago current_time - 60 recent_usage [ tokens for timestamp, tokens in self.usage.get(service_name, []) if timestamp one_minute_ago ] total_recent_usage sum(recent_usage) return total_recent_usage self.limits[service_name] def get_usage_statistics(self) - Dict[str, Dict]: 获取使用统计 stats {} current_time time.time() for service_name, records in self.usage.items(): one_minute_ago current_time - 60 one_hour_ago current_time - 3600 minute_usage sum(tokens for timestamp, tokens in records if timestamp one_minute_ago) hour_usage sum(tokens for timestamp, tokens in records if timestamp one_hour_ago) stats[service_name] { minute_usage: minute_usage, hour_usage: hour_usage, limit: self.limits.get(service_name, 0), within_limit: minute_usage self.limits.get(service_name, float(inf)) } return stats # 使用示例 monitor TokenUsageMonitor({openai: 100, weather: 30}) # 模拟API调用 for i in range(10): if monitor.check_limit(openai): monitor.record_usage(openai, 10) print(f调用 {i1}: 成功) else: print(f调用 {i1}: 超过限额等待...) time.sleep(1) print(使用统计:, monitor.get_usage_statistics())8. 高级Skill开发技巧8.1 技能组合与工作流复杂的任务往往需要多个Skill协同工作class SkillOrchestrator: def __init__(self): self.skills [] def register_skill(self, skill): 注册Skill self.skills.append(skill) def process_query(self, user_input: str) - str: 处理用户查询自动选择合适的Skill # 首先尝试精确匹配 for skill in self.skills: if skill.recognize_intent(user_input): return skill.process(user_input) # 如果没有精确匹配使用相似度匹配 best_match None best_score 0 for skill in self.skills: # 简单的关键词匹配评分实际可以使用更复杂的NLP模型 score sum(1 for keyword in getattr(skill, keywords, []) if keyword in user_input) if score best_score: best_score score best_match skill if best_match and best_score 0: return best_match.process(user_input) return 抱歉我没有理解您的请求 # 创建技能编排器 orchestrator SkillOrchestrator() orchestrator.register_skill(TimeSkill()) orchestrator.register_skill(WeatherSkill(token_manager)) # 测试技能组合 test_queries [ 现在几点了, 北京天气怎么样, 今天有什么新闻 # 这个查询没有对应的Skill ] for query in test_queries: response orchestrator.process_query(query) print(fQ: {query}) print(fA: {response}\n)8.2 错误处理与重试机制class RobustSkill(WeatherSkill): def __init__(self, token_manager, max_retries3): super().__init__(token_manager) self.max_retries max_retries def execute_with_retry(self, parameters: Dict[str, Any]) - str: 带重试机制的技能执行 for attempt in range(self.max_retries): try: result self.execute(parameters) return result except requests.exceptions.RequestException as e: if attempt self.max_retries - 1: return f服务暂时不可用请稍后重试。错误: {str(e)} print(f第{attempt 1}次尝试失败等待重试...) time.sleep(2 ** attempt) # 指数退避 return 服务调用失败 def process(self, user_input: str) - str: if not self.recognize_intent(user_input): return 这不是天气查询请求 parameters self.extract_parameters(user_input) return self.execute_with_retry(parameters)9. 常见问题与排查方法在Skill开发和Token管理过程中经常会遇到各种问题。以下是常见问题的解决方案问题现象可能原因排查方式解决方案Skill无法识别意图关键词不匹配或过于简单检查recognize_intent方法增加同义词使用NLP模型改进意图识别API调用返回403错误Token无效或过期验证Token格式和有效期重新生成Token检查权限设置响应速度慢网络延迟或API限流监控响应时间检查用量统计实现缓存机制优化重试策略内存使用过高技能实例未正确释放使用内存分析工具实现资源清理使用上下文管理器Token泄漏风险Token硬编码在代码中代码安全审查使用环境变量或安全存储9.1 调试技巧class DebuggableSkill(TimeSkill): def __init__(self, debugFalse): super().__init__() self.debug debug def process(self, user_input: str) - str: if self.debug: print(f[DEBUG] 输入: {user_input}) intent_recognized self.recognize_intent(user_input) if self.debug: print(f[DEBUG] 意图识别: {intent_recognized}) if not intent_recognized: return 这不是时间查询请求 parameters self.extract_parameters(user_input) if self.debug: print(f[DEBUG] 提取参数: {parameters}) result self.execute(parameters) if self.debug: print(f[DEBUG] 执行结果: {result}) return result # 启用调试模式 debug_skill DebuggableSkill(debugTrue) result debug_skill.process(现在北京时间几点)10. 性能优化与最佳实践10.1 Skill性能优化import functools import time from cachetools import TTLCache class OptimizedSkill(WeatherSkill): def __init__(self, token_manager, cache_ttl300): # 5分钟缓存 super().__init__(token_manager) self.cache TTLCache(maxsize100, ttlcache_ttl) functools.lru_cache(maxsize50) def recognize_intent_cached(self, user_input: str) - bool: 带缓存的意图识别 return super().recognize_intent(user_input) def execute(self, parameters: Dict[str, Any]) - str: # 生成缓存键 cache_key fweather_{parameters.get(location, default)} # 检查缓存 if cache_key in self.cache: return self.cache[cache_key] # 执行查询 result super().execute(parameters) # 缓存结果 if 失败 not in result and 错误 not in result: self.cache[cache_key] result return result # 性能测试 def benchmark_skill(skill, queries, iterations100): start_time time.time() for i in range(iterations): for query in queries: skill.process(query) end_time time.time() return end_time - start_time # 比较优化前后的性能 basic_skill WeatherSkill(token_manager) optimized_skill OptimizedSkill(token_manager) test_queries [北京天气, 上海天气, 广州天气] basic_time benchmark_skill(basic_skill, test_queries, 10) optimized_time benchmark_skill(optimized_skill, test_queries, 10) print(f基础技能耗时: {basic_time:.2f}秒) print(f优化技能耗时: {optimized_time:.2f}秒) print(f性能提升: {(basic_time - optimized_time) / basic_time * 100:.1f}%)10.2 安全最佳实践Token安全永远不要将Token提交到版本控制系统使用环境变量或密钥管理服务定期轮换Token输入验证对所有用户输入进行验证和清理防止注入攻击限制输入长度和格式错误处理不要向用户暴露敏感错误信息记录详细的调试日志实现适当的错误恢复机制通过本文的实践指导你应该已经掌握了Skill编写和Token管理的基本方法。这些技能是构建实用AI Agent的基础也是进一步学习高级Agent开发的前提。在实际项目中建议从简单的Skill开始逐步增加复杂度同时建立完善的Token管理和监控体系。记住一个好的Agent系统不仅要有强大的功能更要有稳定的性能和可靠的安全保障。
这次我们来深入探讨AI Agent开发中的两个核心概念Skill技能和Token令牌。如果你正在学习Agent开发或者想要理解如何让AI Agent具备特定能力这篇文章将带你从零开始掌握Skill的编写方法和Token的工作原理。在AI Agent体系中Skill是Agent能够执行的具体任务能力而Token则是控制访问和计算资源的关键凭证。理解这两者的关系是构建实用Agent系统的基石。本文将通过实际案例演示如何编写Skill、如何管理Token以及如何让Agent真正理解你的指令。1. 核心能力速览能力项说明Skill类型文本处理、数据分析、API调用、文件操作、自动化任务等Token作用身份验证、API调用配额、资源访问控制、计费单位开发门槛需要基础编程知识熟悉Python或JavaScript更佳测试环境本地开发环境或云平台测试环境适用场景智能助手、自动化流程、数据分析、内容生成等2. Agent开发基础Skill与Token的关系在AI Agent生态中Skill和Token是相辅相成的两个核心组件。Skill定义了Agent能做什么而Token则决定了Agent被允许做什么以及能做多少。2.1 什么是SkillSkill是Agent执行特定任务的能力单元。每个Skill都包含三个基本要素意图识别理解用户想要执行什么操作参数提取从用户输入中提取执行任务所需的信息动作执行调用相应的API或执行具体操作例如一个天气查询Skill需要识别用户询问天气的意图提取地点参数然后调用天气API返回结果。2.2 什么是TokenToken在Agent开发中有多重含义API访问令牌用于身份验证和授权计算资源单位衡量AI模型处理文本的复杂度会话标识维持对话上下文的一致性理解Token的不同含义有助于避免开发过程中的常见错误。3. 环境准备与开发工具开始编写Skill前需要准备合适的开发环境。以下是推荐的配置方案3.1 基础开发环境# 创建Python虚拟环境 python -m venv agent_env source agent_env/bin/activate # Linux/Mac # 或 agent_env\Scripts\activate # Windows # 安装核心依赖 pip install openai pip install requests pip install python-dotenv3.2 常用Agent开发框架根据项目需求选择合适的框架LangChain功能全面的Agent开发框架AutoGPT专注于自主任务执行的Agent自定义框架针对特定需求的轻量级解决方案# 基础环境检查脚本 import sys import requests def check_environment(): print(fPython版本: {sys.version}) try: response requests.get(https://httpbin.org/get, timeout5) print(网络连接: 正常) except: print(网络连接: 异常) if __name__ __main__: check_environment()4. 编写你的第一个Skill让我们通过一个实际的例子来学习Skill的编写方法。我们将创建一个时间查询Skill它能够回答当前时间或指定时区的时间。4.1 Skill基本结构import datetime import pytz from typing import Dict, Any class TimeSkill: def __init__(self): self.skill_name 时间查询 self.description 查询当前时间或指定时区的时间 def recognize_intent(self, user_input: str) - bool: 识别用户是否想要查询时间 time_keywords [时间, 几点, 钟点, 现在几点] return any(keyword in user_input for keyword in time_keywords) def extract_parameters(self, user_input: str) - Dict[str, Any]: 从用户输入中提取参数 parameters {} # 提取时区信息 if 北京时间 in user_input: parameters[timezone] Asia/Shanghai elif 纽约时间 in user_input: parameters[timezone] America/New_York else: parameters[timezone] local # 默认本地时间 return parameters def execute(self, parameters: Dict[str, Any]) - str: 执行时间查询操作 timezone parameters.get(timezone, local) if timezone local: current_time datetime.datetime.now() return f当前时间是: {current_time.strftime(%Y-%m-%d %H:%M:%S)} else: try: tz pytz.timezone(timezone) current_time datetime.datetime.now(tz) return f{timezone} 当前时间是: {current_time.strftime(%Y-%m-%d %H:%M:%S)} except pytz.UnknownTimeZoneError: return 抱歉我不认识这个时区 def process(self, user_input: str) - str: 完整的Skill处理流程 if not self.recognize_intent(user_input): return 这不是时间查询请求 parameters self.extract_parameters(user_input) result self.execute(parameters) return result # 测试Skill if __name__ __main__: time_skill TimeSkill() test_inputs [ 现在几点了, 查询北京时间, 纽约现在几点钟 ] for test_input in test_inputs: print(f输入: {test_input}) print(f输出: {time_skill.process(test_input)}) print(- * 50)4.2 Skill的进阶特性一个成熟的Skill还应该包含以下特性class AdvancedTimeSkill(TimeSkill): def __init__(self): super().__init__() self.supported_timezones [ Asia/Shanghai, America/New_York, Europe/London, Asia/Tokyo ] def validate_parameters(self, parameters: Dict[str, Any]) - bool: 验证参数有效性 timezone parameters.get(timezone) if timezone ! local and timezone not in self.supported_timezones: return False return True def get_skill_info(self) - Dict[str, Any]: 返回Skill的元信息 return { name: self.skill_name, description: self.description, version: 1.0, author: Your Name }5. Token的管理与使用Token管理是Agent开发中的关键环节。不当的Token管理会导致API调用失败、资源浪费或安全风险。5.1 API Token的配置与管理import os from dotenv import load_dotenv import hashlib class TokenManager: def __init__(self): load_dotenv() # 加载环境变量 self.tokens {} def load_tokens(self): 从环境变量加载Token self.tokens[openai] os.getenv(OPENAI_API_KEY) self.tokens[weather] os.getenv(WEATHER_API_KEY) # 添加更多服务的Token def validate_token(self, service_name: str) - bool: 验证Token有效性 token self.tokens.get(service_name) if not token: print(f未找到 {service_name} 的Token) return False # 基础格式验证 if len(token) 10: # 假设Token至少10个字符 print(f{service_name} Token格式异常) return False return True def get_token_hash(self, service_name: str) - str: 获取Token的哈希值用于日志记录不暴露真实Token token self.tokens.get(service_name, ) return hashlib.md5(token.encode()).hexdigest()[:8] # 使用示例 token_manager TokenManager() token_manager.load_tokens() if token_manager.validate_token(openai): print(OpenAI Token有效) else: print(OpenAI Token无效或未配置)5.2 Token的安全最佳实践import keyring # 用于安全存储密码 class SecureTokenManager: def __init__(self, service_name: str): self.service_name service_name def store_token(self, token: str): 安全存储Token keyring.set_password(self.service_name, api_token, token) def retrieve_token(self) - str: 安全获取Token return keyring.get_password(self.service_name, api_token) def clear_token(self): 清除存储的Token keyring.delete_password(self.service_name, api_token) # 环境变量配置示例 (.env文件) # API Tokens OPENAI_API_KEYsk-your-openai-token-here WEATHER_API_KEYyour-weather-api-key CUSTOM_SERVICE_TOKENyour-custom-token # 配置参数 API_TIMEOUT30 MAX_RETRIES3 6. Skill与Token的集成实战现在我们将Skill和Token结合起来创建一个实际可用的天气查询Skill。6.1 天气查询Skill实现import requests import json from typing import Dict, Any class WeatherSkill: def __init__(self, token_manager): self.skill_name 天气查询 self.token_manager token_manager self.base_url https://api.weatherapi.com/v1 def recognize_intent(self, user_input: str) - bool: weather_keywords [天气, 气温, 温度, 天气预报] return any(keyword in user_input for keyword in weather_keywords) def extract_parameters(self, user_input: str) - Dict[str, Any]: parameters {} # 简单的地点提取逻辑实际项目可以使用NLP模型 locations [北京, 上海, 广州, 深圳, 纽约, 伦敦] for location in locations: if location in user_input: parameters[location] location break if location not in parameters: parameters[location] 北京 # 默认地点 return parameters def execute(self, parameters: Dict[str, Any]) - str: 调用天气API查询天气 location parameters.get(location, 北京) api_key self.token_manager.tokens.get(weather) if not api_key: return 天气服务暂不可用 try: url f{self.base_url}/current.json params { key: api_key, q: location, lang: zh } response requests.get(url, paramsparams, timeout10) if response.status_code 200: data response.json() current data[current] result f{location}天气{current[condition][text]}\n result f温度{current[temp_c]}°C\n result f湿度{current[humidity]}%\n result f风速{current[wind_kph]} km/h return result else: return f天气查询失败错误代码{response.status_code} except requests.exceptions.Timeout: return 天气查询超时请稍后重试 except Exception as e: return f天气查询出错{str(e)} def process(self, user_input: str) - str: if not self.recognize_intent(user_input): return 这不是天气查询请求 parameters self.extract_parameters(user_input) return self.execute(parameters) # 集成测试 token_manager TokenManager() token_manager.load_tokens() weather_skill WeatherSkill(token_manager) test_queries [ 北京天气怎么样, 查询上海气温, 今天纽约的天气 ] for query in test_queries: print(f用户: {query}) print(fAgent: {weather_skill.process(query)}) print()7. Token限额与用量监控在实际应用中需要监控Token的使用情况避免超出限额。7.1 Token用量监控器import time from datetime import datetime, timedelta class TokenUsageMonitor: def __init__(self, limits: Dict[str, int]): limits: 服务名称到每分钟限额的映射 {openai: 100, weather: 50} self.limits limits self.usage {service: [] for service in limits.keys()} def record_usage(self, service_name: str, tokens_used: int 1): 记录Token使用情况 current_time time.time() if service_name in self.usage: self.usage[service_name].append((current_time, tokens_used)) # 清理过期的使用记录保留最近1小时 one_hour_ago current_time - 3600 self.usage[service_name] [ record for record in self.usage[service_name] if record[0] one_hour_ago ] def check_limit(self, service_name: str) - bool: 检查是否超过限额 if service_name not in self.limits: return True current_time time.time() one_minute_ago current_time - 60 recent_usage [ tokens for timestamp, tokens in self.usage.get(service_name, []) if timestamp one_minute_ago ] total_recent_usage sum(recent_usage) return total_recent_usage self.limits[service_name] def get_usage_statistics(self) - Dict[str, Dict]: 获取使用统计 stats {} current_time time.time() for service_name, records in self.usage.items(): one_minute_ago current_time - 60 one_hour_ago current_time - 3600 minute_usage sum(tokens for timestamp, tokens in records if timestamp one_minute_ago) hour_usage sum(tokens for timestamp, tokens in records if timestamp one_hour_ago) stats[service_name] { minute_usage: minute_usage, hour_usage: hour_usage, limit: self.limits.get(service_name, 0), within_limit: minute_usage self.limits.get(service_name, float(inf)) } return stats # 使用示例 monitor TokenUsageMonitor({openai: 100, weather: 30}) # 模拟API调用 for i in range(10): if monitor.check_limit(openai): monitor.record_usage(openai, 10) print(f调用 {i1}: 成功) else: print(f调用 {i1}: 超过限额等待...) time.sleep(1) print(使用统计:, monitor.get_usage_statistics())8. 高级Skill开发技巧8.1 技能组合与工作流复杂的任务往往需要多个Skill协同工作class SkillOrchestrator: def __init__(self): self.skills [] def register_skill(self, skill): 注册Skill self.skills.append(skill) def process_query(self, user_input: str) - str: 处理用户查询自动选择合适的Skill # 首先尝试精确匹配 for skill in self.skills: if skill.recognize_intent(user_input): return skill.process(user_input) # 如果没有精确匹配使用相似度匹配 best_match None best_score 0 for skill in self.skills: # 简单的关键词匹配评分实际可以使用更复杂的NLP模型 score sum(1 for keyword in getattr(skill, keywords, []) if keyword in user_input) if score best_score: best_score score best_match skill if best_match and best_score 0: return best_match.process(user_input) return 抱歉我没有理解您的请求 # 创建技能编排器 orchestrator SkillOrchestrator() orchestrator.register_skill(TimeSkill()) orchestrator.register_skill(WeatherSkill(token_manager)) # 测试技能组合 test_queries [ 现在几点了, 北京天气怎么样, 今天有什么新闻 # 这个查询没有对应的Skill ] for query in test_queries: response orchestrator.process_query(query) print(fQ: {query}) print(fA: {response}\n)8.2 错误处理与重试机制class RobustSkill(WeatherSkill): def __init__(self, token_manager, max_retries3): super().__init__(token_manager) self.max_retries max_retries def execute_with_retry(self, parameters: Dict[str, Any]) - str: 带重试机制的技能执行 for attempt in range(self.max_retries): try: result self.execute(parameters) return result except requests.exceptions.RequestException as e: if attempt self.max_retries - 1: return f服务暂时不可用请稍后重试。错误: {str(e)} print(f第{attempt 1}次尝试失败等待重试...) time.sleep(2 ** attempt) # 指数退避 return 服务调用失败 def process(self, user_input: str) - str: if not self.recognize_intent(user_input): return 这不是天气查询请求 parameters self.extract_parameters(user_input) return self.execute_with_retry(parameters)9. 常见问题与排查方法在Skill开发和Token管理过程中经常会遇到各种问题。以下是常见问题的解决方案问题现象可能原因排查方式解决方案Skill无法识别意图关键词不匹配或过于简单检查recognize_intent方法增加同义词使用NLP模型改进意图识别API调用返回403错误Token无效或过期验证Token格式和有效期重新生成Token检查权限设置响应速度慢网络延迟或API限流监控响应时间检查用量统计实现缓存机制优化重试策略内存使用过高技能实例未正确释放使用内存分析工具实现资源清理使用上下文管理器Token泄漏风险Token硬编码在代码中代码安全审查使用环境变量或安全存储9.1 调试技巧class DebuggableSkill(TimeSkill): def __init__(self, debugFalse): super().__init__() self.debug debug def process(self, user_input: str) - str: if self.debug: print(f[DEBUG] 输入: {user_input}) intent_recognized self.recognize_intent(user_input) if self.debug: print(f[DEBUG] 意图识别: {intent_recognized}) if not intent_recognized: return 这不是时间查询请求 parameters self.extract_parameters(user_input) if self.debug: print(f[DEBUG] 提取参数: {parameters}) result self.execute(parameters) if self.debug: print(f[DEBUG] 执行结果: {result}) return result # 启用调试模式 debug_skill DebuggableSkill(debugTrue) result debug_skill.process(现在北京时间几点)10. 性能优化与最佳实践10.1 Skill性能优化import functools import time from cachetools import TTLCache class OptimizedSkill(WeatherSkill): def __init__(self, token_manager, cache_ttl300): # 5分钟缓存 super().__init__(token_manager) self.cache TTLCache(maxsize100, ttlcache_ttl) functools.lru_cache(maxsize50) def recognize_intent_cached(self, user_input: str) - bool: 带缓存的意图识别 return super().recognize_intent(user_input) def execute(self, parameters: Dict[str, Any]) - str: # 生成缓存键 cache_key fweather_{parameters.get(location, default)} # 检查缓存 if cache_key in self.cache: return self.cache[cache_key] # 执行查询 result super().execute(parameters) # 缓存结果 if 失败 not in result and 错误 not in result: self.cache[cache_key] result return result # 性能测试 def benchmark_skill(skill, queries, iterations100): start_time time.time() for i in range(iterations): for query in queries: skill.process(query) end_time time.time() return end_time - start_time # 比较优化前后的性能 basic_skill WeatherSkill(token_manager) optimized_skill OptimizedSkill(token_manager) test_queries [北京天气, 上海天气, 广州天气] basic_time benchmark_skill(basic_skill, test_queries, 10) optimized_time benchmark_skill(optimized_skill, test_queries, 10) print(f基础技能耗时: {basic_time:.2f}秒) print(f优化技能耗时: {optimized_time:.2f}秒) print(f性能提升: {(basic_time - optimized_time) / basic_time * 100:.1f}%)10.2 安全最佳实践Token安全永远不要将Token提交到版本控制系统使用环境变量或密钥管理服务定期轮换Token输入验证对所有用户输入进行验证和清理防止注入攻击限制输入长度和格式错误处理不要向用户暴露敏感错误信息记录详细的调试日志实现适当的错误恢复机制通过本文的实践指导你应该已经掌握了Skill编写和Token管理的基本方法。这些技能是构建实用AI Agent的基础也是进一步学习高级Agent开发的前提。在实际项目中建议从简单的Skill开始逐步增加复杂度同时建立完善的Token管理和监控体系。记住一个好的Agent系统不仅要有强大的功能更要有稳定的性能和可靠的安全保障。
