中科院ChatGPT官网接入指南：从API调用到实战避坑

2次阅读

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

中科院 ChatGPT 官网 API 为开发者提供了智能对话能力，适用于客服系统自动化、教育领域智能辅导以及企业内部知识库问答等场景。与 OpenAI 官方 API 相比，它在认证机制、访问限制和模型参数上存在显著差异，更适合国内开发者使用。本文将带你从零开始实现一个稳定可靠的中科院 ChatGPT 接入方案。

认证方式：中科院采用 JWT+HMAC-SHA256 签名双重验证，而 OpenAI 使用简单的 API Key
QPS 限制：中科院默认每秒 5 次请求（可申请提升），OpenAI 免费版为 3 次 / 分钟
模型参数 ：中科院提供GPT-3.5-CN 专用中文优化模型，OpenAI 则以 text-davinci-003 等通用模型为主

import time
import hmac
import hashlib
from typing import Tuple
from datetime import datetime, timedelta

class AuthManager:
    def __init__(self, api_key: str, secret: str):
        self.api_key = api_key
        self.secret = secret.encode('utf-8')
        self._token_expire = None
        self._current_token = None

    def generate_token(self) -> Tuple[str, datetime]:
        """生成带过期时间的 JWT"""
        expire = datetime.now() + timedelta(minutes=30)
        payload = f"{self.api_key}:{int(expire.timestamp())}"
        signature = hmac.new(
            self.secret, 
            payload.encode('utf-8'), 
            hashlib.sha256
        ).hexdigest()
        return f"{payload}:{signature}", expire

    def get_token(self) -> str:
        """获取有效 token，自动刷新过期凭证"""
        if not self._current_token or datetime.now() >= self._token_expire:
            try:
                self._current_token, self._token_expire = self.generate_token()
            except Exception as e:
                raise RuntimeError(f"Token 生成失败: {str(e)}")
        return self._current_token

import asyncio
import websockets
from typing import AsyncGenerator

async def stream_chat(
    message: str, 
    api_url: str, 
    auth: AuthManager
) -> AsyncGenerator[str, None]:
    """流式对话消息处理"""
    headers = {"Authorization": f"Bearer {auth.get_token()}",
        "X-Request-ID": str(uuid.uuid4())
    }

    async with websockets.connect(api_url) as ws:
        await ws.send(json.dumps({
            "message": message,
            "stream": True
        }))

        while True:
            try:
                data = await asyncio.wait_for(ws.recv(), timeout=30)
                yield json.loads(data)['content']
            except asyncio.TimeoutError:
                break

from threading import Lock
import time

class TokenBucket:
    """令牌桶限流算法"""
    def __init__(self, capacity: int, fill_rate: float):
        self.capacity = capacity
        self._tokens = capacity
        self.fill_rate = fill_rate  # 令牌 / 秒
        self.last_time = time.time()
        self.lock = Lock()

    def consume(self, tokens: int = 1) -> bool:
        """消费令牌，返回是否允许请求"""
        with self.lock:
            self._refill()
            if self._tokens >= tokens:
                self._tokens -= tokens
                return True
            return False

    def _refill(self):
        """根据时间补充令牌"""
        now = time.time()
        elapsed = now - self.last_time
        self._tokens = min(
            self.capacity,
            self._tokens + elapsed * self.fill_rate
        )
        self.last_time = now

在请求头中添加唯一标识：

import logging

logging.basicConfig(format='%(asctime)s [%(trace_id)s] %(message)s',
    level=logging.INFO
)

class RequestLogger:
    def __init__(self):
        self.trace_id = None

    def set_trace_id(self, trace_id: str):
        self.trace_id = trace_id
        logging.LoggerAdapter(logging.getLogger(), 
            {'trace_id': trace_id}
        )

import re

def sanitize_text(text: str) -> str:
    """脱敏手机号、身份证等敏感信息"""
    # 手机号
    text = re.sub(r'(1[3-9])\d{9}', r'\1******', text)
    # 身份证
    text = re.sub(r'[1-9]\d{5}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[0-9Xx]', 
                 'ID_CARD_NUMBER', text)
    return text

对话状态持久化：是否采用 Redis 存储会话上下文？如何设计 TTL 机制？
429 错误处理：基于业务重要性选择重试策略（立即重试 / 退避重试 / 降级响应）
内容合规验证：是否需要部署本地审核模型进行二次过滤？

通过本文的实践方案，我们实现了包含认证管理、流式交互和限流保护的完整接入方案。在实际部署时，建议配合 Connection Pooling 优化 HTTP 性能，并对关键操作实施 Backoff Retry 策略。对于 Zero-shot Prompting 场景，需要特别注意提示词的中英文混合编码问题。

正文完