但大部分人用的时候都是现查文档每次接入一个新模型都要重写一遍。其实抽象一层写个统一的调用封装后面换模型改一行配置就行。先搭基础调用层不管哪个模型调用方式大同小异POST 一个请求传消息列表拿回复。python# llm_client.pyimport jsonfrom dataclasses import dataclass, fieldfrom typing import Optionalimport requestsdataclassclass Message:role: str # system | user | assistantcontent: strdataclassclass LLMResponse:content: strmodel: strusage: dict field(default_factorydict)class LLMClient:def __init__(self, api_key: str, base_url: str, model: str):self.api_key api_keyself.base_url base_url.rstrip(/)self.model modelself.session requests.Session()self.session.headers.update({Authorization: fBearer {api_key},Content-Type: application/json,})def chat(self, messages: list[Message], **kwargs) - LLMResponse:payload {model: self.model,messages: [{role: m.role, content: m.content} for m in messages],**kwargs,}resp self.session.post(f{self.base_url}/chat/completions,jsonpayload,timeout60,)resp.raise_for_status()data resp.json()choice data[choices][0]return LLMResponse(contentchoice[message][content],modeldata.get(model, self.model),usagedata.get(usage, {}),)这段代码所有模型通用区别只在于 api_key、base_url、model 三个参数。配置化换模型改一行把模型配置写在文件里不写死在代码里。python# config.pyimport osfrom dataclasses import dataclassdataclassclass ModelConfig:name: strapi_key: strbase_url: str# 支持的模型配置MODELS {deepseek: ModelConfig(namedeepseek-chat,api_keyos.getenv(DEEPSEEK_API_KEY, ),base_urlhttps://api.deepseek.com,),openai: ModelConfig(namegpt-4o-mini,api_keyos.getenv(OPENAI_API_KEY, ),base_urlhttps://api.openai.com,),qwen: ModelConfig(nameqwen-plus,api_keyos.getenv(QWEN_API_KEY, ),base_urlhttps://dashscope.aliyuncs.com/compatible-mode,),glm: ModelConfig(nameglm-4-flash,api_keyos.getenv(GLM_API_KEY, ),base_urlhttps://open.bigmodel.cn/api/paas/v4,),}def get_client(model_name: str deepseek) - LLMClient:cfg MODELS.get(model_name)if not cfg:raise ValueError(f不支持的模型: {model_name}可选: {list(MODELS.keys())})if not cfg.api_key:raise ValueError(f{model_name} 的 API Key 未设置)return LLMClient(api_keycfg.api_key,base_urlcfg.base_url,modelcfg.name,)用的时候pythonclient get_client(deepseek)resp client.chat([Message(roleuser, content讲个笑话)])print(resp.content)想换成通义千问改一行pythonclient get_client(qwen)API 接口兼容的前提下哪里都不用改。对话管理别每次都从头开始每次聊天只发一条消息模型没有上下文。加上对话管理自动维护历史消息。python# conversation.pyfrom llm_client import Messageclass Conversation:def __init__(self, system_prompt: str ):self.messages []if system_prompt:self.messages.append(Message(rolesystem, contentsystem_prompt))def add_user(self, content: str):self.messages.append(Message(roleuser, contentcontent))def add_assistant(self, content: str):self.messages.append(Message(roleassistant, contentcontent))def get_messages(self, max_history: int 20) - list[Message]:保留 system prompt 最近 N 轮对话if len(self.messages) max_history 1:return self.messagesreturn [self.messages[0]] self.messages[-(max_history):]为了控制 token 消耗get_messages 只保留最近的 N 轮对话最早的历史自动丢弃。流式输出不用等全部返回等模型全部生成完再返回用户要等很久。改成流式输出逐字显示。pythondef chat_stream(self, messages: list[Message], **kwargs):payload {model: self.model,messages: [{role: m.role, content: m.content} for m in messages],stream: True,**kwargs,}with self.session.post(f{self.base_url}/chat/completions,jsonpayload,streamTrue,timeout120,) as resp:resp.raise_for_status()for line in resp.iter_lines():if not line:continueline line.decode(utf-8)if line.startswith(data: ):data_str line[6:]if data_str.strip() [DONE]:breaktry:chunk json.loads(data_str)delta chunk[choices][0].get(delta, {})content delta.get(content, )if content:yield contentexcept json.JSONDecodeError:continue调用方逐段拿到内容可以实时显示pythonfor chunk in client.chat_stream([Message(roleuser, content写一首诗)]):print(chunk, end, flushTrue)加个简单的重试网络请求偶尔会失败加个重试减少手动重试的次数。pythonimport timedef chat_with_retry(client: LLMClient,messages: list[Message],max_retries: int 3,**kwargs,) - LLMResponse:last_error Nonefor attempt in range(max_retries):try:return client.chat(messages, **kwargs)except requests.exceptions.Timeout:last_error 请求超时time.sleep(2 ** attempt)except requests.exceptions.HTTPError as e:status e.response.status_codeif status in (429, 502, 503):last_error f服务暂不可用 ({status})time.sleep(2 ** attempt)else:raiseexcept Exception as e:last_error str(e)if attempt max_retries - 1:raiseraise RuntimeError(f重试 {max_retries} 次后仍然失败: {last_error})# 使用resp chat_with_retry(client, [Message(roleuser, content你好)])print(resp.content)超时、429限流、502网关错误自动重试指数退避。一个完整的使用例子把上面几个功能串起来写一个简单的命令行对话程序python# cli_chat.pyfrom llm_client import LLMClientfrom config import get_clientfrom conversation import Conversationdef main():print( AI 命令行对话输入 /exit 退出)print(支持的模型: deepseek, openai, qwen, glm)model input(选择模型 (回车默认 deepseek): ).strip() or deepseekclient get_client(model)conv Conversation(system_prompt你是一个助手用简洁的方式回答问题。)while True:user_input input(\n ).strip()if user_input.lower() in (/exit, /quit):print(再见)breakif user_input.lower() /model:print(f当前模型: {client.model})continueconv.add_user(user_input)print(\nAI: , end, flushTrue)response_parts []for chunk in client.chat_stream(conv.get_messages()):print(chunk, end, flushTrue)response_parts.append(chunk)print()full_response .join(response_parts)conv.add_assistant(full_response)if __name__ __main__:main()几个注意事项API Key 不要写死在代码里。上面 config.py 用 os.getenv 从环境变量读取在 .env 文件里配置。不同模型的接口有细微差异。上面的封装基于 OpenAI 兼容接口DeepSeek、通义千问、GLM 都支持。如果用非兼容接口需要单独适配。上下文的长度控制。每个模型的上下文长度不同4K、16K、128KConversation 的 max_history 要根据模型的限制调整。太长会被截断或报错。错误处理要区分可重试和不可重试。401 认证失败、400 请求错误这些重试也没用直接报错。429 限流和 5xx 服务端错误可以重试。