举个栗子：做个AI Gateway demo

豌畔丛 · 昨天 10:45

现在ai盛行，各家api gateway 纷纷实现了ai gateway 的功能，作为强大的卖点
2025 年个人用c#造了个api gateway的轮子VKProxy，现在各项功能或扩展能力齐备，性能也不差，个人时间精力也有限，ai gateway 就搞个简单的 demo 做为此轮子篇章的结尾吧
那么AI Gateway有什么作用？

AI Gateway（AI 网关）是专门为 AI/LLM 型服务设计的网关层，负责把应用和底层模型提供者（或自建模型）之间的请求做统一管理、路由、识别、控制与观测。它把模型接入、权限、合规、性能优化和成本控制等横切关注点抽象出来，供上层应用以统一、安全和可控的方式调用 AI 功能。
主要作用（要点）

统一接入：屏蔽不同模型/供应商（如 OpenAI、Anthropic、自建模型等）差异，提供统一 API。
路由与模型选择：根据策略（任务类型、成本、延迟、质量）把请求路由到合适模型或并联/串联多个模型。
认证与授权：集中鉴权、API key 管理、细粒度权限控制与多租户隔离。
请求治理：输入校验、敏感信息脱敏、内容审查（安全/合规）、速率限制、配额管理。
成本与流量控制：按模型/客户限额、自动降级到更低成本模型、token 计费监控与预算告警。
性能优化：请求批量/合并、并发控制、缓存常见响应、流式转发、超时与重试策略。
可观测性与审计：统一日志、指标、分布式追踪、行为审计、响应质量监控、训练数据反馈采集。
隐私与数据治理：数据脱敏、数据留存策略、是否回传训练（opt-out）控制、满足合规要求。
流水线与编排：支持 prompt 管理、预处理/后处理、链式调用（chaining）、融合多模型输出（ensembling）。
插件/适配器：接入检索增强生成（RAG）、知识库、数据库、向量搜索等组件。

典型架构组件

前端 API 层：统一 REST / gRPC / WebSocket 接口，处理认证与速率限制。
路由/策略引擎：根据规则选择模型、拆分/合并请求、决定缓存策略。
模型适配器层：与不同模型提供者的 SDK/HTTP 适配器（支持不同协议与速率限制）。
处理流水线：输入校验、脱敏、prompt 模板、后处理、过滤。
缓存与队列：降低重复调用、支持批处理与异步任务。
观测与审计：日志、指标、请求追踪、合规审计存档。
管理面板：配额、策略配置、模型目录、监控告警、权限管理。

常见使用场景

企业多模型策略：按任务切换最优模型（例如摘要使用小模型、法律审查使用高质量模型）。
多租户 SaaS：不同客户隔离、配额与计费。
合规/安全敏感场景：控制数据流向、审计、内容过滤。
成本优化：高峰期自动降级或批处理以减少 token 消耗。
平滑迁移模型供应商：替换底层模型/供应商而不改应用代码。
构建复杂应用流水线：RAG、后处理规则、多模型投票/融合等。

请求在 AI Gateway 中的简化流程（示例）

客户端调用网关 API，携带 token/用户 id。
网关验证权限、检查配额/速率。
输入做脱敏/校验，选择并填充 prompt 模板。
路由引擎决定使用哪个模型（或并行多模型）。
调用模型适配器（支持流式或同步），可能先做缓存查找。
收到模型响应后做后处理（过滤、格式化）、记录日志与成本信息。
返回给客户端并存储审计记录或反馈数据供后续监控/微调。

选择或设计 AI Gateway 时的注意事项

延迟要求：网关本身需尽量低延迟，注意串联多个后处理步骤会增加延时。
可扩展性：设计水平可扩展的路由与限流，防止单点瓶颈。
多协议/流式支持：是否需要支持流式响应、WebSocket、SSE（Server-Sent Events）。
可配置性与策略表达：路由、降级策略、流量分配规则是否够灵活且可实时调整。
安全与合规：数据加密、日志脱敏、数据驻留与擦除策略。
可观测性：记录 token 使用、错误率、延迟、模型质量指标（如生成可信度）。
失败/回退策略：模型不可用时的备用方案、重试策略与幂等性处理。
成本透明度：按模型/请求细粒度计费、告警与报表支持。

何时直接调用模型 vs 使用 AI Gateway

直接调用模型适合：小规模、单一模型的试验或原型开发（更简单、快速）。
使用 AI Gateway 适合：生产环境、多模型、多租户、需要合规/审计、成本或路由策略的场景。

AI Gateway Demo 实现

这里用 VKProxy 作为实现基础，不过为了演示简单，不做流式响应处理，所以大家也可以非常简单用 asp.net + client 或 yarp 或任意语言任意框架实现
首先定义 api

作为ai gateway ，我们可以直接统一各个 ai 服务的配置，这样用户无需重复配置各项ai 服务
为了演示简单，这里就演示最简单统一适配各项ai服务，提供统一api的方式，定义的配置就可以像如下

{
"Logging": {
"LogLevel": {
"Default": "Information"
}
},
"ServerOptions": {
"AddServerHeader": false
},
"ReverseProxy": {
"ConnectionTimeout": "00:00:01.000",
"Listen": { // 服务监听地址
"http": {
"Protocols": [ "Http1" ],
"Address": [ "127.0.0.1:5001", "[::1]:5001" ]
}
},
"Routes": {
"ai": { // 路由配置，为了简单我们只做非流式响应的 chat
"Match": {
"Hosts": [ "*" ],
"Paths": [ "/v1/chat/completions" ]
},
"ClusterId": "openai",
"Timeout": "00:10:00",
"Metadata": {
"AiMapping": "{"openai":{"DefaultModel": "llama-3.3-70b-versatile","Driver":"openai"},"deepseek":{}}" // 这里用最简单的配置
}
}
},
"Clusters": { // 可复用的ai 服务配置
"deepseek": {
"LoadBalancingPolicy": "RoundRobin",
"Destinations": [
{
"Address": "https://api.deepseek.com"
}
]
},
"openai": {
"LoadBalancingPolicy": "RoundRobin",
"Destinations": [
{
"Address": "https://api.groq.com/groq"
}
]
}
}
}
}

复制代码

api 请求格式

现在各家 ai 请求都很统一相似，这里就演示简单的固定结构，这样大家理解也简单，不过实现最好以比较动态的方式实现，这样适配各家差异也比较容易

public class AiRequest
{
public string? ApiKey { get; set; } // ai 服务key，这里放请求其实是为了避免个人不小心上传key到 GitHub，至于大家是否需要可自己考虑
public bool? Stream { get; set; } // 是否流式响应，当然这里不会演示，不然就太多了
public string? Provider { get; set; } // 哪家服务
public string? Model { get; set; } // 哪个模型
public List? Messages { get; set; } // 你想让ai干什么
}
public class AiMessage
{
public string? Role { get; set; }
public string? Content { get; set; }
}

复制代码

实现 IAIProvider

public interface IAIProvider
{
Task Request(HttpContext context, AiRequest req, AiMapping ai);
}
public class OpenAiProvider : IAIProvider
{
public virtual async Task Request(HttpContext context, AiRequest req, AiMapping ai)
{
if (req.Model == null)
{
req.Model = ai.DefaultModel;
}
context.Request.Headers.Authorization = $"Bearer {req.ApiKey ?? ai.ApiKey}"; // 设置授权，测试的是这家是这样格式，所以为了演示简单，写死了
req.Provider = null;
req.ApiKey = null; // 通过json 忽略多余字段，有些ai 服务商校验比较严格，多一个它没有的字段都不行
var r = new MemoryStream(Encoding.UTF8.GetBytes(System.Text.Json.JsonSerializer.Serialize(req, AiGatewayMiddleware.jsonOptions)));
context.Request.Body = r;
context.Request.ContentLength = r.Length; // 替换body 和 ContentLength
// 产线最好严格处理 Accept / Accept-Encoding 等等影响格式的内容，避免压缩等等各种情况带来的解析问题，这里就不做了
}
}

复制代码

实现 Middleware

public class AiGatewayMiddleware : IMiddleware
{
private readonly ILogger logger;
private readonly IConfigSource<IProxyConfig> configSource;
private readonly IServiceProvider provider;
private readonly ILoadBalancingPolicyFactory loadBalancing;
public static readonly JsonSerializerOptions jsonOptions = new JsonSerializerOptions() { DefaultIgnoreCondition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingNull, PropertyNameCaseInsensitive = true, PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower };
public AiGatewayMiddleware(ILogger logger, IConfigSource<IProxyConfig> configSource, IServiceProvider provider, ILoadBalancingPolicyFactory loadBalancing)
{
this.logger = logger;
this.configSource = configSource;
this.provider = provider;
this.loadBalancing = loadBalancing;
}
public async Task InvokeAsync(HttpContext context, RequestDelegate next)
{
var feature = context.Features.Get<IReverseProxyFeature>();
if (feature is not null)
{
var route = feature.Route;
if (route is not null && route.Metadata is not null
&& route.Metadata.TryGetValue("AiMapping", out var b)) // 这里 VKProxy 已经解析过配置，匹配上对应路由了，只需判断是否开启了ai 功能，如果你用 asp.net core ，则需要自己处理，当然也可以直接用 controller + httpclient 直接写，虽然少了些优化点，不过ai 场景这点性能不太重要了大多
{
var req = await context.Request.ReadFromJsonAsync(context.RequestAborted); // 读取请求
if (req is null)
{
context.Response.StatusCode = StatusCodes.Status400BadRequest;
return;
}
var mapping = System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, AiMapping>>(b); // 读取配置，这里demo 为了简单就不写什么缓存优化了
IAIProvider driver;
if (mapping.TryGetValue(req.Provider ?? "openai", out var ai)) // 获取ai 调用实现
{
driver = provider.GetKeyedService<IAIProvider>(ai.Driver);
}
else
driver = null;
if (driver is null)
{
context.Response.StatusCode = StatusCodes.Status400BadRequest;
await context.Response.WriteAsJsonAsync(new { error = $"Unsupported AI provider: {req.Provider}" }, context.RequestAborted);
return;
}
if (configSource.CurrentSnapshot.Clusters.TryGetValue(ai.Driver, out var cluster)) // 获取服务地址
{
feature.SelectedDestination = loadBalancing.PickDestination(feature, cluster); // 获取有效地址
if (feature.SelectedDestination is null)
{
context.Response.StatusCode = StatusCodes.Status503ServiceUnavailable;
await context.Response.WriteAsJsonAsync(new { error = $"{req.Provider} service unavailable" }, context.RequestAborted);
return;
}
}
await SendAi(context, req, ai, driver); // 处理请求
// 由于VKProxy 再后续已经实现非常高效的代理处理，这里就简单实现复制内容，这样respone 会第一时间就已经送回的 client端，即使后续我们token 记录出问题也不影响用户，不过要修改就要麻烦点了，所以不在乎性能的话，client 直接call 是最简单的
var origin = context.Response.Body;
using var buf = new ResponseCachingStream(origin, 64 * 1024 * 1024 * 10, 81920, () => ValueTask.CompletedTask); // demo这里直接用内存实现的，产线如果也想这么做，推荐小的用内存，大的用磁盘，或者直接镜像复制发送给其他程序做分析统计等等
context.Response.Body = buf;
await next(context); // 代理处理
// 结果解析，并记录 token 使用，这里demo 就简单记录 log 了，真实场景就可以通过log 解析或者直接存入统计db 等，这样可以达到ai gateway最重要的功能之一： token 统计，找出账单超支的罪魁祸首
var resp = buf.GetCachedResponseBody();
using var u = new MemoryStream(resp.Segments.SelectMany(i => i).ToArray());
var aiResp = JsonSerializer.Deserialize(u, jsonOptions);
if (aiResp?.Usage != null)
{
logger.LogWarning("AI TotalTokens: {Usage}", aiResp.Usage.TotalTokens);
}
context.Response.Body = origin;
return;
}
}
await next(context);
}
private async Task SendAi(HttpContext context, AiRequest req, AiMapping ai, IAIProvider driver)
{
await driver.Request(context, req, ai);
}
}

复制代码

除开 SSE ，基本的实现非常简单，其实这里代理实现VKProxy 已经为了性能引入一些复杂度，如果不在乎性能，比如用httpclient call，你们的写法还可以更简单，懒得自己写，还可以叫ai 代劳
全部代码也放在了 VKProxy AiGatewayDemo

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

账号		自动登录	找回密码
密码			立即注册

举个栗子：做个AI Gateway demo

相关帖子

浏览过的版块

签约作者

举个栗子：做个AI Gateway demo

相关帖子

相关推荐

浏览过的版块

签约作者