封装可切换的模型客户端

第 1 章的 hello.py 每次都要手写 api_key、base_url、model 三件套，代码又散又容易出错。智能体的所有能力都建立在「发一条消息、拿回一个回复」之上，所以先把这一步封装成唯一入口：以后所有章节只 import 一个 LLMClient，调用它的 chat() 即可，换服务商只改 .env。本章代码就是 agent_demo/llm_client.py 的完整内容，可直接照抄。

1. 设计目标

  • 构造时从 .env 统一读取 LLM_API_KEY / LLM_BASE_URL / LLM_MODEL / LLM_TIMEOUT；
  • chat(messages, tools=None, stream=False) 一个方法打天下，tools 为函数调用工具列表（第 6 章用），stream 为流式开关（第 11 章细讲）；
  • 超时、限流、网络抖动自动重试，密钥错误直接给出中文提示；
  • 全程类型标注，IDE 提示友好。

2. 完整代码 llm_client.py

"""llm_client.py —— 统一的大模型客户端（OpenAI 兼容接口）。

用法：
    from llm_client import LLMClient, get_message
    llm = LLMClient()
    resp = llm.chat([{"role": "user", "content": "你好"}])
    print(get_message(resp).content)   # 助手文本
"""
import os
import time
from typing import Any, Dict, List, Optional

from dotenv import load_dotenv
from openai import (
    OpenAI,
    APIConnectionError,
    APITimeoutError,
    AuthenticationError,
    OpenAIError,
    RateLimitError,
)

load_dotenv()  # 读取项目根目录 .env（在 agent_demo 下运行）


class LLMClient:
    """一个实例对应 .env 里的一套密钥/地址/模型，可随时多开。"""

    def __init__(self) -> None:
        self.api_key: str = os.getenv("LLM_API_KEY", "")
        self.base_url: str = os.getenv("LLM_BASE_URL", "https://api.openai.com/v1")
        self.model: str = os.getenv("LLM_MODEL", "deepseek-chat")
        self.timeout: float = float(os.getenv("LLM_TIMEOUT", "60"))
        if not self.api_key:
            raise RuntimeError("未配置 LLM_API_KEY，请先在 .env 中填写密钥")
        self.client = OpenAI(
            api_key=self.api_key,
            base_url=self.base_url,
            timeout=self.timeout,   # 秒；也可传 httpx.Timeout 精细区分连接/读取
        )

    def chat(self, messages: List[Dict[str, Any]],
             tools: Optional[List[Dict[str, Any]]] = None,
             stream: bool = False,
             temperature: Optional[float] = None,
             max_tokens: Optional[int] = None,
             retries: int = 2,
             **extra: Any) -> Any:
        """调用模型并返回完整响应对象 resp。

        - 文本内容：resp.choices[0].message.content
        - 函数调用：resp.choices[0].message.tool_calls（第 6 章）
        - stream=True 时返回流对象，逐块取内容（第 11 章）
        - extra 可透传 response_format 等服务商扩展参数
        """
        params: Dict[str, Any] = {
            "model": self.model,
            "messages": messages,
            "stream": stream,
        }
        if tools:
            params["tools"] = tools
        if temperature is not None:
            params["temperature"] = temperature
        if max_tokens is not None:
            params["max_tokens"] = max_tokens
        params.update(extra)

        for attempt in range(retries + 1):
            try:
                return self.client.chat.completions.create(**params)
            except AuthenticationError as e:
                raise RuntimeError(f"API 密钥无效或已过期：{e}") from e
            except RateLimitError as e:
                wait = 2 ** attempt + 1
                print(f"触发限流，{wait} 秒后重试（第 {attempt + 1} 次）")
                time.sleep(wait)
            except (APITimeoutError, APIConnectionError) as e:
                wait = 2 ** attempt
                print(f"网络异常（{type(e).__name__}），{wait} 秒后重试")
                time.sleep(wait)
            except OpenAIError as e:   # 其余 4xx/5xx 不重试，直接报错
                raise RuntimeError(f"模型调用失败：{e}") from e
        raise RuntimeError(f"连续重试 {retries} 次仍失败，请检查网络与服务状态")

    def chat_text(self, messages: List[Dict[str, Any]], **kwargs: Any) -> str:
        """便捷方法：只想要文本时用这个，省去取 message 的一步。"""
        resp = self.chat(messages, **kwargs)
        return get_message(resp).content or ""


def get_message(resp: Any) -> Any:
    """从响应取第一条消息：含 role / content / tool_calls 字段。"""
    return resp.choices[0].message


if __name__ == "__main__":
    llm = LLMClient()
    reply = llm.chat_text([{"role": "user", "content": "你好，用一句话介绍你自己"}])
    print(reply)

3. 运行与用法示例

直接运行模块自测：

python llm_client.py
# 输出示例（随模型与时间变化）：
# 你好！我是 DeepSeek，一个 AI 助手，可以帮你解答问题、编写代码……

在别的文件里这样用（多轮上下文由 messages 列表承载，第 3 章展开）：

from llm_client import LLMClient

llm = LLMClient()
messages = [
    {"role": "system", "content": "你是一个Python导师，回答要简短"},
    {"role": "user", "content": "什么是虚拟环境？"},
]
print(llm.chat_text(messages))
# 输出示例：虚拟环境是独立的 Python 运行空间，用 venv 创建，避免项目间依赖冲突……

4. 错误处理说明

异常触发场景客户端动作
AuthenticationError密钥错误/过期直接抛 RuntimeError，不重试
RateLimitError超出调用频率配额退避后重试（最多 retries 次）
APITimeoutError连接或读取超时退避后重试
APIConnectionError网络不通/断连退避后重试
其他 OpenAIError4xx/5xx 业务错误直接抛 RuntimeError
提醒两点：其一，openai SDK 底层默认对连接类错误自带 2 次重试（client.max_retries 控制），本章又加了一层业务重试，两层叠加在断网时会拖长耗时，属可接受代价；其二，超时时间由 .env 的 LLM_TIMEOUT 控制，调大适合长回答、调小适合快速自测，重试间隔的指数退避策略细节见第 8 章。

5. 为什么这样设计

chat() 返回完整响应对象而不是只返回字符串，是因为第 6 章判断「模型是否想调用工具」要看 message.tool_calls，第 8 章判断回复是否被截断要看 finish_reason——把原始响应完整交还，调用方才拿得到这些信息。类型上 List[Dict[str, Any]] 已足够宽松，任何 OpenAI 兼容服务（DeepSeek、通义等）的消息结构都一致。 小结：LLMClient 把密钥、地址、模型、超时、重试全部收进一个类，对外只暴露 chat()/chat_text() 两个方法；第 3 章起所有程序都从 from llm_client import LLMClient 开始，代码不再关心「调用的是哪家模型」。

笔记加载中…

标签：#Agent实战

返回知识库