ChatGPT智能体开发实战：从零构建可扩展的对话系统架构

1次阅读

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

最近在尝试用 ChatGPT API 开发智能体时，遇到了几个头疼的问题：

上下文丢失：长对话中模型突然『失忆』
业务逻辑耦合：天气查询、日程管理代码搅在一起
token 消耗失控：简单对话就烧掉大量 token

经过多次迭代，终于摸索出一套分层架构方案，今天分享这个可扩展的实现框架。

flowchart TD
    A[请求编排层] -->| 解析意图 | B[记忆管理层]
    B -->| 读取上下文 | C[业务逻辑层]
    C -->| 返回结果 | A

负责对话流程控制，核心功能：

用户输入预处理
意图识别路由
响应结果格式化

关键技术点：

LRU 缓存最近 10 轮对话
关键实体记忆持久化
自动摘要压缩历史

通过函数调用实现：

原子 API 能力封装
独立计费单元
熔断降级机制

import openai
from typing import Literal

# 函数定义遵循 OpenAI 规范
weather_func = {
    "name": "get_current_weather",
    "description": "Get the current weather",
    "parameters": {
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "城市名称"
            },
            "unit": {
                "type": "string",
                "enum": ["celsius", "fahrenheit"]
            }
        },
        "required": ["location"]
    }
}

# 实际业务函数
def get_weather(location: str, unit: Literal['celsius','fahrenheit'] = 'celsius'):
    """模拟天气 API 调用"""
    print(f"查询 {location} 的天气（单位:{unit}）")
    return {"temperature": 25, "unit": unit}

# 对话处理器
class WeatherAgent:
    def __init__(self):
        self.memory = LRUCache(maxsize=5)  # 后文实现

    def chat(self, user_input: str):
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-0613",
            messages=[{"role": "system", "content": "你是一个天气助手"},
                *self.memory.get_context(),
                {"role": "user", "content": user_input}
            ],
            functions=[weather_func]
        )

        # 处理函数调用
        if response.choices[0].finish_reason == "function_call":
            func_name = response.choices[0].message.function_call.name
            args = json.loads(response.choices[0].message.function_call.arguments)

            if func_name == "get_current_weather":
                result = get_weather(**args)
                # 将结果存入记忆
                self.memory.add(
                    role="function",
                    name=func_name,
                    content=json.dumps(result)
                )
                return f"{args['location']}天气: {result['temperature']}°{args.get('unit','celsius')}"

from collections import OrderedDict
import json

class LRUCache:
    """基于 OrderedDict 实现的对话记忆缓存"""
    def __init__(self, maxsize: int = 10):
        self.cache = OrderedDict()
        self.maxsize = maxsize

    def add(self, role: str, content: str, name: str = None):
        """添加对话记录"""
        new_item = {"role": role, "content": content}
        if name:
            new_item["name"] = name

        # 自动清理旧记录
        if len(self.cache) >= self.maxsize:
            self.cache.popitem(last=False)

        self.cache[json.dumps(new_item)] = new_item
        self.cache.move_to_end(json.dumps(new_item))

    def get_context(self) -> list:
        """获取最近 N 条对话上下文"""
        return list(self.cache.values())

def count_tokens(text: str, model: str = "gpt-3.5-turbo") -> int:
    """更精确的 token 计数方法"""
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        encoding = tiktoken.get_encoding("cl100k_base")

    # 特殊字符处理    
    text = text.replace('<|im_end|>', '').replace('<|im_start|>','')
    return len(encoding.encode(text))

# 使用示例
input_text = "今天北京天气怎么样？"
print(f"Token 消耗: {count_tokens(input_text)}")

上下文窗口大小	平均响应延迟(ms)	Token 消耗 / 次
5 轮对话	1200	850
10 轮对话	1800	1500
20 轮对话	超时风险	3000+

记忆压缩算法效果对比（基于 GPT-3.5-turbo）：

原始历史：15 轮对话，占用 2800token
摘要压缩后：等效信息，仅消耗 600token

# 错误示范
async def handle_message(msg):
    global memory  # 多线程竞争风险
    memory.add(msg)

# 正确做法
from threading import Lock
memory_lock = Lock()

async def safe_add(msg):
    with memory_lock:
        memory.add(msg)

推荐处理流程：