从零开始：使用ChatGPT API构建智能体的实践指南

1次阅读

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

智能体（Agent）是一种能够自主执行任务、与用户交互的软件实体。在自然语言处理领域，基于大语言模型（如 ChatGPT）的智能体能够理解用户意图、维护对话状态，并提供连贯的响应。ChatGPT 因其强大的语言理解与生成能力，成为构建智能体的理想选择。

ChatGPT 的优势主要体现在：

上下文理解能力强，能处理多轮对话
生成结果自然流畅，接近人类表达
API 接口简单易用，开发门槛低
支持多种定制方式，适合不同场景

构建 ChatGPT 智能体主要有以下几种方式：

直接 API 调用
优点：实现简单，适合快速验证
缺点：缺乏状态管理，上下文有限
结合后端服务的完整架构
优点：功能完整，可扩展性强
缺点：开发复杂度高
使用 LangChain 等框架
优点：提供现成工具链
缺点：学习曲线较陡

对于大多数应用场景，我们建议采用第二种方式，即在服务端维护对话状态并通过 API 与 ChatGPT 交互。

智能体需要维护对话历史以实现连贯的交互。我们使用一个简单的字典结构来存储会话状态：

class ChatAgent:
    def __init__(self):
        self.sessions = {}  # {session_id: [messages]}

    def create_session(self, session_id):
        self.sessions[session_id] = [{"role": "system", "content": "你是一个有帮助的 AI 助手"}
        ]

ChatGPT API 有 token 限制，需合理管理上下文长度。我们可以采用以下策略：

只保留最近的 N 轮对话
对长消息进行摘要
定期清除无关上下文

示例实现：

def trim_context(messages, max_tokens=4096):
    current_length = sum(len(msg["content"]) for msg in messages)
    while current_length > max_tokens and len(messages) > 1:
        # 保留系统消息，删除最早的对话
        if messages[1]["role"] != "system":
            removed = messages.pop(1)
            current_length -= len(removed["content"])
    return messages

API 调用可能因网络或速率限制失败，应实现自动重试：

import openai
from time import sleep

def safe_completion(messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=messages
            )
            return response
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            sleep(2 ** attempt)  # 指数退避

下面是一个基础智能体的完整实现：

import openai
from typing import List, Dict

class ChatGPTAgent:
    """基于 ChatGPT API 的智能体实现"""

    def __init__(self, api_key: str):
        openai.api_key = api_key
        self.sessions = {}

    def start_session(self, session_id: str, system_prompt: str = None) -> None:
        """初始化新会话"""
        self.sessions[session_id] = [{"role": "system", "content": system_prompt or "你是一个有帮助的 AI 助手"}
        ]

    def chat(self, session_id: str, user_input: str) -> str:
        """处理用户输入并返回 AI 响应"""
        if session_id not in self.sessions:
            self.start_session(session_id)

        # 添加用户消息
        self.sessions[session_id].append({"role": "user", "content": user_input})

        # 修剪上下文
        messages = self._trim_context(self.sessions[session_id])

        # 调用 API
        response = self._safe_completion(messages)

        # 添加 AI 响应
        ai_response = response.choices[0].message.content
        self.sessions[session_id].append({"role": "assistant", "content": ai_response})

        return ai_response

    def _trim_context(self, messages: List[Dict], max_tokens: int = 4096) -> List[Dict]:
        """上下文修剪"""
        # 实现同前
        ...

    def _safe_completion(self, messages: List[Dict], max_retries: int = 3):
        """安全的 API 调用"""
        # 实现同前
        ...