七、LLM 基础设施层与提供商抽象：智能客服系统的模型接入统一架构

在智能客服问答系统中，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 提供商的基础抽象类，设计简洁而富有表现力：

classLLMProvider:name:str# 提供商名称，如 "openai", "deepseek"api_key:str|None# API 密钥base_url:str|None# API 端点 URLmodels: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 模式@propertydefclient(self):ifnotself._client:# 首次调用时初始化 OpenAI 兼容客户端self._client=create_api_client(api_key=self.api_key,base_url=self.base_url)returnself._client

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

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

2.3 子类实现

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

# 伪代码: 子类实现示意classOpenAIProvider(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.1chat()——统一 Chat Completion 入口

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

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

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

# 伪代码: chat() 重试逻辑示意defchat(self,messages,model=None,**kwargs):# 最多重试 3 次forattemptinrange(MAX_RETRIES):try:response=self._call_llm_api(messages,model,**kwargs)returnparse_response(response)exceptTEMPORARY_ERRORSase:ifattempt==MAX_RETRIES-1:raise# 最后一次失败，向上抛出wait=EXPONENTIAL_BACKOFF(attempt)# 指数退避sleep(wait)

3.2chat_stream()——流式输出

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

# 伪代码: 流式 chat 示意defchat_stream(self,messages,model=None,**kwargs):# 发起流式请求stream=self._create_stream(messages,model,**kwargs)forchunkinstream:# 跳过推理/思考过程的中间内容ifchunk.is_reasoning:continue# 仅输出最终回答的部分ifchunk.content:yieldchunk.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 模式，标记启用ifself.thinking:request.enable_thinking_mode()returnsend_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 清理流程defclean_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)returntext ```text=re.sub(r'```\s*','',text)# 第三步：提取纯 JSON（找到第一个 { 和最后一个 }）start=text.find('{')end=text.rfind('}')ifstart!=-1andend!=-1:text=text[start:end+1]# 第四步：尾部截断——删除最后一个 } 之后的无关内容# 处理 LLM 在 JSON 后继续输出解释文字的情况returntext.strip()

5.2 四大处理步骤详解

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={})@contextmanagerdeftoken_stage(stage_name:str):# 保存当前阶段，切换到新阶段previous=_current_stage.set(stage_name)try:yieldfinally:# 恢复上一阶段_current_stage.reset(previous)

6.2 ContextVar 实现并发安全

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

# 请求 A 在处理 retrieval 阶段withtoken_stage("retrieval"):llm.chat(...)# 累计到 retrieval# 请求 B 同时在处理 generation 阶段withtoken_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 调用可能因多种原因失败：

7.2 指数退避实现

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

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

7.3 优雅降级的层级

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

1. 请求级：同一提供商的同一请求重试（指数退避）
2. 模型级：当前模型失败后，切换到同提供商的备用模型（如 gpt-4o → gpt-4o-mini）
3. 提供商级：当前提供商全面不可用时，切换到另一个提供商（如 OpenAI → DeepSeek）

# 伪代码：多提供商逐级降级available_providers=["openai","deepseek","ollama"]forprovider_nameinavailable_providers:try:provider=get_provider(provider_name)returnprovider.chat(messages)exceptException:continue# 当前提供商不可用，尝试下一个# 所有提供商都失败raiseServiceUnavailable("所有 LLM 提供商均不可用")

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

八、日志记录——_log_llm与_log_llm_error

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

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

# 伪代码: LLM 调用日志记录deflog_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 调用错误日志deflog_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.pyclassLLMProvider:defchat(self,...):# 在函数内部导入 config，避免模块顶层的循环依赖fromconfigimportsettings model=settings.LLM_MODEL...

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

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

# 伪代码: 构造时注入classLLMProvider: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 基础设施层与提供商抽象（本文）