企业官网建设流程全解析

做AI应用开发这一年多，我踩得最深的坑不在模型效果上，而是“把大模型调用直接写死在业务代码里”。一开始项目只有一个模型供应商，代码里到处是SDK调用，看着没啥问题。等第二个供应商接进来，立刻乱了：请求参数、返回结构、鉴权方式全不一样，更别提哪家一限流，整个业务跟着抖。后来我专门抽了一层LLM网关和大模型调用层，把所有供应商适配、高可用策略、成本控制全部收拢到这一层，才算是把这块地耕顺了。这篇就把我在实际项目里的做法和踩过的坑整理出来。适合正在做AI应用开发、准备接入多个大模型供应商，或者被线上调用稳定性折腾得睡不好的同学。

1. 为什么需要单独搞一层“大模型调用层”

1.1 直接调SDK写业务代码，前期爽后面痛

很多AI应用一开始都是从“调一个供应商”起步的。业务代码里直接new一个OpenAI客户端，或者用某个国产模型的Python SDK，写到哪调到哪。看起来简单直接，第一版可能两三天就上线了。但这时候你欠下的技术债，会在第二个供应商进来的时候一次性偿还。

举个例子。业务方今天说“我们想同时对比几家模型的效果”，你要接Claude、文心、通义、智谱这些。每家SDK的接口不一样，有的用messages数组，有的用prompt字段；有的叫max_tokens，有的叫max_new_tokens；有的返回在choices[0].message.content里，有的返回在output.text里。如果你在每一个业务接口里都对不同供应商做适配，那这个适配代码会散落在几十个文件中，后面改一个超时参数都要全局搜索，还容易漏。

更麻烦的是异常处理。一个供应商超时了要不要重试？重试会不会重复扣费？另一个返回400是不是因为prompt格式有问题？这些问题散落到业务代码里之后，排查一次线上事故能把人逼疯。我见过最夸张的情况是某业务模块里写满了“if model == 'xxx'”这样的分支，后来要下掉一个渠道，团队讨论了一周都不敢动，怕误伤。

1.2 网关层解决什么问题，什么时候需要上

把大模型调用层独立出来，本质上就是做一套“LLM网关”，它挡在业务和多个模型供应商之间，对外提供一套稳定的接口，对内做统一的模型路由和生命周期管理。

这套网关通常要解决四个问题。第一是协议统一，把不同厂商的请求参数、返回结构、鉴权方式全部归一化；第二是高可用，包括超时、重试、熔断、限流，不能让单个供应商故障拖垮整个业务；第三是成本与配额管理，按业务线、按API Key统计调用量和token消耗，防止预算失控；第四是可观测性，每个请求走了哪个模型、耗时多少、用了多少token、出错在哪个环节，都要有日志和指标。

那是不是所有项目一上来就要做这套东西？我的建议是未必。如果你只是在本地跑个Demo，或者业务规模很小，只有一个供应商，临时调一下也无妨。但出现下面几个信号，就说明你该上了：

• 开始接入第二个供应商做模型对比或灾备；
• 业务方经常提出“这个接口换个模型试试”；
• 某次供应商服务抖动，导致线上大面积超时；
• 月底算账时发现根本说不清每位客户消耗了多少token。

这些信号出现任何一个，花一周时间把LLM网关补齐，都远比后续反复救火划算。

2. 多供应商适配：核心是把“差异”挡在网关里

2.1 统一请求模型和响应模型，先定义家园

做多供应商适配，第一件事不是写代码，而是定义一套“网关自己的”请求和响应数据模型。它不跟任何一家供应商绑定，是所有适配逻辑的共同语言。

我们当时用的是一套精简的聊天补全模型。请求侧包含model、messages、temperature、max_tokens、stream、stop等字段；响应侧则是id、model、choices、usage、created等字段。这套模型只保留业务真正关心的信息，不追求覆盖所有厂商的所有参数。

为什么必须统一？因为你不能让上层业务感知“这个请求要发给OpenAI，所以要用OpenAI的字段格式”。业务只负责把意图发给网关，至于网关内部怎么把temperature翻译成各家参数，那是网关的事。我用一个表格说明当时踩过的典型差异：

这些差异看起来小，但如果不做抽象，业务层写一套兼容所有厂商的解析代码会非常痛苦。我在设计时把各家返回先通过适配器转换成统一的ChatResponse，再提交给上层的业务方。就算某家模型改了返回层级，也只需要改适配器，业务代码完全不用动。

2.2 供应商插件化与动态路由

统一模型有了之后，下一步就是供应商插件化。在我看来，每个供应商就是一个Provider，它要实现几个固定方法：chat_completion、chat_completion_stream、embedding（如果网关还要做向量化）、check_health。上层路由只需要依赖这个接口，不需要关心Provider内部调的是HTTP API还是SDK。

插件化之后，动态路由就顺理成章了。所谓动态，就是供应商列表、模型映射、权重、优先级都不需要改代码，通过配置文件或配置中心动态调整。

路由这件事我建议从最简单的开始，不要一上来就整什么机器学习路由。我们最早期就两个规则：按指定模型路由、按主备降级路由。比如请求里带model=turbo，网关认为这个逻辑名对应OpenAI的gpt-4o-mini和国产某厂的qwen-turbo两个真实模型，平时默认走OpenAI，如果OpenAI健康状态异常，自动切到国产模型。

权重路由也可以作为补充，比如某段时间内60%流量走供应商A，40%走供应商B，用于灰度对比模型效果。但权重路由要小心会话类应用，同一个用户的上下文可能分散到不同模型，导致体验不一致。所以对于聊天场景，我更推荐“按用户或会话绑定供应商”，即在网关里维护一个会话级别的路由亲和性，同一个session_id始终走同一个供应商。

下面这个简化代码片段展示了路由的核心结构：

# router.py import random from typing import List, Optional class RouteDecision: def __init__(self, provider_name: str, model_name: str): self.provider_name = provider_name self.model_name = model_name class GatewayRouter: def __init__(self, provider_registry, health_status): self.provider_registry = provider_registry self.health_status = health_status async def route(self, request): # 请求指定了逻辑模型名，例如"turbo" candidates = self.provider_registry.get_models(request.model) if not candidates: raise ValueError(f"model {request.model} not configured") # 如果指定了要走的供应商，优先用 if request.preferred_provider: for cand in candidates: if cand.provider == request.preferred_provider: return RouteDecision(cand.provider, cand.model_name) # 根据权重轮询或随机 healthy_candidates = [] for cand in candidates: if self.health_status.is_healthy(cand.provider): healthy_candidates.append(cand) if not healthy_candidates: raise RuntimeError("no healthy provider") weights = [c.weight for c in healthy_candidates] chosen = random.choices(healthy_candidates, weights=weights)[0] return RouteDecision(chosen.provider, chosen.model_name)

这里没有做太复杂的东西，但足以支撑一个稳定的路由循环。真正的复杂度其实不在路由算法，而在健康状态管理和错误处理。

2.3 模型能力差异与降级策略

多供应商适配还有一个容易翻车的点：不同模型的能力差异很大。有的模型支持function calling，有的还不支持；有的支持JSON模式，有的只支持字符串输出；有的流式稳定，有的断流概率高。

所以网关里一定要有一份“能力矩阵”，登记每个真实模型的能力标签。比如function_call、json_mode、vision、streaming、long_context等。请求进来时，网关根据业务声明的能力要求，筛选符合条件的供应商，而不是什么请求都往第一个可用模型上丢。

降级策略也要基于能力矩阵设计。一个典型场景：主供应商因为限流或者故障不可用，网关自动切到备供应商。这个“备”不能只看健康状态，还要看能力和输入格式。比如主供应商支持function calling，备供应商不支持，你直接把原始请求丢过去，对方大概率会报错或者忽略函数定义。正确的做法是提供一个降级处理器，在切换时把function calling相关的参数剥掉，并提示应用层“当前模型不支持工具调用”。

我在实际项目中总结了一条原则：降级宁可慢，不能错。永远不要在降级路径上访问一个未知字段，也永远不要在降级时丢掉必要的上下文。更安全的做法是，在降级前用配置中心把模型映射和参数映射临时调整好，再切换流量。人工可控的降级，往往比“全自动聪明降级”更可靠。

3. 高可用大模型调用层的设计要点

3.1 超时、重试与熔断，三个老伙计换个新环境

LLM调用和普通HTTP调用有个很大区别：它真的慢。一个复杂推理请求可能要几十秒，流式更是能持续一分钟以上。所以“超时”这件事不能照搬普通接口的标准。

我一般把超时分成三层。第一层是连接超时，通常设置3到5秒；第二层是首包超时，也就是发完请求后等第一个token返回的时间，这个根据模型不同可能从5秒到30秒不等；第三层是空闲超时，针对流式响应，两个token之间超过15秒没有数据就认为连接已经死了。对于非流式请求，还需要设置一个总超时，比如60秒或120秒。

重试就要更克制。生成型接口不天然幂等，同一个prompt请求两次可能生成内容不同，还会产生两次费用。所以我的重试原则是：只有网络层错误、连接重置、429限流，以及5xx服务端错误才重试，业务参数错误（400、401）一律不重试。重试次数控制在2次以内，并且使用指数退避加随机抖动。重试请求要打上retry_count标签，方便在日志里识别。

熔断是防止雪崩的关键。如果某个供应商连续报错，再多的重试都是浪费。我在网关里为每个Provider单独维护一个熔断器，状态有CLOSED、OPEN和HALF_OPEN。CLOSED代表正常，错误率超过阈值就变成OPEN；OPEN状态下直接快速失败，不再真实请求；过一段时间放少量探测请求，成功则恢复为CLOSED，失败则回到OPEN。

有一个容易被忽略的点：熔断的指标不能只看HTTP状态码，还要把超时和空响应算进去。有些供应商响应200，但request_id异常或者返回的token数量为0，这种也应当算作失败。

3.2 限流、配额与健康检查的联动

LLM网关的限流分为入口和出口两级。入口限流是保护业务后端，按API Key或用户维度限制每秒请求数；出口限流是尊重供应商配额，防止单个渠道被打爆。入口限流我们用的是令牌桶算法，分布式环境用Redis存储令牌，保证多个网关实例共享限流状态。

出口限流更要精细。每家供应商都有每分钟请求数（RPM）和每分钟token数（TPM）限制。网关要动态统计最近一分钟的请求量和token消耗，在接近阈值时主动排队或降级，而不是傻乎乎地硬冲。否则你会在供应商侧收到一堆429，然后被对方限得更狠。

健康检查和限流通常是联动的。我每隔30秒会调用一次供应商的轻量级接口做健康探测，同时把真实请求的错误率、平均延迟也纳入健康判断。如果某个供应商连续5个请求失败，就把它的状态置为“不健康”，路由自动跳过。当它恢复稳定后，再逐步把流量调回去。

这个机制还要考虑“冷却”。不要一恢复就立刻全量放流量，我一般会让恢复后的供应商承担20%流量，观察几分钟没有异常再增加到50%、100%。这本质上是一种手动/半自动的灰度回归。

3.3 缓存、成本与数据安全，一个都不能少

LLM调用成本不低。同样是聊天对话，每天上万次调用，几个月下来账单可能让人肉疼。所以网关层一定要做成本优化，第一招就是缓存。

缓存适合那些“相同或高度相似的请求”。比如智能客服里的常见问题，用户问法接近，完全可以让网关缓存住第一次的回复，后续相同意图直接命中。最简单的缓存key是请求参数的哈希值，精细一点的是把消息内容做向量化，用余弦相似度判断是否命中缓存。但向量缓存要慎重，相似不等于完全一致，如果业务对答案准确性要求极高，建议还是只做精确缓存。

成本优化的第二招是模型选择。同一个任务，用旗舰模型和轻量模型可能效果差距不大，成本却相差10倍。网关可以根据业务优先级设置“成本路由”：高价值用户走贵模型，普通用户走便宜模型；高峰期自动降级到性价比更高的模型。

数据安全同样要在网关层做。日志里绝不能明文记录完整的prompt和response，必须做脱敏或者截断。密钥管理要单独拎出来，存放在专门的密钥管理服务中，不要在配置文件里写死密钥，更不要把密钥打进容器镜像。另外，不同供应商的数据驻留和隐私协议不同，接入前要让安全团队评估是否符合自己公司的合规要求。

4. 实操：从零构建一个轻量LLM网关

4.1 技术选型：FastAPI + httpx + Redis，够了

自研LLM网关不一定要搞很重的框架。我选的组合是FastAPI作为HTTP服务框架，httpx作为异步HTTP客户端，Redis承担限流和分布式锁，Prometheus + Grafana做监控指标展示。如果公司已经有完善的监控体系，也可以直接上报到现有平台。

为什么不用某些大而全的网关中间件？因为LLM调用层和传统流量网关有一个关键区别：它需要理解模型语义，要做流式转发、token统计、模型降级。自研虽然要写一些代码，但胜在灵活，能按照自己的业务需求调整路由策略。当然，如果你们团队时间紧、供应商也不多，也可以先基于开源方案改，哪怕后面再自研，也能少走很多弯路。

目录结构我会按“领域”划分，而不是按“技术层”划分：

llm-gateway/ main.py core/ config.py schemas.py router.py circuit_breaker.py providers/ base.py openai_provider.py anthropic_provider.py custom_provider.py middleware/ auth.py rate_limit.py logging.py storage/ redis_client.py

这样做的目的是让“供应商”这一维度非常清晰。新增一个供应商，只需要在providers目录里加一个文件，然后在配置里登记，不用动业务代码。

4.2 统一数据模型与配置文件

我用Pydantic定义请求和响应模型。这里要特别注意流式和非流式需要复用同一套消息格式。

# core/schemas.py from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class ChatMessage(BaseModel): role: str content: str class ChatRequest(BaseModel): model: str = Field(..., description="逻辑模型名") messages: List[ChatMessage] temperature: float = 0.7 max_tokens: int = 1024 stream: bool = False stop: Optional[List[str]] = None preferred_provider: Optional[str] = None user_id: Optional[str] = None request_id: Optional[str] = None capabilities: Optional[List[str]] = Field( default_factory=lambda: ["chat"] ) class ChatChoice(BaseModel): index: int message: Optional[ChatMessage] = None delta: Optional[ChatMessage] = None finish_reason: str = "" class ChatResponse(BaseModel): id: str model: str provider: str choices: List[ChatChoice] usage: Dict[str, Any] = Field(default_factory=dict) created: int = 0

配置文件我用YAML。里面维护两个重要映射：一个是逻辑模型到供应商真实模型的映射，另一个是供应商实例的连接参数。

# config/providers.yaml providers: - name: openai base_url: https://api.openai.com api_key_env: OPENAI_API_KEY timeout: 60 health_check_path: /models weights: turbo: model_name: gpt-4o-mini weight: 80 pro: model_name: gpt-4o weight: 50 - name: qwen base_url: https://dashscope.aliyuncs.com/compatible-mode/v1 api_key_env: DASHSCOPE_API_KEY timeout: 30 weights: turbo: model_name: qwen-turbo weight: 20 models: turbo: fallback_order: [openai, qwen] capabilities: [chat, streaming] pro: fallback_order: [openai, qwen] capabilities: [chat, streaming, function_call]

这里有个小细节：不同供应商的base_url、鉴权方式差异大，所以我让Provider实现类去解析这些配置，而不是在配置里强行统一。

4.3 转发路由与流式处理的实现

下面这个代码是一个关键片段，展示了Provider基类和OpenAI风格Provider的实现。

# providers/base.py from abc import ABC, abstractmethod from typing import AsyncIterator, Dict, Any class BaseProvider(ABC): def __init__(self, name: str, config: Dict[str, Any]): self.name = name self.config = config @abstractmethod async def chat_completion(self, request, params) -> Dict[str, Any]: pass @abstractmethod async def chat_completion_stream(self, request, params) -> AsyncIterator[Dict[str, Any]]: pass async def check_health(self) -> bool: return True

OpenAI兼容接口的Provider实现起来最省事，因为国内很多模型也提供OpenAI兼容协议。但即使如此，我还是建议在Provider内做一层响应转换，不直接依赖它“恰好兼容”。

# providers/openai_provider.py import httpx from .base import BaseProvider from core.schemas import ChatResponse, ChatChoice from typing import AsyncIterator, Dict class OpenAIProvider(BaseProvider): def __init__(self, name, config): super().__init__(name, config) self.api_key = os.getenv(config["api_key_env"]) self.client = httpx.AsyncClient( base_url=config["base_url"], timeout=config.get("timeout", 60), headers={"Authorization": f"Bearer {self.api_key}"}, ) async def chat_completion(self, request, params): payload = { "model": params["model_name"], "messages": [m.dict() for m in request.messages], "temperature": request.temperature, "max_tokens": request.max_tokens, "stream": False, } resp = await self.client.post("/chat/completions", json=payload) resp.raise_for_status() data = resp.json() # 这里只提取网关需要的字段 return { "id": data.get("id"), "provider": self.name, "model": data.get("model"), "choices": [ { "index": item.get("index", 0), "message": item.get("message", {}), "finish_reason": item.get("finish_reason", ""), } for item in data.get("choices", []) ], "usage": data.get("usage", {}), "created": data.get("created", 0), }

流式处理是LLM网关最容易翻车的点。我在实现时选用了Server-Sent Events（SSE），也就是text/event-stream。网关要做的不是等完整响应再返回，而是边收供应商的token边转发给客户端。

async def chat_completion_stream(self, request, params): payload = { "model": params["model_name"], "messages": [m.dict() for m in request.messages], "temperature": request.temperature, "max_tokens": request.max_tokens, "stream": True, } async with self.client.stream("POST", "/chat/completions", json=payload) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): if not line.startswith("data:"): continue data = line[5:].strip() if data == "[DONE]": break yield data # 这里可以直接向上游转发

流式转发要注意，网关在转发时不能随便吞掉事件。有些模型的流式事件里携带usage信息，通常放在最后一个chunk里，网关要尽量原样透传，同时自己也解析一份用于指标统计。如果中途断了，网关要记得给客户端发送一个自定义的error事件，而不是默默断开。

4.4 熔断、重试与监控的落地编码

熔断器我不推荐引入很重的中间件，自己实现一个简洁版就够了。下面是一个滑动窗口计数的基础版本，只保留最近10秒的错误率。

# core/circuit_breaker.py import time from collections import deque class CircuitBreaker: def __init__(self, name, failure_threshold=0.5, window_seconds=10, open_time=30): self.name = name self.failure_threshold = failure_threshold self.window = deque() self.window_seconds = window_seconds self.state = "CLOSED" self.open_since = 0 self.open_time = open_time def record(self, success: bool): now = time.time() self.window.append((now, success)) while self.window and now - self.window[0][0] > self.window_seconds: self.window.popleft() if self.state == "CLOSED" and len(self.window) >= 10: rate = sum(1 for _, s in self.window if not s) / len(self.window) if rate >= self.failure_threshold: self.state = "OPEN" self.open_since = now def allow_request(self) -> bool: if self.state == "OPEN": if time.time() - self.open_since >= self.open_time: self.state = "HALF_OPEN" return True return False return True def on_success(self): self.record(True) if self.state == "HALF_OPEN": self.state = "CLOSED" self.window.clear() def on_failure(self): if self.state == "HALF_OPEN": self.state = "OPEN" self.open_since = time.time() else: self.record(False)

熔断器需要和重试逻辑搭配。我的处理流程是：先查熔断器是否允许请求；发送请求；根据结果更新熔断器；如果允许重试且是对应错误类型，重试前先退避一段时间。请求要携带request_id，贯穿整个链路，方便在日志里串联。

监控方面，我每个请求会在结束时记录这些指标：provider_name、model_name、succeed/failed、http_status、latency_ms、total_tokens、prompt_tokens、completion_tokens、retry_count、circuit_breaker_state。这些指标打进Prometheus，配合Grafana做面板。告警规则至少要有三条：单供应商错误率5分钟超过20%；P95延迟超过10秒；日调用成本超过设定阈值。

5. 常见问题与故障排查实录

5.1 路由结果和预期不一致，先查配置和健康状态

有一次我们线上流量突然全部切到了备用供应商，业务反馈响应变慢了，但主供应商看起来没挂。查日志才发现，我们在配置中心改权重的时候，不小心把主供应商的权重改成了0。看起来权重是0，路由就把所有流量分给了备用。这个问题排查了半天，最后是直接看RouteDecision的日志信息才定位到。

所以我强烈建议，网关里每个请求都要打印路由决策详情，至少包括：逻辑模型名、候选供应商列表、每个候选的健康状态、最终选择的供应商和原因。不要只打印一个最终选择，否则出了问题你都不知道它是怎么选出来的。

另外要检查健康检查本身是否误判。有些供应商的/models接口很轻，但推理接口可能不稳定。如果你只依赖/models做健康检查，很可能接口健康、推理超时。

5.2 流式响应中途断开，重试要非常小心

流式断连在所有AI应用里都让人头疼。我们遇到过一种典型场景：客户端从网关读取SSE流，读到一半网关和供应商的连接断了，客户端卡住直到超时才报错。

排查时发现是网关到供应商之间没有设置空闲超时，恰好某个模型生成长文本时两个token之间的间隔超过了默认超时，连接被系统回收。后来我们把空闲超时加大到30秒，并增加了首包超时，问题缓解很多。

但要提醒一句：流式请求一旦已经向客户端推送了部分内容，就尽量不要做自动重试。因为客户端可能已经渲染了部分文本，再从头来一次会造成内容重复甚至前后不一致。正确做法是，上游断了就立刻给客户端发一个错误事件，并把这个请求标记为失败，让应用层决定是提示用户重试还是继续等待。

5.3 高并发下供应商被限流，别想着硬扛

某次活动流量上来后，大量请求打到主供应商，触发了对端的TPM限制，返回了一堆429。当时网关的自动降级没有配置好，导致所有请求都在等主供应商的429重试，反而加剧了服务端压力。后来我们调整了策略：当检测到429比例升高时，直接把部分流量降级到备用供应商，同时在网关出口限流中降低该供应商的每秒并发数。

这里有个经验：429响应里通常带Retry-After头，网关一定要读取并遵守。不能对每个429都立刻重试。更好的做法是把这些请求放入一个带延迟的队列，一点点放出去，而不是在同一秒内全部冲回去。

5.4 生产环境排障速查表

我整理了一个简单的速查表，线上出问题时可以照着看：

这张表不是万能的，但大多数LLM网关的线上故障都能归到这六类里。

5.5 上线前一定要做故障注入测试

最后说一个我的习惯。每次新增供应商或者改路由策略，我都会在测试环境做几轮故障注入测试：

• 直接关掉主供应商的服务，看网关能否在30秒内自动切换到备用供应商；
• 人为制造超时，看超时设置和重试次数是否符合预期；
• 把某个备用供应商的API Key改成错的，看健康检查能否及时摘除它；
• 用大流量压测，看出口限流会不会触发雪崩。

这些测试不需要很复杂的工具，用脚本模拟失败请求就行。但一定要做。因为LLM网关的高可用，不是写在代码里就算数，而是要真的能扛住一次突如其来的故障才算数。

我个人的体会是，LLM网关最难的并不是某个技术点有多高深，而是你愿不愿意把“供应商差异”“失败场景”“成本指标”这些脏活累活提前想清楚。把这一层做扎实之后，业务换模型就像切换数据库连接池一样自然，供应商出故障也不再是半夜夺命call的理由。希望这篇实战记录能给你一些可复用的思路，少走点弯路。