在 AI 应用开发领域Agent 已经从一个学术概念变成了工程实践中的核心组件。很多开发者第一次接触 Agent 框架时会被 Skill、Token、Action、Tool 等术语搞混更不清楚如何从零开始构建一个真正可用的智能体。实际项目中一个配置不当的 Skill 或者对 Token 机制的误解都可能导致整个 Agent 无法正常工作而错误信息往往晦涩难懂比如 token exchange failed: token endpoint returned status 403 或者 your access token could not be refreshed。本文将以实践为导向通过一个完整的天气查询 Agent 案例解释 Skill 的编写方法、Token 的作用机制以及如何避免常见的认证和配置问题。无论你是刚开始学习 AI 应用开发还是已经在实际项目中遇到过 Agent 部署问题都能通过本文理解核心概念并掌握排查方法。1. 理解 Agent 的基本架构为什么需要 Skill 和 Token1.1 Agent 是什么解决了什么问题Agent 本质上是一个能够理解用户意图、制定计划、执行动作并返回结果的智能程序。与传统的聊天机器人不同真正的 Agent 具备自主决策能力能够根据上下文选择不同的工具和策略完成任务。在实际项目中Agent 通常包含三个核心组件大脑Brain负责理解用户输入、制定计划、决策下一步动作技能SkillAgent 可以调用的具体能力如查询天气、发送邮件、分析数据记忆Memory保存对话历史和上下文确保连贯性1.2 Skill 在 Agent 中的角色定位Skill 是 Agent 的能力单元每个 Skill 都封装了一个特定的功能。比如天气查询 Skill、文件读写 Skill、数据库查询 Skill 等。Skill 的设计质量直接决定了 Agent 的实用性和可靠性。一个设计良好的 Skill 应该具备以下特点单一职责每个 Skill 只负责一个明确的功能领域清晰接口输入输出定义明确便于其他组件调用错误处理能够妥善处理异常情况并给出有意义的错误信息可测试性支持独立测试不依赖完整的 Agent 环境1.3 Token 的作用和常见问题Token 在 Agent 系统中主要承担两个角色身份认证和用量控制。身份认证 Token如 JWT Token、API Key 等用于验证 Agent 是否有权访问某个服务或资源。常见的错误包括Token 过期或失效Token 权限不足Token 格式错误网络问题导致的 Token 交换失败用量控制 Token在大语言模型场景中Token 也指文本处理的基本单位用于计算 API 调用成本。开发者需要关注输入输出的 Token 数量估算上下文窗口限制成本控制策略下面是一个典型的 Agent 系统架构图展示了各组件之间的关系用户输入 → Agent大脑 → Skill选择 → Token验证 → 外部API → 结果返回2. 环境准备与依赖配置2.1 选择适合的 Agent 开发框架目前主流的 Agent 开发框架包括 LangChain、AutoGPT、CrewAI 等。对于初学者建议从 LangChain 开始因为它有丰富的文档和社区支持。创建项目并安装基础依赖# 创建项目目录 mkdir weather-agent cd weather-agent # 创建虚拟环境 python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 安装核心依赖 pip install langchain-openai langchain-core python-dotenv requests2.2 配置环境变量和认证信息为了避免在代码中硬敏感信息使用环境变量管理 API Key 等配置# 创建 .env 文件 echo OPENAI_API_KEYyour_openai_api_key_here .env echo WEATHER_API_KEYyour_weather_api_key_here .env创建配置文件加载逻辑# config.py import os from dotenv import load_dotenv load_dotenv() class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) WEATHER_API_KEY os.getenv(WEATHER_API_KEY) classmethod def validate(cls): 验证必要配置是否完整 missing [] if not cls.OPENAI_API_KEY: missing.append(OPENAI_API_KEY) if not cls.WEATHER_API_KEY: missing.append(WEATHER_API_KEY) if missing: raise ValueError(f缺少必要环境变量: {, .join(missing)})2.3 项目结构设计合理的项目结构有助于后续维护和扩展weather-agent/ ├── src/ │ ├── skills/ │ │ ├── __init__.py │ │ ├── weather_skill.py │ │ └── base_skill.py │ ├── agents/ │ │ ├── __init__.py │ │ └── weather_agent.py │ └── utils/ │ ├── __init__.py │ └── token_manager.py ├── tests/ ├── requirements.txt ├── .env.example └── main.py3. 编写第一个 Skill天气查询功能3.1 设计 Skill 基类首先创建一个基础的 Skill 类定义统一的接口规范# src/skills/base_skill.py from abc import ABC, abstractmethod from typing import Dict, Any, Optional class BaseSkill(ABC): Skill 基类定义统一接口 def __init__(self, name: str, description: str): self.name name self.description description self._token_manager None abstractmethod async def execute(self, parameters: Dict[str, Any]) - Dict[str, Any]: 执行 Skill 的核心逻辑 pass def validate_parameters(self, parameters: Dict[str, Any]) - bool: 验证输入参数是否有效 return True def set_token_manager(self, token_manager): 设置 Token 管理器 self._token_manager token_manager def get_skill_info(self) - Dict[str, str]: 获取 Skill 描述信息 return { name: self.name, description: self.description, parameters: self.get_parameters_schema() } abstractmethod def get_parameters_schema(self) - Dict[str, Any]: 定义 Skill 所需的参数格式 pass3.2 实现天气查询 Skill基于基类实现具体的天气查询功能# src/skills/weather_skill.py import requests from typing import Dict, Any from .base_skill import BaseSkill class WeatherSkill(BaseSkill): 天气查询 Skill def __init__(self): super().__init__( nameweather_query, description查询指定城市的天气情况 ) self.api_url http://api.weatherapi.com/v1/current.json def get_parameters_schema(self) - Dict[str, Any]: return { type: object, properties: { city: { type: string, description: 城市名称 } }, required: [city] } def validate_parameters(self, parameters: Dict[str, Any]) - bool: if not parameters.get(city): return False return True async def execute(self, parameters: Dict[str, Any]) - Dict[str, Any]: if not self.validate_parameters(parameters): return { success: False, error: 参数验证失败缺少城市名称, data: None } try: # 获取 API Key这里演示 Token 的使用 api_key self._get_api_key() if not api_key: return { success: False, error: API Token 配置错误, data: None } # 调用天气 API response requests.get( self.api_url, params{ key: api_key, q: parameters[city], aqi: no }, timeout10 ) if response.status_code 200: data response.json() return { success: True, error: None, data: self._format_weather_data(data) } elif response.status_code 403: return { success: False, error: API Token 无效或权限不足, data: None } else: return { success: False, error: fAPI 请求失败: {response.status_code}, data: None } except requests.exceptions.Timeout: return { success: False, error: 请求超时请检查网络连接, data: None } except Exception as e: return { success: False, error: f系统错误: {str(e)}, data: None } def _get_api_key(self) - str: 从配置或 Token 管理器获取 API Key # 这里可以扩展为从 Token 管理器动态获取 from config import Config return Config.WEATHER_API_KEY def _format_weather_data(self, raw_data: Dict[str, Any]) - Dict[str, Any]: 格式化天气数据 current raw_data.get(current, {}) location raw_data.get(location, {}) return { city: location.get(name, 未知), temperature: current.get(temp_c, 未知), condition: current.get(condition, {}).get(text, 未知), humidity: current.get(humidity, 未知), wind_speed: current.get(wind_kph, 未知) }3.3 测试 Skill 功能编写单元测试验证 Skill 的正确性# tests/test_weather_skill.py import pytest from src.skills.weather_skill import WeatherSkill class TestWeatherSkill: def setup_method(self): self.skill WeatherSkill() def test_parameter_validation(self): 测试参数验证 # 有效参数 assert self.skill.validate_parameters({city: 北京}) True # 无效参数 assert self.skill.validate_parameters({}) False assert self.skill.validate_parameters({city: }) False def test_skill_info(self): 测试 Skill 信息获取 info self.skill.get_skill_info() assert info[name] weather_query assert city in info[parameters][required]4. Token 管理机制详解4.1 Token 的生命周期管理在 Agent 系统中Token 需要完整的生命周期管理# src/utils/token_manager.py import time from typing import Optional, Dict, Any from datetime import datetime, timedelta class TokenManager: Token 管理器负责 Token 的获取、刷新和验证 def __init__(self): self._tokens {} self._refresh_callbacks {} def store_token(self, service_name: str, token: str, expires_in: int 3600): 存储 Token 信息 expires_at datetime.now() timedelta(secondsexpires_in) self._tokens[service_name] { token: token, expires_at: expires_at, created_at: datetime.now() } def get_token(self, service_name: str) - Optional[str]: 获取有效的 Token token_info self._tokens.get(service_name) if not token_info: return None # 检查 Token 是否过期 if datetime.now() token_info[expires_at]: if service_name in self._refresh_callbacks: # 自动刷新 Token new_token self._refresh_callbacks[service_name]() if new_token: self.store_token(service_name, new_token) return new_token return None return token_info[token] def register_refresh_callback(self, service_name: str, callback): 注册 Token 刷新回调函数 self._refresh_callbacks[service_name] callback def is_token_valid(self, service_name: str) - bool: 检查 Token 是否有效 token self.get_token(service_name) return token is not None def clear_token(self, service_name: str): 清除 Token if service_name in self._tokens: del self._tokens[service_name]4.2 处理常见的 Token 错误在实际项目中需要妥善处理各种 Token 相关错误错误现象可能原因检查方式处理建议token exchange failed: status 403Token 无效、权限不足、区域限制检查 API Key 格式、权限设置、服务区域重新生成 Token确认服务区域配置token could not be refreshed刷新 Token 失败、网络问题检查刷新逻辑、网络连接、认证服务器状态实现重试机制添加降级方案token endpoint returned error认证服务器异常、请求格式错误检查请求参数、服务器状态码验证请求格式联系服务提供商your access token could not be refreshedToken 已撤销、账户异常检查账户状态、账单信息登录账户确认状态更新支付信息4.3 实现安全的 Token 存储在生产环境中Token 存储需要额外的安全措施# src/utils/secure_token_manager.py import base64 import os from cryptography.fernet import Fernet from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC from .token_manager import TokenManager class SecureTokenManager(TokenManager): 安全的 Token 管理器支持加密存储 def __init__(self, password: str, salt: bytes None): super().__init__() self.password password.encode() self.salt salt or os.urandom(16) self.fernet self._create_fernet() def _create_fernet(self) - Fernet: 创建 Fernet 加密实例 kdf PBKDF2HMAC( algorithmhashes.SHA256(), length32, saltself.salt, iterations100000, ) key base64.urlsafe_b64encode(kdf.derive(self.password)) return Fernet(key) def store_token(self, service_name: str, token: str, expires_in: int 3600): 加密存储 Token encrypted_token self.fernet.encrypt(token.encode()) super().store_token(service_name, encrypted_token.decode(), expires_in) def get_token(self, service_name: str) - Optional[str]: 解密获取 Token encrypted_token super().get_token(service_name) if encrypted_token: try: return self.fernet.decrypt(encrypted_token.encode()).decode() except Exception: # Token 解密失败可能是密码更改或数据损坏 self.clear_token(service_name) return None return None5. 构建完整的 Weather Agent5.1 集成 Skill 和 Token 管理现在将 Skill 和 Token 管理整合到完整的 Agent 中# src/agents/weather_agent.py from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from typing import List, Dict, Any from src.skills.base_skill import BaseSkill from src.utils.token_manager import TokenManager class WeatherAgent: 天气查询 Agent def __init__(self): self.llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) self.skills: List[BaseSkill] [] self.token_manager TokenManager() self.agent_executor None def add_skill(self, skill: BaseSkill): 添加 Skill 到 Agent skill.set_token_manager(self.token_manager) self.skills.append(skill) def _create_tools(self): 将 Skill 转换为 LangChain Tools from langchain.tools import Tool tools [] for skill in self.skills: tool Tool( nameskill.name, descriptionskill.description, funcself._create_skill_wrapper(skill) ) tools.append(tool) return tools def _create_skill_wrapper(self, skill: BaseSkill): 创建 Skill 的包装函数 async def skill_wrapper(**kwargs): return await skill.execute(kwargs) return skill_wrapper def initialize(self): 初始化 Agent tools self._create_tools() prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的天气查询助手。根据用户需求使用合适的工具查询天气信息。 可用工具 {tools} 使用要求 1. 明确用户要查询的城市 2. 只使用提供的工具查询天气 3. 如果用户没有指定城市请主动询问 4. 结果要清晰易懂包含温度、天气状况等关键信息 ), (human, {input}), ]) agent create_tool_calling_agent(self.llm, tools, prompt) self.agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) async def query(self, user_input: str) - str: 处理用户查询 if not self.agent_executor: self.initialize() try: result await self.agent_executor.ainvoke({input: user_input}) return result[output] except Exception as e: return f查询过程中出现错误: {str(e)}5.2 创建主程序入口# main.py import asyncio from config import Config from src.agents.weather_agent import WeatherAgent from src.skills.weather_skill import WeatherSkill async def main(): # 验证配置 try: Config.validate() except ValueError as e: print(f配置错误: {e}) return # 创建 Agent agent WeatherAgent() # 添加天气查询 Skill weather_skill WeatherSkill() agent.add_skill(weather_skill) # 测试查询 queries [ 北京天气怎么样, 查询上海的天气情况, 今天纽约的温度是多少 ] for query in queries: print(f\n用户: {query}) response await agent.query(query) print(fAgent: {response}) await asyncio.sleep(1) # 避免请求过快 if __name__ __main__: asyncio.run(main())6. 常见问题排查与解决方案6.1 Skill 执行失败排查问题现象Skill 返回错误或超时排查步骤检查参数验证逻辑验证 API Token 有效性测试网络连接和超时设置查看完整的错误日志# 添加详细的日志记录 import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class DebugWeatherSkill(WeatherSkill): async def execute(self, parameters: Dict[str, Any]) - Dict[str, Any]: logger.info(f执行天气查询参数: {parameters}) try: result await super().execute(parameters) logger.info(f查询结果: {result}) return result except Exception as e: logger.error(f查询失败: {str(e)}) return { success: False, error: f系统错误: {str(e)}, data: None }6.2 Token 相关错误处理问题现象Token 失效、刷新失败、权限错误解决方案实现 Token 自动刷新机制添加降级方案和备用 Token完善的错误提示和日志记录class RobustTokenManager(TokenManager): 增强的 Token 管理器支持重试和降级 async def get_token_with_retry(self, service_name: str, max_retries: int 3) - Optional[str]: 带重试的 Token 获取 for attempt in range(max_retries): token self.get_token(service_name) if token: return token # Token 无效尝试刷新 if service_name in self._refresh_callbacks: try: new_token await self._refresh_callbacks[service_name]() if new_token: self.store_token(service_name, new_token) return new_token except Exception as e: logger.warning(f第 {attempt 1} 次 Token 刷新失败: {e}) if attempt max_retries - 1: await asyncio.sleep(2 ** attempt) # 指数退避 return None6.3 性能优化建议Token 缓存合理设置 Token 缓存时间避免频繁刷新连接池对频繁调用的 API 使用连接池异步处理使用异步编程避免阻塞主线程限流控制实现请求限流避免超过 API 限制7. 生产环境部署最佳实践7.1 安全配置清单部署到生产环境前必须检查以下安全项目[ ] API Key 和 Token 是否通过环境变量管理[ ] 是否实现了 Token 加密存储[ ] 网络请求是否使用 HTTPS[ ] 是否设置了合理的超时时间[ ] 错误信息是否避免泄露敏感数据[ ] 是否实现了访问日志记录[ ] 是否有权限控制机制7.2 监控和告警配置生产环境需要完善的监控体系# monitoring/config.yaml metrics: - name: skill_execution_time description: Skill 执行时间 thresholds: warning: 5000 # 5秒 critical: 10000 # 10秒 - name: token_refresh_failures description: Token 刷新失败次数 thresholds: warning: 3 critical: 10 - name: api_error_rate description: API 错误率 thresholds: warning: 0.05 # 5% critical: 0.1 # 10%7.3 扩展方向和建议掌握了基础 Skill 开发后可以进一步学习复杂 Skill 开发集成数据库操作、文件处理等复杂功能Skill 组合实现多个 Skill 的协同工作自定义 LLM 集成接入私有化部署的大模型持久化记忆实现对话历史和上下文的长期保存可视化界面为 Agent 开发 Web 界面或聊天机器人接口在实际项目中建议先从简单的单个 Skill 开始逐步验证每个组件的可靠性再扩展到复杂的多 Skill 协作场景。Token 管理要特别注意安全性和可靠性避免因为认证问题导致整个系统不可用。通过本文的实践案例你应该已经掌握了 Agent 开发的核心概念和基本流程。下一步可以尝试为你的 Agent 添加更多实用的 Skill或者优化现有的 Token 管理机制让系统更加健壮可靠。
