大模型 API 遍地都是。OpenAI、DeepSeek、通义千问、GLM……每个都给 API 接口，连上去就能用。
但大部分人用的时候都是现查文档，每次接入一个新模型都要重写一遍。其实抽象一层，写个统一的调用封装，后面换模型改一行配置就行。
先搭基础调用层
不管哪个模型，调用方式大同小异：POST 一个请求，传消息列表，拿回复。
```python
# llm_client.py
import json
from dataclasses import dataclass, field
from typing import Optional
import requests
@dataclass
class Message:
role: str # "system" | "user" | "assistant"
content: str
@dataclass
class LLMResponse:
content: str
model: str
usage: dict = field(default_factory=dict)
class LLMClient:
def __init__(self, api_key: str, base_url: str, model: str):
self.api_key = api_key
self.base_url = base_url.rstrip("/")
self.model = model
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {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",
json=payload,
timeout=60,
)
resp.raise_for_status()
data = resp.json()
choice = data["choices"][0]
return LLMResponse(
content=choice["message"]["content"],
model=data.get("model", self.model),
usage=data.get("usage", {}),
)
```
这段代码所有模型通用，区别只在于 api_key、base_url、model 三个参数。
配置化，换模型改一行
把模型配置写在文件里，不写死在代码里。
```python
# config.py
import os
from dataclasses import dataclass
@dataclass
class ModelConfig:
name: str
api_key: str
base_url: str
# 支持的模型配置
MODELS = {
"deepseek": ModelConfig(
name="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY", ""),
base_url="https://api.deepseek.com",
),
"openai": ModelConfig(
name="gpt-4o-mini",
api_key=os.getenv("OPENAI_API_KEY", ""),
base_url="https://api.openai.com",
),
"qwen": ModelConfig(
name="qwen-plus",
api_key=os.getenv("QWEN_API_KEY", ""),
base_url="https://dashscope.aliyuncs.com/compatible-mode",
),
"glm": ModelConfig(
name="glm-4-flash",
api_key=os.getenv("GLM_API_KEY", ""),
base_url="https://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_key=cfg.api_key,
base_url=cfg.base_url,
model=cfg.name,
)
```
用的时候：
```python
client = get_client("deepseek")
resp = client.chat([Message(role="user", content="讲个笑话")])
print(resp.content)
```
想换成通义千问，改一行：
```python
client = get_client("qwen")
```
API 接口兼容的前提下，哪里都不用改。
对话管理，别每次都从头开始
每次聊天只发一条消息，模型没有上下文。加上对话管理，自动维护历史消息。
```python
# conversation.py
from llm_client import Message
class Conversation:
def __init__(self, system_prompt: str = ""):
self.messages = []
if system_prompt:
self.messages.append(Message(role="system", content=system_prompt))
def add_user(self, content: str):
self.messages.append(Message(role="user", content=content))
def add_assistant(self, content: str):
self.messages.append(Message(role="assistant", content=content))
def get_messages(self, max_history: int = 20) -> list[Message]:
"""保留 system prompt + 最近 N 轮对话"""
if len(self.messages) <= max_history + 1:
return self.messages
return [self.messages[0]] + self.messages[-(max_history):]
```
为了控制 token 消耗，get_messages 只保留最近的 N 轮对话，最早的历史自动丢弃。
流式输出，不用等全部返回
等模型全部生成完再返回，用户要等很久。改成流式输出，逐字显示。
```python
def 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",
json=payload,
stream=True,
timeout=120,
) as resp:
resp.raise_for_status()
for line in resp.iter_lines():
if not line:
continue
line = line.decode("utf-8")
if line.startswith("data: "):
data_str = line[6:]
if data_str.strip() == "[DONE]":
break
try:
chunk = json.loads(data_str)
delta = chunk["choices"][0].get("delta", {})
content = delta.get("content", "")
if content:
yield content
except json.JSONDecodeError:
continue
```
调用方逐段拿到内容，可以实时显示：
```python
for chunk in client.chat_stream([Message(role="user", content="写一首诗")]):
print(chunk, end="", flush=True)
```
加个简单的重试
网络请求偶尔会失败，加个重试减少手动重试的次数。
```python
import time
def chat_with_retry(
client: LLMClient,
messages: list[Message],
max_retries: int = 3,
**kwargs,
) -> LLMResponse:
last_error = None
for 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_code
if status in (429, 502, 503):
last_error = f"服务暂不可用 ({status})"
time.sleep(2 ** attempt)
else:
raise
except Exception as e:
last_error = str(e)
if attempt == max_retries - 1:
raise
raise RuntimeError(f"重试 {max_retries} 次后仍然失败: {last_error}")
# 使用
resp = chat_with_retry(client, [Message(role="user", content="你好")])
print(resp.content)
```
超时、429限流、502网关错误自动重试，指数退避。
一个完整的使用例子
把上面几个功能串起来，写一个简单的命令行对话程序：
```python
# cli_chat.py
from llm_client import LLMClient
from config import get_client
from conversation import Conversation
def main():
print("=== AI 命令行对话（输入 /exit 退出）===")
print("支持的模型: deepseek, openai, qwen, glm")
model = input("选择模型 (回车默认 deepseek): ").strip() or "deepseek"
client = get_client(model)
conv = Conversation(system_prompt="你是一个助手，用简洁的方式回答问题。")
while True:
user_input = input("\n>> ").strip()
if user_input.lower() in ("/exit", "/quit"):
print("再见")
break
if user_input.lower() == "/model":
print(f"当前模型: {client.model}")
continue
conv.add_user(user_input)
print("\nAI: ", end="", flush=True)
response_parts = []
for chunk in client.chat_stream(conv.get_messages()):
print(chunk, end="", flush=True)
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、128K），Conversation 的 max_history 要根据模型的限制调整。太长会被截断或报错。
错误处理要区分可重试和不可重试。401 认证失败、400 请求错误这些重试也没用，直接报错。429 限流和 5xx 服务端错误可以重试。