企业里做大模型落地,最容易被低估的一环不是模型选型,而是"网关"。我见过太多团队一开始直接让业务代码裸调各家 API,等到要换模型、要限流、要审计、要计费的时候,才发现改一处牵动全身。这篇就围绕大模型网关和自动化编程这两条线,把从基础搭建到真正落地的完整路径讲清楚,包括 Agent、CLI、API 这几个关键词背后的实际工程含义。不管你是刚接触这块的开发者,还是已经在做企业内部 AI 平台的负责人,都能从里面找到可以直接抄作业的部分。
1. 为什么企业一定要有大模型网关这层
1.1 裸调 API 的三个致命问题
先说清楚网关到底解决什么问题。很多团队初期为了快,业务代码里直接写openai.ChatCompletion.create(...)或者requests.post("https://api.xxx.com/v1/chat/completions", ...),跑起来确实没问题。但只要规模稍微上来,三个问题必然暴露。
第一个是密钥扩散。每个调用方都持有真实 API Key,一旦某个服务被入侵或者日志打印了请求头,密钥就泄露了。企业里几十个服务共用一套密钥,出了事根本查不到是谁泄露的。
第二个是模型切换成本。今天用 A 家的模型,明天老板说 B 家便宜一半要换,你得改所有业务代码。不同厂商的请求体格式、鉴权方式、流式返回格式都不一样,改起来是灾难。
第三个是可观测性缺失。谁在调、调了多少次、花了多少钱、响应多慢、有没有报错,这些数据散落在各个服务里,根本没法统一统计。等到财务问"这个月 AI 花了多少钱",你只能干瞪眼。
网关这层就是把这些横切关注点全部收拢到一个统一入口。业务方只需要知道网关地址和一个内部 token,剩下的鉴权、路由、限流、计费、日志全在网关内部完成。
1.2 网关的核心能力清单
一个能上生产的大模型网关,至少要具备下面这些能力,我按优先级排一下:
| 能力 | 优先级 | 说明 |
|---|---|---|
| 统一鉴权 | P0 | 内部 token 换真实密钥,密钥不出网关 |
| 多模型路由 | P0 | 按模型名/策略转发到不同厂商 |
| 流式转发 | P0 | SSE 透传,不能破坏流式体验 |
| 限流限速 | P1 | 按用户/应用/模型维度限流 |
| 用量计费 | P1 | token 统计、成本核算 |
| 日志审计 | P1 | 请求响应留痕,可追溯 |
| 失败重试与降级 | P2 | 主模型挂了自动切备用 |
| 缓存 | P2 | 相同请求命中缓存省钱 |
这里要特别强调流式转发。大模型的响应是 SSE(Server-Sent Events)流式返回的,网关如果处理不当,比如先把整个响应读完再返回,用户就会看到"卡半天然后一次性蹦出来",体验直接崩掉。正确的做法是边收边转发,用流式管道处理。
1.3 网关的部署形态选择
网关部署形态主要有三种,各有适用场景:
- 进程内 SDK 模式:把网关逻辑做成一个库,业务直接引入。优点是零网络开销,缺点是每个语言都要实现一遍,且升级困难。
- 独立服务模式:网关是一个独立部署的服务,业务通过 HTTP 调用。这是最主流的做法,语言无关,升级方便。
- Sidecar 模式:每个业务 Pod 旁边挂一个网关容器。适合 K8s 环境,隔离性好但资源开销大。
对绝大多数企业,我推荐独立服务模式。用 Go 或 Rust 写,性能足够,单机扛几千 QPS 没问题。下面给一个用 Go 写的极简网关核心逻辑示意:
func handleChatCompletion(w http.ResponseWriter, r *http.Request) { // 1. 校验内部 token internalToken := r.Header.Get("X-Internal-Token") app, err := auth.Verify(internalToken) if err != nil { http.Error(w, "unauthorized", 401) return } // 2. 解析请求,确定目标模型 var req ChatRequest json.NewDecoder(r.Body).Decode(&req) provider := router.Select(req.Model) // 3. 限流检查 if !limiter.Allow(app.ID, req.Model) { http.Error(w, "rate limited", 429) return } // 4. 替换为真实密钥,转发请求 upstreamReq := buildUpstreamRequest(req, provider.APIKey) resp, err := httpClient.Do(upstreamReq) if err != nil { // 5. 失败降级到备用模型 resp, err = fallback(req, provider) } // 6. 流式透传响应 streamCopy(w, resp.Body) }这段代码看着简单,但每一步都有坑。比如第 6 步的streamCopy,必须用http.Flusher强制刷新缓冲区,否则 Go 的默认缓冲会让流式变成"伪流式"。
提示:网关转发时一定要设置合理的超时。大模型首 token 延迟可能到几秒甚至十几秒,超时设太短会误杀正常请求,设太长又会拖垮连接池。我的经验是首 token 超时 30s,整体超时按模型最大输出长度估算。
2. 自动化编程里的 Agent 与 CLI 到底怎么配合
2.1 Agent 和 CLI 不是一回事
这两个词经常被混着用,但它们的定位完全不同。Agent 是"决策者",CLI 是"执行者"。
Agent 负责理解任务、拆解步骤、决定下一步做什么。它本质上是一个循环:观察当前状态 → 思考 → 选择动作 → 执行 → 观察结果 → 继续。而 CLI 是 Agent 可以调用的一个具体工具,比如git、npm、docker,或者专门为 AI 编程设计的命令行工具。
打个比方,Agent 像是一个项目经理,CLI 像是他手下的各种专业工具。项目经理不会自己去拧螺丝,而是决定"现在该用螺丝刀了",然后调用螺丝刀。
现在市面上有不少专门给 AI 用的 CLI 工具,比如一些代码生成 CLI、文件操作 CLI。它们的共同特点是:输入输出结构化、幂等性好、错误信息清晰。这三点是 Agent 能可靠调用它们的前提。
2.2 Agent 调用 CLI 的典型循环
一个自动化编程 Agent 的完整工作循环大概是这样:
- 接收任务:比如"给这个项目加上单元测试"
- 探索环境:调用
ls、cat、grep等 CLI 了解项目结构 - 制定计划:决定先看哪些文件,再写哪些测试
- 执行动作:调用文件读写 CLI 创建测试文件
- 验证结果:调用测试运行 CLI,看是否通过
- 修正迭代:如果失败,读错误信息,回到第 4 步
这个循环里,第 5 步的验证是灵魂。没有验证的 Agent 就是瞎猜,有了验证才能自我纠错。这也是为什么好的 Agent 框架都强调"工具返回结果要可解析"。
下面是一个 Agent 调用 CLI 的伪代码结构:
def agent_loop(task, max_steps=20): context = [{"role": "system", "content": SYSTEM_PROMPT}] context.append({"role": "user", "content": task}) for step in range(max_steps): # 让模型决定下一步动作 response = llm.chat(context, tools=AVAILABLE_TOOLS) if response.is_final: return response.content # 执行模型选择的工具 tool_name = response.tool_call.name tool_args = response.tool_call.args result = execute_tool(tool_name, tool_args) # 把结果喂回上下文 context.append(response.message) context.append({"role": "tool", "content": result}) return "达到最大步数限制"这里有个关键细节:上下文会越来越长。每轮工具调用都会往 context 里塞内容,几十轮下来 token 消耗惊人。所以生产级 Agent 必须做上下文管理,比如只保留最近 N 轮、对历史做摘要、把大文件内容截断等。
2.3 工具设计的三条铁律
给 Agent 设计 CLI 工具时,我总结了三条铁律,踩过坑的都懂:
第一,输出要精简且结构化。你让 Agent 跑一个ls -la,返回几百行带权限、时间、大小的信息,模型要花大量 token 去解析。更好的做法是提供一个list_files工具,只返回文件名列表。
第二,错误信息要能指导下一步。工具失败时,返回的不该是"Error: failed",而应该是"文件 xxx.py 第 42 行语法错误:缺少冒号"。模型看到后者才知道怎么修。
第三,操作要幂等或可回滚。Agent 可能会重复调用同一个工具,如果工具不幂等,就会产生副作用。比如"创建文件"应该是覆盖式的,而不是追加式的。
注意:涉及删除、覆盖、执行系统命令的工具,一定要加确认机制或沙箱隔离。我见过 Agent 误删整个目录的案例,血的教训。
3. 把网关和 Agent 串起来:一个完整的落地架构
3.1 整体架构分层
把前面两块拼起来,一个企业级的自动化编程平台大概分四层:
- 接入层:Web 界面、IDE 插件、CI/CD 钩子,用户从这里发起任务
- 编排层:Agent 运行时,负责任务拆解、工具调度、上下文管理
- 网关层:统一的大模型网关,处理所有 LLM 调用
- 执行层:各种 CLI 工具、代码仓库、测试环境
这个分层的好处是职责清晰。编排层不关心用的是哪家模型,网关层不关心任务是什么,执行层不关心谁在调用。任何一层要替换或升级,都不影响其他层。
3.2 请求的完整生命周期
一个"帮我修复这个 bug"的请求,走完整个链路是这样的:
- 用户在 IDE 插件里输入任务,插件把当前文件内容和任务发给编排层
- 编排层启动 Agent,构造初始上下文
- Agent 调用网关,请求模型生成下一步动作
- 网关鉴权、限流、路由到具体模型,返回结果
- Agent 解析出要调用的工具,比如"读取 test.py"
- 执行层执行工具,返回文件内容
- Agent 把结果加入上下文,再次调用网关
- 循环直到 Agent 认为任务完成,返回最终结果
- 编排层把结果返回给 IDE 插件
这个链路里,网关是唯一的模型出口,所有 token 消耗、延迟、错误都在这里被记录。这对成本控制和问题排查至关重要。
3.3 关键配置示例
网关的路由配置我一般用 YAML 管理,方便运维改:
providers: - name: primary base_url: https://api.provider-a.com/v1 api_key: ${PROVIDER_A_KEY} models: [gpt-4-class, gpt-3.5-class] timeout: 60s - name: backup base_url: https://api.provider-b.com/v1 api_key: ${PROVIDER_B_KEY} models: [claude-class] timeout: 60s routes: - match: "gpt-4-class" primary: primary fallback: backup rate_limit: 100/min - match: "claude-class" primary: backup rate_limit: 50/min billing: currency: CNY rates: gpt-4-class: { input: 0.03, output: 0.06 } # 每千 token claude-class: { input: 0.02, output: 0.04 }这份配置里,fallback字段实现了自动降级,rate_limit实现了限流,billing实现了计费。运维改配置不用动代码,重启网关即可生效。
4. 实操中真正会踩的坑
4.1 上下文长度超限的处理
大模型都有上下文窗口限制,Agent 跑久了必然超。报错信息通常是这样的:
API error: 400 This model's maximum context length is 1048576 tokens. However, your messages resulted in 1200000 tokens.处理这个问题的策略有三层:
第一层,预防。在往上下文里塞内容前,先估算 token 数。大文件不要整个塞进去,只塞相关片段。可以用简单的字符数除以 4 来粗估 token 数(英文),中文大概除以 1.5。
第二层,压缩。当上下文接近上限时,对历史消息做摘要。把前面十几轮的对话压缩成一段总结,保留关键决策和结论。
第三层,截断。实在不行就丢弃最早的几轮,但要保留 system prompt 和最近几轮。丢弃时最好保留工具调用的结果摘要,而不是直接删。
def manage_context(messages, max_tokens=100000): total = estimate_tokens(messages) if total < max_tokens * 0.8: return messages # 保留 system + 最近 5 轮 system = messages[0] recent = messages[-10:] # 中间部分做摘要 middle = messages[1:-10] if middle: summary = summarize(middle) return [system, {"role": "system", "content": f"历史摘要:{summary}"}] + recent return [system] + recent4.2 流式响应的中断与重连
流式响应最烦人的是中途断掉。网络抖动、模型服务重启、网关超时都可能导致流中断。用户看到的是"回答到一半没了"。
处理方案是在网关层做流式重试。具体做法是:网关记录已经转发给客户端的 token 数,如果上游断了,用相同的 prompt 重新请求,但要求模型跳过已发送的部分。不过这个方案实现复杂,且不是所有模型都支持。
更实用的方案是客户端侧容错。前端检测到流中断后,把已收到的内容作为上下文,发起一个"继续"请求。虽然会多花点 token,但实现简单可靠。
async function streamWithRetry(messages, maxRetries = 3) { for (let i = 0; i < maxRetries; i++) { try { const response = await fetch('/api/chat', { method: 'POST', body: JSON.stringify({ messages, stream: true }) }); return await processStream(response); } catch (e) { if (i === maxRetries - 1) throw e; // 把已收到的内容加入上下文,继续请求 messages.push({ role: 'assistant', content: partialContent }); messages.push({ role: 'user', content: '请继续' }); } } }4.3 密钥与权限的常见错误
配置网关时,密钥相关的错误特别多。最常见的两类:
第一类,环境变量没生效。报错长这样:
llm-deepseek: no api key for provider route "deepseek-official"这通常是配置文件里写了${DEEPSEEK_KEY},但环境变量没导出,或者导出在了错误的 shell 里。排查方法是在网关启动脚本里加一行env | grep KEY确认。
第二类,权限范围不匹配。比如某些 API 需要在控制台声明 scope,没声明就会报:
choosemedia:fail api scope is not declared in the privacy agreement这类问题只能去对应平台的控制台检查应用权限配置,代码层面无解。
提示:所有密钥统一用密钥管理服务(如 Vault、KMS)管理,不要写在配置文件或环境变量里。环境变量在容器里容易被
docker inspect看到。
4.4 Docker 权限问题
网关如果用 Docker 部署,经常会遇到:
permission denied while trying to connect to the docker api这是因为当前用户不在 docker 组里。解决方法是sudo usermod -aG docker $USER,然后重新登录。但生产环境更推荐用 rootless Docker 或者把网关跑在 K8s 里,避免直接暴露 docker socket。
5. 性能与成本的优化空间
5.1 缓存能省多少钱
大模型调用里,有相当比例的请求是重复或高度相似的。比如 Agent 反复读取同一个文件、反复问同样的问题。加一层语义缓存,命中率能到 20%-40%。
缓存分两种:
- 精确缓存:请求内容完全一致才命中。实现简单,用 Redis 存
hash(prompt) -> response即可。 - 语义缓存:请求语义相似就命中。需要向量化 prompt,做相似度检索。命中率更高但实现复杂。
对大多数场景,精确缓存就够了。注意缓存要设置合理的 TTL,模型更新后旧缓存要失效。
5.2 模型分级路由
不是所有请求都需要最强模型。Agent 的很多步骤,比如"判断文件类型""提取函数名",用便宜的小模型完全够用。只有关键推理步骤才需要大模型。
网关可以按请求的复杂度做分级路由:
def select_model(request): # 简单任务用小模型 if request.task_type in ["classify", "extract", "format"]: return "small-model" # 复杂推理用大模型 if request.task_type in ["reason", "plan", "debug"]: return "large-model" return "medium-model"这个策略实测能省 50% 以上的成本,而任务成功率几乎不受影响。
5.3 并发与连接池
网关作为所有请求的入口,并发能力是瓶颈。几个关键点:
- HTTP 客户端要复用连接,不要每次请求都新建。Go 的
http.Client默认就复用,但要注意MaxIdleConnsPerHost要调大。 - 上游连接数要限制,避免把模型服务打挂。用信号量或连接池控制。
- 流式请求要单独管理,因为流式连接占用时间长,不能和普通请求共用连接池。
transport := &http.Transport{ MaxIdleConns: 1000, MaxIdleConnsPerHost: 200, IdleConnTimeout: 90 * time.Second, // 流式请求需要禁用响应缓冲 DisableCompression: false, } client := &http.Client{ Transport: transport, Timeout: 0, // 流式请求不设总超时,用 context 控制 }6. 从能跑到好用的几个进阶点
6.1 可观测性建设
网关跑起来后,最重要的就是"看得见"。至少要采集这几类指标:
| 指标类型 | 具体指标 | 用途 |
|---|---|---|
| 请求量 | QPS、按模型/应用分组 | 容量规划 |
| 延迟 | P50/P95/P99、首 token 延迟 | 体验监控 |
| 错误 | 错误率、错误类型分布 | 故障排查 |
| 成本 | token 消耗、费用 | 成本控制 |
| 限流 | 触发次数、被限流应用 | 配额调整 |
这些指标用 Prometheus 采集,Grafana 展示。首 token 延迟这个指标特别重要,它直接决定用户体感,但很多团队只监控总延迟,忽略了它。
6.2 灰度与回滚
模型切换、网关升级都要能灰度。做法是给请求打标签,按比例分流到新旧版本。比如 10% 流量走新模型,观察指标正常后再逐步放大。
回滚要能秒级完成。配置化的路由让回滚变成改一行配置的事,这是网关架构相比硬编码的最大优势。
6.3 Agent 的安全边界
Agent 能执行命令、读写文件,安全边界必须划清楚:
- 文件系统隔离:Agent 只能访问指定目录,用容器或 chroot 限制
- 命令白名单:只允许执行预定义的安全命令,禁止
rm -rf、curl外网等 - 网络隔离:Agent 执行环境不能访问外网,防止数据外泄
- 资源限制:CPU、内存、执行时间都要设上限,防止死循环
这些限制在网关层和编排层都要做,不能只靠一层。
6.4 多租户隔离
企业里多个团队共用一套平台,隔离是刚需。隔离维度包括:
- 配额隔离:每个团队有独立的 token 配额和 QPS 上限
- 数据隔离:A 团队的对话历史 B 团队看不到
- 密钥隔离:每个团队用自己的密钥,便于独立计费
- 模型隔离:某些高级模型只对特定团队开放
这些都在网关的鉴权环节实现,通过内部 token 关联到租户信息,后续所有操作都带上租户上下文。
7. 一些实战中的零碎经验
最后分享几个散落但很实用的点。
关于 CLI 工具的选择,优先选那些有--json输出选项的。文本输出解析起来太脆弱,模型稍微换个格式就崩。JSON 输出配合 schema 校验,稳定性高一个数量级。
关于 Agent 的步数限制,不要设太大。我见过设 100 步的,结果 Agent 陷入循环,烧了几百万 token 才发现。一般任务 20 步足够,复杂任务 50 步封顶,超了就报错让人介入。
关于网关的日志,请求和响应内容要脱敏后再存。用户可能在里面输入了敏感信息,直接存明文有合规风险。至少要把身份证、手机号、邮箱这类模式做正则替换。
关于模型降级,降级不是简单换个模型就行。不同模型的输出格式可能不同,Agent 的解析逻辑要能兼容。最好在网关层做输出格式归一化,让上层感知不到模型差异。
关于测试,网关和 Agent 都要有 mock 能力。不能每次测试都真调模型,又慢又贵。用录制回放的方式,把真实响应录下来,测试时回放,既快又稳定。
关于版本管理,prompt 也要版本化。Agent 的 system prompt 改一个字,行为可能天差地别。把 prompt 纳入 Git 管理,每次改动都有记录,出问题能快速定位是哪次改动导致的。
这套东西我从零搭过两遍,第一遍踩了无数坑,第二遍就顺多了。核心体会是:网关这层越早建越好,Agent 的能力越晚放越好。网关是基础设施,早建早省心;Agent 的能力边界要慢慢放开,每放开一个权限都要配套的监控和回滚机制。急着让 Agent 什么都能干,最后往往是收拾烂摊子花的时间比省下来的还多。