需要统一调用接口,因为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 + 配置驱动):
from typing import List, Dict, Any, Optional import httpx from abc import ABC, abstractmethodclass 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./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./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"]) # 直接拿到纯文本回复
实际落地时,可逐步增强鲁棒性和可观测性:
tenacity 库)tiktoken 或 transformers 的 tokenizer)async llm_chat_async(...) 版本,适配 FastAPI 等异步框架