企业官网建设流程全解析

这次我们来深入探讨AI Agent开发中的两个核心概念：Skill（技能）和Token（令牌）。如果你正在学习Agent开发，或者想要理解如何让AI Agent具备特定能力，这篇文章将带你从零开始掌握Skill的编写方法和Token的工作原理。

在AI Agent体系中，Skill是Agent能够执行的具体任务能力，而Token则是控制访问和计算资源的关键凭证。理解这两者的关系，是构建实用Agent系统的基石。本文将通过实际案例演示如何编写Skill、如何管理Token，以及如何让Agent真正"理解"你的指令。

1. 核心能力速览

2. Agent开发基础：Skill与Token的关系

在AI Agent生态中，Skill和Token是相辅相成的两个核心组件。Skill定义了Agent"能做什么"，而Token则决定了Agent"被允许做什么"以及"能做多少"。

2.1 什么是Skill？

Skill是Agent执行特定任务的能力单元。每个Skill都包含三个基本要素：

• 意图识别：理解用户想要执行什么操作
• 参数提取：从用户输入中提取执行任务所需的信息
• 动作执行：调用相应的API或执行具体操作

例如，一个"天气查询"Skill需要识别用户询问天气的意图，提取地点参数，然后调用天气API返回结果。

2.2 什么是Token？

Token在Agent开发中有多重含义：

1. API访问令牌：用于身份验证和授权
2. 计算资源单位：衡量AI模型处理文本的复杂度
3. 会话标识：维持对话上下文的一致性

理解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-dotenv

3.2 常用Agent开发框架

根据项目需求选择合适的框架：

• LangChain：功能全面的Agent开发框架
• AutoGPT：专注于自主任务执行的Agent
• 自定义框架：针对特定需求的轻量级解决方案

# 基础环境检查脚本 import sys import requests def check_environment(): print(f"Python版本: {sys.version}") try: response = requests.get('https://httpbin.org/get', timeout=5) 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_KEY=sk-your-openai-token-here WEATHER_API_KEY=your-weather-api-key CUSTOM_SERVICE_TOKEN=your-custom-token # 配置参数 API_TIMEOUT=30 MAX_RETRIES=3 """

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, params=params, timeout=10) 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(f"Agent: {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"调用 {i+1}: 成功") else: print(f"调用 {i+1}: 超过限额，等待...") 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(f"Q: {query}") print(f"A: {response}\n")

8.2 错误处理与重试机制

class RobustSkill(WeatherSkill): def __init__(self, token_manager, max_retries=3): 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管理过程中，经常会遇到各种问题。以下是常见问题的解决方案：

9.1 调试技巧

class DebuggableSkill(TimeSkill): def __init__(self, debug=False): 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(debug=True) 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_ttl=300): # 5分钟缓存 super().__init__(token_manager) self.cache = TTLCache(maxsize=100, ttl=cache_ttl) @functools.lru_cache(maxsize=50) 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 = f"weather_{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, iterations=100): 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 安全最佳实践

1. Token安全

   • 永远不要将Token提交到版本控制系统
   • 使用环境变量或密钥管理服务
   • 定期轮换Token

2. 输入验证

   • 对所有用户输入进行验证和清理
   • 防止注入攻击
   • 限制输入长度和格式

3. 错误处理

   • 不要向用户暴露敏感错误信息
   • 记录详细的调试日志
   • 实现适当的错误恢复机制

通过本文的实践指导，你应该已经掌握了Skill编写和Token管理的基本方法。这些技能是构建实用AI Agent的基础，也是进一步学习高级Agent开发的前提。

在实际项目中，建议从简单的Skill开始，逐步增加复杂度，同时建立完善的Token管理和监控体系。记住，一个好的Agent系统不仅要有强大的功能，更要有稳定的性能和可靠的安全保障。