ChatGPT智能体构建实战：从零搭建可对话AI系统的技术解析

1次阅读

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

在自建 ChatGPT 智能体的过程中，开发者通常会遇到三个主要痛点：

对话断层 （Dialogue Breakdown）：当对话轮次增多时，模型容易丢失上下文，导致回复不连贯。
意图漂移 （Intent Drift）：在多轮对话中，用户的真实意图可能被模型误解或忽略。
API 调用成本 （API Cost）：频繁调用 GPT API 会导致高昂的费用，尤其在并发场景下。

这些问题直接影响智能体的用户体验和生产可用性。

纯 API 调用 ：简单直接，但成本高且难以控制对话流程。
混合架构 （API + 本地逻辑）：通过本地模块管理对话状态和意图识别，减少 API 调用次数。

sequenceDiagram
    participant User
    participant Agent
    participant LocalModule
    participant GPTAPI

    User->>Agent: 发送消息
    Agent->>LocalModule: 意图识别
    LocalModule-->>Agent: 意图结果
    Agent->>GPTAPI: 调用 API（必要时）GPTAPI-->>Agent: API 响应
    Agent->>User: 返回回复

from functools import lru_cache

class DialogueManager:
    def __init__(self, max_history=5):
        self.max_history = max_history

    @lru_cache(maxsize=100)
    def get_recent_history(self, user_id: str) -> list:
        """获取最近对话历史，LRU 缓存优化"""
        # 实现从数据库或缓存获取历史
        return []

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

class IntentRecognizer:
    def __init__(self):
        self.vectorizer = TfidfVectorizer()
        self.intent_templates = {"greeting": ["你好", "嗨"],
            "goodbye": ["再见", "拜拜"]
        }

    def recognize(self, text: str) -> str:
        """识别用户意图"""
        # 训练向量化器
        templates = sum(self.intent_templates.values(), [])
        self.vectorizer.fit(templates)

        # 计算相似度
        vec = self.vectorizer.transform([text])
        for intent, phrases in self.intent_templates.items():
            template_vec = self.vectorizer.transform(phrases)
            if cosine_similarity(vec, template_vec).max() > 0.7:
                return intent
        return "unknown"

import asyncio
from openai import AsyncOpenAI

class GPTAsyncClient:
    def __init__(self):
        self.client = AsyncOpenAI()

    async def batch_call(self, messages_list: list) -> list:
        """批量处理 API 调用"""
        tasks = [self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=msgs
        ) for msgs in messages_list]
        return await asyncio.gather(*tasks)

Rate Limiting（速率限制）：实现令牌桶算法控制请求频率
降级方案 ：
缓存常用回复
在达到限额时切换到轻量级模型

from prometheus_client import Counter, Gauge

# 定义指标
API_CALLS = Counter('gpt_api_calls', 'API 调用次数')
TOKENS_USED = Gauge('gpt_tokens_used', '消耗的令牌数')

# 在 API 调用处添加指标记录
API_CALLS.inc()
TOKENS_USED.set(token_count)