企业官网建设流程全解析

在 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 和 Token

1.1 Agent 是什么，解决了什么问题

Agent 本质上是一个能够理解用户意图、制定计划、执行动作并返回结果的智能程序。与传统的聊天机器人不同，真正的 Agent 具备自主决策能力，能够根据上下文选择不同的工具和策略完成任务。

在实际项目中，Agent 通常包含三个核心组件：

• 大脑（Brain）：负责理解用户输入、制定计划、决策下一步动作
• 技能（Skill）：Agent 可以调用的具体能力，如查询天气、发送邮件、分析数据
• 记忆（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 requests

2.2 配置环境变量和认证信息

为了避免在代码中硬敏感信息，使用环境变量管理 API Key 等配置：

# 创建 .env 文件 echo "OPENAI_API_KEY=your_openai_api_key_here" > .env echo "WEATHER_API_KEY=your_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.py

3. 编写第一个 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 所需的参数格式""" pass

3.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__( name="weather_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" }, timeout=10 ) 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": f"API 请求失败: {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(seconds=expires_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 相关错误：

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( algorithm=hashes.SHA256(), length=32, salt=self.salt, iterations=100000, ) 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 None

5. 构建完整的 Weather Agent

5.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(model="gpt-3.5-turbo", temperature=0) 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( name=skill.name, description=skill.description, func=self._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(agent=agent, tools=tools, verbose=True) 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(f"Agent: {response}") await asyncio.sleep(1) # 避免请求过快 if __name__ == "__main__": asyncio.run(main())

6. 常见问题排查与解决方案

6.1 Skill 执行失败排查

问题现象：Skill 返回错误或超时

排查步骤：

1. 检查参数验证逻辑
2. 验证 API Token 有效性
3. 测试网络连接和超时设置
4. 查看完整的错误日志

# 添加详细的日志记录 import logging logging.basicConfig(level=logging.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 失效、刷新失败、权限错误

解决方案：

1. 实现 Token 自动刷新机制
2. 添加降级方案和备用 Token
3. 完善的错误提示和日志记录

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 None

6.3 性能优化建议

1. Token 缓存：合理设置 Token 缓存时间，避免频繁刷新
2. 连接池：对频繁调用的 API 使用连接池
3. 异步处理：使用异步编程避免阻塞主线程
4. 限流控制：实现请求限流，避免超过 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 开发后，可以进一步学习：

1. 复杂 Skill 开发：集成数据库操作、文件处理等复杂功能
2. Skill 组合：实现多个 Skill 的协同工作
3. 自定义 LLM 集成：接入私有化部署的大模型
4. 持久化记忆：实现对话历史和上下文的长期保存
5. 可视化界面：为 Agent 开发 Web 界面或聊天机器人接口

在实际项目中，建议先从简单的单个 Skill 开始，逐步验证每个组件的可靠性，再扩展到复杂的多 Skill 协作场景。Token 管理要特别注意安全性和可靠性，避免因为认证问题导致整个系统不可用。

通过本文的实践案例，你应该已经掌握了 Agent 开发的核心概念和基本流程。下一步可以尝试为你的 Agent 添加更多实用的 Skill，或者优化现有的 Token 管理机制，让系统更加健壮可靠。