Python调用硅基流动ChatGPT API实战指南：从认证到流式响应处理

1次阅读

没有评论

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

在人工智能应用开发中，调用大模型 API 已成为常见需求。本文将详细介绍如何使用 Python 高效调用硅基流动 ChatGPT API，解决开发者在实际集成过程中遇到的典型问题。

在对接硅基流动 ChatGPT API 时，开发者常会遇到以下三大问题：

认证令牌过期：OAuth2.0 令牌的有效期有限，需要妥善处理刷新逻辑
流式响应截断 ：Server-Sent Events(SSE) 数据流的解析和异常处理较为复杂
速率限制：API 有严格的调用频率限制，需要完善的错误重试机制

在处理长连接和流式响应时，Python 常用的 HTTP 客户端各有优劣：

requests：同步请求，简单易用，但不适合高并发场景
aiohttp：异步 IO 支持好，适合高并发，但学习曲线较陡
httpx：同步 / 异步双模式，兼容 requests API，推荐首选

import requests
from datetime import datetime, timedelta

class AuthManager:
    def __init__(self, client_id, client_secret):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token = None
        self.expires_at = None
        self.session = requests.Session()

    def get_token(self):
        if self.token and datetime.now() < self.expires_at:
            return self.token

        # 获取新令牌
        auth_url = "https://api.example.com/oauth/token"
        response = self.session.post(
            auth_url,
            auth=(self.client_id, self.client_secret),
            data={"grant_type": "client_credentials"}
        )
        response.raise_for_status()

        token_data = response.json()
        self.token = token_data["access_token"]
        self.expires_at = datetime.now() + timedelta(seconds=token_data["expires_in"] - 60)  # 提前 60 秒刷新
        return self.token

from typing import List, Dict

class MessageQueue:
    def __init__(self, max_length: int = 10):
        self.messages: List[Dict] = []
        self.max_length = max_length

    def add_message(self, role: str, content: str) -> None:
        self.messages.append({"role": role, "content": content})
        if len(self.messages) > self.max_length:
            self.messages.pop(0)

    def clear(self) -> None:
        self.messages.clear()

    def to_list(self) -> List[Dict]:
        return self.messages.copy()

import json
from typing import Iterator, Optional

def parse_sse_stream(response) -> Iterator[Optional[dict]]:
    buffer = ""
    for chunk in response.iter_content(chunk_size=1024):
        if not chunk:
            continue

        buffer += chunk.decode("utf-8")
        while "\n\n" in buffer:
            event, buffer = buffer.split("\n\n", 1)
            if not event.startswith("data:"):
                continue

            event_data = event[6:]  # 去掉 "data:" 前缀
            if event_data == "[DONE]":
                yield None
                return

            try:
                yield json.loads(event_data)
            except json.JSONDecodeError:
                continue

from functools import wraps
import time
import requests
from typing import Iterator, Optional, Dict, Any

class ChatGPTClient:
    def __init__(self, client_id: str, client_secret: str):
        self.auth = AuthManager(client_id, client_secret)
        self.session = requests.Session()
        self.base_url = "https://api.example.com/chat/completions"

    def retry_on_failure(max_retries=3, delay=1):
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                retries = 0
                while retries < max_retries:
                    try:
                        return func(*args, **kwargs)
                    except requests.exceptions.HTTPError as e:
                        if e.response.status_code == 429:  # 速率限制
                            retries += 1
                            time.sleep(delay * (2 ** retries))  # 指数退避
                        else:
                            raise
                raise Exception(f"Max retries ({max_retries}) exceeded")
            return wrapper
        return decorator

    @retry_on_failure()
    def chat_stream(self, messages: List[Dict], model: str = "gpt-3.5-turbo") -> Iterator[Optional[Dict]]:
        headers = {"Authorization": f"Bearer {self.auth.get_token()}",
            "Accept": "text/event-stream",
            "Cache-Control": "no-cache",
            "Connection": "keep-alive"
        }

        payload = {
            "model": model,
            "messages": messages,
            "stream": True,
            "temperature": 0.7
        }

        with self.session.post(
            self.base_url,
            headers=headers,
            json=payload,
            stream=True,
            timeout=30
        ) as response:
            response.raise_for_status()
            yield from parse_sse_stream(response)