尧图网页设计 
在AI应用开发中很多开发者对Agent、Skill和Token这些核心概念的理解停留在表面导致在实际项目中遇到架构设计混乱、API调用超限、权限控制失效等问题。本文基于EP3-6系列内容系统拆解Agent框架中Skill的编写方法和Token的底层原理通过完整可运行的代码示例帮助开发者从本质掌握这些关键技术点。1. Agent、Skill与Token的核心概念解析1.1 什么是Agent智能体Agent在AI领域中指能够自主感知环境、做出决策并执行行动的智能实体。一个完整的Agent通常包含以下核心组件感知模块接收外部输入用户请求、传感器数据、API响应等决策引擎基于内部逻辑或AI模型分析信息并制定行动计划执行单元调用具体的工具或技能完成任务记忆系统存储历史交互和上下文信息在现代AI应用架构中Agent通常作为协调中心通过组合多个Skill来实现复杂功能。比如一个客服Agent可能包含查询Skill、转接Skill、情感分析Skill等。1.2 Skill技能的定义与作用Skill是Agent能够执行的特定任务或能力单元。每个Skill都应该具备以下特征单一职责一个Skill只负责一个明确的功能领域可复用性可以在不同的Agent中被重复使用标准化接口提供统一的输入输出规范独立配置拥有自己的参数设置和依赖管理例如一个天气查询Skill只需要关注如何获取和返回天气信息而不需要处理用户身份验证或数据存储等无关功能。1.3 Token的本质与重要性Token是AI系统中进行权限控制、资源计量和API调用的基本单位。理解Token需要从三个层面入手技术层面Token是服务端生成的凭证字符串用于标识用户身份和权限范围。常见的Token类型包括Access Token短期有效的访问令牌Refresh Token用于更新Access Token的长期令牌API Token第三方服务调用的认证令牌经济层面在AI服务中Token通常作为计费单位比如OpenAI的API按Token数量收费。安全层面Token实现了无状态的身份验证避免了传统的Session管理复杂性。2. 开发环境准备与工具链配置2.1 基础环境要求在开始编写Skill之前需要确保开发环境满足以下要求操作系统Windows 10/11、macOS 10.15 或 Ubuntu 18.04Python版本3.8推荐3.9或3.10Node.js16如果涉及JavaScript Skill开发Git版本控制工具2.2 核心开发工具安装# 创建虚拟环境Python项目 python -m venv agent_venv source agent_venv/bin/activate # Linux/macOS # agent_venv\Scripts\activate # Windows # 安装基础依赖 pip install openai python-dotenv requests flask # 如果使用LangChain框架 pip install langchain langchain-community # 安装代码质量工具 pip install black flake8 mypy2.3 项目结构规划一个标准的Agent项目应该采用模块化设计my_agent_project/ ├── agents/ # Agent核心逻辑 │ ├── __init__.py │ └── chat_agent.py ├── skills/ # Skill模块 │ ├── __init__.py │ ├── weather.py │ ├── calculator.py │ └── base_skill.py ├── config/ # 配置文件 │ ├── __init__.py │ └── settings.py ├── utils/ # 工具函数 │ ├── __init__.py │ └── token_manager.py ├── tests/ # 测试代码 │ ├── __init__.py │ ├── test_skills.py │ └── test_agents.py ├── requirements.txt # 依赖列表 └── main.py # 入口文件3. Skill编写实战从基础到高级3.1 创建基础Skill类所有Skill都应该继承自一个基础类确保接口一致性# skills/base_skill.py from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional class BaseSkill(ABC): Skill基类定义统一接口 def __init__(self, name: str, description: str): self.name name self.description description self.required_params [] abstractmethod def execute(self, **kwargs) - Dict[str, Any]: 执行Skill的核心方法 pass def validate_params(self, **kwargs) - bool: 验证输入参数是否完整 for param in self.required_params: if param not in kwargs: return False return True def get_info(self) - Dict[str, Any]: 返回Skill的元信息 return { name: self.name, description: self.description, required_params: self.required_params }3.2 实现具体Skill示例计算器Skill# skills/calculator.py import math from typing import Dict, Any from .base_skill import BaseSkill class CalculatorSkill(BaseSkill): 简单的计算器Skill支持基础数学运算 def __init__(self): super().__init__( namecalculator, description执行数学计算支持加减乘除、幂运算等 ) self.required_params [operation, numbers] def execute(self, **kwargs) - Dict[str, Any]: # 参数验证 if not self.validate_params(**kwargs): return {error: 缺少必要参数, required: self.required_params} operation kwargs[operation] numbers kwargs[numbers] try: result self._calculate(operation, numbers) return { success: True, result: result, operation: operation, numbers: numbers } except Exception as e: return {error: f计算失败: {str(e)}} def _calculate(self, operation: str, numbers: List[float]) - float: 执行具体的计算逻辑 operations { add: lambda nums: sum(nums), subtract: lambda nums: nums[0] - sum(nums[1:]), multiply: lambda nums: math.prod(nums), divide: lambda nums: nums[0] / math.prod(nums[1:]) if all(n ! 0 for n in nums[1:]) else None, power: lambda nums: math.pow(nums[0], nums[1]) if len(nums) 2 else None } if operation not in operations: raise ValueError(f不支持的操作: {operation}) if not numbers: raise ValueError(数字列表不能为空) result operations[operation](numbers) if result is None: raise ValueError(计算错误检查数字有效性) return result3.3 实现天气查询Skill# skills/weather.py import requests from typing import Dict, Any from .base_skill import BaseSkill class WeatherSkill(BaseSkill): 天气查询Skill调用第三方API获取天气信息 def __init__(self, api_key: str): super().__init__( nameweather, description查询指定城市的天气信息 ) self.required_params [city] self.api_key api_key self.base_url http://api.weatherapi.com/v1 def execute(self, **kwargs) - Dict[str, Any]: if not self.validate_params(**kwargs): return {error: 缺少城市参数} city kwargs[city] try: # 构建API请求 url f{self.base_url}/current.json params { key: self.api_key, q: city, aqi: no } response requests.get(url, paramsparams, timeout10) response.raise_for_status() data response.json() # 提取关键信息 current data[current] location data[location] return { success: True, city: location[name], country: location[country], temperature: current[temp_c], condition: current[condition][text], humidity: current[humidity], wind_speed: current[wind_kph] } except requests.exceptions.RequestException as e: return {error: fAPI请求失败: {str(e)}} except KeyError as e: return {error: f数据解析失败: {str(e)}}3.4 Skill注册与管理机制# skills/__init__.py from typing import Dict, Type from .base_skill import BaseSkill from .calculator import CalculatorSkill from .weather import WeatherSkill class SkillRegistry: Skill注册表管理所有可用的Skill def __init__(self): self._skills: Dict[str, Type[BaseSkill]] {} self._register_builtin_skills() def _register_builtin_skills(self): 注册内置Skill self.register(calculator, CalculatorSkill) # WeatherSkill需要API key在实例化时传入 self.register(weather, WeatherSkill) def register(self, name: str, skill_class: Type[BaseSkill]): 注册新的Skill if name in self._skills: raise ValueError(fSkill {name} 已注册) self._skills[name] skill_class def get_skill(self, name: str, **kwargs) - BaseSkill: 获取Skill实例 if name not in self._skills: raise KeyError(f未找到Skill: {name}) skill_class self._skills[name] return skill_class(**kwargs) def list_skills(self) - Dict[str, str]: 列出所有可用的Skill return {name: skill_class.__doc__ or for name, skill_class in self._skills.items()}4. Token机制深度解析与实践4.1 Token的生命周期管理Token的有效管理是AI应用稳定运行的关键。一个完整的Token生命周期包括# utils/token_manager.py import time from typing import Optional, Dict, Any import jwt from datetime import datetime, timedelta class TokenManager: Token管理器负责生成、验证和刷新Token def __init__(self, secret_key: str, algorithm: str HS256): self.secret_key secret_key self.algorithm algorithm self._blacklist set() # 简单的Token黑名单 def generate_access_token(self, user_id: str, expires_delta: Optional[timedelta] None) - str: 生成访问Token if expires_delta: expire datetime.utcnow() expires_delta else: expire datetime.utcnow() timedelta(minutes15) payload { user_id: user_id, exp: expire, iat: datetime.utcnow(), type: access } return jwt.encode(payload, self.secret_key, algorithmself.algorithm) def generate_refresh_token(self, user_id: str) - str: 生成刷新Token长期有效 payload { user_id: user_id, exp: datetime.utcnow() timedelta(days30), iat: datetime.utcnow(), type: refresh } return jwt.encode(payload, self.secret_key, algorithmself.algorithm) def verify_token(self, token: str) - Optional[Dict[str, Any]]: 验证Token有效性 if token in self._blacklist: return None try: payload jwt.decode(token, self.secret_key, algorithms[self.algorithm]) return payload except jwt.ExpiredSignatureError: return None except jwt.InvalidTokenError: return None def revoke_token(self, token: str): 撤销Token加入黑名单 self._blacklist.add(token) def refresh_access_token(self, refresh_token: str) - Optional[str]: 使用Refresh Token获取新的Access Token payload self.verify_token(refresh_token) if not payload or payload.get(type) ! refresh: return None user_id payload.get(user_id) if not user_id: return None # 撤销旧的refresh token可选 self.revoke_token(refresh_token) # 生成新的token对 new_access_token self.generate_access_token(user_id) new_refresh_token self.generate_refresh_token(user_id) return { access_token: new_access_token, refresh_token: new_refresh_token }4.2 API调用中的Token使用策略在实际的AI服务调用中Token管理需要特别注意以下策略# utils/api_client.py import requests from typing import Optional, Dict, Any import time from .token_manager import TokenManager class APIClient: 智能API客户端自动处理Token管理和重试机制 def __init__(self, base_url: str, token_manager: TokenManager, max_retries: int 3): self.base_url base_url self.token_manager token_manager self.max_retries max_retries self._current_token: Optional[str] None self._refresh_token: Optional[str] None def authenticate(self, username: str, password: str) - bool: 用户认证获取初始Token # 模拟认证过程 auth_data {username: username, password: password} try: response requests.post(f{self.base_url}/auth, jsonauth_data, timeout10) if response.status_code 200: token_data response.json() self._current_token token_data[access_token] self._refresh_token token_data[refresh_token] return True except requests.RequestException: pass return False def _ensure_valid_token(self) - bool: 确保当前Token有效必要时刷新 if self._current_token and self.token_manager.verify_token(self._current_token): return True if self._refresh_token: new_tokens self.token_manager.refresh_access_token(self._refresh_token) if new_tokens: self._current_token new_tokens[access_token] self._refresh_token new_tokens[refresh_token] return True return False def call_api(self, endpoint: str, data: Dict[str, Any]) - Optional[Dict[str, Any]]: 调用API自动处理Token验证和重试 for attempt in range(self.max_retries): if not self._ensure_valid_token(): return {error: 认证失败请重新登录} headers {Authorization: fBearer {self._current_token}} try: response requests.post( f{self.base_url}/{endpoint}, jsondata, headersheaders, timeout30 ) if response.status_code 200: return response.json() elif response.status_code 401: # Token过期 self._current_token None # 强制刷新Token continue else: return {error: fAPI调用失败: {response.status_code}} except requests.RequestException as e: if attempt self.max_retries - 1: # 最后一次尝试 return {error: f网络错误: {str(e)}} time.sleep(2 ** attempt) # 指数退避 return {error: 达到最大重试次数}4.3 Token在AI服务中的计量与优化在使用商业AI API时Token计量直接关系到成本控制# utils/token_counter.py import tiktoken # OpenAI的Token计数库 from typing import List, Dict, Any class TokenCounter: Token计数器用于估算API调用成本 def __init__(self, model: str gpt-3.5-turbo): self.model model try: self.encoding tiktoken.encoding_for_model(model) except KeyError: self.encoding tiktoken.get_encoding(cl100k_base) def count_tokens(self, text: str) - int: 计算文本的Token数量 return len(self.encoding.encode(text)) def estimate_api_cost(self, messages: List[Dict[str, str]], max_tokens: int 1000) - Dict[str, Any]: 估算API调用成本 total_input_tokens 0 for message in messages: total_input_tokens self.count_tokens(message.get(content, )) estimated_output_tokens max_tokens total_tokens total_input_tokens estimated_output_tokens # 成本估算以GPT-3.5-turbo为例 cost_per_1k_tokens 0.002 # 美元 estimated_cost (total_tokens / 1000) * cost_per_1k_tokens return { input_tokens: total_input_tokens, output_tokens: estimated_output_tokens, total_tokens: total_tokens, estimated_cost_usd: round(estimated_cost, 4) } def optimize_prompt(self, messages: List[Dict[str, str]], max_tokens: int 4000) - List[Dict[str, str]]: 优化提示词确保不超过Token限制 optimized_messages [] current_tokens 0 for message in reversed(messages): # 从最新消息开始 message_tokens self.count_tokens(message.get(content, )) if current_tokens message_tokens max_tokens: optimized_messages.insert(0, message) # 保持顺序 current_tokens message_tokens else: break # 达到限制停止添加 return optimized_messages5. 完整Agent系统集成实战5.1 构建核心Agent类# agents/chat_agent.py from typing import Dict, Any, List, Optional from skills import SkillRegistry from utils.token_manager import TokenManager from utils.token_counter import TokenCounter class ChatAgent: 聊天Agent集成多个Skill处理用户请求 def __init__(self, skill_registry: SkillRegistry, token_manager: TokenManager): self.skill_registry skill_registry self.token_manager token_manager self.token_counter TokenCounter() self.conversation_history: List[Dict[str, Any]] [] def process_message(self, user_input: str, user_id: str, context: Optional[Dict[str, Any]] None) - Dict[str, Any]: 处理用户输入选择合适的Skill执行 # 分析用户意图 intent self._analyze_intent(user_input) # 记录对话历史 self.conversation_history.append({ user_id: user_id, input: user_input, intent: intent, timestamp: self._get_timestamp() }) # 根据意图选择Skill if intent[type] calculation: return self._execute_calculator_skill(user_input, intent) elif intent[type] weather_query: return self._execute_weather_skill(user_input, intent, context) elif intent[type] general_chat: return self._handle_general_chat(user_input) else: return {response: 抱歉我还没有学会处理这个类型的请求} def _analyze_intent(self, text: str) - Dict[str, Any]: 分析用户意图简化版 text_lower text.lower() if any(word in text_lower for word in [计算, 算一下, 加减, 乘除]): return {type: calculation, confidence: 0.9} elif any(word in text_lower for word in [天气, 气温, 下雨, 晴天]): return {type: weather_query, confidence: 0.8} else: return {type: general_chat, confidence: 0.5} def _execute_calculator_skill(self, text: str, intent: Dict[str, Any]) - Dict[str, Any]: 执行计算器Skill try: calculator self.skill_registry.get_skill(calculator) # 简单的文本解析实际项目中可以使用更复杂的NLP numbers self._extract_numbers(text) operation self._detect_operation(text) if not numbers or not operation: return {response: 请提供有效的计算表达式如计算 2 加 3} result calculator.execute(operationoperation, numbersnumbers) if result.get(success): return { response: f计算结果{result[result]}, type: calculation_result, data: result } else: return {response: 计算失败请检查输入格式} except Exception as e: return {response: f计算服务暂时不可用{str(e)}} def _execute_weather_skill(self, text: str, intent: Dict[str, Any], context: Optional[Dict[str, Any]]) - Dict[str, Any]: 执行天气查询Skill try: # 需要API key从配置或上下文中获取 api_key context.get(weather_api_key) if context else None if not api_key: return {response: 天气服务需要配置API密钥} weather_skill self.skill_registry.get_skill(weather, api_keyapi_key) city self._extract_city(text) if not city: return {response: 请指定要查询的城市名称} result weather_skill.execute(citycity) if result.get(success): weather_info ( f{result[city]}的天气 f温度{result[temperature]}°C f{result[condition]} f湿度{result[humidity]}% f风速{result[wind_speed]}km/h ) return { response: weather_info, type: weather_result, data: result } else: return {response: 天气查询失败请检查城市名称或稍后重试} except Exception as e: return {response: f天气服务暂时不可用{str(e)}} def _handle_general_chat(self, text: str) - Dict[str, Any]: 处理一般聊天请求 # 这里可以集成大语言模型 simple_responses { 你好: 你好我是您的AI助手可以帮您计算、查询天气等。, 谢谢: 不客气很高兴能帮助您, 再见: 再见祝您有美好的一天 } response simple_responses.get(text, 我还在学习中目前主要擅长数学计算和天气查询哦) return {response: response} def _extract_numbers(self, text: str) - List[float]: 从文本中提取数字简化版 import re numbers re.findall(r[-]?\d*\.\d|\d, text) return [float(num) for num in numbers] def _detect_operation(self, text: str) - str: 检测数学操作类型 text_lower text.lower() if 加 in text_lower or in text_lower: return add elif 减 in text_lower or - in text_lower: return subtract elif 乘 in text_lower or * in text_lower: return multiply elif 除 in text_lower or / in text_lower: return divide else: return add # 默认加法 def _extract_city(self, text: str) - Optional[str]: 提取城市名称简化版 # 实际项目中可以使用NER实体识别 import re cities [北京, 上海, 广州, 深圳, 杭州, 成都] for city in cities: if city in text: return city return None def _get_timestamp(self) - str: 获取当前时间戳 from datetime import datetime return datetime.now().isoformat()5.2 系统集成与测试# main.py from skills import SkillRegistry from utils.token_manager import TokenManager from agents.chat_agent import ChatAgent import os from dotenv import load_dotenv def main(): 主函数演示完整的Agent系统 # 加载环境变量 load_dotenv() # 初始化组件 skill_registry SkillRegistry() token_manager TokenManager(secret_keyos.getenv(SECRET_KEY, default-secret-key)) # 创建Agent实例 agent ChatAgent(skill_registry, token_manager) # 测试用例 test_cases [ 计算一下 2 加 3 等于多少, 北京天气怎么样, 你好, 计算 10 乘以 5, 上海今天气温如何 ] context { weather_api_key: os.getenv(WEATHER_API_KEY, test-key) } print( Agent系统测试 ) for i, test_input in enumerate(test_cases, 1): print(f\n测试 {i}: {test_input}) response agent.process_message(test_input, test_user, context) print(f响应: {response[response]}) if data in response: print(f详细数据: {response[data]}) if __name__ __main__: main()6. 常见问题与解决方案6.1 Skill开发中的典型问题问题1Skill执行超时或阻塞现象Agent调用Skill时长时间无响应原因网络请求超时、死循环、资源竞争解决方案# 为Skill添加超时控制 import signal from contextlib import contextmanager contextmanager def timeout_handler(seconds: int): def timeout_handler(signum, frame): raise TimeoutError(Skill执行超时) signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(seconds) try: yield finally: signal.alarm(0) # 在Skill执行中使用 try: with timeout_handler(30): # 30秒超时 result skill.execute(**params) except TimeoutError: return {error: Skill执行超时}问题2Skill参数验证不充分现象传入无效参数导致Skill崩溃解决方案加强参数验证def validate_params(self, **kwargs) - Dict[str, Any]: 增强版参数验证 errors [] # 检查必需参数 for param in self.required_params: if param not in kwargs: errors.append(f缺少必需参数: {param}) # 参数类型验证 if numbers in kwargs and not isinstance(kwargs[numbers], list): errors.append(numbers参数必须是列表) # 参数范围验证 if city in kwargs and len(kwargs[city]) 50: errors.append(城市名称过长) return {valid: len(errors) 0, errors: errors}6.2 Token管理中的安全隐患问题Token泄露导致未授权访问风险Access Token被窃取攻击者可以冒充用户防护措施# 增强Token安全性 class SecureTokenManager(TokenManager): def __init__(self, secret_key: str, max_attempts: int 5): super().__init__(secret_key) self.failed_attempts {} self.max_attempts max_attempts def verify_token_with_security(self, token: str, client_ip: str) - Optional[Dict[str, Any]]: 带安全验证的Token检查 # 检查尝试次数 if self.failed_attempts.get(client_ip, 0) self.max_attempts: return None payload self.verify_token(token) if not payload: # 记录失败尝试 self.failed_attempts[client_ip] self.failed_attempts.get(client_ip, 0) 1 return None # 重置失败计数 self.failed_attempts[client_ip] 0 # 检查Token是否在有效设备上使用 expected_device payload.get(device_id) if expected_device and expected_device ! self.get_current_device_id(): return None return payload def get_current_device_id(self) - str: 获取当前设备标识简化版 import hashlib import platform device_info f{platform.node()}-{platform.system()} return hashlib.md5(device_info.encode()).hexdigest()6.3 Agent系统性能优化性能瓶颈Skill执行顺序串行导致响应慢优化方案并行执行无依赖的Skillimport asyncio from concurrent.futures import ThreadPoolExecutor class AsyncChatAgent(ChatAgent): 支持异步执行的Agent def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.executor ThreadPoolExecutor(max_workers5) async def process_message_async(self, user_input: str, user_id: str, context: Dict[str, Any]) - Dict[str, Any]: 异步处理消息 intent self._analyze_intent(user_input) # 并行执行多个可能相关的Skill if intent[type] complex_query: tasks [ self._execute_calculator_skill_async(user_input, intent), self._execute_weather_skill_async(user_input, intent, context) ] results await asyncio.gather(*tasks, return_exceptionsTrue) return self._combine_results(results) return await super().process_message_async(user_input, user_id, context) async def _execute_calculator_skill_async(self, text: str, intent: Dict[str, Any]): 异步执行计算器Skill loop asyncio.get_event_loop() return await loop.run_in_executor( self.executor, self._execute_calculator_skill, text, intent )7. 生产环境最佳实践7.1 Skill开发规范代码质量要求每个Skill必须有完整的单元测试代码覆盖率不低于80%遵循PEP 8编码规范提供清晰的文档字符串# tests/test_skills.py import unittest from skills.calculator import CalculatorSkill class TestCalculatorSkill(unittest.TestCase): def setUp(self): self.calculator CalculatorSkill() def test_addition(self): result self.calculator.execute(operationadd, numbers[2, 3]) self.assertTrue(result[success]) self.assertEqual(result[result], 5) def test_division_by_zero(self): result self.calculator.execute(operationdivide, numbers[1, 0]) self.assertIn(error, result) def test_missing_parameters(self): result self.calculator.execute(operationadd) # 缺少numbers参数 self.assertIn(error, result) if __name__ __main__: unittest.main()7.2 Token安全部署策略生产环境Token管理使用HSM硬件安全模块存储密钥实现Token自动轮换建立完整的审计日志设置合理的Token过期时间# 生产环境Token配置 PRODUCTION_TOKEN_CONFIG { access_token_expiry: timedelta(minutes15), # 短期Access Token refresh_token_expiry: timedelta(days7), # 中期Refresh Token max_concurrent_sessions: 5, # 最大并发会话 token_rotation_enabled: True, # 启用Token轮换 audit_logging: True # 审计日志 }7.3 监控与日志体系建立完整的可观测性体系# utils/monitoring.py import logging import time from functools import wraps from typing import Callable, Any def monitor_performance(func: Callable) - Callable: 性能监控装饰器 wraps(func) def wrapper(*args, **kwargs) - Any: start_time time.time() try: result func(*args, **kwargs) execution_time time.time() - start_time # 记录性能指标 logging.info(f{func.__name__} 执行时间: {execution_time:.3f}秒) # 超过阈值告警 if execution_time 5.0: # 5秒阈值 logging.warning(f{func.__name__} 执行缓慢: {execution_time:.3f}秒) return result except Exception as e: logging.error(f{func.__name__} 执行失败: {str(e)}) raise return wrapper # 在关键方法上使用 monitor_performance def critical_skill_operation(self, data): # 重要业务逻辑 pass7.4 错误处理与降级策略** graceful degradation** 设计class ResilientAgent(ChatAgent): 具备容错能力的Agent def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.circuit_breaker {} # 熔断器状态 def process_message_with_fallback(self, user_input: str, user_id: str, context: Dict[str, Any]) - Dict[str, Any]: 带降级策略的消息处理 try: # 检查熔断器状态 if self._is_circuit_open(primary_skill): return self._fallback_response(user_input) result self.process_message(user_input, user_id, context) # 成功则重置熔断器 self._reset_circuit(primary_skill) return result except Exception as e: # 记录失败触发熔断 self._record_failure(primary_skill) return self._fallback_response(user_input) def _fallback_response(self, user_input: str) - Dict[str, Any]: 降级响应 return { response: 系统正在维护 AI Agent开发实战：Skill编写与Token管理核心技术解析
在AI应用开发中很多开发者对Agent、Skill和Token这些核心概念的理解停留在表面导致在实际项目中遇到架构设计混乱、API调用超限、权限控制失效等问题。本文基于EP3-6系列内容系统拆解Agent框架中Skill的编写方法和Token的底层原理通过完整可运行的代码示例帮助开发者从本质掌握这些关键技术点。1. Agent、Skill与Token的核心概念解析1.1 什么是Agent智能体Agent在AI领域中指能够自主感知环境、做出决策并执行行动的智能实体。一个完整的Agent通常包含以下核心组件感知模块接收外部输入用户请求、传感器数据、API响应等决策引擎基于内部逻辑或AI模型分析信息并制定行动计划执行单元调用具体的工具或技能完成任务记忆系统存储历史交互和上下文信息在现代AI应用架构中Agent通常作为协调中心通过组合多个Skill来实现复杂功能。比如一个客服Agent可能包含查询Skill、转接Skill、情感分析Skill等。1.2 Skill技能的定义与作用Skill是Agent能够执行的特定任务或能力单元。每个Skill都应该具备以下特征单一职责一个Skill只负责一个明确的功能领域可复用性可以在不同的Agent中被重复使用标准化接口提供统一的输入输出规范独立配置拥有自己的参数设置和依赖管理例如一个天气查询Skill只需要关注如何获取和返回天气信息而不需要处理用户身份验证或数据存储等无关功能。1.3 Token的本质与重要性Token是AI系统中进行权限控制、资源计量和API调用的基本单位。理解Token需要从三个层面入手技术层面Token是服务端生成的凭证字符串用于标识用户身份和权限范围。常见的Token类型包括Access Token短期有效的访问令牌Refresh Token用于更新Access Token的长期令牌API Token第三方服务调用的认证令牌经济层面在AI服务中Token通常作为计费单位比如OpenAI的API按Token数量收费。安全层面Token实现了无状态的身份验证避免了传统的Session管理复杂性。2. 开发环境准备与工具链配置2.1 基础环境要求在开始编写Skill之前需要确保开发环境满足以下要求操作系统Windows 10/11、macOS 10.15 或 Ubuntu 18.04Python版本3.8推荐3.9或3.10Node.js16如果涉及JavaScript Skill开发Git版本控制工具2.2 核心开发工具安装# 创建虚拟环境Python项目 python -m venv agent_venv source agent_venv/bin/activate # Linux/macOS # agent_venv\Scripts\activate # Windows # 安装基础依赖 pip install openai python-dotenv requests flask # 如果使用LangChain框架 pip install langchain langchain-community # 安装代码质量工具 pip install black flake8 mypy2.3 项目结构规划一个标准的Agent项目应该采用模块化设计my_agent_project/ ├── agents/ # Agent核心逻辑 │ ├── __init__.py │ └── chat_agent.py ├── skills/ # Skill模块 │ ├── __init__.py │ ├── weather.py │ ├── calculator.py │ └── base_skill.py ├── config/ # 配置文件 │ ├── __init__.py │ └── settings.py ├── utils/ # 工具函数 │ ├── __init__.py │ └── token_manager.py ├── tests/ # 测试代码 │ ├── __init__.py │ ├── test_skills.py │ └── test_agents.py ├── requirements.txt # 依赖列表 └── main.py # 入口文件3. Skill编写实战从基础到高级3.1 创建基础Skill类所有Skill都应该继承自一个基础类确保接口一致性# skills/base_skill.py from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional class BaseSkill(ABC): Skill基类定义统一接口 def __init__(self, name: str, description: str): self.name name self.description description self.required_params [] abstractmethod def execute(self, **kwargs) - Dict[str, Any]: 执行Skill的核心方法 pass def validate_params(self, **kwargs) - bool: 验证输入参数是否完整 for param in self.required_params: if param not in kwargs: return False return True def get_info(self) - Dict[str, Any]: 返回Skill的元信息 return { name: self.name, description: self.description, required_params: self.required_params }3.2 实现具体Skill示例计算器Skill# skills/calculator.py import math from typing import Dict, Any from .base_skill import BaseSkill class CalculatorSkill(BaseSkill): 简单的计算器Skill支持基础数学运算 def __init__(self): super().__init__( namecalculator, description执行数学计算支持加减乘除、幂运算等 ) self.required_params [operation, numbers] def execute(self, **kwargs) - Dict[str, Any]: # 参数验证 if not self.validate_params(**kwargs): return {error: 缺少必要参数, required: self.required_params} operation kwargs[operation] numbers kwargs[numbers] try: result self._calculate(operation, numbers) return { success: True, result: result, operation: operation, numbers: numbers } except Exception as e: return {error: f计算失败: {str(e)}} def _calculate(self, operation: str, numbers: List[float]) - float: 执行具体的计算逻辑 operations { add: lambda nums: sum(nums), subtract: lambda nums: nums[0] - sum(nums[1:]), multiply: lambda nums: math.prod(nums), divide: lambda nums: nums[0] / math.prod(nums[1:]) if all(n ! 0 for n in nums[1:]) else None, power: lambda nums: math.pow(nums[0], nums[1]) if len(nums) 2 else None } if operation not in operations: raise ValueError(f不支持的操作: {operation}) if not numbers: raise ValueError(数字列表不能为空) result operations[operation](numbers) if result is None: raise ValueError(计算错误检查数字有效性) return result3.3 实现天气查询Skill# skills/weather.py import requests from typing import Dict, Any from .base_skill import BaseSkill class WeatherSkill(BaseSkill): 天气查询Skill调用第三方API获取天气信息 def __init__(self, api_key: str): super().__init__( nameweather, description查询指定城市的天气信息 ) self.required_params [city] self.api_key api_key self.base_url http://api.weatherapi.com/v1 def execute(self, **kwargs) - Dict[str, Any]: if not self.validate_params(**kwargs): return {error: 缺少城市参数} city kwargs[city] try: # 构建API请求 url f{self.base_url}/current.json params { key: self.api_key, q: city, aqi: no } response requests.get(url, paramsparams, timeout10) response.raise_for_status() data response.json() # 提取关键信息 current data[current] location data[location] return { success: True, city: location[name], country: location[country], temperature: current[temp_c], condition: current[condition][text], humidity: current[humidity], wind_speed: current[wind_kph] } except requests.exceptions.RequestException as e: return {error: fAPI请求失败: {str(e)}} except KeyError as e: return {error: f数据解析失败: {str(e)}}3.4 Skill注册与管理机制# skills/__init__.py from typing import Dict, Type from .base_skill import BaseSkill from .calculator import CalculatorSkill from .weather import WeatherSkill class SkillRegistry: Skill注册表管理所有可用的Skill def __init__(self): self._skills: Dict[str, Type[BaseSkill]] {} self._register_builtin_skills() def _register_builtin_skills(self): 注册内置Skill self.register(calculator, CalculatorSkill) # WeatherSkill需要API key在实例化时传入 self.register(weather, WeatherSkill) def register(self, name: str, skill_class: Type[BaseSkill]): 注册新的Skill if name in self._skills: raise ValueError(fSkill {name} 已注册) self._skills[name] skill_class def get_skill(self, name: str, **kwargs) - BaseSkill: 获取Skill实例 if name not in self._skills: raise KeyError(f未找到Skill: {name}) skill_class self._skills[name] return skill_class(**kwargs) def list_skills(self) - Dict[str, str]: 列出所有可用的Skill return {name: skill_class.__doc__ or for name, skill_class in self._skills.items()}4. Token机制深度解析与实践4.1 Token的生命周期管理Token的有效管理是AI应用稳定运行的关键。一个完整的Token生命周期包括# utils/token_manager.py import time from typing import Optional, Dict, Any import jwt from datetime import datetime, timedelta class TokenManager: Token管理器负责生成、验证和刷新Token def __init__(self, secret_key: str, algorithm: str HS256): self.secret_key secret_key self.algorithm algorithm self._blacklist set() # 简单的Token黑名单 def generate_access_token(self, user_id: str, expires_delta: Optional[timedelta] None) - str: 生成访问Token if expires_delta: expire datetime.utcnow() expires_delta else: expire datetime.utcnow() timedelta(minutes15) payload { user_id: user_id, exp: expire, iat: datetime.utcnow(), type: access } return jwt.encode(payload, self.secret_key, algorithmself.algorithm) def generate_refresh_token(self, user_id: str) - str: 生成刷新Token长期有效 payload { user_id: user_id, exp: datetime.utcnow() timedelta(days30), iat: datetime.utcnow(), type: refresh } return jwt.encode(payload, self.secret_key, algorithmself.algorithm) def verify_token(self, token: str) - Optional[Dict[str, Any]]: 验证Token有效性 if token in self._blacklist: return None try: payload jwt.decode(token, self.secret_key, algorithms[self.algorithm]) return payload except jwt.ExpiredSignatureError: return None except jwt.InvalidTokenError: return None def revoke_token(self, token: str): 撤销Token加入黑名单 self._blacklist.add(token) def refresh_access_token(self, refresh_token: str) - Optional[str]: 使用Refresh Token获取新的Access Token payload self.verify_token(refresh_token) if not payload or payload.get(type) ! refresh: return None user_id payload.get(user_id) if not user_id: return None # 撤销旧的refresh token可选 self.revoke_token(refresh_token) # 生成新的token对 new_access_token self.generate_access_token(user_id) new_refresh_token self.generate_refresh_token(user_id) return { access_token: new_access_token, refresh_token: new_refresh_token }4.2 API调用中的Token使用策略在实际的AI服务调用中Token管理需要特别注意以下策略# utils/api_client.py import requests from typing import Optional, Dict, Any import time from .token_manager import TokenManager class APIClient: 智能API客户端自动处理Token管理和重试机制 def __init__(self, base_url: str, token_manager: TokenManager, max_retries: int 3): self.base_url base_url self.token_manager token_manager self.max_retries max_retries self._current_token: Optional[str] None self._refresh_token: Optional[str] None def authenticate(self, username: str, password: str) - bool: 用户认证获取初始Token # 模拟认证过程 auth_data {username: username, password: password} try: response requests.post(f{self.base_url}/auth, jsonauth_data, timeout10) if response.status_code 200: token_data response.json() self._current_token token_data[access_token] self._refresh_token token_data[refresh_token] return True except requests.RequestException: pass return False def _ensure_valid_token(self) - bool: 确保当前Token有效必要时刷新 if self._current_token and self.token_manager.verify_token(self._current_token): return True if self._refresh_token: new_tokens self.token_manager.refresh_access_token(self._refresh_token) if new_tokens: self._current_token new_tokens[access_token] self._refresh_token new_tokens[refresh_token] return True return False def call_api(self, endpoint: str, data: Dict[str, Any]) - Optional[Dict[str, Any]]: 调用API自动处理Token验证和重试 for attempt in range(self.max_retries): if not self._ensure_valid_token(): return {error: 认证失败请重新登录} headers {Authorization: fBearer {self._current_token}} try: response requests.post( f{self.base_url}/{endpoint}, jsondata, headersheaders, timeout30 ) if response.status_code 200: return response.json() elif response.status_code 401: # Token过期 self._current_token None # 强制刷新Token continue else: return {error: fAPI调用失败: {response.status_code}} except requests.RequestException as e: if attempt self.max_retries - 1: # 最后一次尝试 return {error: f网络错误: {str(e)}} time.sleep(2 ** attempt) # 指数退避 return {error: 达到最大重试次数}4.3 Token在AI服务中的计量与优化在使用商业AI API时Token计量直接关系到成本控制# utils/token_counter.py import tiktoken # OpenAI的Token计数库 from typing import List, Dict, Any class TokenCounter: Token计数器用于估算API调用成本 def __init__(self, model: str gpt-3.5-turbo): self.model model try: self.encoding tiktoken.encoding_for_model(model) except KeyError: self.encoding tiktoken.get_encoding(cl100k_base) def count_tokens(self, text: str) - int: 计算文本的Token数量 return len(self.encoding.encode(text)) def estimate_api_cost(self, messages: List[Dict[str, str]], max_tokens: int 1000) - Dict[str, Any]: 估算API调用成本 total_input_tokens 0 for message in messages: total_input_tokens self.count_tokens(message.get(content, )) estimated_output_tokens max_tokens total_tokens total_input_tokens estimated_output_tokens # 成本估算以GPT-3.5-turbo为例 cost_per_1k_tokens 0.002 # 美元 estimated_cost (total_tokens / 1000) * cost_per_1k_tokens return { input_tokens: total_input_tokens, output_tokens: estimated_output_tokens, total_tokens: total_tokens, estimated_cost_usd: round(estimated_cost, 4) } def optimize_prompt(self, messages: List[Dict[str, str]], max_tokens: int 4000) - List[Dict[str, str]]: 优化提示词确保不超过Token限制 optimized_messages [] current_tokens 0 for message in reversed(messages): # 从最新消息开始 message_tokens self.count_tokens(message.get(content, )) if current_tokens message_tokens max_tokens: optimized_messages.insert(0, message) # 保持顺序 current_tokens message_tokens else: break # 达到限制停止添加 return optimized_messages5. 完整Agent系统集成实战5.1 构建核心Agent类# agents/chat_agent.py from typing import Dict, Any, List, Optional from skills import SkillRegistry from utils.token_manager import TokenManager from utils.token_counter import TokenCounter class ChatAgent: 聊天Agent集成多个Skill处理用户请求 def __init__(self, skill_registry: SkillRegistry, token_manager: TokenManager): self.skill_registry skill_registry self.token_manager token_manager self.token_counter TokenCounter() self.conversation_history: List[Dict[str, Any]] [] def process_message(self, user_input: str, user_id: str, context: Optional[Dict[str, Any]] None) - Dict[str, Any]: 处理用户输入选择合适的Skill执行 # 分析用户意图 intent self._analyze_intent(user_input) # 记录对话历史 self.conversation_history.append({ user_id: user_id, input: user_input, intent: intent, timestamp: self._get_timestamp() }) # 根据意图选择Skill if intent[type] calculation: return self._execute_calculator_skill(user_input, intent) elif intent[type] weather_query: return self._execute_weather_skill(user_input, intent, context) elif intent[type] general_chat: return self._handle_general_chat(user_input) else: return {response: 抱歉我还没有学会处理这个类型的请求} def _analyze_intent(self, text: str) - Dict[str, Any]: 分析用户意图简化版 text_lower text.lower() if any(word in text_lower for word in [计算, 算一下, 加减, 乘除]): return {type: calculation, confidence: 0.9} elif any(word in text_lower for word in [天气, 气温, 下雨, 晴天]): return {type: weather_query, confidence: 0.8} else: return {type: general_chat, confidence: 0.5} def _execute_calculator_skill(self, text: str, intent: Dict[str, Any]) - Dict[str, Any]: 执行计算器Skill try: calculator self.skill_registry.get_skill(calculator) # 简单的文本解析实际项目中可以使用更复杂的NLP numbers self._extract_numbers(text) operation self._detect_operation(text) if not numbers or not operation: return {response: 请提供有效的计算表达式如计算 2 加 3} result calculator.execute(operationoperation, numbersnumbers) if result.get(success): return { response: f计算结果{result[result]}, type: calculation_result, data: result } else: return {response: 计算失败请检查输入格式} except Exception as e: return {response: f计算服务暂时不可用{str(e)}} def _execute_weather_skill(self, text: str, intent: Dict[str, Any], context: Optional[Dict[str, Any]]) - Dict[str, Any]: 执行天气查询Skill try: # 需要API key从配置或上下文中获取 api_key context.get(weather_api_key) if context else None if not api_key: return {response: 天气服务需要配置API密钥} weather_skill self.skill_registry.get_skill(weather, api_keyapi_key) city self._extract_city(text) if not city: return {response: 请指定要查询的城市名称} result weather_skill.execute(citycity) if result.get(success): weather_info ( f{result[city]}的天气 f温度{result[temperature]}°C f{result[condition]} f湿度{result[humidity]}% f风速{result[wind_speed]}km/h ) return { response: weather_info, type: weather_result, data: result } else: return {response: 天气查询失败请检查城市名称或稍后重试} except Exception as e: return {response: f天气服务暂时不可用{str(e)}} def _handle_general_chat(self, text: str) - Dict[str, Any]: 处理一般聊天请求 # 这里可以集成大语言模型 simple_responses { 你好: 你好我是您的AI助手可以帮您计算、查询天气等。, 谢谢: 不客气很高兴能帮助您, 再见: 再见祝您有美好的一天 } response simple_responses.get(text, 我还在学习中目前主要擅长数学计算和天气查询哦) return {response: response} def _extract_numbers(self, text: str) - List[float]: 从文本中提取数字简化版 import re numbers re.findall(r[-]?\d*\.\d|\d, text) return [float(num) for num in numbers] def _detect_operation(self, text: str) - str: 检测数学操作类型 text_lower text.lower() if 加 in text_lower or in text_lower: return add elif 减 in text_lower or - in text_lower: return subtract elif 乘 in text_lower or * in text_lower: return multiply elif 除 in text_lower or / in text_lower: return divide else: return add # 默认加法 def _extract_city(self, text: str) - Optional[str]: 提取城市名称简化版 # 实际项目中可以使用NER实体识别 import re cities [北京, 上海, 广州, 深圳, 杭州, 成都] for city in cities: if city in text: return city return None def _get_timestamp(self) - str: 获取当前时间戳 from datetime import datetime return datetime.now().isoformat()5.2 系统集成与测试# main.py from skills import SkillRegistry from utils.token_manager import TokenManager from agents.chat_agent import ChatAgent import os from dotenv import load_dotenv def main(): 主函数演示完整的Agent系统 # 加载环境变量 load_dotenv() # 初始化组件 skill_registry SkillRegistry() token_manager TokenManager(secret_keyos.getenv(SECRET_KEY, default-secret-key)) # 创建Agent实例 agent ChatAgent(skill_registry, token_manager) # 测试用例 test_cases [ 计算一下 2 加 3 等于多少, 北京天气怎么样, 你好, 计算 10 乘以 5, 上海今天气温如何 ] context { weather_api_key: os.getenv(WEATHER_API_KEY, test-key) } print( Agent系统测试 ) for i, test_input in enumerate(test_cases, 1): print(f\n测试 {i}: {test_input}) response agent.process_message(test_input, test_user, context) print(f响应: {response[response]}) if data in response: print(f详细数据: {response[data]}) if __name__ __main__: main()6. 常见问题与解决方案6.1 Skill开发中的典型问题问题1Skill执行超时或阻塞现象Agent调用Skill时长时间无响应原因网络请求超时、死循环、资源竞争解决方案# 为Skill添加超时控制 import signal from contextlib import contextmanager contextmanager def timeout_handler(seconds: int): def timeout_handler(signum, frame): raise TimeoutError(Skill执行超时) signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(seconds) try: yield finally: signal.alarm(0) # 在Skill执行中使用 try: with timeout_handler(30): # 30秒超时 result skill.execute(**params) except TimeoutError: return {error: Skill执行超时}问题2Skill参数验证不充分现象传入无效参数导致Skill崩溃解决方案加强参数验证def validate_params(self, **kwargs) - Dict[str, Any]: 增强版参数验证 errors [] # 检查必需参数 for param in self.required_params: if param not in kwargs: errors.append(f缺少必需参数: {param}) # 参数类型验证 if numbers in kwargs and not isinstance(kwargs[numbers], list): errors.append(numbers参数必须是列表) # 参数范围验证 if city in kwargs and len(kwargs[city]) 50: errors.append(城市名称过长) return {valid: len(errors) 0, errors: errors}6.2 Token管理中的安全隐患问题Token泄露导致未授权访问风险Access Token被窃取攻击者可以冒充用户防护措施# 增强Token安全性 class SecureTokenManager(TokenManager): def __init__(self, secret_key: str, max_attempts: int 5): super().__init__(secret_key) self.failed_attempts {} self.max_attempts max_attempts def verify_token_with_security(self, token: str, client_ip: str) - Optional[Dict[str, Any]]: 带安全验证的Token检查 # 检查尝试次数 if self.failed_attempts.get(client_ip, 0) self.max_attempts: return None payload self.verify_token(token) if not payload: # 记录失败尝试 self.failed_attempts[client_ip] self.failed_attempts.get(client_ip, 0) 1 return None # 重置失败计数 self.failed_attempts[client_ip] 0 # 检查Token是否在有效设备上使用 expected_device payload.get(device_id) if expected_device and expected_device ! self.get_current_device_id(): return None return payload def get_current_device_id(self) - str: 获取当前设备标识简化版 import hashlib import platform device_info f{platform.node()}-{platform.system()} return hashlib.md5(device_info.encode()).hexdigest()6.3 Agent系统性能优化性能瓶颈Skill执行顺序串行导致响应慢优化方案并行执行无依赖的Skillimport asyncio from concurrent.futures import ThreadPoolExecutor class AsyncChatAgent(ChatAgent): 支持异步执行的Agent def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.executor ThreadPoolExecutor(max_workers5) async def process_message_async(self, user_input: str, user_id: str, context: Dict[str, Any]) - Dict[str, Any]: 异步处理消息 intent self._analyze_intent(user_input) # 并行执行多个可能相关的Skill if intent[type] complex_query: tasks [ self._execute_calculator_skill_async(user_input, intent), self._execute_weather_skill_async(user_input, intent, context) ] results await asyncio.gather(*tasks, return_exceptionsTrue) return self._combine_results(results) return await super().process_message_async(user_input, user_id, context) async def _execute_calculator_skill_async(self, text: str, intent: Dict[str, Any]): 异步执行计算器Skill loop asyncio.get_event_loop() return await loop.run_in_executor( self.executor, self._execute_calculator_skill, text, intent )7. 生产环境最佳实践7.1 Skill开发规范代码质量要求每个Skill必须有完整的单元测试代码覆盖率不低于80%遵循PEP 8编码规范提供清晰的文档字符串# tests/test_skills.py import unittest from skills.calculator import CalculatorSkill class TestCalculatorSkill(unittest.TestCase): def setUp(self): self.calculator CalculatorSkill() def test_addition(self): result self.calculator.execute(operationadd, numbers[2, 3]) self.assertTrue(result[success]) self.assertEqual(result[result], 5) def test_division_by_zero(self): result self.calculator.execute(operationdivide, numbers[1, 0]) self.assertIn(error, result) def test_missing_parameters(self): result self.calculator.execute(operationadd) # 缺少numbers参数 self.assertIn(error, result) if __name__ __main__: unittest.main()7.2 Token安全部署策略生产环境Token管理使用HSM硬件安全模块存储密钥实现Token自动轮换建立完整的审计日志设置合理的Token过期时间# 生产环境Token配置 PRODUCTION_TOKEN_CONFIG { access_token_expiry: timedelta(minutes15), # 短期Access Token refresh_token_expiry: timedelta(days7), # 中期Refresh Token max_concurrent_sessions: 5, # 最大并发会话 token_rotation_enabled: True, # 启用Token轮换 audit_logging: True # 审计日志 }7.3 监控与日志体系建立完整的可观测性体系# utils/monitoring.py import logging import time from functools import wraps from typing import Callable, Any def monitor_performance(func: Callable) - Callable: 性能监控装饰器 wraps(func) def wrapper(*args, **kwargs) - Any: start_time time.time() try: result func(*args, **kwargs) execution_time time.time() - start_time # 记录性能指标 logging.info(f{func.__name__} 执行时间: {execution_time:.3f}秒) # 超过阈值告警 if execution_time 5.0: # 5秒阈值 logging.warning(f{func.__name__} 执行缓慢: {execution_time:.3f}秒) return result except Exception as e: logging.error(f{func.__name__} 执行失败: {str(e)}) raise return wrapper # 在关键方法上使用 monitor_performance def critical_skill_operation(self, data): # 重要业务逻辑 pass7.4 错误处理与降级策略** graceful degradation** 设计class ResilientAgent(ChatAgent): 具备容错能力的Agent def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.circuit_breaker {} # 熔断器状态 def process_message_with_fallback(self, user_input: str, user_id: str, context: Dict[str, Any]) - Dict[str, Any]: 带降级策略的消息处理 try: # 检查熔断器状态 if self._is_circuit_open(primary_skill): return self._fallback_response(user_input) result self.process_message(user_input, user_id, context) # 成功则重置熔断器 self._reset_circuit(primary_skill) return result except Exception as e: # 记录失败触发熔断 self._record_failure(primary_skill) return self._fallback_response(user_input) def _fallback_response(self, user_input: str) - Dict[str, Any]: 降级响应 return { response: 系统正在维护