在智能客服问答系统中，LLM（大语言模型）是核心推理引擎。然而，现实中的 LLM 服务生态碎片化严重——OpenAI、Anthropic、DeepSeek、Ollama 本地模型各有不同的 API 格式、认证方式和能力边界。本文深入解析一个智能问答系统的 LLM 基础设施层设计，展示如何通过统一抽象将六种以上 LLM 提供商无缝接入同一套问答管线。

一、为什么需要 LLM 基础设施层？

FAQ 问答系统初期可能只接入一个 LLM 提供商——比如 OpenAI 的 GPT系列。但随着业务演进，你会面临以下挑战：

  • 成本优化：简单查询用便宜模型，复杂推理用高端模型
  • 可用性保障：单一提供商可能因 API 故障、配额耗尽导致服务中断
  • 特殊能力需求：DeepSeek 的推理模式支持思维链展示，Ollama 支持完全本地化部署
  • 区域合规：某些客户要求数据不出境，必须使用国内模型提供商

如果没有统一的 LLM 基础设施层，每个调用点都需要处理不同提供商的认证、重试、错误处理和响应解析，代码会迅速陷入不可维护的泥潭。

核心设计目标：让上层业务代码永远只面对一个 chat() 接口，底层自动处理模型选择、认证、重试、流式解析和 token 统计。

二、LLMProvider 类设计——统一抽象的核心

2.1 类属性与构造

LLMProvider 是所有 LLM 提供商的基础抽象类，设计简洁而富有表现力：

class LLMProvider:
    name: str                    # 提供商名称，如 "openai", "deepseek"
    api_key: str | None          # API 密钥
    base_url: str | None         # API 端点 URL
    models: list[str]            # 可用模型列表
    thinking: bool               # 是否支持思考/推理模式

这四个属性已经足够描述任何 LLM 提供商的核心特征：

  • name：用于配置路由和日志记录
  • api_key：认证凭证，支持 None 以兼容无需认证的本地模型
  • base_url：兼容 OpenAI API 格式的所有提供商只需要修改端点 URL
  • models：允许运行时发现可用模型列表
  • thinking：标记是否支持 DeepSeek 风格的 thinking/推理模式，控制流式输出行为

2.2 Lazy OpenAI Client

值得注意的是，OpenAI 客户端的初始化采用了 lazy 模式——并非在构造时创建，而是在首次调用 chat() 或 chat_stream() 时按需创建：

# 伪代码: lazy client 模式
@property
def client(self):
    if not self._client:
        # 首次调用时初始化 OpenAI 兼容客户端
        self._client = create_api_client(
            api_key=self.api_key,
            base_url=self.base_url
        )
    return self._client

这种设计有两个实际好处：

  1. 减少空启动开销：如果某个提供商在运行中未被实际调用（例如 fallback 未触发），不会浪费资源初始化客户端
  2. 配置灵活性：允许在创建 Provider 实例之后再动态修改 api_key 或 base_url

2.3 子类实现

以 OpenAI 提供商为例，子类实现非常简洁：

# 伪代码: 子类实现示意
class OpenAIProvider(LLMProvider):
    def __init__(self):
        super().__init__(
            name="openai",
            api_key=read_env("OPENAI_API_KEY"),
            base_url=read_env("OPENAI_BASE_URL"),
            models=["gpt-4o", "gpt-4o-mini", "..."],
            thinking=False  # 不支持 thinking 模式
        )

Ollama（本地模型）甚至不需要 api_key，只需要 base_url 指向本地服务地址。这种一致性使得在配置文件中切换提供商成为可能。

三、统一 Chat 接口——chat() 与 chat_stream()

3.1 chat()——统一 Chat Completion 入口

chat() 方法是整个系统调用 LLM 的唯一入口。它的职责边界非常清晰：

接收: messages, model, temperature, max_tokens, response_format, ...
处理: 自动重试、JSON 清理、token 统计
返回: ChatCompletionMessage 对象

关键设计决策：所有异常处理都在这一层收敛。上层业务代码（如 Agent 循环、路由决策）永远不需要关心网络错误或 JSON 解析失败。

# 伪代码: chat() 重试逻辑示意
def chat(self, messages, model=None, **kwargs):
    # 最多重试 3 次
    for attempt in range(MAX_RETRIES):
        try:
            response = self._call_llm_api(messages, model, **kwargs)
            return parse_response(response)
        except TEMPORARY_ERRORS as e:
            if attempt == MAX_RETRIES - 1:
                raise  # 最后一次失败，向上抛出
            wait = EXPONENTIAL_BACKOFF(attempt)  # 指数退避
            sleep(wait)

3.2 chat_stream()——流式输出

流式版本的设计重点是 过滤掉 thinking/reasoning 内容，只向调用方 yield 最终的文本内容：

# 伪代码: 流式 chat 示意
def chat_stream(self, messages, model=None, **kwargs):
    # 发起流式请求
    stream = self._create_stream(messages, model, **kwargs)
    
    for chunk in stream:
        # 跳过推理/思考过程的中间内容
        if chunk.is_reasoning:
            continue
        # 仅输出最终回答的部分
        if chunk.content:
            yield chunk.content

这种做法确保下游消费者（Web 前端、API 调用者）收到的只有纯净的最终回答，不需要自己处理推理过程的中间 token。

四、深入 _do_chat()——思考模式与流式处理的底层实现

4.1 实际执行 Chat 请求

_do_chat() 是 chat() 和 chat_stream() 的底层执行函数。它负责构建实际的 API 请求参数，处理 thinking 模式等特殊场景。

对于支持 thinking 模式的提供商（如 DeepSeek），请求中会注入特殊参数：

# 伪代码: thinking 模式下的请求构造
def _do_chat(self, messages, model, **kwargs):
    request = build_request(messages, model, **kwargs)
    
    # 如果提供商支持 thinking 模式，标记启用
    if self.thinking:
        request.enable_thinking_mode()
    
    return send_request(request)

4.2 Thinking 模式的流式处理

DeepSeek 的 thinking 模式会在流式响应中包含 reasoning_content 字段，表示模型内部的思考过程。_do_chat() 的处理逻辑如下：

流式模式（stream=True）：

  • 如果 SHOW_THINKING 环境变量为 true，在终端显示思考过程（带特殊格式标记）
  • 否则直接跳过 reasoning_content，只积累 content
  • 最终 yield 完整的 content

非流式模式：

  • choices[0].message 中同时包含 content 和 reasoning_content
  • 系统架构中有一项重要约束：所有手动构造的 ChatCompletionMessage 必须保留 reasoning_content 字段（DeepSeek 官方要求）

4.3 SHOW_THINKING 控制

这是一个由环境变量控制的开关，用于调试和观察场景：

SHOW_THINKING=true  → 在终端用 [thinking] 标记展示推理过程
SHOW_THINKING=false → 完全隐藏推理过程，只返回最终内容

这种设计平衡了开发调试需求与终端用户体验——普通用户只关心最终答案，而开发者需要观察模型推理链路以调整 prompt 和检索策略。

五、clean_json_response()——LLM 输出的 JSON 净化器

LLM 返回的 JSON 经常包含各种"杂质"：思考标签、Markdown 代码块包裹、尾部多余的文本等。clean_json_response() 是专门处理这些问题的函数。

5.1 处理流程

# 伪代码: JSON 清理流程
def clean_json_response(text: str) -> str:
    # 第一步：剥离模型思考过程（如 <think> 标签包裹的内容）
    text = strip_thinking_tags(text)
    
    # 第二步：剥离 Markdown 代码块标记（```json, ```等）
    text = strip_code_block_markers(text)
    
    # 第三步：找到 JSON 的起始位置（{ 或 [）
    # 第四步：截断 JSON 尾部之后的无关内容
    # 第五步：返回纯净的 JSON 字符串
    text = extract_json_body(text)
    return text
```text = re.sub(r'```\s*', '', text)
    
    # 第三步：提取纯 JSON（找到第一个 { 和最后一个 }）
    start = text.find('{')
    end = text.rfind('}')
    if start != -1 and end != -1:
        text = text[start:end + 1]
    
    # 第四步：尾部截断——删除最后一个 } 之后的无关内容
    # 处理 LLM 在 JSON 后继续输出解释文字的情况
    
    return text.strip()

5.2 四大处理步骤详解

步骤 目标 典型输入示例
剥离 <think> 标签 移除 DeepSeek 推理内容 <think>我需要查找...<｜end▁of▁thinking｜> > 最终内容
移除 Markdown 代码块 清理 ` ```json … ````包裹 ` ```json {“key”: “value”} ````
提取纯 JSON 片段 定位 JSON 起止位置 回复内容：{"result": "ok"} 说明完毕
尾部截断 删除 JSON 后的冗余文本 {"data": [...]} 以上就是全部内容

5.3 为什么需要尾部截断？

这是一个从实际生产数据中发现的问题。当 LLM 被要求"只返回 JSON"时，有时它会在输出完 JSON 后继续补充一段文字解释。尾部截断确保 } 之后的任何内容都被安全丢弃，程序只会解析到最后的 } 处。

六、Token 统计——_token_stage 与 _token_accumulator

6.1 按阶段累计的设计

在一个复杂的问答管线中，LLM 可能被多次调用，每次调用扮演不同的角色：

  • Query Analysis 阶段：分析用户问题
  • Retrieval 阶段：调用 LLM 生成检索查询
  • Generation 阶段：基于检索结果生成最终回答
  • Tool Use 阶段：Agent 调用工具时的小型 LLM 调用

系统设计了 _token_stage 上下文管理器和 _token_accumulator 来分别统计不同阶段的 token 消耗：

# 伪代码: Token 统计的阶段切换机制
# 使用 ContextVar 实现并发安全，不同请求互不干扰

_current_stage = ContextVar('current_stage', default='default')
_token_ledger = ContextVar('token_ledger', default={})

@contextmanager
def token_stage(stage_name: str):
    # 保存当前阶段，切换到新阶段
    previous = _current_stage.set(stage_name)
    try:
        yield
    finally:
        # 恢复上一阶段
        _current_stage.reset(previous)

6.2 ContextVar 实现并发安全

使用 ContextVar 而非全局变量的原因是：系统的 Web 模式需要同时处理多个请求。

# 请求 A 在处理 retrieval 阶段
with token_stage("retrieval"):
    llm.chat(...)  # 累计到 retrieval

# 请求 B 同时在处理 generation 阶段
with token_stage("generation"):
    llm.chat(...)  # 不会污染请求 A 的统计

ContextVar 是 Python 3.7+ 标准库的一部分，它为每个协程/线程维护独立的上下文，天然避免并发冲突。

6.3 Token 数据结构

每个阶段的 token 数据按 prompt_tokens、completion_tokens、reasoning_tokens 三个维度累计：

{
    "retrieval": {
        "prompt_tokens": 1245,
        "completion_tokens": 89,
        "reasoning_tokens": 0
    },
    "generation": {
        "prompt_tokens": 3456,
        "completion_tokens": 512,
        "reasoning_tokens": 340  # DeepSeek thinking 模式
    },
    "total": {
        "prompt_tokens": 4701,
        "completion_tokens": 601,
        "reasoning_tokens": 340
    }
}

这种精细化统计让运营团队能精确了解每个阶段的成本分布，从而优化管线设计——例如，如果 retrieval 阶段的 prompt_tokens 异常高，说明检索引擎返回的上下文过长，需要调整 chunk 大小。

七、重试策略——指数退避与优雅降级

7.1 三种需要重试的异常

在生产环境中，LLM API 调用可能因多种原因失败：

异常类型 典型原因 重试策略
ConnectionError 网络断开、DNS 解析失败 指数退避，最长等待 30s
TimeoutError 请求超时（如 60s 无响应） 退避重试，最多 3 次
APIStatusError (429) 速率限制（Rate Limit） 带 Retry-After 头信息的退避
APIStatusError (503) 服务临时不可用 指数退避，最多 3 次

7.2 指数退避实现

# 伪代码: 指数退避 + 随机抖动
def calc_backoff(attempt: int) -> float:
    # 初始等待 1 秒，每次翻倍，最长 30 秒
    wait = min(1.0 * (2 ** attempt), 30.0)
    # 加入随机抖动（jitter），防止并发重试风暴
    wait += random.uniform(0, 0.5)
    return wait

指数退避的核心公式是 base * 2^attempt，加上随机抖动避免"惊群效应"。429 错误还额外读取 Retry-After 响应头，如果服务器指明了等待时间则优先使用。

7.3 优雅降级的层级

重试不是唯一的容错手段。系统的容错策略分为三个层级：

  1. 请求级：同一提供商的同一请求重试（指数退避）
  2. 模型级：当前模型失败后，切换到同提供商的备用模型（如 gpt-4o → gpt-4o-mini）
  3. 提供商级：当前提供商全面不可用时，切换到另一个提供商（如 OpenAI → DeepSeek）
# 伪代码：多提供商逐级降级
available_providers = ["openai", "deepseek", "ollama"]

for provider_name in available_providers:
    try:
        provider = get_provider(provider_name)
        return provider.chat(messages)
    except Exception:
        continue  # 当前提供商不可用，尝试下一个

# 所有提供商都失败
raise ServiceUnavailable("所有 LLM 提供商均不可用")

第三级降级是系统的最后防线，确保即使主流云服务全面宕机，本地部署的 Ollama 模型仍能提供服务。

八、日志记录——_log_llm 与 _log_llm_error

8.1 _log_llm——正常调用的完整日志

每次 LLM 调用都会记录以下信息：

# 伪代码: LLM 调用日志记录
def log_llm_call(provider, model, stage, token_usage, latency_ms):
    # 记录每次 LLM 调用的关键指标
    logger.info({
        "event": "llm_call",
        "provider": provider,
        "model": model,
        "stage": stage,              # 哪个处理阶段发起的调用
        "tokens": token_usage,       # 包含 prompt/completion/reasoning
        "latency_ms": latency_ms,    # 请求耗时
    })

关键字段包括：提供商、模型、阶段、各类 token 数、延迟。这些数据为后续的成本分析和性能优化提供了精确的原始数据。

8.2 _log_llm_error——异常日志

# 伪代码: LLM 调用错误日志
def log_llm_error(provider, model, error, attempt):
    logger.error({
        "event": "llm_error",
        "provider": provider,
        "model": model,
        "error_type": type(error).__name__,
        "message": str(error),
        "attempt": attempt,  # 第几次重试
    })

异常日志记录了完整的错误信息和重试次数，方便运维团队在发生故障时快速定位问题根因。

8.3 日志的两种消费场景

  • 实时监控：通过日志聚合系统（如 ELK、Datadog）监控 LLM 调用成功率、平均延迟、Token 消耗趋势
  • 事后分析：回放日志排查特定会话中的异常行为，例如某个问题为什么触发了三次重试

九、跨模块循环依赖处理

9.1 问题的本质

config.py 需要引用 LLM 提供商定义来初始化默认模型，而 llm.py 需要引用 config 来获取配置项。两个模块相互引用，形成了循环依赖。

9.2 延迟导入方案

系统采用延迟导入（Lazy Import）来打破循环：

# 伪代码: 延迟导入方案
# llm.py
class LLMProvider:
    def chat(self, ...):
        # 在函数内部导入 config，避免模块顶层的循环依赖
        from config import settings
        model = settings.LLM_MODEL
        ...

使用延迟导入的代价是轻微的性能开销（每次调用时做 import），但 Python 的 import cache 让这个开销几乎可以忽略。

另一种方案是通过构造时注入配置来完全消除循环依赖：

# 伪代码: 构造时注入
class LLMProvider:
    def __init__(self, config: dict):
        self.config = config  # 配置由外部注入，不依赖 config 模块

但这种方式需要改动更多代码，延迟导入作为阶段性方案是性价比最高的选择。

十、生产实践与经验总结

10.1 兼容性是第一优先级

所有 LLM 提供商都通过兼容 OpenAI API 格式的方式接入。即使使用 Anthropic 的 Claude，也通过 Anthropic 提供的 OpenAI-compatible 代理端点接入。这让 LLMProvider 的核心代码保持在 150 行以内。

10.2 Thinking 模式的特殊约束

使用 DeepSeek 的 thinking 模式时，有一条容易踩坑的约束：

所有手动构造的 ChatCompletionMessage 必须保留 reasoning_content 字段。 DeepSeek 官方要求消息格式与返回格式保持一致，如果手动构建消息时丢弃了 reasoning_content，后续调用可能会报格式错误。

这意味着在 Agent 循环中拼接消息时，不能简单只提取 content：

# 伪代码: 手动构造消息时的注意事项
# 错误做法 ❌  — 忘记保留推理内容
message = {"role": "assistant", "content": result_text}

# 正确做法 ✅ — 保留推理内容（某些 API 强制要求）
message = {
    "role": "assistant",
    "content": result_text,
    "reasoning_content": result.reasoning,  # DeepSeek 等 API 要求此字段
}

10.3 多 Provider 的配置发现

系统在启动时自动扫描环境变量，发现所有已配置的 LLM 提供商：

OPENAI_API_KEY=sk-xxx          → 启用 OpenAI
DEEPSEEK_API_KEY=sk-xxx        → 启用 DeepSeek
ANTHROPIC_API_KEY=sk-xxx       → 启用 Anthropic
OLLAMA_BASE_URL=http://...     → 启用 Ollama
CUSTOM_1_API_KEY=sk-xxx        → 启用自定义提供商

这种零配置发现机制意味着：部署时只需要在 .env 文件中填入对应的 API Key，系统就会自动接入该提供商，无需修改代码或配置文件。

十一、总结

LLM 基础设施层是智能客服问答系统的"水电煤"——它为上层所有业务逻辑提供了稳定、统一的模型调用能力。核心设计原则概括如下：

  1. 统一抽象：LLMProvider 类用四个属性（name, api_key, base_url, models）统一了所有 LLM 提供商，子类只需填充各自的参数
  2. 统一入口：chat() 和 chat_stream() 是两个唯一的调用接口，屏蔽了底层提供商的差异
  3. 自动容错：指数退避重试 + 三级降级策略（重试→换模型→换提供商），保障服务可用性
  4. 净化输出：clean_json_response() 处理 LLM 输出的各种"杂质"，确保 JSON 解析零失败
  5. 精细计量：ContextVar 实现并发安全的按阶段 token 统计，为成本优化提供数据基础
  6. 完全可观测：每次 LLM 调用都被日志记录，支持实时监控和事后分析

这套基础设施层经过FAQ 系统的生产验证，在 99.5%+ 的 API 调用中实现了零手动干预的自动故障恢复。

下一篇文章将深入解析 JSON Schema 驱动的结构化输出与工具调用（Tool Use）机制。

本系列文章索引：

  • 01: Agent 记忆分层设计实践
  • 02: 多策略路由的智能客服系统设计
  • 03: 五种检索引擎与混合匹配架构
  • 04: Agentic RAG 质量自检与自适应重试
  • 05: 多意图查询拆分与递归分解
  • 06: 配置驱动架构与双模式管线
  • 07: LLM 基础设施层与提供商抽象（本文）
# 人工智能 # RAG
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

WalkingLab 组织入驻 AtomGit，释放高校智能体开源科研成果

avatar AtomGit开源社区
cover

拼手速！GLM-5.2免费Token每天10点准点开抢！

avatar AtomGit开源社区
cover

圆满落幕｜智驱迭代・昇腾赋能 AI Agent 行业实践 Meetup 上海站，全栈落地干货一次吃透

avatar AtomGit开源社区