企业官网建设流程全解析

1. 项目概述：从单点调用到多模型编排的实战演进

最近在折腾一个内部的知识库问答系统，最初图省事，直接接入了某一家大模型的API，上线跑了一段时间后，问题就来了：有些复杂逻辑推理的任务，它表现得很出色；但一到需要生成特定格式（比如严格的JSON）或者处理超长文本总结时，就时不时给你“抽风”，要么格式错乱，要么直接超时。这让我意识到，把鸡蛋放在一个篮子里，在LLM（大语言模型）应用开发里是行不通的。现在的模型生态百花齐放，各有专长，有的长于代码，有的精于创意，有的则在特定语言或成本控制上优势明显。于是，“多种LLM的API使用及开发流程”就成了我必须要趟过去的一个坎。

这不仅仅是为了简单的故障转移（A挂了切B），更深层的价值在于，我们可以根据不同的任务类型，智能地选择最合适、最经济或性能最好的模型，甚至将复杂任务拆解，让不同的模型各司其职，协同完成。比如，让一个模型负责理解用户意图并拆解任务，另一个模型负责检索信息，再找一个模型专门做格式规整和输出。要实现这套“模型编排”的愿景，第一步就是得把这些不同厂商、不同协议的API给统一管起来，平滑地集成到我们的开发流程里。接下来，我就结合自己的踩坑经验，聊聊如何系统性地搞定这件事。

2. 核心设计：构建一个可扩展的LLM网关层

直接在我们的业务代码里到处写requests.post(‘api.openai.com/v1/...’)，或者散落着各种client.chat.completions.create，是项目走向混乱的开端。一旦需要换模型、加降级、统一日志或计费，改动点就会散落一地。因此，核心思路是抽象出一个统一的LLM服务层或网关。这个层向上对业务代码提供一致的调用接口，向下则适配和管理各个不同的LLM提供商。

2.1 抽象接口设计

首先，我们需要定义一套与具体模型无关的核心抽象。无论底层是GPT、Claude还是国内的各种大模型，业务方关心的核心操作无非是“聊天补全”和“文本嵌入”。我们可以设计一个基础客户端接口。

from abc import ABC, abstractmethod from typing import List, Dict, Any, Optional class BaseLLMClient(ABC): """大语言模型客户端抽象基类""" @abstractmethod async def chat_completion( self, messages: List[Dict[str, str]], model: Optional[str] = None, temperature: float = 0.7, max_tokens: Optional[int] = None, **kwargs ) -> Dict[str, Any]: """ 聊天补全核心方法 :param messages: 消息列表，格式如 [{"role": "user", "content": "你好"}] :param model: 可选，指定使用的模型名称。若不指定，使用客户端默认模型。 :param temperature: 温度参数，控制随机性。 :param max_tokens: 生成的最大token数。 :param kwargs: 其他模型特定的参数。 :return: 包含响应内容、使用量等信息的字典。 """ pass @abstractmethod async def create_embedding( self, text: str, model: Optional[str] = None, **kwargs ) -> List[float]: """ 生成文本嵌入向量 :param text: 输入文本。 :param model: 可选，指定使用的嵌入模型。 :return: 嵌入向量列表。 """ pass

这个接口定义了最核心的两个功能。注意，这里将model参数设为可选，是因为每个具体的客户端实例在初始化时，通常会绑定一个默认模型（如gpt-4-turbo-preview或claude-3-sonnet-20240229）。这样设计既保持了灵活性，也提供了默认便利。

2.2 配置与工厂模式

接下来，我们需要一个统一的地方来管理所有LLM供应商的配置（API Key、Base URL、默认模型等），并且能根据一个标识符（如"openai","anthropic","deepseek"）动态创建对应的客户端实例。这里适合使用工厂模式。

import os from typing import Dict, Any from openai import AsyncOpenAI import anthropic class LLMClientFactory: """LLM客户端工厂，负责创建和管理具体的客户端实例""" _client_cache = {} # 简单缓存，避免重复创建 @classmethod def get_client(cls, provider: str, **kwargs) -> BaseLLMClient: """ 获取指定供应商的客户端 :param provider: 供应商标识，如 'openai', 'anthropic', 'zhipu' 等。 :param kwargs: 可覆盖默认配置，如 api_key, base_url。 :return: 实现了BaseLLMClient的具体客户端实例。 """ cache_key = f"{provider}:{str(kwargs.get('model', 'default'))}" if cache_key in cls._client_cache: return cls._client_cache[cache_key] # 从环境变量或配置中心读取默认配置 config = cls._get_default_config(provider) config.update(kwargs) # 用户传入的参数优先级最高 if provider == "openai": client = OpenAIClient(**config) elif provider == "anthropic": client = AnthropicClient(**config) elif provider == "deepseek": client = DeepSeekClient(**config) # ... 可以继续扩展其他供应商 else: raise ValueError(f"Unsupported LLM provider: {provider}") cls._client_cache[cache_key] = client return client @staticmethod def _get_default_config(provider: str) -> Dict[str, Any]: """获取各供应商的默认配置""" configs = { "openai": { "api_key": os.getenv("OPENAI_API_KEY"), "base_url": os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"), "default_model": "gpt-4o-mini" }, "anthropic": { "api_key": os.getenv("ANTHROPIC_API_KEY"), "default_model": "claude-3-5-sonnet-20241022" }, "deepseek": { "api_key": os.getenv("DEEPSEEK_API_KEY"), "base_url": "https://api.deepseek.com", "default_model": "deepseek-chat" } } return configs.get(provider, {}).copy()

注意：在实际项目中，不建议将敏感配置硬编码或直接写在工厂类里。这里为了示例清晰，从环境变量读取。生产环境应使用专门的配置管理服务（如Consul、AWS Parameter Store）或安全的密钥管理服务。

2.3 统一响应格式

不同API的返回结构差异很大。OpenAI返回的可能是.choices[0].message.content，而Anthropic返回的是.content[0].text。我们需要在抽象层将它们归一化，让业务代码无需关心这些细节。

class UnifiedResponse: """统一响应格式""" def __init__(self, content: str, raw_response: Any, usage: Optional[Dict] = None, model: str = ""): self.content = content # 模型生成的文本内容 self.raw_response = raw_response # 原始响应对象，用于调试或高级处理 self.usage = usage or {} # token使用情况，如 {'prompt_tokens': 10, 'completion_tokens': 20} self.model = model # 实际使用的模型名称 def __str__(self): return self.content

这样，无论底层调用哪个模型，业务层拿到的都是一个UnifiedResponse对象，直接访问response.content即可。

3. 核心细节：不同LLM供应商的适配与调优

有了顶层设计，接下来就是为每个供应商实现具体的适配器。这是工作量最大、也最需要细心的地方，因为每个API都有其“脾气”。

3.1 OpenAI系列适配详解

OpenAI的API是目前事实上的标准，协议相对规范。但即便是它，也有不少细节需要注意。

import httpx from openai import AsyncOpenAI, APIError, APITimeoutError class OpenAIClient(BaseLLMClient): """OpenAI系列客户端（兼容Azure OpenAI及其他兼容API）""" def __init__(self, api_key: str, base_url: str = "https://api.openai.com/v1", default_model: str = "gpt-4o-mini", timeout: float = 30.0): self.client = AsyncOpenAI(api_key=api_key, base_url=base_url, http_client=httpx.AsyncClient(timeout=timeout)) self.default_model = default_model async def chat_completion(self, messages, model=None, temperature=0.7, max_tokens=None, **kwargs): model = model or self.default_model try: # 构建请求参数 request_params = { "model": model, "messages": messages, "temperature": temperature, **kwargs # 允许传入其他OpenAI特有参数，如 top_p, presence_penalty } if max_tokens: request_params["max_tokens"] = max_tokens # 发起请求 response = await self.client.chat.completions.create(**request_params) # 提取内容并统一格式 content = response.choices[0].message.content usage = { "prompt_tokens": response.usage.prompt_tokens, "completion_tokens": response.usage.completion_tokens, "total_tokens": response.usage.total_tokens } if response.usage else None return UnifiedResponse( content=content, raw_response=response, usage=usage, model=model ) except APITimeoutError: # 处理超时，可触发重试或降级逻辑 raise LLMTimeoutError(f"OpenAI API request timeout for model {model}") except APIError as e: # 处理其他API错误，如额度不足、模型不可用等 raise LLMAPIError(f"OpenAI API error: {e}")

实操要点与避坑指南：

1. 超时设置：务必设置合理的超时时间（如30秒）。对于长文本或复杂推理，模型响应可能较慢，但也不能无限等待。httpx.AsyncClient(timeout=30.0)是一个好选择。
2. 流式响应处理：如果支持流式输出（stream=True），需要单独处理。代码会变得异步迭代，适用于需要实时显示生成结果的场景（如聊天界面）。但要注意，流式响应无法在响应头中直接获取usage信息，通常需要在流结束后通过单独的API查询。
3. 非OpenAI官方端点：许多国产模型或开源模型部署服务提供了与OpenAI兼容的API。使用它们时，只需将base_url替换为对应的地址，并传入正确的api_key（有时可能是一个占位符）。这是目前生态中最方便的集成方式。
4. Token计算与成本：response.usage里的token数非常关键，直接关联成本。对于长对话，需要自己维护历史消息的token计数，防止超出模型上下文限制。可以使用tiktoken库（针对OpenAI模型）进行本地估算。

3.2 Anthropic Claude系列适配

Anthropic的Claude模型在长上下文和逻辑推理上表现优异，但其API设计与OpenAI有显著不同。

class AnthropicClient(BaseLLMClient): """Anthropic Claude 客户端""" def __init__(self, api_key: str, default_model: str = "claude-3-5-sonnet-20241022", timeout: float = 60.0): # Anthropic官方SDK self.client = anthropic.AsyncAnthropic(api_key=api_key, timeout=timeout) self.default_model = default_model # Claude的消息格式要求更严格，system prompt是独立参数 self._system_prompt = None def set_system_prompt(self, prompt: str): """设置系统提示词，在Claude API中这是一个独立参数""" self._system_prompt = prompt async def chat_completion(self, messages, model=None, temperature=0.7, max_tokens=4096, **kwargs): model = model or self.default_model # 需要将通用的messages格式转换为Claude格式，并分离出system消息 claude_messages = [] system_content = self._system_prompt # 遍历消息，识别并提取system角色内容 for msg in messages: if msg["role"] == "system": # 如果messages里本身有system，且未通过set_system_prompt设置，则用它 if system_content is None: system_content = msg["content"] # 注意：Claude的system是独立参数，不能混在messages里，所以这里跳过该条消息 continue # 其他角色（user, assistant）直接转换 claude_messages.append({"role": msg["role"], "content": msg["content"]}) try: request_params = { "model": model, "messages": claude_messages, "temperature": temperature, "max_tokens": max_tokens, # Claude必须指定max_tokens **kwargs } if system_content: request_params["system"] = system_content response = await self.client.messages.create(**request_params) # Claude响应内容在 response.content[0].text content = response.content[0].text usage = { "input_tokens": response.usage.input_tokens, "output_tokens": response.usage.output_tokens } return UnifiedResponse( content=content, raw_response=response, usage=usage, model=model ) except anthropic.APITimeoutError: raise LLMTimeoutError(f"Anthropic API timeout for model {model}") except anthropic.APIError as e: raise LLMAPIError(f"Anthropic API error: {e}")

关键差异与注意事项：

1. System Prompt处理：这是与OpenAI最大的不同。Claude的system是一个独立的顶级参数，而不是放在messages列表里的一条消息。在适配时，需要从通用的messages列表中提取role为system的内容，或者通过单独的方法（如set_system_prompt）设置。如果混在messages里发送，Claude API会报错。
2. 必须的max_tokens：Claude API强制要求指定max_tokens参数，即你期望模型生成的最大token数。这需要根据任务合理预估，设置过低会导致回答被截断，过高则可能浪费资源。对于未知长度的生成，可以设置一个较大的安全值（如4096），并结合停止序列（stop_sequences）来控制。
3. 消息结构：Claude的消息内容字段也叫content，但响应结构是response.content[0].text，需要小心提取。
4. 长上下文优势：Claude 3.5 Sonnet支持200K上下文，在处理超长文档时优势明显。在适配时，可以设计逻辑：当输入文本超过某个阈值（如GPT-4 Turbo的128K）时，自动切换到Claude进行处理。

3.3 国内主流模型适配（以DeepSeek为例）

国内模型的API接入方式正在快速向OpenAI标准靠拢，这大大降低了集成成本。以DeepSeek为例，其最新版本提供了高度兼容的接口。

class DeepSeekClient(BaseLLMClient): """DeepSeek客户端（使用OpenAI兼容格式）""" def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com", default_model: str = "deepseek-chat", timeout: float = 30.0): # 直接复用OpenAI客户端，因为API格式兼容 self.client = AsyncOpenAI(api_key=api_key, base_url=base_url, http_client=httpx.AsyncClient(timeout=timeout)) self.default_model = default_model async def chat_completion(self, messages, model=None, temperature=0.7, max_tokens=None, **kwargs): # 调用方式与OpenAI客户端几乎完全一致 model = model or self.default_model try: request_params = { "model": model, "messages": messages, "temperature": temperature, **kwargs } if max_tokens: request_params["max_tokens"] = max_tokens response = await self.client.chat.completions.create(**request_params) content = response.choices[0].message.content usage = { "prompt_tokens": response.usage.prompt_tokens, "completion_tokens": response.usage.completion_tokens, "total_tokens": response.usage.total_tokens } if response.usage else None return UnifiedResponse( content=content, raw_response=response, usage=usage, model=model ) except Exception as e: # 注意：DeepSeek的错误类型可能与OpenAI不完全一致，这里做通用捕获 raise LLMAPIError(f"DeepSeek API error: {e}")

针对国内模型的通用经验：

1. 网络与延迟：这是最大的挑战。务必在客户端设置合理的超时和重试机制。考虑在业务部署地区（如国内）使用代理或选择提供国内加速节点的服务商。
2. 内容审核：国内模型通常有更严格的内容安全过滤。如果生成的文本触发了安全策略，可能会返回空内容或特定错误码。在业务逻辑中需要处理这种情况，例如进行重试（使用更温和的表述）或降级到其他模型。
3. 计费模式：仔细阅读计费文档。有些模型按token计费，有些按调用次数计费，还有的提供套餐包。在统一网关里，可以集成计费统计模块，为每个供应商的调用记录token消耗或次数，便于成本分析和优化。
4. 模型更新：国内模型迭代速度快，模型名称可能频繁变化（如从deepseek-chat到deepseek-v3）。建议将模型名称配置化，而不是硬编码在代码里。

4. 高级编排与策略实现

当多个模型客户端就绪后，我们就可以玩出更多花样了，不再仅仅是简单的替换，而是智能的调度与组合。

4.1 故障转移与降级策略

最基本的策略是故障转移。当主模型服务不可用、超时或返回错误时，自动切换到备选模型。

class FallbackLLMOrchestrator: """带故障转移的LLM编排器""" def __init__(self, primary_provider: str, fallback_providers: list, **client_kwargs): """ :param primary_provider: 主用供应商，如 'openai' :param fallback_providers: 降级供应商列表，如 ['anthropic', 'deepseek'] """ self.primary_provider = primary_provider self.fallback_providers = fallback_providers self.client_kwargs = client_kwargs self._clients = {} # 缓存客户端 async def chat_completion_with_fallback(self, messages, **kwargs): """带自动降级的聊天补全""" providers_to_try = [self.primary_provider] + self.fallback_providers last_error = None for idx, provider in enumerate(providers_to_try): try: client = self._get_client(provider) # 如果是降级模型，可以调整参数，比如降低temperature以保证稳定性 if idx > 0: # 降级模型 kwargs_fallback = kwargs.copy() kwargs_fallback['temperature'] = min(kwargs.get('temperature', 0.7), 0.3) response = await client.chat_completion(messages, **kwargs_fallback) else: response = await client.chat_completion(messages, **kwargs) # 记录本次实际使用的模型 response.metadata = {"used_provider": provider, "was_fallback": idx > 0} return response except (LLMTimeoutError, LLMAPIError) as e: print(f"Provider {provider} failed: {e}") last_error = e continue # 尝试下一个 # 所有提供商都失败 raise LLMServiceUnavailable(f"All LLM providers failed. Last error: {last_error}") def _get_client(self, provider: str): """获取或创建客户端""" if provider not in self._clients: self._clients[provider] = LLMClientFactory.get_client(provider, **self.client_kwargs) return self._clients[provider]

这个编排器会按顺序尝试供应商列表。一个更高级的策略可以是基于错误的类型进行降级：如果是超时错误，可能只是网络波动，可以短暂重试主模型；如果是认证错误或额度不足，则应立即切换到备用模型。

4.2 基于任务类型的路由策略

更精细的策略是根据任务本身的特点来选择模型。我们可以预先定义一些任务类型和对应的模型偏好。

class SmartRouterLLMOrchestrator: """基于任务类型的智能路由编排器""" # 任务类型与模型偏好映射 TASK_ROUTING_TABLE = { "creative_writing": ["openai/gpt-4", "anthropic/claude-3-opus"], # 创意写作，需要想象力 "code_generation": ["openai/gpt-4", "openai/gpt-4o"], # 代码生成，GPT系列传统强项 "logical_reasoning": ["anthropic/claude-3-5-sonnet", "openai/gpt-4"], # 逻辑推理 "long_document_qa": ["anthropic/claude-3-5-sonnet-200k", "openai/gpt-4-turbo"], # 长文档问答 "structured_output": ["openai/gpt-4o", "anthropic/claude-3-5-sonnet"], # 需要严格JSON/XML输出 "cost_sensitive": ["openai/gpt-4o-mini", "deepseek/deepseek-chat"], # 成本敏感型任务 } def __init__(self, **client_kwargs): self.client_kwargs = client_kwargs self._client_cache = {} async def route_and_complete(self, task_type: str, messages, **kwargs): """ 根据任务类型路由到最合适的模型 :param task_type: 任务类型，如 'code_generation' """ if task_type not in self.TASK_ROUTING_TABLE: # 未知任务类型，使用默认路由或抛出错误 task_type = "default" preferred_models = self.TASK_ROUTING_TABLE.get(task_type, ["openai/gpt-4o-mini"]) # 默认 last_error = None for model_identifier in preferred_models: try: # model_identifier 格式如 "openai/gpt-4" provider, model = model_identifier.split('/') client = self._get_client(provider) # 调用时指定具体模型 response = await client.chat_completion(messages, model=model, **kwargs) response.metadata = { "task_type": task_type, "selected_model": model_identifier } return response except (LLMTimeoutError, LLMAPIError) as e: print(f"Model {model_identifier} failed for task {task_type}: {e}") last_error = e continue raise LLMServiceUnavailable(f"No suitable model available for task {task_type}. Last error: {last_error}")

这个路由表需要根据实际测试效果和业务反馈不断调整。例如，经过实测，我们发现对于“生成严格遵循Schema的JSON”这类任务，GPT-4o的指令跟随能力极强，几乎每次都能输出完美格式，而其他模型则可能需要更复杂的后处理。那么就可以将structured_output任务优先路由给它。

4.3 串联与并联工作流

对于复杂任务，我们可以让多个模型协同工作，形成工作流。

• 串联（Chain）：一个模型的输出作为下一个模型的输入。例如，先用一个擅长总结的模型（如Claude）阅读长文档并生成摘要，再将摘要交给一个擅长推理的模型（如GPT-4）来回答问题。
• 并联（Parallel/Fan-out）：将同一个问题同时发给多个模型，然后综合它们的答案。这可以用于提高可靠性（如投票选择最佳答案）或生成多样化的创意。

实现一个简单的串联工作流示例：

async def summarize_and_qa(long_document: str, question: str): """串联工作流：先总结，再基于总结回答""" # 第一步：用Claude总结长文档（假设其长上下文能力强） summarizer = LLMClientFactory.get_client("anthropic") summary_prompt = f"请用中文简要总结以下文档的核心内容：\n\n{long_document}" summary_response = await summarizer.chat_completion( messages=[{"role": "user", "content": summary_prompt}], max_tokens=500 ) summary = summary_response.content # 第二步：用GPT-4基于总结来回答问题 qa_agent = LLMClientFactory.get_client("openai") qa_prompt = f"基于以下摘要回答问题。\n摘要：{summary}\n\n问题：{question}" answer_response = await qa_agent.chat_completion( messages=[{"role": "user", "content": qa_prompt}], model="gpt-4o" ) return { "summary": summary, "answer": answer_response.content, "usage": { "summary_tokens": summary_response.usage, "qa_tokens": answer_response.usage } }

这种模式将不同模型的优势结合起来，往往能取得比单一模型更好的效果，但代价是更高的延迟和成本。

5. 生产环境考量与运维实践

将多模型API集成到生产环境，远不止写通适配代码那么简单。下面是一些关键的运维考量点。

5.1 监控、日志与可观测性

必须对每一次LLM调用进行详尽的记录和监控。

1. 日志记录：记录每次调用的时间戳、供应商、模型、输入消息（可脱敏）、输出内容（可截断）、token使用量、耗时、是否成功。这有助于分析成本、性能和模型质量。
2. 性能指标：监控平均响应时间、每秒请求数（QPS）、错误率（按供应商和模型细分）。设置告警，当某个模型的错误率或延迟超过阈值时，及时通知。
3. 成本监控：实时聚合各供应商的token消耗，换算成费用。可以设置每日/每周预算告警，防止意外开销。
4. 链路追踪：在微服务架构中，使用OpenTelemetry等工具为每次LLM调用生成唯一的Trace ID，便于在复杂工作流中定位问题。

可以在统一的客户端基类或网关层注入这些监控逻辑。

5.2 限流、重试与熔断

面对外部API，必须考虑其稳定性。

1. 限流（Rate Limiting）：每个API都有调用频率限制。需要在网关层实现全局或按API Key的限流，防止触发供应商的限流导致429错误。可以使用令牌桶等算法。
2. 重试策略：对于网络超时（Timeout）、服务器内部错误（5xx）等暂时性故障，应实施有退避策略的重试（如指数退避）。但对于客户端错误（如4xx，无效请求）则不应重试。
3. 熔断器（Circuit Breaker）：当某个供应商的API在短时间内失败率达到阈值（如50%），应快速熔断，在一段时间内直接拒绝发往该供应商的请求，转而使用降级方案，给下游服务恢复的时间。这可以防止系统资源被拖垮。

5.3 密钥管理与安全

API Key是最高权限的凭证，必须妥善管理。

1. 严禁硬编码：绝对不要将API Key写在源代码或配置文件中提交到代码仓库。
2. 使用密钥管理服务：如AWS Secrets Manager、HashiCorp Vault、Azure Key Vault等，动态获取密钥。
3. 密钥轮转：定期轮换API Key，并在网关层支持多Key轮询，既能分散风险，也能应对单个Key的额度限制。
4. 按需授权：在架构上，业务服务不直接持有LLM API Key，而是向统一的LLM网关发起请求，由网关负责鉴权和转发。这样可以将密钥管理集中在网关层。

5.4 测试与评估

如何确保多模型策略真的提升了效果？

1. A/B测试：对于关键任务，可以同时将请求发送给新旧两种策略（如单一模型 vs. 智能路由），在线上进行小流量对比，评估指标（如回答准确率、用户满意度、成本）的变化。
2. 评估数据集：构建一个涵盖各种任务类型的测试集，定期（如每周）用不同的模型/策略跑一遍，自动评估生成质量（可以通过另一个LLM打分，或基于规则匹配），形成质量报告。
3. 影子测试（Shadow Testing）：将生产流量复制一份，发送给新的模型或策略进行处理，但不将结果返回给用户。通过对比新老结果的差异，来评估新策略的效果，无风险地收集数据。

6. 常见问题与实战排坑记录

在实际开发和运维中，我遇到了不少坑，这里记录一些典型问题和解决方法。

6.1 上下文长度管理与Token计算

问题：不同模型上下文窗口大小不同（从4K到200K不等）。当对话历史很长时，如何确保不超出限制？

解决方案：

1. 本地Token计数：对于OpenAI系模型，使用tiktoken编码器进行精确计数。对于其他模型，查找其官方编码方式或使用近似估算（如按字符数/4估算）。在网关层维护一个“对话管理器”，在每次添加新消息前计算总token数。
2. 智能截断策略：当历史即将超限时，不是简单丢弃最早的消息，而是采用更智能的策略：
   • 总结压缩：用模型将早期的多轮对话总结成一条精简的系统提示。
   • 滑动窗口：只保留最近N条消息。
   • 关键信息提取：从历史中提取与当前问题最相关的实体、事实，作为新的上下文注入。
3. 模型路由：如前所述，识别出需要超长上下文的请求，直接路由到Claude 200K或GPT-4 Turbo 128K这类模型。

6.2 处理非标准API响应与错误

问题：即便使用统一接口，底层API可能返回各种非标准错误或内容过滤结果。

排查与处理：

1. 详细日志：务必记录完整的错误响应体，而不仅仅是状态码。很多信息藏在错误消息里。
2. 错误分类处理：
   • 内容安全拒绝：返回内容可能为空或被替换。需要检查响应中是否有特定标记（如finish_reason为content_filter），并设计重试逻辑（如修改提问措辞）。
   • 模型过载/不可用：错误码如503,429。触发重试和降级逻辑。
   • 无效请求：如参数错误、模型不存在。不应重试，应记录并告警，检查配置。
3. 设置默认响应：对于非关键路径，当所有降级都失败时，可以返回一个友好的默认回复，如“服务正在思考中，请稍后再试”，而不是直接向用户抛出技术错误。

6.3 流式输出的统一处理

问题：部分场景（如聊天）需要流式输出以提升用户体验，但不同API的流式响应格式不同。

解决方案：

1. 在抽象层定义流式接口：在BaseLLMClient中增加stream_chat_completion方法，返回一个异步生成器（AsyncGenerator）。
2. 适配不同流式格式：在每个具体客户端中，处理各自的流式数据包，提取出“增量文本”（delta），并统一包装成{"content": delta_text, "finish_reason": null}这样的格式yield出去。
3. 最终消息聚合：流式结束时，通常还会返回完整的usage信息。需要在生成器最后一并返回，或通过其他机制传递给调用方。

6.4 成本控制与优化

问题：多模型调用后，成本可能失控。

优化策略：

1. 分层模型策略：将任务分为高、中、低价值。高价值任务（如直接面向客户的关键回答）用最强但最贵的模型（如GPT-4）；中价值任务（如内容分类、标签生成）用性价比高的模型（如GPT-4o-mini、Claude Haiku）；低价值或内部任务用成本最低的模型（如小型开源模型API）。
2. 缓存：对于频繁出现的、结果确定的查询（如“今天的天气怎么样？”），可以将LLM的响应结果缓存起来（设置合理的TTL），直接返回缓存结果，大幅节省token。
3. 输出限制：在调用时明确设置max_tokens，避免模型生成冗长无关的内容。对于摘要等任务，可以要求模型“用不超过100字总结”。
4. 定期审计：通过监控数据，定期分析“成本最高的任务类型是哪些？”、“哪个模型的性价比突然变低了？”，持续调整路由策略和模型选型。

整合多模型API的旅程，是从一个简单的API调用者，向一个资源调度者和质量保障者角色的转变。初期会觉得繁琐，但一旦这套体系搭建起来，你会发现面对模型世界的快速变化时，你拥有了前所未有的灵活性和主动权。不再被单一供应商绑定，可以冷静地根据任务需求、性能表现和成本预算，选择最合适的工具。这套架构的核心价值，不在于支持了多少个模型，而在于它赋予了你一种“模型即插即用”的能力，让LLM真正成为你手中可组合、可调配的智能组件。