拓冰网站定制 
做AI应用最烦的一件事就是各家模型厂商的API长得都不一样。OpenAI的聊天补全格式、Anthropic的消息格式、通义的qwen格式、文心的ERNIE格式再加上各家流式返回的差异前端接一个还好接多个直接能把后端代码写成屎山。我最早是写了一个又一个Provider类硬编码在业务代码里后来接口数量上来了模型也要频繁切换才发现这条路走不通于是动手用C#和YARP做了这套多模型统一接入与路由网关。这个网关解决的核心问题很简单业务代码只认一套OpenAI风格接口后面实际是GPT、Claude、通义还是文心由网关去路由、转换和分发同时把密钥管理、限流、日志、负载均衡这些事一并收编。无论你是自己在做聚合产品还是团队内部想统一管理模型访问这套东西都很值得参考。1. 为什么需要一条AI网关而不是直接调各家API1.1 这个项目的起点和要解决的问题接多个大模型这件事最早看起来不复杂——无非就是多写几个SDK调用对吧但真正跑起来后发现问题集中在四个层面第一是协议不统一。OpenAI的接口格式和Anthropic不一样国内厂商又各有各的兼容姿势有的说兼容OpenAI格式实际又不完全兼容流式返回的结构更是五花八门。第二是密钥管理和安全性很头疼。多个模型的API Key散落在各业务服务里前端如果直连模型API密钥等于裸奔即使后端调用每个服务都要配密钥一旦泄漏或者要轮换就是一场灾难。第三是路由和容灾。同一个模型可能有多个供应商或者你有主备两个通道需要在某个通道故障时自动切换还得考虑不同业务线用不同模型、不同预算配额。第四是观测能力模型调用了多少次、花了多少Token、哪个业务线在烧钱这些如果不统一收口根本没法统计。这三四年里我见过很多团队用Python写AI网关也有直接用云厂商API网关的但作为C#技术栈的团队我们更希望保持技术栈统一于是才有了这个方案——基于YARP构建一个轻量但完整的AI网关。1.2 网关要承担的四个基本职责我梳理下来一个合格的AI网关至少要承担四个职责统一接口屏蔽差异。对外部业务系统只暴露一套稳定的接口协议内部把请求翻译成各家模型的真实格式。这样业务代码永远不会因为换模型而修改只改网关配置就行。路由和分发。根据请求里的模型名、业务线标识或者自定义Header把请求送到指定的上游服务。上游可以是OpenAI官方、Azure OpenAI、国内厂商也可以是内网私有化部署的模型服务。保护和治理。在网关这一层做API Key的统一存储与替换、限流、访问控制、审计日志。比如给每个业务线分配不同的网关Key按Key和模型维度做配额管理。观测和统计。记录每次请求的耗时、状态码、Token消耗估算把数据落到存储里方便后面做成本分摊和异常告警。这四个职责如果散落在业务代码里每个服务都要实现一遍那基本就是无尽的重复劳动而且标准还不统一。收敛到网关这一层是性价比最高的做法。1.3 选型时为什么绕了一圈还是用了YARP做AI网关业界常见的选型有Kong、APISIX这类通用API网关或者直接用云厂商的网关产品再或者用Nginx做七层转发。这些方案各有优势但对我们这种以C#为核心的团队来说有个更自然的选项——微软开源的YARPYet Another Reverse Proxy。YARP是微软官方维护的、完全跑在ASP.NET Core上的反向代理组件优点特别实在第一它是纯C#实现和我们的技术栈无缝衔接中间要加逻辑不需要跨语言调试。Kong和APISIX核心是Lua/Go体系Nginx是C加Lua团队不熟的话维护成本很高。第二YARP的管道本身就是ASP.NET Core中间件管道可以在转发前后插入任意自定义逻辑比如请求改写、响应拦截、限流、鉴权这些对AI网关来说都是必需品。第三YARP的配置是标准IConfiguration体系支持JSON、内存、数据库各种配置源我们后面做成配置驱动很方便。第四YARP支持负载均衡、健康检查、会话亲和性等代理特性做多供应商负载和故障转移时不用自己造轮子。实际上YARP在微软内部就是用来承载Bing等大规模流量的性能不用担心。唯一需要注意的是YARP默认更适合HTTP转发场景AI调用里的SSE流式响应需要在转发时保留原样输出这个在实操中要稍微处理一下后面会专门讲。2. 整体架构与数据流设计2.1 网关的分层结构这套网关我按三层来组织接入层对外暴露统一的REST API兼容OpenAI的/v1/chat/completions、/v1/models、/v1/embeddings等路径格式同时支持流式和非流式两种调用方式。这一层同时也是鉴权和限流的入口。路由与转换层这是网关的核心负责三件事。第一按请求Header或路径参数选择目标模型第二把统一的OpenAI格式请求转换成上游模型真正需要的格式比如Anthropic的messages格式或者通义的input格式第三把上游返回的响应再统一转换成OpenAI格式返回给业务方。上游适配层封装对各家模型供应商HTTP端点的实际调用包括认证、超时、重试、SSE解析等。这一层通过统一的IModelProvider接口抽象每个模型一个实现类。数据流是这样的业务系统发起请求 - 网关鉴权限流 - 路由匹配 - 请求体改写 - YARP转发到上游适配层 - 上游真正调用模型服务 - 响应体统一转换 - 返回业务系统同时异步记录日志。2.2 多模型接入的统一抽象设计要让网关支持多模型第一件事就是定义统一抽象。我设计了一个核心接口所有上游模型适配器都实现它public interface IModelProvider { string ModelName { get; } TaskChatCompletionResponse CompleteAsync(ChatCompletionRequest request, CancellationToken ct); IAsyncEnumerablestring StreamCompleteAsync(ChatCompletionRequest request, CancellationToken ct); }ChatCompletionRequest和ChatCompletionResponse是网关定义的统一传输对象结构上对齐OpenAI格式但字段做了更宽泛的定义比如parameters字典可以携带模型特有的一些参数。每个Provider负责把自己的ModelName映射到上游API业务侧永远只传统一格式。为什么不直接用OpenAI官方SDK的ChatCompletionRequest因为那会把网关死死绑定在OpenAI协议上Anthropic的system字段、通义的parameters这些差异全部要硬塞后面维护会很痛苦。自己定义一个中性对象反而灵活得多。2.3 配置驱动的路由理念路由规则我全部做成配置不写死在代码里。配置用JSON存储格式大致如下{ Routes: [ { RouteName: chat, MatchPath: /v1/chat/completions, DefaultModel: gpt-4o, ModelRouter: { Headers: { X-Model: model }, Params: { model: model } } } ], Models: { gpt-4o: { Provider: OpenAI, Endpoint: https://api.openai.com/v1, ApiKeySecret: secret:openai-key, TimeoutSeconds: 60 }, claude-3-5-sonnet: { Provider: Anthropic, Endpoint: https://api.anthropic.com/v1, ApiKeySecret: secret:anthropic-key }, qwen-max: { Provider: DashScope, Endpoint: https://dashscope.aliyuncs.com/api/v1, ApiKeySecret: secret:dashscope-key } } }路由决策简单直接请求进来先看表单参数或JSON Body里的model字段再看X-ModelHeader都没指定就用DefaultModel。命中ModelRouter后把选定的模型名写入请求上下文后续的转换和转发阶段都能用到。这样做的最大好处是业务方想切换模型时完全不用改代码改配置或者加一个新的Header即可。比如灰度发布一个新模型只需要在Models里增加一个条目然后指定一小部分请求走新模型其余走老模型一天之内就能完成切换。3. 第一步搭建YARP反向代理并跑通OpenAI格式3.1 项目初始化和基础配置创建一个空的ASP.NET Core Web API项目引入Yarp.ReverseProxy包。我用的是.NET 8YARP版本2.x稳定性和性能都相当成熟。启动配置去掉默认模板的Controllers精简成一个纯粹的代理加中间件的管道var builder WebApplication.CreateBuilder(args); builder.Services.AddReverseProxy() .LoadFromConfig(builder.Configuration.GetSection(ReverseProxy)); // 注册网关自己的服务 builder.Services.AddSingletonIModelRegistry, ModelRegistry(); builder.Services.AddSingletonIModelProviderFactory, ModelProviderFactory(); builder.Services.AddSingletonIPromptTransformer, PromptTransformer(); builder.Services.AddSingletonITokenEstimator, TokenEstimator(); var app builder.Build(); app.UseMiddlewareApiKeyAuthMiddleware(); app.UseMiddlewareRateLimitMiddleware(); app.MapReverseProxy(); app.Run();YARP的配置直接挂在ReverseProxy配置节下包括Routes和Clusters两部分。一个Cluster就是一个上游服务集合可以包含多个目的地Destinations也就是同一个模型服务在多个节点上的地址。这样负载均衡的底子就有了。这里有个小坑LoadFromConfig默认监听ReverseProxy节如果你把它放在其他节下需要在LoadFromConfig里传入对应的IConfigurationSection否则启动时YARP找不到任何路由网关静默失败请求全部404。我一开始就栽在这个上面查了半天才发现是配置节路径的问题。3.2 配置模型端点和ClusterCluster的配置大概长这样{ ReverseProxy: { Routes: { chat-route: { ClusterId: openai-cluster, Match: { Path: /v1/chat/completions }, Transforms: [ { PathPattern: v1/chat/completions } ] } }, Clusters: { openai-cluster: { Destinations: { openai-primary: { Address: https://api.openai.com/ }, openai-backup: { Address: https://openai-backup.example.internal/ } }, LoadBalancingPolicy: RoundRobin } } } }YARP路由匹配用的是ASP.NET Core的端点路由语法Match.Path支持通配和参数。转发的目标路径用Transforms里的PathPattern控制。比如我对外暴露的是/v1/chat/completions而上游OpenAI真实路径也是这个那直接原样转发即可。如果上游地址和对外路径不同比如Azure OpenAI路径是/openai/deployments/{deployment}/chat/completions?api-versionxxx就需要用自定义Transformer来做路径拼接和QueryString追加。YARP的Transformer是自定义转发逻辑的官方入口后面再展开。3.3 用Transformer做请求头和密钥改写AI网关一个非常重要的能力是密钥改写。外部请求打到网关时业务方不应该也不需要携带上游模型的Key而是用网关自己签发的Key。网关在转发前根据目标模型从配置中心取出真正的上游API Key替换掉请求Header里的鉴权信息。自定义Transformer继承IRequestTransformer接口即可public class ApiKeyTransform : IRequestTransformer { private readonly IModelRegistry _modelRegistry; public ApiKeyTransform(IModelRegistry modelRegistry) { _modelRegistry modelRegistry; } public async ValueTask TransformRequestAsync(HttpContext context, ProxyRequest proxyRequest, RequestProxyState state, CancellationToken ct) { var targetModel context.Items[TargetModel]?.ToString(); if (string.IsNullOrEmpty(targetModel)) return; var modelConfig await _modelRegistry.GetModelConfigAsync(targetModel); if (modelConfig null) return; // 移除外部传入的可能存在的上游Key防止绕过网关 proxyRequest.Headers.Remove(Authorization); // 根据不同Provider写入不同的鉴权Header if (modelConfig.Provider ProviderType.OpenAI) { proxyRequest.Headers.Authorization new AuthenticationHeaderValue(Bearer, modelConfig.ApiKey); } else if (modelConfig.Provider ProviderType.Anthropic) { proxyRequest.Headers.Add(x-api-key, modelConfig.ApiKey); proxyRequest.Headers.Add(anthropic-version, 2023-06-01); } await ValueTask.CompletedTask; } }然后在YARP路由上挂这个TransformerTransforms: [ { RequestHeadersCopy: true }, { RequestTransformer: ApiKeyTransform } ]注意RequestHeadersCopy要设置为true否则YARP默认会复制原始请求的所有Header如果你在Transformer里移除Authorization就一定要基于复制后的请求修改避免原始Header直接裸奔到上游。密钥本身不要明文写在配置文件里我这次用的是环境变量占位符secret:xxx的形式然后在代码里解析并从密钥管理服务拉取真正的值。即便你的模型Key只放在内网也应该遵循配置不存秘钥这条铁律。4. 核心环节模型路由与负载均衡是怎么实现的4.1 基于Header和Body参数的模型路由实现路由是整个网关的灵魂。我的实现思路是在一开始就用中间件解析请求确定目标模型把结果放进HttpContext.Items后面的Transformer和上游适配层直接消费这个结果。判断模型名有优先级顺序URL路由参数比如/models/{model}/chat这种REST风格最优先其次是X-ModelHeader然后是JSON Body里的model字段最后是路由默认模型。落到代码上大致是public class ModelRouteMiddleware { private readonly RequestDelegate _next; private readonly IModelRegistry _registry; public async Task InvokeAsync(HttpContext context) { var route context.Request.Path.ToString(); var modelName ResolveModelFromRoute(route); if (string.IsNullOrEmpty(modelName)) modelName context.Request.Headers[X-Model].FirstOrDefault(); if (string.IsNullOrEmpty(modelName) context.Request.Method HttpMethod.Post.Method) { // 只尝试读取一次Body并缓存下来避免和后面的流式读取冲突 context.Request.EnableBuffering(); using var reader new StreamReader(context.Request.Body, Encoding.UTF8, leaveOpen: true); var body await reader.ReadToEndAsync(); context.Request.Body.Position 0; using var doc JsonDocument.Parse(body); if (doc.RootElement.TryGetProperty(model, out var modelProp)) modelName modelProp.GetString(); } if (string.IsNullOrEmpty(modelName)) { context.Response.StatusCode 400; await context.Response.WriteAsJsonAsync(new { error model is required }); return; } context.Items[TargetModel] modelName; context.Items[RequestBody] body; // 后续改写直接用不用二次读流 await _next(context); } }这里有个非常重要的细节请求体只能读一次。ASP.NET Core里Request.Body是一个前向流默认读完就没有了。如果中间件这次读了后面的Transformer又要读就会拿到空内容。解决办法是EnableBuffering()加上Body.Position 0复位。我封装那会儿踩过一次后来干脆连Body都缓存到HttpContext.Items里后面所有环节共用。4.2 请求体改写把统一格式转换成各家协议几乎所有模型厂商的接口参数名和对内容的组织方式都不一样所以网关内部默认用OpenAI风格接收请求往外转发前再做一次格式转换。比如Anthropic格式最重要的差异是用户和系统的提示词被拆成了system和messages两个部分其中messages数组里的每条content可以是字符串也可以是结构化的content blocks。OpenAI格式里messages的第一条rolesystem完全可以转换成Anthropic的顶层system字段其余对话消息则原样保留。代码粗略如下public static class AnthropicRequestBuilder { public static string ConvertFromOpenAi(string openAiJson) { using var doc JsonDocument.Parse(openAiJson); var root doc.RootElement; var messages new Listobject(); string system null; if (root.TryGetProperty(messages, out var msgArr)) { foreach (var msg in msgArr.EnumerateArray()) { var role msg.GetProperty(role).GetString(); var content msg.GetProperty(content).ToString(); if (role system) { system content; } else { messages.Add(new { role, content }); } } } var payload new Dictionarystring, object { [model] root.TryGetProperty(model, out var m) ? m.GetString() : null, [messages] messages, [max_tokens] root.TryGetProperty(max_tokens, out var mt) ? mt.GetInt32() : 4096, [stream] root.TryGetProperty(stream, out var st) ? st.GetBoolean() : false, [temperature] root.TryGetProperty(temperature, out var tp) ? tp.GetDouble() : 1.0 }; if (!string.IsNullOrEmpty(system)) payload[system] system; return JsonSerializer.Serialize(payload); } }注意这里我用了Dictionarystring, object而不是强类型DTO是因为各家格式的额外参数差异太大强类型DTO反而每个模型都要定义一套泛型化的字典更灵活。缺点是运行期少了一些编译检查但搭配单元测试把常见格式都覆盖上完全可控。转换层我建议设计为 先解析成中间对象再序列化成目标格式而不是字符串替换。字符串替换看着简单实际上一遇到转义、Unicode、嵌套结构就崩AI模型返回的内容里反斜杠、引号、换行符多得要命正则去处理这种结构本身就是个灾难。4.3 自定义DestinationSelector做负载均衡与故障转移YARP内置了RoundRobin、LeastRequests、PowerOfTwoChoices等负载均衡策略。但我的需求是多模型、多供应商的容灾切换——比如同样一个qwen-max可能同时配了阿里云官网通道和某个内网私有化通道当官网通道连续错误超过阈值把流量自动切到内网通道。这个用YARP默认策略做不到需要自定义IDestinationSelector。public class HealthyDestinationSelector : IDestinationSelector { private readonly IDestinationHealthTracker _healthTracker; public bool TrySelectDestination(ClusterState cluster, ref DestinationState destination) { var healthyDestinations cluster.DestinationsState.AllDestinations .Where(d _healthTracker.IsHealthy(d.DestinationId)) .ToList(); if (healthyDestinations.Count 0) return false; // 简单的轮询选择 destination healthyDestinations[Random.Shared.Next(healthyDestinations.Count)]; return true; } }注册的时候要注意自定义策略的命名要跟配置里严格一致builder.Services.AddSingletonIDestinationSelector, HealthyDestinationSelector(); builder.Services.AddReverseProxy() .LoadFromConfig(builder.Configuration.GetSection(ReverseProxy)); // 然后在配置的 Cluster 上指定 // LoadBalancingPolicy: HealthyDestinationSelector健康检查的状态从哪里来我调研了下YARP有不少自带的检测机制比如IMultipleDestinationHealthCheckService可以做主动轮询健康检查还有被动探测也就是根据请求的响应状态码动态判定失败。对于AI场景我建议用被动探测为主因为主动健康检查会额外消耗Token没事儿去打一个Completion接口太浪费了。一个更实际的办法是每隔一段时间给模型服务发一个轻量的GET /models请求做探测既便宜又能感知服务是否在线。被动探测的实现不难给YARP加一个IForwarderErrorHandler或者干脆在响应中间件里监听状态码当5xx或超时计数超过阈值就认为该目标不健康。这里要小心误伤——模型接口偶发429限流不应该直接判死要区分429和5xx的阈值。4.4 流式SSE转发的处理细节构建AI网关最容易被坑的就是SSE流式响应。业务侧绝大多数AI应用都是流式打字机效果网关作为中间层如果处理不当常见的症状是前端一个字一个字接收实际上网关攒完了整个响应才开始吐这等于把流式交互彻底毁了。YARP在底层其实支持响应流透传它本身就是高性能代理流式转发没问题。真正的坑在两点第一是不要在你自己的代码里对HttpResponse.Body做缓存或Buffering。很多人为了做响应格式统一习惯读出全部内容再写入这在普通API没问题但在SSE场景下就是灾难。流式响应必须边收边转不能等全部。我的做法是用StreamReader在ReadLineAsync循环里逐行处理data: ...块解析出增量内容按OpenAI的SSE格式重新封装后直接写入context.Response.Body用完立刻FlushAsync。第二是响应缓冲中间件比如ResponseCompression或OutputCache在SSE场景下可能造成整段缓冲。所以我特别强调在YARP管道的SSE请求路径上关掉压缩和缓存否则要么前端拿到gzip后解不开要么响应被攒包后延迟到达。流式的统一转换要区分两种情况上游OpenAI格式基本上不用改原样转发上游是Anthropic格式时它的SSE事件格式和OpenAI不同OpenAI是data: {choices: [{delta: {content: ...}}]}Anthropic是data: {type: content_block_delta, delta: {text: ...}}所以需要写一个转换器public async IAsyncEnumerablestring ConvertAnthropicSseToOpenAi( IAsyncEnumerablestring anthropicLines, CancellationToken ct) { await foreach (var line in anthropicLines) { if (!line.StartsWith(data:)) continue; var json line.Substring(5).Trim(); if (json [DONE]) { yield return data: [DONE]\n\n; break; } using var doc JsonDocument.Parse(json); if (doc.RootElement.GetProperty(type).GetString() content_block_delta) { var text doc.RootElement.GetProperty(delta).GetProperty(text).GetString(); var openAiEvent new { choices new[] { new { delta new { content text }, index 0 } } }; yield return $data: {JsonSerializer.Serialize(openAiEvent)}\n\n; } } }这套方案跑下来普通调用和流式调用的兼容性都很稳定。业务侧统一收OpenAI格式前端的openai-node、LangChain.js甚至直接fetch都能无缝接入。5. 网关的附加能力鉴权、限流、日志与用量统计5.1 API Key管理与鉴权中间件网关自己的Key体系我按照业务线 配额两个维度来设计。每个业务线分配一组Key比如ak_live_pv3k9f是生产环境的Keyak_dev_8v2ksd是测试环境的Key。Key的存储是一张表本地先用SQLite字段包括Key、业务线名、状态、可用模型列表、日限额、月限额。鉴权中间件的逻辑非常简单先从Authorization: Bearer xxx中取出Key查库确认Key存在且状态为启用把Key对应的业务线信息写入HttpContext.Items[ClientId]校验这个业务线是否允许访问目标模型路由中间件已经确定了TargetModel最后进入限流环节。需要特别注意的是用Hash存储Key的明文还是只存Hash。我的做法是数据库存Hash网关实例内存里缓存一份Key到业务线的映射避免每次请求都查库。Key本身发给业务方时只展示一次丢失就重新生成。这种设计解决的是多团队共用模型时造成的Key滥用问题。以前开放一个OpenAI Key出去一个研发拿走全公司都用出了问题也不知道谁在调成本失控。现在每个业务线一个Key谁调用了多少一目了然不给某条业务线开新模型它也调不动。5.2 限流固定窗口还是令牌桶限流这块又要多模型又要每业务线配额还要有突发容忍度。我设计的规则是三层网关整体维度每秒最多N个请求单Key维度每秒钟最多M个请求单Key单模型维度每日Token消耗上限。实现层面框架自带的固定窗口限流足够简单但尖刺流量一来容易瞬间打死下游。我最后用了令牌桶桶容量对应突发请求数填充速率对应平滑速率。基于内存的实现很简单几百行代码的事没必要引入Redis除非你要多实例部署共享计数。如果上K8s多副本部署内存令牌桶就不准了你得把计数丢到Redis里用Lua脚本做原子扣减。Token估算方面很多人以为是模型返回里带的usage字段直接拿来用就行。但网关在鉴权限流时请求还没转发出去哪来的真实用量所以要自己做估算按字符数估算一个大概值精确实时数据等上游返回后再修正。中英文混合的估算公式大概是这样中文按1.5个Token/字英文按0.25个Token/字符整个再乘一个1.1的经验系数。虽然不精确但做预扣足够用。5.3 请求日志与Token用量统计每次模型调用都是钱日志如果没有结构化记录月底对账的时候就会很难受。我做的请求日志中间件在每个请求结束后把以下信息异步写入日志请求ID、业务线、目标模型、上游供应商、请求Token估算、状态码、耗时、是否流式调用、错误信息。日志用Serilog输出结构化字段直接推给ElasticSearch或ClickHouse然后再在后台做聚合报表。特别强调异步写入三个字不要在请求主链路上同步写数据库不然高并发下数据库IO会直接拖垮网关延迟。我的方案是用Channel做生产者消费者队列日志写入全异步请求主链路零阻塞。5.4 用SQLite做本地用量统计存储轻量场景下SQLite其实是个被低估的选择。单个文件、零部署、读写性能可接受做单机网关的统计存储绰绰有余。我用EF Core连SQLite建表如下public class UsageRecord { public long Id { get; set; } public string ClientId { get; set; } public string Model { get; set; } public string Provider { get; set; } public int EstimatedTokens { get; set; } public int? ActualTokens { get; set; } public int StatusCode { get; set; } public long DurationMs { get; set; } public DateTime CreatedAt { get; set; } }每条请求写一条记录后台再按ClientId Model聚合出日用量和趋势。用SQLite时要注意默认连接不支持并发写多线程写入会报database is locked所以我用一个独立的Channel消费者线程集中写入或者启用WAL模式这两个都能解决问题。6. 实操中遇到的坑与排查实录6.1 流式响应被网关攒包了最早实现SSE转发时我图省事直接在中间件里把HttpResponse.Body换成了一个MemoryStream打算拿到全部内容之后再统一写出去。结果前端流式效果全废答一句完整的话要等十几秒体验极差。排查思路先用curl直接打上游模型API确认上游SSE正常再跳过网关直连接口也正常最后定位到是我中间件把响应体缓存了。修正方式只对非SSE请求做Body缓冲SSE请求直接往原始响应流写写完立即FlushAsync。另外一个隐藏问题如果有Response.Headers里设置了Content-Length在流式模式下会被框架拒绝所以SSE响应不要设置长度。6.2 偶发超时上游读取超时设置太短有段时间生产环境老报超时查看日志发现集中在模型高峰期响应时间偶尔飙到30秒以上。原因是模型推理本身就是慢操作尤其是大模型在峰值时排队动辄几十秒才返回第一个token而我把HttpClient的超时设成了15秒导致频繁提前断开。修正思路区分三段时间。连接超时设置5秒读取响应头超时设置30秒读取内容流超时不设上限或者设置到300秒。原因是大模型流式响应只要第一个字节到了说明服务端已经在生成多等等没关系。这个经验在接国内一些厂商的API时尤其重要它们的排队时长可能比OpenAI还长。6.3 JSON转义把提示词弄坏了在写PromptTransformer时我为了省事把整个请求体用JsonSerializer.Serialize序列化后直接作为字符串嵌入到目标格式里结果发现提示词里的引号、换行被双重转义模型看到了一大堆\\n而不是真正的换行符。这个问题本质上是对JSON序列化的机制理解不透彻。把序列化后的结果当字符串用等于做了两次序列化。正确做法是始终操作JsonNode或JsonDocument对象树通过节点赋值来构造目标结构最后再做一次整体序列化。既避免了转义问题性能还更好。6.4 并发连接数冲击默认HttpClient限死了性能ASP.NET Core默认的HttpClient如果不显式配置线程池会为每个转发请求建立一个连接高并发下Socket耗尽表现为大量No such device or address和超时的混合错误。YARP虽然自带连接管理但如果自定义Provider里自己new了HttpClient这个坑就跑不掉。正确做法所有自定义调用一律注入IHttpClientFactory创建的客户端同时开启连接复用并且区分不同上游服务的连接池。还要注意HTTP/2的支持——OpenAI和Anthropic现在都支持HTTP/2打开之后TLS握手开销小很多在高并发下能省掉一大截连接建立时间。6.5 上游返回错误时错误信息原样抛给业务方早期设计里上游模型API返回4xx/5xx时网关直接把上游错误透传给业务方。比如供应商那边的限流提示是英文的、字段名也和OpenAI不同业务方收到后没法处理。后来我在网关加了一层错误规范化上游错误先解析提取错误状态码、错误类型、错误信息然后重新封装成OpenAI风格的错误结构返回。比如上游返回429网关会返回{error: {message: rate limit exceeded, retry after 12s, type: rate_limit_error}}。这样业务方无论是做重试还是告警都有统一的规范和字段不用为每一家厂商做一遍兼容。7. 这个网关后续还能怎么扩展目前这套网关整体跑得很稳定但它绝不是终点。经过这段实践我觉得在AI基础设施这条路上网关的价值会越来越大而且发展方向也很明确模型自动降级是下一步优先要做的方向。比如配置一个主模型gpt-4o当它持续限流或报错时自动把流量切到备用的qwen-max或claude-3-5-sonnet并返回一个可追踪的头信息让业务方知道这次实际用的是哪个模型。这个我做了一半后续整完再分享。多模态与多协议扩展。现在网关统一的是文本聊天格式但图片生成、Embedding、语音转文字都在快速普及每种能力的协议都不太一样。把这个网关扩展成一切AI能力的中枢那覆盖面就又上一个台阶。计费与配额的商品化。当前是按业务线做配额下一步希望能按项目、按功能模块甚至按最终用户做多层配额嵌套方便做内部结算和成本归因。这需要引入一套树形结构的预算模型技术上比现在复杂不少但收益也大。我在实际开发里的一个体会是做这类中间层基础设施最容易出错的地方其实不是技术选型而是需求的边界。AI模型更新迭代太快了今天接五家三个月后又多三家网关的抽象层如果做得太死每次接新模型都要动核心代码那就又回到了当年屎山代码的老路。所以设计的时候一定要把新增一个模型Provider做成只加一个类、改一段配置的事这也正是C#的接口、委托和泛型最拿手的场景。最后再分享一个压箱底的小技巧网关上线前一定先写一套完整的、基于真实上游API的集成测试覆盖非流式、流式、上游4xx、上游5xx、超时、限流命中这些场景。不要只做Mock测试因为Mock永远模拟不出真实供应商API的诡异行为。跑通这套测试以后后面每次加模型心里都有底改代码也不会慌。构建AI网关这件事越早做后面越省心。 基于YARP构建多模型统一接入与路由网关的实践
做AI应用最烦的一件事就是各家模型厂商的API长得都不一样。OpenAI的聊天补全格式、Anthropic的消息格式、通义的qwen格式、文心的ERNIE格式再加上各家流式返回的差异前端接一个还好接多个直接能把后端代码写成屎山。我最早是写了一个又一个Provider类硬编码在业务代码里后来接口数量上来了模型也要频繁切换才发现这条路走不通于是动手用C#和YARP做了这套多模型统一接入与路由网关。这个网关解决的核心问题很简单业务代码只认一套OpenAI风格接口后面实际是GPT、Claude、通义还是文心由网关去路由、转换和分发同时把密钥管理、限流、日志、负载均衡这些事一并收编。无论你是自己在做聚合产品还是团队内部想统一管理模型访问这套东西都很值得参考。1. 为什么需要一条AI网关而不是直接调各家API1.1 这个项目的起点和要解决的问题接多个大模型这件事最早看起来不复杂——无非就是多写几个SDK调用对吧但真正跑起来后发现问题集中在四个层面第一是协议不统一。OpenAI的接口格式和Anthropic不一样国内厂商又各有各的兼容姿势有的说兼容OpenAI格式实际又不完全兼容流式返回的结构更是五花八门。第二是密钥管理和安全性很头疼。多个模型的API Key散落在各业务服务里前端如果直连模型API密钥等于裸奔即使后端调用每个服务都要配密钥一旦泄漏或者要轮换就是一场灾难。第三是路由和容灾。同一个模型可能有多个供应商或者你有主备两个通道需要在某个通道故障时自动切换还得考虑不同业务线用不同模型、不同预算配额。第四是观测能力模型调用了多少次、花了多少Token、哪个业务线在烧钱这些如果不统一收口根本没法统计。这三四年里我见过很多团队用Python写AI网关也有直接用云厂商API网关的但作为C#技术栈的团队我们更希望保持技术栈统一于是才有了这个方案——基于YARP构建一个轻量但完整的AI网关。1.2 网关要承担的四个基本职责我梳理下来一个合格的AI网关至少要承担四个职责统一接口屏蔽差异。对外部业务系统只暴露一套稳定的接口协议内部把请求翻译成各家模型的真实格式。这样业务代码永远不会因为换模型而修改只改网关配置就行。路由和分发。根据请求里的模型名、业务线标识或者自定义Header把请求送到指定的上游服务。上游可以是OpenAI官方、Azure OpenAI、国内厂商也可以是内网私有化部署的模型服务。保护和治理。在网关这一层做API Key的统一存储与替换、限流、访问控制、审计日志。比如给每个业务线分配不同的网关Key按Key和模型维度做配额管理。观测和统计。记录每次请求的耗时、状态码、Token消耗估算把数据落到存储里方便后面做成本分摊和异常告警。这四个职责如果散落在业务代码里每个服务都要实现一遍那基本就是无尽的重复劳动而且标准还不统一。收敛到网关这一层是性价比最高的做法。1.3 选型时为什么绕了一圈还是用了YARP做AI网关业界常见的选型有Kong、APISIX这类通用API网关或者直接用云厂商的网关产品再或者用Nginx做七层转发。这些方案各有优势但对我们这种以C#为核心的团队来说有个更自然的选项——微软开源的YARPYet Another Reverse Proxy。YARP是微软官方维护的、完全跑在ASP.NET Core上的反向代理组件优点特别实在第一它是纯C#实现和我们的技术栈无缝衔接中间要加逻辑不需要跨语言调试。Kong和APISIX核心是Lua/Go体系Nginx是C加Lua团队不熟的话维护成本很高。第二YARP的管道本身就是ASP.NET Core中间件管道可以在转发前后插入任意自定义逻辑比如请求改写、响应拦截、限流、鉴权这些对AI网关来说都是必需品。第三YARP的配置是标准IConfiguration体系支持JSON、内存、数据库各种配置源我们后面做成配置驱动很方便。第四YARP支持负载均衡、健康检查、会话亲和性等代理特性做多供应商负载和故障转移时不用自己造轮子。实际上YARP在微软内部就是用来承载Bing等大规模流量的性能不用担心。唯一需要注意的是YARP默认更适合HTTP转发场景AI调用里的SSE流式响应需要在转发时保留原样输出这个在实操中要稍微处理一下后面会专门讲。2. 整体架构与数据流设计2.1 网关的分层结构这套网关我按三层来组织接入层对外暴露统一的REST API兼容OpenAI的/v1/chat/completions、/v1/models、/v1/embeddings等路径格式同时支持流式和非流式两种调用方式。这一层同时也是鉴权和限流的入口。路由与转换层这是网关的核心负责三件事。第一按请求Header或路径参数选择目标模型第二把统一的OpenAI格式请求转换成上游模型真正需要的格式比如Anthropic的messages格式或者通义的input格式第三把上游返回的响应再统一转换成OpenAI格式返回给业务方。上游适配层封装对各家模型供应商HTTP端点的实际调用包括认证、超时、重试、SSE解析等。这一层通过统一的IModelProvider接口抽象每个模型一个实现类。数据流是这样的业务系统发起请求 - 网关鉴权限流 - 路由匹配 - 请求体改写 - YARP转发到上游适配层 - 上游真正调用模型服务 - 响应体统一转换 - 返回业务系统同时异步记录日志。2.2 多模型接入的统一抽象设计要让网关支持多模型第一件事就是定义统一抽象。我设计了一个核心接口所有上游模型适配器都实现它public interface IModelProvider { string ModelName { get; } TaskChatCompletionResponse CompleteAsync(ChatCompletionRequest request, CancellationToken ct); IAsyncEnumerablestring StreamCompleteAsync(ChatCompletionRequest request, CancellationToken ct); }ChatCompletionRequest和ChatCompletionResponse是网关定义的统一传输对象结构上对齐OpenAI格式但字段做了更宽泛的定义比如parameters字典可以携带模型特有的一些参数。每个Provider负责把自己的ModelName映射到上游API业务侧永远只传统一格式。为什么不直接用OpenAI官方SDK的ChatCompletionRequest因为那会把网关死死绑定在OpenAI协议上Anthropic的system字段、通义的parameters这些差异全部要硬塞后面维护会很痛苦。自己定义一个中性对象反而灵活得多。2.3 配置驱动的路由理念路由规则我全部做成配置不写死在代码里。配置用JSON存储格式大致如下{ Routes: [ { RouteName: chat, MatchPath: /v1/chat/completions, DefaultModel: gpt-4o, ModelRouter: { Headers: { X-Model: model }, Params: { model: model } } } ], Models: { gpt-4o: { Provider: OpenAI, Endpoint: https://api.openai.com/v1, ApiKeySecret: secret:openai-key, TimeoutSeconds: 60 }, claude-3-5-sonnet: { Provider: Anthropic, Endpoint: https://api.anthropic.com/v1, ApiKeySecret: secret:anthropic-key }, qwen-max: { Provider: DashScope, Endpoint: https://dashscope.aliyuncs.com/api/v1, ApiKeySecret: secret:dashscope-key } } }路由决策简单直接请求进来先看表单参数或JSON Body里的model字段再看X-ModelHeader都没指定就用DefaultModel。命中ModelRouter后把选定的模型名写入请求上下文后续的转换和转发阶段都能用到。这样做的最大好处是业务方想切换模型时完全不用改代码改配置或者加一个新的Header即可。比如灰度发布一个新模型只需要在Models里增加一个条目然后指定一小部分请求走新模型其余走老模型一天之内就能完成切换。3. 第一步搭建YARP反向代理并跑通OpenAI格式3.1 项目初始化和基础配置创建一个空的ASP.NET Core Web API项目引入Yarp.ReverseProxy包。我用的是.NET 8YARP版本2.x稳定性和性能都相当成熟。启动配置去掉默认模板的Controllers精简成一个纯粹的代理加中间件的管道var builder WebApplication.CreateBuilder(args); builder.Services.AddReverseProxy() .LoadFromConfig(builder.Configuration.GetSection(ReverseProxy)); // 注册网关自己的服务 builder.Services.AddSingletonIModelRegistry, ModelRegistry(); builder.Services.AddSingletonIModelProviderFactory, ModelProviderFactory(); builder.Services.AddSingletonIPromptTransformer, PromptTransformer(); builder.Services.AddSingletonITokenEstimator, TokenEstimator(); var app builder.Build(); app.UseMiddlewareApiKeyAuthMiddleware(); app.UseMiddlewareRateLimitMiddleware(); app.MapReverseProxy(); app.Run();YARP的配置直接挂在ReverseProxy配置节下包括Routes和Clusters两部分。一个Cluster就是一个上游服务集合可以包含多个目的地Destinations也就是同一个模型服务在多个节点上的地址。这样负载均衡的底子就有了。这里有个小坑LoadFromConfig默认监听ReverseProxy节如果你把它放在其他节下需要在LoadFromConfig里传入对应的IConfigurationSection否则启动时YARP找不到任何路由网关静默失败请求全部404。我一开始就栽在这个上面查了半天才发现是配置节路径的问题。3.2 配置模型端点和ClusterCluster的配置大概长这样{ ReverseProxy: { Routes: { chat-route: { ClusterId: openai-cluster, Match: { Path: /v1/chat/completions }, Transforms: [ { PathPattern: v1/chat/completions } ] } }, Clusters: { openai-cluster: { Destinations: { openai-primary: { Address: https://api.openai.com/ }, openai-backup: { Address: https://openai-backup.example.internal/ } }, LoadBalancingPolicy: RoundRobin } } } }YARP路由匹配用的是ASP.NET Core的端点路由语法Match.Path支持通配和参数。转发的目标路径用Transforms里的PathPattern控制。比如我对外暴露的是/v1/chat/completions而上游OpenAI真实路径也是这个那直接原样转发即可。如果上游地址和对外路径不同比如Azure OpenAI路径是/openai/deployments/{deployment}/chat/completions?api-versionxxx就需要用自定义Transformer来做路径拼接和QueryString追加。YARP的Transformer是自定义转发逻辑的官方入口后面再展开。3.3 用Transformer做请求头和密钥改写AI网关一个非常重要的能力是密钥改写。外部请求打到网关时业务方不应该也不需要携带上游模型的Key而是用网关自己签发的Key。网关在转发前根据目标模型从配置中心取出真正的上游API Key替换掉请求Header里的鉴权信息。自定义Transformer继承IRequestTransformer接口即可public class ApiKeyTransform : IRequestTransformer { private readonly IModelRegistry _modelRegistry; public ApiKeyTransform(IModelRegistry modelRegistry) { _modelRegistry modelRegistry; } public async ValueTask TransformRequestAsync(HttpContext context, ProxyRequest proxyRequest, RequestProxyState state, CancellationToken ct) { var targetModel context.Items[TargetModel]?.ToString(); if (string.IsNullOrEmpty(targetModel)) return; var modelConfig await _modelRegistry.GetModelConfigAsync(targetModel); if (modelConfig null) return; // 移除外部传入的可能存在的上游Key防止绕过网关 proxyRequest.Headers.Remove(Authorization); // 根据不同Provider写入不同的鉴权Header if (modelConfig.Provider ProviderType.OpenAI) { proxyRequest.Headers.Authorization new AuthenticationHeaderValue(Bearer, modelConfig.ApiKey); } else if (modelConfig.Provider ProviderType.Anthropic) { proxyRequest.Headers.Add(x-api-key, modelConfig.ApiKey); proxyRequest.Headers.Add(anthropic-version, 2023-06-01); } await ValueTask.CompletedTask; } }然后在YARP路由上挂这个TransformerTransforms: [ { RequestHeadersCopy: true }, { RequestTransformer: ApiKeyTransform } ]注意RequestHeadersCopy要设置为true否则YARP默认会复制原始请求的所有Header如果你在Transformer里移除Authorization就一定要基于复制后的请求修改避免原始Header直接裸奔到上游。密钥本身不要明文写在配置文件里我这次用的是环境变量占位符secret:xxx的形式然后在代码里解析并从密钥管理服务拉取真正的值。即便你的模型Key只放在内网也应该遵循配置不存秘钥这条铁律。4. 核心环节模型路由与负载均衡是怎么实现的4.1 基于Header和Body参数的模型路由实现路由是整个网关的灵魂。我的实现思路是在一开始就用中间件解析请求确定目标模型把结果放进HttpContext.Items后面的Transformer和上游适配层直接消费这个结果。判断模型名有优先级顺序URL路由参数比如/models/{model}/chat这种REST风格最优先其次是X-ModelHeader然后是JSON Body里的model字段最后是路由默认模型。落到代码上大致是public class ModelRouteMiddleware { private readonly RequestDelegate _next; private readonly IModelRegistry _registry; public async Task InvokeAsync(HttpContext context) { var route context.Request.Path.ToString(); var modelName ResolveModelFromRoute(route); if (string.IsNullOrEmpty(modelName)) modelName context.Request.Headers[X-Model].FirstOrDefault(); if (string.IsNullOrEmpty(modelName) context.Request.Method HttpMethod.Post.Method) { // 只尝试读取一次Body并缓存下来避免和后面的流式读取冲突 context.Request.EnableBuffering(); using var reader new StreamReader(context.Request.Body, Encoding.UTF8, leaveOpen: true); var body await reader.ReadToEndAsync(); context.Request.Body.Position 0; using var doc JsonDocument.Parse(body); if (doc.RootElement.TryGetProperty(model, out var modelProp)) modelName modelProp.GetString(); } if (string.IsNullOrEmpty(modelName)) { context.Response.StatusCode 400; await context.Response.WriteAsJsonAsync(new { error model is required }); return; } context.Items[TargetModel] modelName; context.Items[RequestBody] body; // 后续改写直接用不用二次读流 await _next(context); } }这里有个非常重要的细节请求体只能读一次。ASP.NET Core里Request.Body是一个前向流默认读完就没有了。如果中间件这次读了后面的Transformer又要读就会拿到空内容。解决办法是EnableBuffering()加上Body.Position 0复位。我封装那会儿踩过一次后来干脆连Body都缓存到HttpContext.Items里后面所有环节共用。4.2 请求体改写把统一格式转换成各家协议几乎所有模型厂商的接口参数名和对内容的组织方式都不一样所以网关内部默认用OpenAI风格接收请求往外转发前再做一次格式转换。比如Anthropic格式最重要的差异是用户和系统的提示词被拆成了system和messages两个部分其中messages数组里的每条content可以是字符串也可以是结构化的content blocks。OpenAI格式里messages的第一条rolesystem完全可以转换成Anthropic的顶层system字段其余对话消息则原样保留。代码粗略如下public static class AnthropicRequestBuilder { public static string ConvertFromOpenAi(string openAiJson) { using var doc JsonDocument.Parse(openAiJson); var root doc.RootElement; var messages new Listobject(); string system null; if (root.TryGetProperty(messages, out var msgArr)) { foreach (var msg in msgArr.EnumerateArray()) { var role msg.GetProperty(role).GetString(); var content msg.GetProperty(content).ToString(); if (role system) { system content; } else { messages.Add(new { role, content }); } } } var payload new Dictionarystring, object { [model] root.TryGetProperty(model, out var m) ? m.GetString() : null, [messages] messages, [max_tokens] root.TryGetProperty(max_tokens, out var mt) ? mt.GetInt32() : 4096, [stream] root.TryGetProperty(stream, out var st) ? st.GetBoolean() : false, [temperature] root.TryGetProperty(temperature, out var tp) ? tp.GetDouble() : 1.0 }; if (!string.IsNullOrEmpty(system)) payload[system] system; return JsonSerializer.Serialize(payload); } }注意这里我用了Dictionarystring, object而不是强类型DTO是因为各家格式的额外参数差异太大强类型DTO反而每个模型都要定义一套泛型化的字典更灵活。缺点是运行期少了一些编译检查但搭配单元测试把常见格式都覆盖上完全可控。转换层我建议设计为 先解析成中间对象再序列化成目标格式而不是字符串替换。字符串替换看着简单实际上一遇到转义、Unicode、嵌套结构就崩AI模型返回的内容里反斜杠、引号、换行符多得要命正则去处理这种结构本身就是个灾难。4.3 自定义DestinationSelector做负载均衡与故障转移YARP内置了RoundRobin、LeastRequests、PowerOfTwoChoices等负载均衡策略。但我的需求是多模型、多供应商的容灾切换——比如同样一个qwen-max可能同时配了阿里云官网通道和某个内网私有化通道当官网通道连续错误超过阈值把流量自动切到内网通道。这个用YARP默认策略做不到需要自定义IDestinationSelector。public class HealthyDestinationSelector : IDestinationSelector { private readonly IDestinationHealthTracker _healthTracker; public bool TrySelectDestination(ClusterState cluster, ref DestinationState destination) { var healthyDestinations cluster.DestinationsState.AllDestinations .Where(d _healthTracker.IsHealthy(d.DestinationId)) .ToList(); if (healthyDestinations.Count 0) return false; // 简单的轮询选择 destination healthyDestinations[Random.Shared.Next(healthyDestinations.Count)]; return true; } }注册的时候要注意自定义策略的命名要跟配置里严格一致builder.Services.AddSingletonIDestinationSelector, HealthyDestinationSelector(); builder.Services.AddReverseProxy() .LoadFromConfig(builder.Configuration.GetSection(ReverseProxy)); // 然后在配置的 Cluster 上指定 // LoadBalancingPolicy: HealthyDestinationSelector健康检查的状态从哪里来我调研了下YARP有不少自带的检测机制比如IMultipleDestinationHealthCheckService可以做主动轮询健康检查还有被动探测也就是根据请求的响应状态码动态判定失败。对于AI场景我建议用被动探测为主因为主动健康检查会额外消耗Token没事儿去打一个Completion接口太浪费了。一个更实际的办法是每隔一段时间给模型服务发一个轻量的GET /models请求做探测既便宜又能感知服务是否在线。被动探测的实现不难给YARP加一个IForwarderErrorHandler或者干脆在响应中间件里监听状态码当5xx或超时计数超过阈值就认为该目标不健康。这里要小心误伤——模型接口偶发429限流不应该直接判死要区分429和5xx的阈值。4.4 流式SSE转发的处理细节构建AI网关最容易被坑的就是SSE流式响应。业务侧绝大多数AI应用都是流式打字机效果网关作为中间层如果处理不当常见的症状是前端一个字一个字接收实际上网关攒完了整个响应才开始吐这等于把流式交互彻底毁了。YARP在底层其实支持响应流透传它本身就是高性能代理流式转发没问题。真正的坑在两点第一是不要在你自己的代码里对HttpResponse.Body做缓存或Buffering。很多人为了做响应格式统一习惯读出全部内容再写入这在普通API没问题但在SSE场景下就是灾难。流式响应必须边收边转不能等全部。我的做法是用StreamReader在ReadLineAsync循环里逐行处理data: ...块解析出增量内容按OpenAI的SSE格式重新封装后直接写入context.Response.Body用完立刻FlushAsync。第二是响应缓冲中间件比如ResponseCompression或OutputCache在SSE场景下可能造成整段缓冲。所以我特别强调在YARP管道的SSE请求路径上关掉压缩和缓存否则要么前端拿到gzip后解不开要么响应被攒包后延迟到达。流式的统一转换要区分两种情况上游OpenAI格式基本上不用改原样转发上游是Anthropic格式时它的SSE事件格式和OpenAI不同OpenAI是data: {choices: [{delta: {content: ...}}]}Anthropic是data: {type: content_block_delta, delta: {text: ...}}所以需要写一个转换器public async IAsyncEnumerablestring ConvertAnthropicSseToOpenAi( IAsyncEnumerablestring anthropicLines, CancellationToken ct) { await foreach (var line in anthropicLines) { if (!line.StartsWith(data:)) continue; var json line.Substring(5).Trim(); if (json [DONE]) { yield return data: [DONE]\n\n; break; } using var doc JsonDocument.Parse(json); if (doc.RootElement.GetProperty(type).GetString() content_block_delta) { var text doc.RootElement.GetProperty(delta).GetProperty(text).GetString(); var openAiEvent new { choices new[] { new { delta new { content text }, index 0 } } }; yield return $data: {JsonSerializer.Serialize(openAiEvent)}\n\n; } } }这套方案跑下来普通调用和流式调用的兼容性都很稳定。业务侧统一收OpenAI格式前端的openai-node、LangChain.js甚至直接fetch都能无缝接入。5. 网关的附加能力鉴权、限流、日志与用量统计5.1 API Key管理与鉴权中间件网关自己的Key体系我按照业务线 配额两个维度来设计。每个业务线分配一组Key比如ak_live_pv3k9f是生产环境的Keyak_dev_8v2ksd是测试环境的Key。Key的存储是一张表本地先用SQLite字段包括Key、业务线名、状态、可用模型列表、日限额、月限额。鉴权中间件的逻辑非常简单先从Authorization: Bearer xxx中取出Key查库确认Key存在且状态为启用把Key对应的业务线信息写入HttpContext.Items[ClientId]校验这个业务线是否允许访问目标模型路由中间件已经确定了TargetModel最后进入限流环节。需要特别注意的是用Hash存储Key的明文还是只存Hash。我的做法是数据库存Hash网关实例内存里缓存一份Key到业务线的映射避免每次请求都查库。Key本身发给业务方时只展示一次丢失就重新生成。这种设计解决的是多团队共用模型时造成的Key滥用问题。以前开放一个OpenAI Key出去一个研发拿走全公司都用出了问题也不知道谁在调成本失控。现在每个业务线一个Key谁调用了多少一目了然不给某条业务线开新模型它也调不动。5.2 限流固定窗口还是令牌桶限流这块又要多模型又要每业务线配额还要有突发容忍度。我设计的规则是三层网关整体维度每秒最多N个请求单Key维度每秒钟最多M个请求单Key单模型维度每日Token消耗上限。实现层面框架自带的固定窗口限流足够简单但尖刺流量一来容易瞬间打死下游。我最后用了令牌桶桶容量对应突发请求数填充速率对应平滑速率。基于内存的实现很简单几百行代码的事没必要引入Redis除非你要多实例部署共享计数。如果上K8s多副本部署内存令牌桶就不准了你得把计数丢到Redis里用Lua脚本做原子扣减。Token估算方面很多人以为是模型返回里带的usage字段直接拿来用就行。但网关在鉴权限流时请求还没转发出去哪来的真实用量所以要自己做估算按字符数估算一个大概值精确实时数据等上游返回后再修正。中英文混合的估算公式大概是这样中文按1.5个Token/字英文按0.25个Token/字符整个再乘一个1.1的经验系数。虽然不精确但做预扣足够用。5.3 请求日志与Token用量统计每次模型调用都是钱日志如果没有结构化记录月底对账的时候就会很难受。我做的请求日志中间件在每个请求结束后把以下信息异步写入日志请求ID、业务线、目标模型、上游供应商、请求Token估算、状态码、耗时、是否流式调用、错误信息。日志用Serilog输出结构化字段直接推给ElasticSearch或ClickHouse然后再在后台做聚合报表。特别强调异步写入三个字不要在请求主链路上同步写数据库不然高并发下数据库IO会直接拖垮网关延迟。我的方案是用Channel做生产者消费者队列日志写入全异步请求主链路零阻塞。5.4 用SQLite做本地用量统计存储轻量场景下SQLite其实是个被低估的选择。单个文件、零部署、读写性能可接受做单机网关的统计存储绰绰有余。我用EF Core连SQLite建表如下public class UsageRecord { public long Id { get; set; } public string ClientId { get; set; } public string Model { get; set; } public string Provider { get; set; } public int EstimatedTokens { get; set; } public int? ActualTokens { get; set; } public int StatusCode { get; set; } public long DurationMs { get; set; } public DateTime CreatedAt { get; set; } }每条请求写一条记录后台再按ClientId Model聚合出日用量和趋势。用SQLite时要注意默认连接不支持并发写多线程写入会报database is locked所以我用一个独立的Channel消费者线程集中写入或者启用WAL模式这两个都能解决问题。6. 实操中遇到的坑与排查实录6.1 流式响应被网关攒包了最早实现SSE转发时我图省事直接在中间件里把HttpResponse.Body换成了一个MemoryStream打算拿到全部内容之后再统一写出去。结果前端流式效果全废答一句完整的话要等十几秒体验极差。排查思路先用curl直接打上游模型API确认上游SSE正常再跳过网关直连接口也正常最后定位到是我中间件把响应体缓存了。修正方式只对非SSE请求做Body缓冲SSE请求直接往原始响应流写写完立即FlushAsync。另外一个隐藏问题如果有Response.Headers里设置了Content-Length在流式模式下会被框架拒绝所以SSE响应不要设置长度。6.2 偶发超时上游读取超时设置太短有段时间生产环境老报超时查看日志发现集中在模型高峰期响应时间偶尔飙到30秒以上。原因是模型推理本身就是慢操作尤其是大模型在峰值时排队动辄几十秒才返回第一个token而我把HttpClient的超时设成了15秒导致频繁提前断开。修正思路区分三段时间。连接超时设置5秒读取响应头超时设置30秒读取内容流超时不设上限或者设置到300秒。原因是大模型流式响应只要第一个字节到了说明服务端已经在生成多等等没关系。这个经验在接国内一些厂商的API时尤其重要它们的排队时长可能比OpenAI还长。6.3 JSON转义把提示词弄坏了在写PromptTransformer时我为了省事把整个请求体用JsonSerializer.Serialize序列化后直接作为字符串嵌入到目标格式里结果发现提示词里的引号、换行被双重转义模型看到了一大堆\\n而不是真正的换行符。这个问题本质上是对JSON序列化的机制理解不透彻。把序列化后的结果当字符串用等于做了两次序列化。正确做法是始终操作JsonNode或JsonDocument对象树通过节点赋值来构造目标结构最后再做一次整体序列化。既避免了转义问题性能还更好。6.4 并发连接数冲击默认HttpClient限死了性能ASP.NET Core默认的HttpClient如果不显式配置线程池会为每个转发请求建立一个连接高并发下Socket耗尽表现为大量No such device or address和超时的混合错误。YARP虽然自带连接管理但如果自定义Provider里自己new了HttpClient这个坑就跑不掉。正确做法所有自定义调用一律注入IHttpClientFactory创建的客户端同时开启连接复用并且区分不同上游服务的连接池。还要注意HTTP/2的支持——OpenAI和Anthropic现在都支持HTTP/2打开之后TLS握手开销小很多在高并发下能省掉一大截连接建立时间。6.5 上游返回错误时错误信息原样抛给业务方早期设计里上游模型API返回4xx/5xx时网关直接把上游错误透传给业务方。比如供应商那边的限流提示是英文的、字段名也和OpenAI不同业务方收到后没法处理。后来我在网关加了一层错误规范化上游错误先解析提取错误状态码、错误类型、错误信息然后重新封装成OpenAI风格的错误结构返回。比如上游返回429网关会返回{error: {message: rate limit exceeded, retry after 12s, type: rate_limit_error}}。这样业务方无论是做重试还是告警都有统一的规范和字段不用为每一家厂商做一遍兼容。7. 这个网关后续还能怎么扩展目前这套网关整体跑得很稳定但它绝不是终点。经过这段实践我觉得在AI基础设施这条路上网关的价值会越来越大而且发展方向也很明确模型自动降级是下一步优先要做的方向。比如配置一个主模型gpt-4o当它持续限流或报错时自动把流量切到备用的qwen-max或claude-3-5-sonnet并返回一个可追踪的头信息让业务方知道这次实际用的是哪个模型。这个我做了一半后续整完再分享。多模态与多协议扩展。现在网关统一的是文本聊天格式但图片生成、Embedding、语音转文字都在快速普及每种能力的协议都不太一样。把这个网关扩展成一切AI能力的中枢那覆盖面就又上一个台阶。计费与配额的商品化。当前是按业务线做配额下一步希望能按项目、按功能模块甚至按最终用户做多层配额嵌套方便做内部结算和成本归因。这需要引入一套树形结构的预算模型技术上比现在复杂不少但收益也大。我在实际开发里的一个体会是做这类中间层基础设施最容易出错的地方其实不是技术选型而是需求的边界。AI模型更新迭代太快了今天接五家三个月后又多三家网关的抽象层如果做得太死每次接新模型都要动核心代码那就又回到了当年屎山代码的老路。所以设计的时候一定要把新增一个模型Provider做成只加一个类、改一段配置的事这也正是C#的接口、委托和泛型最拿手的场景。最后再分享一个压箱底的小技巧网关上线前一定先写一套完整的、基于真实上游API的集成测试覆盖非流式、流式、上游4xx、上游5xx、超时、限流命中这些场景。不要只做Mock测试因为Mock永远模拟不出真实供应商API的诡异行为。跑通这套测试以后后面每次加模型心里都有底改代码也不会慌。构建AI网关这件事越早做后面越省心。