以下内容节选自我的实战课程《从0到1教你搭建一个基于微信小程序的AI智能体应用平台》，课程包含详细代码和讲解，链接如下： https://edu.csdn.net/course/detail/40753

课程目标

• 集成 Coze 等 AI 平台接口，完成统一请求与格式化响应
• 管理 多智能体路由与参数
• 建立 稳定可靠的 AI 服务调用体系

知识要点

配置管理

• bots_config.json：智能体配置文件，包含各 AI 平台的参数
• 环境变量：API 密钥、端点地址等敏感信息管理
• 密钥管理：支持多环境、多平台的密钥轮换
• 配置热更新：支持运行时配置更新，无需重启服务

统一客户端设计

• 请求封装：统一的请求格式和参数处理
• 重试机制：指数退避算法，支持自定义重试策略
• 超时控制：连接超时、读取超时、总超时时间管理
• 日志埋点：详细的请求日志，包含性能指标和错误信息
• 异常分级：根据错误类型进行分级处理和响应

响应规范化

• 内容提取：从不同平台响应中提取统一格式内容
• 引用处理：处理 AI 回复中的引用和来源信息
• Token 统计：准确统计输入和输出的 token 数量
• 敏感内容拦截：实时检测和过滤敏感内容

实战步骤详解

1. AI 客户端模块设计

基础客户端类

import requests
import json
import time
import hashlib
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Generator
from dataclasses import dataclass
from enum import Enum

class AIProvider(Enum):
    COZE = "coze"
    OPENAI = "openai"
    CLAUDE = "claude"
    CUSTOM = "custom"

@dataclass
class AIRequest:
    """AI 请求数据结构"""
    user_id: str
    bot_id: str
    message: str
    image_url: Optional[str] = None
    session_id: Optional[str] = None
    temperature: float = 0.7
    max_tokens: int = 2000
    stream: bool = True

@dataclass
class AIResponse:
    """AI 响应数据结构"""
    content: str
    provider: AIProvider
    model: str
    input_tokens: int
    output_tokens: int
    cost: float
    response_time: float
    request_id: str
    references: List[Dict] = None
    is_sensitive: bool = False

class AIClientBase(ABC):
    """AI 客户端基类"""
   
    def __init__(self, config: Dict):
        self.config = config
        self.api_key = config.get('api_key')
        self.base_url = config.get('base_url')
        self.model = config.get('model')
        self.timeout = config.get('timeout', 30)
        self.max_retries = config.get('max_retries', 3)
        self.retry_delay = config.get('retry_delay', 1)
       
    @abstractmethod
    def generate_response(self, request: AIRequest) -> AIResponse:
        """生成响应"""
        pass
   
    @abstractmethod
    def generate_stream_response(self, request: AIRequest) -> Generator[Dict, None, None]:
        """生成流式响应"""
        pass
   
    def _make_request(self, url: str, data: Dict, headers: Dict) -> requests.Response:
        """发起 HTTP 请求"""
        for attempt in range(self.max_retries):
            try:
                response = requests.post(
                    url,
                    json=data,
                    headers=headers,
                    timeout=self.timeout
                )
                response.raise_for_status()
                return response
            except requests.exceptions.RequestException as e:
                if attempt == self.max_retries - 1:
                    raise
                time.sleep(self.retry_delay * (2 ** attempt))
       
    def _generate_request_id(self, request: AIRequest) -> str:
        """生成请求 ID"""
        content = f"{request.user_id}_{request.bot_id}_{request.message}_{time.time()}"
        return hashlib.md5(content.encode()).hexdigest()

2. 多智能体路由系统

智能体配置管理

class BotConfigManager:
    """智能体配置管理器"""
   
    def __init__(self, config_file: str = 'bots_config.json'):
        self.config_file = config_file
        self.configs = self._load_configs()
       
    def _load_configs(self) -> Dict:
        """加载配置文件"""
        try:
            with open(self.config_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        except FileNotFoundError:
            return {}
        except json.JSONDecodeError as e:
            raise ConfigError(f"Invalid config file: {e}")
   
    def get_bot_config(self, bot_id: str) -> Dict:
        """获取指定智能体配置"""
        if bot_id not in self.configs:
            raise ConfigError(f"Bot {bot_id} not found")
       
        config = self.configs[bot_id]
       
        # 验证必需字段
        required_fields = ['provider', 'model', 'api_key']
        for field in required_fields:
            if field not in config:
                raise ConfigError(f"Missing required field: {field}")
       
        return config

3. 响应格式化与处理

响应标准化

class ResponseFormatter:
    """响应格式化器"""
   
    @staticmethod
    def format_response(response: AIResponse) -> Dict:
        """格式化响应为标准格式"""
        return {
            'content': response.content,
            'provider': response.provider.value,
            'model': response.model,
            'usage': {
                'input_tokens': response.input_tokens,
                'output_tokens': response.output_tokens,
                'total_tokens': response.input_tokens + response.output_tokens
            },
            'cost': response.cost,
            'response_time': response.response_time,
            'request_id': response.request_id,
            'references': response.references or [],
            'is_sensitive': response.is_sensitive
        }

4. 保护机制实现

限流器

class RateLimiter:
    """限流器"""
   
    def __init__(self, max_requests: int = 100, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = {}  # user_id -> [timestamps]
   
    def is_allowed(self, user_id: str) -> bool:
        """检查是否允许请求"""
        now = time.time()
        user_requests = self.requests.get(user_id, [])
       
        # 清理过期请求
        user_requests = [req_time for req_time in user_requests
                        if now - req_time < self.time_window]
       
        if len(user_requests) >= self.max_requests:
            return False
       
        # 记录新请求
        user_requests.append(now)
        self.requests[user_id] = user_requests
       
        return True

验收标准

功能验收

• 支持多个 AI 平台的统一调用
• 智能体路由正确工作
• 响应格式统一规范
• 敏感内容检测有效

性能验收

• 常见错误（超时、网络、配额）被正确捕获
• 重试机制工作正常
• 限流和熔断机制有效
• 响应时间在可接受范围内

日志验收

• 包含 request_id、botId、耗时等关键信息
• 错误分布统计准确
• 性能指标记录完整
• 支持日志查询和分析

连接池实现方案说明：

为提升 AI API 调用的并发性能与资源利用率，推荐采用连接池（如 requests.Session 或第三方库如 httpx.AsyncClient 的池化管理）统一复用 HTTP 连接。每个 AI 平台可为其分配独立的 Session/Client 实例，集中管理连接生命周期，避免频繁建立和销毁 TCP 连接带来的开销。连接池支持最大连接数、空闲超时等参数配置，能自动回收失效连接，提升稳定性。对于高并发场景，建议结合异步池（如 aiohttp.TCPConnector）实现高效的异步请求。统一的连接池管理还便于后续扩展熔断、重试等高级特性，保障服务高可用和低延迟。