Python调用硅基流动ChatGPT接口的完整指南：从API封装到异常处理

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

硅基流动 API 概述

硅基流动的 ChatGPT API 提供了基于自然语言处理的服务，适用于智能客服、内容生成、代码辅助等多种场景。与 OpenAI 官方 API 相比，其特点包括：

• 支持中文语境优化
• 提供企业级 QPS 配额管理
• 可定制化敏感词过滤规则

HTTP 调用 vs 官方 SDK

直接使用 HTTP 调用的优势在于：

1. 避免 SDK 依赖升级问题
2. 更灵活的请求定制
3. 更好的性能调优空间

而官方 SDK 的优势则是：

• 开箱即用的认证流程
• 内置重试和错误处理
• 类型安全的接口定义

核心实现

带 JWT 认证的请求封装

import time
import jwt
from typing import Dict, Optional

class AuthClient:
    """封装 JWT 认证的 HTTP 客户端"""
    def __init__(self, api_key: str, secret: str):
        self.api_key = api_key
        self.secret = secret

    def generate_token(self) -> str:
        """生成 JWT 认证 Token"""
        payload = {
            'api_key': self.api_key,
            'exp': int(time.time()) + 300
        }
        return jwt.encode(payload, self.secret, algorithm='HS256')

    def auth_header(self) -> Dict[str, str]:
        """生成认证头"""
        return {'Authorization': f'Bearer {self.generate_token()}'}

Streaming 响应处理

import json
from typing import Iterator

class StreamProcessor:
    """处理流式响应数据"""
    @staticmethod
    def parse_lines(response) -> Iterator[str]:
        """解析服务器推送的事件流"""
        for line in response.iter_lines():
            if line.startswith(b'data:'):
                yield json.loads(line[6:])

自动重试装饰器

import random
from functools import wraps
from time import sleep

def retry(max_retries=3, base_delay=1):
    """指数退避重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    retries += 1
                    delay = base_delay * (2 ** retries) + random.uniform(0, 1)
                    sleep(delay)
            raise Exception(f'Max retries ({max_retries}) exceeded')
        return wrapper
    return decorator

异步实现版本

import aiohttp
from typing import AsyncIterator

class AsyncChatClient:
    """异步 API 客户端"""
    def __init__(self, auth: AuthClient):
        self.auth = auth

    async def stream_chat(self, prompt: str) -> AsyncIterator[dict]:
        """异步流式聊天接口"""
        async with aiohttp.ClientSession() as session:
            async with session.post(
                'https://api.siliconflow.ai/v1/chat/stream',
                headers={**self.auth.auth_header(), 'Accept': 'text/event-stream'},
                json={'messages': [{'role': 'user', 'content': prompt}]}
            ) as resp:
                async for line in resp.content:
                    if line.startswith(b'data:'):
                        yield json.loads(line[6:])

关键技巧

控制响应格式

通过设置 HTTP 头控制返回格式：

• Accept: application/json – 普通 JSON 响应
• Accept: text/event-stream – 流式响应

对话上下文管理

推荐实现方式：

1. 维护对话历史数组
2. 限制最大 token 数（通常 4096）
3. 使用 FIFO 队列淘汰旧消息

Token 计算

def count_tokens(text: str) -> int:
    """近似计算 token 数量"""
    # 中文按字计算，英文按空格分词
    ch_count = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
    en_count = len(text.split()) - ch_count
    return ch_count + en_count // 2

常见问题

403 错误排查

1. API 密钥失效
2. 请求频率超限
3. 地域限制
4. 账户欠费

长文本处理

推荐策略：

1. 自动分段发送
2. 摘要前文内容
3. 使用 [继续] 标记衔接

敏感词过滤

服务端会返回：

{
    "error": "content_blocked",
    "filtered_words": ["关键词"]
}

进阶建议

1. 将会话状态保存到 Redis
2. 使用令牌桶算法实现限流
3. 暴露 /metrics 端点供 Prometheus 采集

完整的示例项目已开源在 GitHub，包含单元测试和 Docker 部署配置。实际使用中建议添加请求日志和监控告警，当 QPS 接近配额时自动降级。