OpenClaw接入ChatGPT实战：从架构设计到性能调优

5次阅读

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

在将 OpenClaw 系统与 ChatGPT 集成时，开发者通常会遇到几个核心挑战：

认证复杂性 ：ChatGPT API 使用基于 token 的认证机制，需要处理密钥轮换和权限管理
会话状态维护 ：多轮对话场景需要保持上下文连贯性，而 HTTP 协议本身是无状态的
API 调用限制 ：ChatGPT 对免费和付费账户都有严格的速率限制（如 3.5 版本每分钟 20 次请求）
响应延迟 ：AI 模型推理时间不稳定，高峰期可能达到 2 - 3 秒 / 请求

我们评估了三种主流集成方案：

直接 API 调用
优点：实现简单，无需额外组件
缺点：难以处理限流和重试，业务逻辑与 API 调用耦合
官方 SDK 封装
优点：错误处理内置，支持类型提示
缺点：灵活性差，版本更新可能滞后
中间件代理（本文方案）
优点：业务解耦，可集中处理认证 / 限流 / 监控
缺点：需要额外开发维护

import backoff
import openai
from tenacity import retry, stop_after_attempt

class ChatGPTWrapper:
    def __init__(self, api_key):
        openai.api_key = api_key
        self.session_map = {}  # 会话 ID 到消息历史的映射

    @retry(stop=stop_after_attempt(3))
    async def get_completion(self, prompt, session_id=None):
        messages = self._build_messages(prompt, session_id)
        try:
            resp = await openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=messages,
                timeout=10
            )
            self._update_session(session_id, prompt, resp.choices[0].message)
            return resp
        except openai.error.RateLimitError:
            # 触发指数退避重试
            raise

    def _build_messages(self, prompt, session_id):
        """构建包含历史上下文的 message 数组"""
        if session_id and session_id in self.session_map:
            return self.session_map[session_id] + [{"role": "user", "content": prompt}]
        return [{"role": "user", "content": prompt}]

关键设计点：

使用 tenacity 库实现带指数退避的自动重试
通过 session_id 维护多轮对话上下文
异步 IO 提升并发能力

import redis
from pickle import dumps, loads

class SessionManager:
    def __init__(self, redis_url):
        self.redis = redis.from_url(redis_url)
        self.ttl = 3600  # 1 小时过期

    def save_session(self, session_id, messages):
        """序列化存储消息历史"""
        self.redis.setex(f"chat:{session_id}",
            self.ttl,
            dumps(messages)
        )

    def load_session(self, session_id):
        """反序列化加载消息历史"""
        data = self.redis.get(f"chat:{session_id}")
        return loads(data) if data else None

使用 Redis 令牌桶算法实现每分钟限流：

class RateLimiter:
    def __init__(self, redis, max_tokens=20):
        self.redis = redis
        self.max_tokens = max_tokens

    async def acquire_token(self, key):
        """返回剩余令牌数，0 表示被限流"""
        lua_script = """local current = tonumber(redis.call('get', KEYS[1]))
        if current == nil then
            current = ARGV[1]
            redis.call('setex', KEYS[1], 60, current-1)
            return current
        end
        if current > 0 then
            redis.call('decr', KEYS[1])
            return current
        end
        return current
        """
        return await self.redis.eval(lua_script, 1, key, self.max_tokens)

使用 FastAPI 构建异步端点：

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.post("/chat")
async def chat_endpoint(request: ChatRequest, background_tasks: BackgroundTasks):
    # 限流检查
    remaining = await rate_limiter.acquire_token("global_rate_limit")
    if remaining <= 0:
        return {"error": "Rate limit exceeded"}

    # 异步执行避免阻塞事件循环
    background_tasks.add_task(chat_service.process, request)
    return {"status": "processing"}

对相同 prompt 进行 MD5 缓存：

import hashlib

def get_cache_key(prompt):
    return hashlib.md5(prompt.encode()).hexdigest()

async def get_cached_response(prompt):
    key = f"cache:{get_cache_key(prompt)}"
    cached = await redis.get(key)
    return loads(cached) if cached else None

调整 OpenAI 客户端参数：

import httpx

transport = httpx.AsyncHTTPTransport(
    retries=3,
    max_connections=100,
    max_keepalive_connections=20
)

client = openai.AsyncOpenAI(http_client=httpx.AsyncClient(transport=transport)
)