需要统一调用接口，因为OpenAI、Qwen、GLM等厂商API在鉴权、路径、参数、响应上差异大，硬编码导致维护难、切换难、错误处理散；统一接口通过标准化输入（messages列表）、一致化输出（content/usage/model）、厂商适配器解耦、异常归一化来解决。
为什么需要统一调用接口
不同大模型厂商（如OpenAI、Qwen、GLM、Moonshot）的API格式差异大：鉴权方式、请求路径、参数名、响应结构各不相同。直接在业务代码里硬编码各家调用逻辑，会导致维护成本高、切换模型困难、错误处理分散。封装一层统一接口，能让上层只关心“输入提示词、获取回复”，不必操心底层细节。
核心设计原则
统一接口不是简单做一层转发，关键在于抽象出稳定、可扩展的契约：
-
输入标准化：统一使用
messages列表（角色+内容），支持 system/user/assistant；额外参数如temperature、max_tokens作为可选关键字传入 -
输出结构一致：无论后端是哪个模型，都返回含
content（字符串回复）、usage（token统计）、model（实际调用模型名）的字典 -
厂商解耦：每个厂商实现独立的适配器类（如
OpenAIAdapter、QwenAdapter），通过工厂函数按配置自动选择 -
异常归一化：网络超时、认证失败、限流等底层异常统一转为自定义异常（如
LLMConnectionError、LLMAuthError），业务层无需识别 HTTP 状态码
一个轻量可用的封装示例
以下是一个精简但生产可用的骨架（基于 httpx + 配置驱动）：
安全 Python 代码沙箱：在隔离的沙箱中执行任意 Python 代码，预装 requests、NumPy、pandas 库。返回 stdout、stderr。
from typing import List, Dict, Any, Optional
import httpx
from abc import ABC, abstractmethod
<p>class LLMAdapter(ABC):
@abstractmethod
def chat(self, messages: List[Dict], **kwargs) -> Dict[str, Any]:
pass</p><p>class OpenAIAdapter(LLMAdapter):
def <strong>init</strong>(self, api_key: str, base_url: str = "<a href="https://www.php.cn/link/8e5ac35e1661cc14cb518fa9b9364854">https://www.php.cn/link/8e5ac35e1661cc14cb518fa9b9364854</a>"):
self.client = httpx.Client(headers={"Authorization": f"Bearer {api_key}"})
self.base_url = base_url</p><pre class='brush:python;toolbar:false;'>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 数，自动截断过长历史消息（可用
tiktoken或transformers的 tokenizer） - 调用日志与指标：记录耗时、模型、token 数，上报到 Prometheus 或写入日志，便于监控延迟与成本
-
异步支持：提供
async llm_chat_async(...)版本，适配 FastAPI 等异步框架
