ChatGPT智能体开发实战：从零构建到生产环境部署

1次阅读

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

在开发基于 ChatGPT 的智能体时，开发者常遇到几个核心挑战：

对话状态丢失 ：API 默认无状态，用户刷新页面后历史对话消失
长上下文处理 ：超过 token 限制时关键信息被截断
意图识别漂移 ：多轮对话中用户意图可能发生偏移
生产环境稳定性 ：突发流量下的 API 限流和错误恢复

这些痛点直接影响用户体验，本文将从工程角度提供解决方案。

# 最简实现（存在状态丢失问题）import openai

def chat(prompt):
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

缺点：
1. 每次请求都是独立会话
2. 无法利用对话历史
3. 无错误恢复机制

flowchart LR
    A[用户输入] --> B[对话状态管理器]
    B --> C{是否需要检索}
    C -->| 是 | D[向量数据库查询]
    C -->| 否 | E[GPT 上下文组装]
    D --> E
    E --> F[API 调用]
    F --> G[响应处理]
    G --> H[状态持久化]

优势：
1. 支持知识库扩展
2. 可控制上下文长度
3. 实现意图识别增强

from uuid import uuid4
from datetime import datetime
import redis  # 需要 pip 安装

class DialogManager:
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379, db=0)

    def create_session(self, user_id: str) -> str:
        """生成带时间戳的会话 ID"""
        session_id = f"{user_id}_{datetime.now().timestamp()}"
        self.redis.hset(session_id, "created_at", datetime.now().isoformat())
        return session_id

    def save_context(self, session_id: str, messages: list[dict]):
        """存储对话上下文"""
        self.redis.hset(session_id, "context", pickle.dumps(messages))

    def load_context(self, session_id: str) -> list[dict]:
        """加载历史对话"""
        data = self.redis.hget(session_id, "context")
        return pickle.loads(data) if data else []

import backoff  # 需要 pip 安装
from openai.error import APIConnectionError, RateLimitError

@backoff.on_exception(backoff.expo, 
                     (APIConnectionError, RateLimitError),
                     max_tries=3)
def safe_chat_completion(messages):
    try:
        return openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0.7  # 控制创造性
        )
    except Exception as e:
        log_error(e)
        return get_fallback_response()

def trim_context(messages: list[dict], max_tokens=4096) -> list[dict]:
    """
    基于 token 计算的上下文裁剪
    假设：- 英文 1token≈4 字符
    - 中文 1token≈2 字符
    """
    total = 0
    trimmed = []

    for msg in reversed(messages):
        content = msg["content"]
        tokens = len(content.encode()) // 3  # 简化估算
        if total + tokens > max_tokens:
            break
        trimmed.insert(0, msg)
        total += tokens

    return trimmed

TP99 响应时间 ：控制在 2 秒内
并发吞吐量 ：单实例至少支持 50RPS
错误率 ：<0.5%

from presidio_analyzer import AnalyzerEngine  # 需要 pip 安装

def sanitize_input(text: str) -> str:
    analyzer = AnalyzerEngine()
    results = analyzer.analyze(text=text, language="en")

    for result in results:
        text = text.replace(result.text, "[REDACTED]")
    return text