企业官网建设流程全解析

1. 从单模型到多模型：为什么我决定自建调度网关

去年下半年，我负责的一个智能问答类项目遇到了一个很典型的问题：所有请求都打给同一个大模型，结果就是成本失控、高峰期延迟飙升、部分场景回答质量还不稳定。最要命的是，老板问我“到底哪个模型在什么场景下表现最好、花了多少钱”，我拿不出任何数据。

那段时间我试过几种方案。最粗暴的是在业务代码里写 if-else，根据请求类型手动切换模型。代码很快就变成了意大利面，新增一个模型要改五六个地方，测试覆盖不过来。后来想用现成的API网关做转发，但那些网关对模型调用的语义支持太弱，没法按 token 数计费、没法做语义缓存、更没法根据实时延迟动态路由。

折腾了两周之后，我决定自己搭一个多模型调度网关。核心目标就三个：成本透明、延迟透明、质量透明。成本透明是指每次调用都能追溯到具体模型和 token 消耗；延迟透明是指能实时看到每个模型在不同场景下的响应时间分布；质量透明是指能通过反馈机制持续评估各模型在具体任务上的表现。

这个网关我最终基于蓝耘元生代平台搭建，跑通之后效果超出预期。下面我把整个设计思路、核心实现和踩过的坑完整分享出来，适合正在做多模型接入、成本优化或者AI应用架构的同行参考。即使你目前只用了一个模型，这套思路也能帮你提前把可观测性做好，后面扩展时不会手忙脚乱。

2. 整体架构设计：调度网关到底该管什么

2.1 核心需求拆解与边界划定

在动手写代码之前，我先花了一天时间把需求理清楚。一个多模型调度网关，本质上要解决四个层面的问题：

第一层是接入层。不同模型的API协议不一样，有的用OpenAI兼容格式，有的用自定义HTTP接口，参数命名和返回结构都有差异。网关要做的第一件事就是把这些差异屏蔽掉，对上提供统一的调用接口。

第二层是路由层。这是网关的核心价值所在。路由策略不能是简单的轮询或者随机，而应该基于请求的特征（比如任务类型、输入长度、是否需要多轮对话）和模型的实时状态（延迟、错误率、当前配额余量）来做决策。

第三层是可观测层。每次调用都要记录完整的上下文：请求ID、模型名称、输入输出token数、首token延迟、总延迟、是否命中缓存、用户反馈等。这些数据要能实时聚合，也要能按时间窗口回溯。

第四层是控制层。包括限流、熔断、降级、A/B测试分流等。当某个模型出现大面积超时或者错误率飙升时，网关要能自动把它从可用池里摘掉，等恢复后再重新加入。

我给自己定的边界是：网关只负责调度和观测，不负责模型本身的推理优化，也不做复杂的业务逻辑。这样职责清晰，后续维护成本低。

2.2 为什么选择蓝耘元生代作为底座

选蓝耘元生代作为底座，主要看中三点。一是它对多模型接入的支持比较友好，提供了统一的鉴权和配额管理，省去了我自己维护多套API Key的麻烦。二是它的调用日志和用量统计可以直接对接，我不用从零搭建数据采集管道。三是它的弹性伸缩能力，在流量波动时不用我手动干预。

当然，这不是说其他平台做不到，而是当时我的时间预算有限，需要一个能快速跑通MVP的方案。如果你用的是其他云平台或者自建推理集群，下面的设计思路同样适用，只是接入层的适配代码需要相应调整。

2.3 网关的整体数据流

整个网关的数据流是这样的：客户端请求先到网关的接入层，接入层做参数校验和标准化，然后交给路由层。路由层根据当前的路由策略选择一个目标模型，同时检查该模型是否在熔断状态。如果命中语义缓存，直接返回缓存结果，不调用模型。否则发起模型调用，记录调用前后的时间戳和token消耗。调用完成后，把结果返回给客户端，同时异步写入观测数据。如果调用失败，触发重试或降级逻辑。

这个流程看起来简单，但每个环节都有细节需要处理。比如语义缓存的命中判断、流式响应的延迟测量、重试时的幂等性保证等。下面我会逐个展开。

3. 核心模块实现：路由、缓存与可观测性

3.1 统一接入层：屏蔽模型差异的关键设计

接入层的核心任务是定义一套内部统一的请求和响应格式。我参考了OpenAI的Chat Completions格式，因为它的结构比较通用，大部分模型都能映射过来。内部请求对象包含这些字段：

• request_id：全局唯一标识，用于链路追踪
• task_type：任务类型，比如问答、摘要、翻译、代码生成
• messages：对话消息列表，每条包含role和content
• max_tokens：最大输出token数
• temperature：采样温度
• stream：是否流式返回
• metadata：业务侧附加信息，比如用户ID、会话ID

响应对象则包含：

• request_id：对应请求的ID
• model_name：实际调用的模型
• content：生成的文本
• prompt_tokens：输入token数
• completion_tokens：输出token数
• first_token_latency_ms：首token延迟
• total_latency_ms：总延迟
• cached：是否命中缓存

每个模型适配器负责把内部格式转换成目标模型的API格式，再把返回结果转换回内部格式。适配器用策略模式实现，新增模型时只需要写一个适配器类，注册到工厂里就行。

注意：token数的计算不要依赖模型返回的usage字段，因为有些模型不返回或者返回不准。我建议在网关侧用统一的tokenizer做一次预计算，虽然会多消耗一点CPU，但数据一致性有保障。

3.2 路由策略：成本、延迟、质量的三维权衡

路由策略是整个网关最核心也最复杂的部分。我最终实现的是一个加权评分模型，每个候选模型会得到一个综合分数，分数最高的被选中。评分公式如下：

score = w_cost * cost_score + w_latency * latency_score + w_quality * quality_score

其中cost_score、latency_score、quality_score都是归一化到0到1之间的值。权重w_cost、w_latency、w_quality可以根据业务场景动态调整。比如对成本敏感的场景，把w_cost调高；对实时性要求高的场景，把w_latency调高。

cost_score的计算方式是：先算出该模型在当前请求预估token数下的费用，然后取所有候选模型中费用的倒数，再归一化。latency_score用的是该模型过去5分钟内的P95延迟，同样取倒数归一化。quality_score则来自人工反馈和自动评估的综合得分，初始时所有模型给相同的默认值，随着反馈数据积累逐渐分化。

这里有个细节：quality_score的更新不能太激进。我一开始用最近10次调用的平均分，结果发现波动太大，路由决策频繁跳变。后来改成指数移动平均，平滑系数取0.1，稳定了很多。

3.3 语义缓存：不只是字符串匹配

语义缓存是我认为投入产出比最高的一个模块。普通的KV缓存只能命中完全相同的请求，但实际业务中大量请求是语义相似的。比如“今天天气怎么样”和“今天天气如何”，字符串不同但语义相同。

我的实现方案是：对请求的messages内容做embedding，然后在向量数据库里做相似度检索。如果相似度超过阈值（我设的是0.92），就认为命中缓存，直接返回之前的结果。阈值不能设太低，否则会把不同意图的请求误判为相同；也不能设太高，否则命中率上不去。0.92是我实测下来比较平衡的值。

缓存的有效期也要考虑。对于事实类问答，缓存可以设长一点，比如1小时；对于时效性强的请求，比如新闻摘要，缓存有效期要短，甚至不缓存。我在metadata里加了一个cache_ttl字段，由业务侧指定。

实操心得：语义缓存的embedding计算本身也有成本。如果请求量很大，建议用一个轻量级的embedding模型，或者对短文本做截断。我试过用完整的对话历史做embedding，延迟增加了80ms，后来改成只对最后一条用户消息做embedding，延迟降到15ms以内，命中率只下降了不到5%。

3.4 可观测性：让每一分钱和每一毫秒都有迹可循

可观测性模块我分了三个维度来建设。

成本维度：每次调用记录prompt_tokens和completion_tokens，乘以该模型的单价，得到本次调用的费用。这些数据按小时、按天、按模型、按业务方聚合，写入时序数据库。我用的是一张简单的宽表，字段包括时间戳、模型名、业务方、token数、费用。查询时用SQL聚合就行，不需要上复杂的OLAP引擎。

延迟维度：记录首token延迟和总延迟。首token延迟对用户体验影响最大，尤其是流式场景。总延迟则影响吞吐量。这两个指标都按P50、P95、P99分位数统计。我用的是滑动窗口算法，窗口大小5分钟，每10秒更新一次。

质量维度：质量评估分两种。一种是显式反馈，比如用户点赞点踩；另一种是隐式反馈，比如用户是否重新生成了回答、是否在回答后继续追问。显式反馈准确但稀疏，隐式反馈密集但有噪声。我把两者加权结合，显式反馈权重0.7，隐式反馈权重0.3。

这三个维度的数据最终汇聚到一个看板上，可以按模型、按时间段、按业务方筛选。看板是我用Grafana搭的，数据源直接连时序数据库。这样老板再问我“哪个模型划算”，我直接截图发过去就行。

4. 实操过程：从零搭建网关的完整步骤

4.1 环境准备与依赖安装

我用的技术栈是Python 3.11 + FastAPI + Redis + PostgreSQL + Milvus。FastAPI负责HTTP服务，Redis做限流和短期缓存，PostgreSQL存调用日志和配置，Milvus存语义缓存的向量。

安装依赖的命令如下：

pip install fastapi uvicorn redis psycopg2-binary pymilvus httpx numpy

如果你不想用Milvus，可以用FAISS做本地向量检索，省去一个中间件。但FAISS不支持分布式，多实例部署时缓存不共享。我一开始用FAISS，后来流量上来后换成了Milvus。

数据库建表语句我放在这里，主要是三张表：call_logs记录每次调用，model_configs存模型配置，routing_weights存路由权重。

CREATE TABLE call_logs ( id BIGSERIAL PRIMARY KEY, request_id VARCHAR(64) UNIQUE, model_name VARCHAR(64), task_type VARCHAR(32), prompt_tokens INT, completion_tokens INT, cost DECIMAL(10,6), first_token_latency_ms INT, total_latency_ms INT, cached BOOLEAN, success BOOLEAN, created_at TIMESTAMP DEFAULT NOW() ); CREATE TABLE model_configs ( model_name VARCHAR(64) PRIMARY KEY, api_endpoint VARCHAR(256), api_key_encrypted VARCHAR(512), input_price_per_1k DECIMAL(10,6), output_price_per_1k DECIMAL(10,6), max_context_tokens INT, enabled BOOLEAN DEFAULT TRUE ); CREATE TABLE routing_weights ( task_type VARCHAR(32) PRIMARY KEY, w_cost DECIMAL(3,2), w_latency DECIMAL(3,2), w_quality DECIMAL(3,2) );

4.2 模型适配器的编写与注册

适配器我定义了一个基类，所有具体模型的适配器都继承它：

from abc import ABC, abstractmethod class BaseModelAdapter(ABC): @abstractmethod async def call(self, request: InternalRequest) -> InternalResponse: pass @abstractmethod def estimate_cost(self, request: InternalRequest) -> float: pass

以某个OpenAI兼容格式的模型为例，适配器的核心逻辑是构造请求体、发送HTTP请求、解析响应。这里要注意超时设置，我设的是连接超时5秒，读取超时60秒。流式请求的读取超时要更长，因为模型可能生成很长的内容。

适配器写好后注册到工厂：

ADAPTER_REGISTRY = {} def register_adapter(model_name): def decorator(cls): ADAPTER_REGISTRY[model_name] = cls return cls return decorator @register_adapter("model_a") class ModelAAdapter(BaseModelAdapter): ...

这样新增模型时只需要加一个类，不用改路由层的代码。

4.3 路由决策的代码实现

路由决策的核心函数大概长这样：

async def select_model(request: InternalRequest) -> str: candidates = get_enabled_models(request.task_type) weights = get_routing_weights(request.task_type) scores = {} for model in candidates: if is_circuit_open(model): continue cost = estimate_cost(model, request) latency = get_p95_latency(model) quality = get_quality_score(model, request.task_type) cost_score = 1.0 / (cost + 1e-6) latency_score = 1.0 / (latency + 1e-6) quality_score = quality scores[model] = ( weights.w_cost * normalize(cost_score, scores) + weights.w_latency * normalize(latency_score, scores) + weights.w_quality * quality_score ) return max(scores, key=scores.get)

归一化函数normalize需要等所有候选模型的分数都算出来后再做，所以实际代码里是两轮循环。第一轮算原始分数，第二轮归一化并加权求和。

注意：当所有模型都熔断时，要有一个兜底策略。我的做法是选一个默认模型强制调用，同时打日志告警。这个默认模型通常选最稳定的那个，哪怕贵一点。

4.4 流式响应的延迟测量

流式响应的延迟测量比非流式麻烦。首token延迟是从发起请求到收到第一个chunk的时间，总延迟是从发起请求到收到最后一个chunk的时间。我在适配器里用异步迭代器处理流式响应，每收到一个chunk就记录时间戳。

async def call_stream(self, request): start = time.monotonic() first_token_time = None async for chunk in self._stream_request(request): if first_token_time is None: first_token_time = time.monotonic() yield chunk end = time.monotonic() first_token_latency = (first_token_time - start) * 1000 total_latency = (end - start) * 1000 await self._log_latency(request.request_id, first_token_latency, total_latency)

这里有个坑：如果客户端提前断开连接，异步迭代器会抛出异常，日志就写不进去了。我的处理是在finally块里写日志，保证即使异常也能记录。

5. 常见问题与排查技巧实录

5.1 路由抖动：为什么模型选择总是跳来跳去

这是我最开始遇到的最头疼的问题。表现是同一个任务类型的请求，一会儿走模型A，一会儿走模型B，导致成本和质量都不稳定。

排查下来发现两个原因。一是延迟数据更新太频繁，每来一个请求就更新一次P95，导致分数波动大。二是质量分数初始值相同，但一旦某个模型被选中几次后分数略有变化，就会导致它被更频繁地选中，形成正反馈。

解决办法：延迟数据改成滑动窗口，每30秒更新一次，而不是每次请求都更新。质量分数用指数移动平均，并且设置一个最小更新间隔，比如每100次调用才更新一次。另外给每个模型加一个探索因子，即使分数低也有小概率被选中，避免陷入局部最优。

5.2 缓存穿透：相似度阈值设错了有多可怕

我一开始把语义缓存的相似度阈值设成了0.85，想着命中率高一点。结果上线第一天就出问题了：用户问“如何重置密码”，系统返回了“如何修改密码”的答案。虽然语义相近，但操作步骤完全不同。

后来我把阈值调到0.92，并且加了一个规则：对于操作类、步骤类的问题，缓存命中后还要做一次意图分类校验，只有意图完全一致才返回缓存。这个校验用一个轻量级分类模型做，延迟增加不到10ms，但避免了大部分误命中。

实操心得：语义缓存不是万能的。对于事实类、常识类问题，缓存效果好；对于操作类、时效类问题，要么不缓存，要么加校验。我现在的策略是，只有task_type为“faq”和“knowledge”的请求才走语义缓存，其他类型直接跳过。

5.3 成本统计对不上：token计算的那些坑

成本统计对不上账，是我被财务追问最多的问题。排查后发现几个原因：

第一，有些模型返回的usage字段里，prompt_tokens不包含系统提示词的token。但计费时是包含的。所以我在网关侧统一用tokenizer重新计算，不信任模型返回的usage。

第二，流式响应的completion_tokens在最后一个chunk才返回，如果客户端提前断开，这个数据就丢了。我的处理是在网关侧累计每个chunk的token数，不依赖模型的最终统计。

第三，不同模型的tokenizer不一样，同样的文本token数可能差20%以上。所以成本估算时要用对应模型的tokenizer，不能用统一的。

下面这张表是我整理的常见问题速查：

5.4 熔断与降级的边界条件

熔断逻辑我踩过一个坑：一开始设置的是错误率超过50%就熔断，结果某个模型因为一次网络抖动，连续5个请求失败，错误率瞬间到100%，直接被熔断。但其实第6个请求就恢复了。

后来我改成基于滑动窗口的错误率，窗口大小20个请求，错误率超过60%才熔断，并且熔断后每隔30秒放一个探测请求过去，成功了就恢复。这样既避免了误熔断，又能快速恢复。

降级策略也要提前设计好。我的降级顺序是：先重试同一个模型（最多2次），再切换到备用模型，最后返回兜底话术。兜底话术虽然体验不好，但比直接报错强。

6. 运行效果与后续优化方向

这套网关上线跑了三个月，效果比我预期的好。成本方面，通过路由策略把简单任务分流到低价模型，整体费用下降了约35%。延迟方面，P95延迟从原来的2.8秒降到了1.6秒，主要是因为语义缓存命中率达到了22%，这部分请求直接返回，不消耗模型推理时间。质量方面，通过持续的质量反馈，高价值任务逐渐集中到表现最好的两个模型上，用户点赞率提升了12个百分点。

后续我打算在几个方向继续优化。一是引入更细粒度的质量评估，比如按知识领域分类，而不是笼统地按任务类型。二是把路由策略从静态权重改成基于强化学习的动态调整，让系统自己学习最优的权重组合。三是把网关的能力开放出去，让业务方可以自定义路由规则，比如指定某些请求必须走特定模型。

不过这些都是后话了。当前这套方案已经足够支撑日均百万级的调用量，而且维护成本很低。如果你也在做多模型接入，建议先从可观测性做起，把成本、延迟、质量的数据采集做好，再逐步引入路由和缓存。数据是决策的基础，没有数据，再复杂的路由策略都是拍脑袋。

最后分享一个我在实际运维中总结的小技巧：每周花十分钟看一下路由权重的变化趋势。如果某个模型的权重持续下降，说明它可能出了问题，或者有更优的替代品出现了。这个习惯帮我提前发现了好几次模型性能退化的情况，比等用户投诉再处理主动得多。