基于ChatGPT构建智能体的架构设计与工程实践

1次阅读

共计 2014 个字符，预计需要花费 6 分钟才能阅读完成。

在构建基于 ChatGPT 的智能体时，开发者常面临三大核心挑战：

对话状态持久化 ：多轮对话中需要维护用户会话状态，传统无状态 API 难以跟踪复杂的对话上下文
长上下文处理 ：当对话轮次超过 10 轮时，直接拼接历史消息会导致 token 数爆炸（超过模型限制的 4k/8k/32k）
API 限流与稳定性 ：GPT- 4 的免费层每分钟仅有 3 次调用额度，生产环境需要智能的流量控制

Monolithic 架构
优点：开发调试简单，适合初期快速验证
缺点：扩展性差，对话引擎阻塞会影响整个服务

Microservices 架构 （推荐方案）

graph TD
  A[客户端] --> B[API Gateway]
  B --> C[对话引擎服务]
  B --> D[状态存储服务]
  C --> E[限流中间件]
  E --> F[ChatGPT API]

对话引擎服务
负责上下文组装和响应生成
内置对话摘要生成模块
状态存储服务
采用 Redis 存储对话状态
过期时间设置为 24 小时
限流中间件
基于令牌桶算法实现
支持多级降级策略

class DialogueStateMachine:
    def __init__(self, redis_conn):
        self.redis = redis_conn

    def get_state(self, session_id: str) -> dict:
        """ 获取当前对话状态，包含以下字段：- context: 压缩后的历史对话
        - last_active: 最后活动时间
        - metadata: 业务自定义数据
        """
        try:
            raw_data = self.redis.get(f'dialogue:{session_id}')
            return json.loads(raw_data) if raw_data else None
        except redis.RedisError as e:
            # 记录异常后降级返回空状态
            logging.error(f"Redis error: {str(e)}")
            return None

    def update_state(self, session_id: str, new_state: dict):
        """更新状态并设置 24 小时过期"""
        try:
            self.redis.setex(f'dialogue:{session_id}',
                86400,  # 24 小时
                json.dumps(new_state)
            )
        except redis.RedisError as e:
            logging.error(f"State update failed: {str(e)}")
            raise

def compress_context(messages: List[dict], max_tokens=3000) -> List[dict]:
    """
    压缩策略：1. 保留最近 3 轮完整对话
    2. 对更早的历史生成摘要
    时间复杂度：O(n)，空间复杂度：O(1)
    """
    if calculate_tokens(messages) <= max_tokens:
        return messages

    # 分割新旧消息
    recent = messages[-6:]  # 3 轮对话 (每轮 user+assistant)
    older = messages[:-6]

    # 生成摘要
    summary = generate_summary(older)

    # 组合结果
    return [{"role": "system", "content": summary}] + recent

from locust import HttpUser, task, between

class ChatbotUser(HttpUser):
    wait_time = between(1, 3)

    @task
    def send_message(self):
        self.client.post("/chat", json={
            "session_id": "test_user",
            "message": "Hello world"
        })