ChatGPT智能体开发实战：从零构建你的第一个AI助手

1次阅读

没有评论

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

传统的聊天机器人主要依赖规则匹配和意图识别，需要人工编写大量对话树和响应模板。这种方式存在几个核心痛点：

对话灵活性差，无法处理用户超出预设范围的提问
上下文理解能力弱，多轮对话容易丢失信息
维护成本高，每次新增功能都需要修改对话逻辑

基于 GPT 的智能体通过大语言模型的涌现能力，从根本上改变了这种状况。它们可以：

理解自然语言表达的复杂意图
自主生成符合语境的连贯回复
通过 prompt engineering 实现零样本学习 (few-shot learning)

开发智能体时常见的技术方案对比：

方案	响应延迟	成本控制	灵活性	适用场景
原生 OpenAI API	低	中等	高	快速原型开发
LangChain	中	高	极高	复杂业务流程编排
自部署模型	高	低 (长期)	极高	数据敏感型业务

对于初学者，建议从原生 API 开始：

不需要处理模型部署的复杂性
API 文档完善，社区支持好
按使用量计费，试错成本低

import openai
from typing import List, Dict

class ChatAgent:
    """
    GPT 智能体基础实现

    Attributes:
        api_key: OpenAI 账户密钥
        model: 使用的模型版本
        temperature: 控制生成随机性 (0-2)
    """def __init__(self, api_key: str, model: str ="gpt-3.5-turbo"):
        openai.api_key = api_key
        self.model = model
        self.conversation_history: List[Dict] = []

    def add_message(self, role: str, content: str) -> None:
        """添加对话历史记录"""
        self.conversation_history.append({"role": role, "content": content})

保持 3 轮对话记忆的关键技术：

维护对话历史数组
智能截断过长的上下文
自动清理无关对话片段

def generate_response(self, prompt: str, max_tokens: int = 150) -> str:
    """
    生成智能体回复

    Args:
        prompt: 用户输入内容
        max_tokens: 限制生成长度

    Returns:
        生成的回复文本
    """self.add_message("user", prompt)

    try:
        response = openai.ChatCompletion.create(
            model=self.model,
            messages=self._get_optimized_context(),
            max_tokens=max_tokens,
            temperature=0.7
        )
        reply = response.choices[0].message.content
        self.add_message("assistant", reply)
        return reply
    except Exception as e:
        print(f"API 调用失败: {str(e)}")
        return "服务暂时不可用"

def _get_optimized_context(self) -> List[Dict]:
    """优化对话上下文避免 token 超限"""
    total_length = sum(len(m["content"]) for m in self.conversation_history)

    # 简易截断策略
    while total_length > 3000 and len(self.conversation_history) > 3:
        removed = self.conversation_history.pop(1)  # 保留系统 prompt
        total_length -= len(removed["content"])

    return self.conversation_history

在返回最终响应前添加安全检查：

from better_profanity import profanity

class SafeChatAgent(ChatAgent):
    """带内容安全检查的智能体"""
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        profanity.load_censor_words()

    def generate_response(self, prompt: str, *args, **kwargs) -> str:
        # 先检查用户输入
        if self._contains_sensitive_info(prompt):
            return "您的问题包含不合适内容，请重新输入"

        reply = super().generate_response(prompt, *args, **kwargs)

        # 再检查模型输出
        return profanity.censor(reply)

    def _contains_sensitive_info(self, text: str) -> bool:
        """检查敏感内容（示例简化版）"""
        forbidden_phrases = ["信用卡号", "密码", "身份证"]
        return any(phrase in text for phrase in forbidden_phrases)

from flask import Flask, request, jsonify
import asyncio
from concurrent.futures import ThreadPoolExecutor

app = Flask(__name__)
executor = ThreadPoolExecutor(4)
agent = SafeChatAgent("your_api_key")

@app.route('/chat', methods=['POST'])
def chat_endpoint():
    """异步处理聊天请求"""
    data = request.json
    loop = asyncio.new_event_loop()

    try:
        # 使用线程池避免阻塞
        reply = loop.run_in_executor(
            executor, 
            lambda: agent.generate_response(data["message"])
        ).result()

        return jsonify({"reply": reply})
    except Exception as e:
        app.logger.error(f"处理失败: {str(e)}")
        return jsonify({"error": "处理请求时出错"}), 500

关键优化点：

使用线程池处理同步 API 调用
添加完善的错误日志记录
实现 HTTP 层级的重试机制

建议在项目中集成以下监控：

import tiktoken

def calculate_token_usage(messages: List[Dict]) -> int:
    """计算对话消耗的 token 数量"""
    encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")
    total = 0

    for msg in messages:
        total += len(encoding.encode(msg["content"]))

    return total

# 在 generate_response 方法中添加统计
self.daily_token_usage += calculate_token_usage(self.conversation_history)

temperature：
客服场景建议 0.2-0.5（稳定）
创意生成建议 0.7-1.0（多样）
max_tokens：
对话场景限制在 150-300
长文本生成可设为 800-1000

使用令牌桶算法控制请求频率：

from ratelimit import limits, sleep_and_retry

# 每分钟最多 30 次调用
@sleep_and_retry
@limits(calls=30, period=60)
def rate_limited_generate(*args, **kwargs):
    return agent.generate_response(*args, **kwargs)

现象：API 返回 ”context length exceeded” 错误

解决方案 ：