Python大模型API封装教程_构建统一调用接口

7次阅读

需要统一调用接口,因为 OpenAI、Qwen、GLM 等厂商 API 在鉴权、路径、参数、响应上差异大,硬编码导致维护难、切换难、错误处理散;统一接口通过标准化输入(messages 列表)、一致化输出(content/usage/model)、厂商适配器解耦、异常归一化来解决。

Python 大模型 API 封装教程_构建统一调用接口

为什么 需要统一调用接口

不同 大模型 厂商(如 OpenAI、Qwen、GLM、Moonshot)的 API 格式差异大:鉴权方式、请求路径、参数名、响应结构各不相同。直接在业务代码里硬 编码 各家调用逻辑,会导致维护成本高、切换模型困难、错误处理分散。封装一层统一接口,能让上层只关心“输入提示词、获取回复”,不必操心底层细节。

核心设计原则

统一接口不是简单做一层转发,关键在于抽象出稳定、可扩展的契约:

  • 输入标准化:统一使用 messages 列表(角色 + 内容),支持 system/user/assistant;额外参数如 temperaturemax_tokens 作为可选关键字传入
  • 输出结构一致 :无论 后端 是哪个模型,都返回含 content(字符串回复)、usage(token 统计)、model(实际调用模型名)的字典
  • 厂商解耦:每个厂商实现独立的适配器类(如 OpenAIAdapterQwenAdapter),通过工厂函数按配置自动选择
  • 异常归一化 :网络超时、认证失败、限流等底层异常统一转为自定义异常(如 LLMConnectionErrorLLMAuthError),业务层无需识别 HTTP 状态码

一个轻量可用的封装示例

以下是一个精简但生产可用的骨架(基于 httpx + 配置驱动):

from typing import List, Dict, Any, Optional import httpx from abc import ABC, abstractmethod 

class LLMAdapter(ABC): @abstractmethod def chat(self, messages: List[Dict], **kwargs) -> Dict[str, Any]: pass

class OpenAIAdapter(LLMAdapter): def init(self, api_key: str, base_url: str = "https://www.php.cn/link/8e5ac35e1661cc14cb518fa9b9364854"): self.client = httpx.Client(headers={"Authorization": f"Bearer {api_key}"}) self.base_url = base_url

def chat(self, messages: List[Dict], model="gpt-4o", **kwargs) -> Dict[str, Any]:     resp = self.client.post(f"{self.base_url}/chat/completions",         json={"model": model, "messages": messages, **kwargs}     )     resp.raise_for_status()     data = resp.json()     return {"content": data["choices"][0]["message"]["content"],         "usage": data.get("usage", {}),         "model": model     }

工厂函数:根据配置加载对应适配器

def get_llm_adapter(provider: str, **config) -> LLMAdapter: if provider == "openai": return OpenAIAdapter(api_key=config["api_key"], base_url=config.get("base_url")) elif provider == "qwen": return QwenAdapter(api_key=config["api_key"], base_url=config.get("base_url", "https://www.php.cn/link/1128b9108f8997cf4b24e3b20e5ecceb"))

…… 其他厂商

raise ValueError(f"Unsupported provider: {provider}")

统一入口

def llm_chat(messages: List[Dict], provider: str = "openai", kwargs ) -> Dict[str, Any]: adapter = get_llm_adapter(provider, kwargs) return adapter.chat(messages, **kwargs)

调用时只需:

立即学习Python 免费学习笔记(深入)”;

resp = llm_chat(messages=[{"role": "user", "content": "你好,请用一句话介绍自己"}],     provider="qwen",     api_key="sk-xxx",     model="qwen-max" ) print(resp["content"])  # 直接拿到纯文本回复

进阶建议

实际落地时,可逐步增强鲁棒性和可观测性:

  • 自动重试与退避:对 429(限流)、5xx 错误内置指数退避重试(推荐用 tenacity 库)
  • Token 自动截断:根据模型最大上下文和预估 prompt token 数,自动截断过长历史消息(可用 tiktokentransformers 的 tokenizer)
  • 调用日志与指标:记录耗时、模型、token 数,上报到 Prometheus 或写入日志,便于监控延迟与成本
  • 异步支持:提供 async llm_chat_async(……) 版本,适配 FastAPI 等异步框架

text=ZqhQzanResources