首页 > 编程语言 >Python多模型统一调用封装类，兼容GPT/Claude/Grok

Python多模型统一调用封装类，兼容GPT/Claude/Grok

来源：互联网 2026-07-21 07:49:02
Claude资讯合集

面向GPT、Claude、Grok三种大模型，设计统一调用封装类，仅需切换模型标识即可自动适配各SDK的入参、响应解析与异常捕获，内置超时与重试机制，返回标准化字典结构，大幅降低多模型切换的维护成本，适合批量开发与API网关场景。

做AI开发的朋友都知道，切换不同大模型时各家SDK的入参、返回格式、异常处理差异巨大，每次都要改大量代码，维护成本高得离谱。无论是OpenAI GPT、Anthropic Claude，还是XAI Grok，没有一个统一的标准。今天分享一个统一调用类的封装，仅需切换模型标识，底层自动适配三家模型的参数、响应解析与异常捕获，同时内置超时、重试和统一返回结构。适合批量开发、API网关中转等场景，不用再为每个模型单独写一套调用逻辑。

Python多模型统一调用封装类，兼容GPT/Claude/Grok

长期稳定更新的攒劲资源： >>>点此立即查看<<<

环境依赖安装

# 安装三家官方SDK
pip install openai anthropic xai python-dotenv

环境版本：

  • Python：3.10+
  • openai：>=1.35.0
  • anthropic：>=0.25.0
  • xai：>=0.1.0

配置说明

使用.env文件统一管理各家密钥，避免硬编码密钥，方便多环境切换。新建 .env 文件：

# OpenAI GPT
OPENAI_API_KEY=sk-xxx
# Claude
ANTHROPIC_API_KEY=sk-ant-xxx
# Grok
XAI_API_KEY=xai-xxx

完整统一封装类（带详细注释）

import os
import time
from dotenv import load_dotenv
from openai import OpenAI, APIError, APIConnectionError, RateLimitError
from anthropic import Anthropic, APIError as AnthropicAPIError
from xai import Xai, APIError as XaiAPIError

# 加载环境变量
load_dotenv()

class UnifiedAIClient:
    """
    多模型统一调用封装
    支持：GPT / Claude / Grok
    对外统一 chat_completion 方法，入参、返回格式标准化
    """
    def __init__(self, model_type: str):
        """
        初始化客户端
        :param model_type: 模型类型，可选值 gpt / claude / grok
        """
        self.model_type = model_type.lower()
        self.client = None
        self._init_client()

    def _init_client(self):
        """根据模型类型初始化对应SDK客户端"""
        if self.model_type == "gpt":
            self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        elif self.model_type == "claude":
            self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        elif self.model_type == "grok":
            self.client = Xai(api_key=os.getenv("XAI_API_KEY"))
        else:
            raise ValueError("model_type 仅支持 gpt / claude / grok")

    def chat_completion(
        self,
        messages: list,
        model_name: str,
        temperature: float = 0.1,
        max_tokens: int = 1024,
        retry_times: int = 2,
        timeout: int = 30
    ) -> dict:
        """
        统一对话入口
        :param messages: 消息列表 [{"role":"user","content":"xxx"}]
        :param model_name: 模型名称 gpt-3.5-turbo / claude-3-haiku / grok-2
        :param temperature: 随机性 0~1
        :param max_tokens: 最大输出长度
        :param retry_times: 失败重试次数
        :param timeout: 请求超时时间
        :return: 标准化返回字典 {"code":0, "content":"文本", "usage":{消耗统计}}
        """
        err_msg = ""
        for i in range(retry_times + 1):
            try:
                if self.model_type == "gpt":
                    return self._call_gpt(messages, model_name, temperature, max_tokens, timeout)
                elif self.model_type == "claude":
                    return self._call_claude(messages, model_name, temperature, max_tokens, timeout)
                elif self.model_type == "grok":
                    return self._call_grok(messages, model_name, temperature, max_tokens, timeout)
            except (RateLimitError, APIConnectionError, APIError, AnthropicAPIError, XaiAPIError) as e:
                err_msg = str(e)
                # 限流/网络异常延时重试
                time.sleep(1.5)
                continue
        # 多次重试全部失败
        return {
            "code": -1,
            "content": f"请求失败，重试{retry_times}次均报错：{err_msg}",
            "usage": {}
        }

    def _call_gpt(self, messages, model, temp, max_tok, timeout):
        """内部：调用GPT接口并标准化返回"""
        resp = self.client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temp,
            max_tokens=max_tok,
            timeout=timeout
        )
        return {
            "code": 0,
            "content": resp.choices[0].message.content.strip(),
            "usage": {
                "prompt_tokens": resp.usage.prompt_tokens,
                "completion_tokens": resp.usage.completion_tokens,
                "total_tokens": resp.usage.total_tokens
            }
        }

    def _call_claude(self, messages, model, temp, max_tok, timeout):
        """内部：调用Claude并适配参数、标准化返回"""
        resp = self.client.messages.create(
            model=model,
            messages=messages,
            temperature=temp,
            max_tokens=max_tok,
            timeout=timeout
        )
        return {
            "code": 0,
            "content": resp.content[0].text.strip(),
            "usage": {
                "prompt_tokens": resp.usage.input_tokens,
                "completion_tokens": resp.usage.output_tokens,
                "total_tokens": resp.usage.input_tokens + resp.usage.output_tokens
            }
        }

    def _call_grok(self, messages, model, temp, max_tok, timeout):
        """内部：调用Grok，参数格式与GPT完全兼容"""
        resp = self.client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temp,
            max_tokens=max_tok,
            timeout=timeout
        )
        return {
            "code": 0,
            "content": resp.choices[0].message.content.strip(),
            "usage": {
                "prompt_tokens": resp.usage.prompt_tokens,
                "completion_tokens": resp.usage.completion_tokens,
                "total_tokens": resp.usage.total_tokens
            }
        }

# ==================== 使用示例 ====================
if __name__ == "__main__":
    test_msg = [
        {"role": "user", "content": "用Python写一个批量读取Excel的工具类"}
    ]

    # 1. 调用GPT
    gpt_client = UnifiedAIClient(model_type="gpt")
    gpt_res = gpt_client.chat_completion(messages=test_msg, model_name="gpt-3.5-turbo")
    print("GPT返回：", gpt_res["content"])
    print("消耗统计：", gpt_res["usage"])

    # 2. 调用Claude
    claude_client = UnifiedAIClient(model_type="claude")
    claude_res = claude_client.chat_completion(messages=test_msg, model_name="claude-3-haiku-20240307")
    print("Claude返回：", claude_res["content"])

    # 3. 调用Grok
    grok_client = UnifiedAIClient(model_type="grok")
    grok_res = grok_client.chat_completion(messages=test_msg, model_name="grok-2")
    print("Grok返回：", grok_res["content"])

核心功能说明

1. 统一入参结构

三家模型消息列表 messages 格式完全通用，不用适配各家差异化 role 参数。直接传入标准的 [{"role": "user", "content": "..."}] 即可，底层自动映射。

2. 标准化返回体

无论调用哪个模型，返回字典结构一致：code 状态码、content 文本内容、usage token 消耗统计。上层业务无需分别解析，直接拿 result["content"] 就能用。

3. 内置自动重试机制

捕获限流（429）、网络超时、接口报错，自动延时重试。批量脚本里大幅降低崩溃概率，再也不用担心单次请求失败导致整个流程中断。

4. 统一异常封装

屏蔽各家 SDK 独有的异常类，对外只通过 code=-1 标识失败，错误信息统一输出。上层代码只需判断 code 即可，不需要 import 一堆异常类型。

5. 易扩展

后续新增模型（比如 Gemini、DeepSeek 等）只需在类里新增 _call_xxx 方法，对外调用函数 chat_completion 完全不用改动。这种设计堪称“开闭原则”的典范。

实操踩坑记录

坑 1：Claude 参数不兼容，直接复用 GPT 代码报错

报错：InvalidRequestError: Unexpected parameter messages
原因：早期 Claude SDK 入参结构和 GPT 差异极大，不能直接复用 GPT 调用逻辑。
解决方案：单独封装 _call_claude 做参数适配，统一上层调用入口。注意 messages 的格式以及 max_tokens 参数名别搞混。

坑 2：各家 usage 字段名称不一致，统计 token 混乱

GPT：prompt_tokens / completion_tokens
Claude：input_tokens / output_tokens
统一封装后自动映射字段，对外输出相同 key（prompt_tokens, completion_tokens, total_tokens），方便做计费统计、额度管控。

坑 3：批量循环调用频繁触发 429 限流

单模型单密钥并发有限，单纯加 sleep 效率极低。拓展思路：可在此类封装基础上搭建密钥池调度，自动切换可用渠道，解决高频调用限流问题。比如维护一个 key_list，遇到限流自动轮换下一个 key。

拓展优化方向

这套封装非常适合作为中转网关的基础组件，以下几个方向可以进一步打磨：

  1. 增加密钥池管理，多 key 自动负载均衡，避免单 key 被限流。
  2. 增加本地缓存，重复请求（比如相同 prompt 的短时间重复调用）直接返回缓存结果，减少 token 消耗。
  3. 封装 HTTP 接口，做成统一中转网关服务，对外暴露一个 REST API 即可切换多种模型。
  4. 增加日志持久化，记录每一次调用的模型、消耗额度、耗时，方便后续审计和成本分析。
  5. 增加输入内容长度校验，提前拦截超长 prompt 避免扣费，防止用户传几万 token 的文本直接调用。

总结

这套统一封装的核心价值在于：消除多模型切换的重复代码，让开发者专注于业务逻辑而非底层 API 适配。标准化返回结构极大降低了上层业务维护成本，而自带的重试与异常捕获机制，让稳定性远高于原生单模型调用。如果是高频批量调用场景，可在此基础上扩展多渠道调度池，解决限流、成本、网络三大痛点。一句话：一次封装，搞定所有主流大模型。

侠游戏发布此文仅为了传递信息，不代表侠游戏网站认同其观点或证实其描述

相关攻略

更多

同类更新

更多

热游推荐

更多

精彩专题

更多
手游 专题 攻略 App下载 web3
返回首页 回到顶部
湘ICP备14008430号-1 湘公网安备 43070302000280号
All Rights Reserved
本站为非盈利网站，不接受任何广告。本站所有软件，都由网友
上传，如有侵犯你的版权，请发邮件给xiayx666@163.com
抵制不良色情、反动、暴力游戏。注意自我保护，谨防受骗上当。
适度游戏益脑，沉迷游戏伤身。合理安排时间，享受健康生活。