用 Magnitude 给编码智能体搭自优化推理层

编码 Agent 的请求其实很杂：单行补全、函数级生成、跨文件重构、单测生成、失败日志归因，彼此对延迟和智力的要求差得很远。全部走同一个旗舰模型，账单和延迟都难看；全部走小模型，复杂任务返工率又会上来。这篇教程做的事，是在 Agent 和模型之间插一层推理编排：按任务打标签路由、记录指标、用真实数据反复压测，再把压测结果反写回路由表。

一个前置说明：下文出现的配置字段名、CLI 子命令、指标名都是通用形态，不同版本可能有差异。落地时以官方文档当前版本的字段名为准，判断标准很简单——网关能正常转发并返回 200，/metrics 能读出数据。

适用场景

适合已经有一个跑起来的编码 Agent（IDE 插件、PR 机器人、CLI agent 都行），每天有稳定调用量，想按任务类型分流到不同模型和参数上的团队。也适合还没接入但在做选型的人：先用一层网关把"哪种任务适合哪个模型"这件事量化出来，再决定要不要自建推理。

如果每天只有几十次调用、或者所有任务都是同一类，这层编排带来的复杂度大于收益，可以先不上。

环境与前置条件

• 操作系统：Linux（x86_64 / arm64）或 macOS。Windows 建议走 WSL2，避免路径和信号处理上的坑。
• 运行时：Python 3.10 及以上，用独立虚拟环境隔离依赖；容器方式运行则需 Docker 24+ 与 compose 插件。
• 内存：只做路由与代理，2 核 4G 起步即可。如果要在同一台机器上跑本地小模型，显存按模型实际需求另算，7B 级别的量化模型经验上要预留 6~8GB 显存，内存再留一倍余量。
• 磁盘：预留 20GB 以上，给镜像、结构化日志和压测数据集。压测集建议直接放本地 SSD，回放时 I/O 会是瓶颈之一。
• 上游：至少两个 OpenAI 兼容的模型服务地址与 API Key（一个偏快偏小，一个偏强偏慢）。Key 一律用环境变量注入，不要写进配置文件提交到仓库。
• 可观测：Prometheus + Grafana 可选但推荐。没有的话，至少保证网关把每次请求的结构化日志落盘，否则"自优化"没有数据来源。
• 网络：能出网访问上游。内网部署要提前配好 HTTP(S)_PROXY，并确认网关容器的 DNS 能解析上游域名。

分步骤部署

步骤 1：建目录、装运行时

```bash

mkdir -p ~/mag-layer/{config,data,logs}

cd ~/mag-layer

python3 -m venv .venv && source .venv/bin/activate

pip install -U pip

pip install magnitude httpx # 包名与安装方式以官方文档当前版本为准

```

这步在做的事：准备一个干净的运行环境和三个目录——config 放路由表，data 放压测集，logs 放请求日志。

成功标志：magnitude --version（或对应的版本子命令）能打印出版本号，没有 command not found。

步骤 2：写路由表

路由表是整个方案的核心。它回答的问题是：什么样的请求，交给谁，花多少预算。

```yaml

~/mag-layer/config/magnitude.yaml

字段名以官方文档当前版本为准

listen: "0.0.0.0:8080"

upstreams:

• name: small-coder # 快、便宜，用于补全与改写

base_url: "https://<上游地址>/v1"

api_key_env: "UPSTREAM_KEY_SMALL"

model: "<小模型名>"

• name: frontier # 强、慢，留给跨文件重构与疑难归因

base_url: "https://<上游地址>/v1"

api_key_env: "UPSTREAM_KEY_FRONTIER"

model: "<大模型名>"

defaults:

timeout_ms: 30000

max_retries: 2

retry_on: [429, 500, 502, 503]

concurrency:

small-coder: 16 # 每个上游的并发上限，防止把上游打到限流

frontier: 4

routes:

• name: inline_completion

match:

task_type: completion

context_tokens: {lte: 4000}

upstream: small-coder

params: {temperature: 0.2, max_tokens: 256}

timeout_ms: 4000

• name: explain_or_test

match:

task_type: [explain, test_gen]

context_tokens: {lte: 32000}

upstream: small-coder

params: {temperature: 0.3, max_tokens: 1024}

timeout_ms: 20000

• name: multi_file_refactor

match:

task_type: refactor

files_touched: {gte: 2}

upstream: frontier

params: {temperature: 0.1, max_tokens: 4096}

timeout_ms: 60000

fallback: # 没命中任何路由时兜底，宁可慢也别报错

upstream: frontier

params: {temperature: 0.2}

budget:

max_context_tokens: 128000 # 超过就截断或拒绝，避免上游直接 400

daily_cost_units: 100000 # 单位由网关定义，用来兜底防失控

```

注意 daily_cost_units 用的是"额度单位"而不是具体金额——这样换上游、调价都不用改配置，只需要改单位换算。

成功标志：配置文件能被解析。先本地校验一下：

```bash

python -c "import yaml,sys; yaml.safe_load(open('config/magnitude.yaml')); print('yaml ok')"

```

步骤 3：给请求打标签，这一步决定自优化能不能做

没有标签，日志里只有一堆匿名请求，没法按任务类型算 p95，也就无从优化。约定四个请求头：

在 Agent 侧统一注入这几个头，别让业务代码各自拼。可以封一个 SDK 包装函数，所有调用走同一个出口。

步骤 4：把编码 Agent 接到网关上

Agent 侧只改两处：base_url 指向网关，api_key 用网关自己的本地 key（不是上游 key）。

```python

from openai import OpenAI

client = OpenAI(base_url="http://127.0.0.1:8080/v1", api_key="sk-local-test")

raw = client.chat.completions.with_raw_response.create(

model="auto", # 让网关按路由表决定真实模型

messages=[

{"role": "system", "content": "你是代码助手，只输出补丁。"},

{"role": "user", "content": "把 utils.py 里的 read_config 改成支持环境变量覆盖。"},

],

extra_headers={

"x-task-type": "refactor",

"x-repo-id": "svc-payment",

"x-files-touched": "3",

"x-agent": "pr-bot",

},

)

print("命中路由:", raw.headers.get("x-route"))

print("上游模型:", raw.headers.get("x-upstream"))

print(raw.parse().choices[0].message.content)

```

with_raw_response 是新版 openai-python 提供的写法，能拿到响应头；如果你的版本不支持，直接用 requests/httpx 发也可以。响应头里的 x-route 就是这次命中的路由名，后面排查"为什么这条走了大模型"全靠它。

成功标志：同一段代码，把 x-task-type 从 completion 改成 refactor，x-route 跟着从 inline_completion 变成 multi_file_refactor。

步骤 5：启动网关并打开指标

```bash

export UPSTREAM_KEY_SMALL="..."

export UPSTREAM_KEY_FRONTIER="..."

magnitude serve --config ./config/magnitude.yaml --log-dir ./logs

容器方式：docker run --rm -p 8080:8080 -v $PWD/config:/config <镜像>

```

容器方式运行时记得 -e UPSTREAM_KEY_SMALL=...，或者用 --env-file，不要把 Key 打进镜像。

成功标志：启动日志里打印出 listen on 0.0.0.0:8080、已加载的路由条数、上游健康检查结果。

步骤 6：准备压测数据集

从真实调用日志里采样，脱敏后存成 jsonl，一行一个任务：

```json

{"id": "t-001", "task_type": "completion", "files_touched": 1, "repo_id": "svc-payment", "messages": [{"role": "user", "content": "补全这个函数：def parse_ts(s):"}]}

{"id": "t-002", "task_type": "refactor", "files_touched": 3, "repo_id": "svc-payment", "messages": [{"role": "user", "content": "把三处重复的校验逻辑抽成 validate()"}]}

```

关键点：分布要接近真实。如果线上 70% 是补全、10% 是重构，压测集也按这个比例来，否则压出来的 p95 没有参考价值。建议至少 200 条，覆盖每类任务。

脱敏规则提前定好：把密钥、内网域名、用户 ID 替换成占位符，再落进数据集。

步骤 7：跑并发梯度压测，拿到基线

不要只测单并发。真实流量是叠加上来的，延迟曲线往往在某个并发点之后开始非线性抬升。

```python

~/mag-layer/data/bench.py

import asyncio, json, time, statistics

import httpx

GATEWAY = "http://127.0.0.1:8080/v1/chat/completions"

KEY = "sk-local-test"

CONCURRENCY = 8 # 依次改成 1 / 4 / 8 / 16

REPEAT = 3 # 每个样本重复跑几轮

async def one(client, sem, item, out):

async with sem:

t0 = time.perf_counter()

try:

r = await client.post(

GATEWAY,

headers={

"Authorization": f"Bearer {KEY}",

"x-task-type": item["task_type"],

"x-files-touched": str(item.get("files_touched", 1)),

"x-repo-id": item.get("repo_id", "bench"),

"x-agent": "bench",

},

json={"model": "auto", "messages": item["messages"]},

timeout=120,

)

ms = (time.perf_counter() - t0) * 1000

usage = {}

try:

usage = r.json().get("usage", {}) or {}

except Exception:

pass

out.append({

"id": item["id"], "task_type": item["task_type"],

"ok": r.status_code == 200, "status": r.status_code,

"latency_ms": ms, "route": r.headers.get("x-route"),

"out_tokens": usage.get("completion_tokens", 0),

})

except Exception as e:

out.append({

"id": item["id"], "task_type": item["task_type"],

"ok": False, "status": type(e).__name__,

"latency_ms": (time.perf_counter() - t0) * 1000,

})

def pct(xs, p):

xs = sorted(xs)

if not xs:

return 0.0

idx = min(len(xs) - 1, int(len(xs) * p))

return xs[idx]

async def main():

items = [json.loads(l) for l in open("data/tasks.jsonl", encoding="utf-8")]

items = items * REPEAT

sem = asyncio.Semaphore(CONCURRENCY)

out = []

t0 = time.perf_counter()

async with httpx.AsyncClient() as client:

await asyncio.gather(*(one(client, sem, it, out) for it in items))

wall = time.perf_counter() - t0

ok = [r for r in out if r["ok"]]

lats = [r["latency_ms"] for r in out]

tok = sum(r.get("out_tokens", 0) for r in ok)

print(f"并发={CONCURRENCY} 总请求={len(out)} 成功率={len(ok)/len(out):.2%}")

print(f"墙钟={wall:.1f}s 吞吐={len(out)/wall:.2f} req/s 输出吞吐={tok/wall:.1f} tok/s")

print(f"p50={pct(lats,0.50):.0f}ms p95={pct(lats,0.95):.0f}ms max={max(lats):.0f}ms")

按路由拆分，看是不是某一条路由拖了后腿

by_route = {}

for r in out:

by_route.setdefault(r.get("route") or "unrouted", []).append(r)

for k, v in sorted(by_route.items()):

ls = [x["latency_ms"] for x in v]

err = sum(1 for x in v if not x["ok"]) / len(v)

print(f" {k:24s} n={len(v):4d} p95={pct(ls,0.95):7.0f}ms err={err:.1%}")

with open(f"logs/bench_c{CONCURRENCY}.jsonl", "w", encoding="utf-8") as f:

for r in out:

f.write(json.dumps(r, ensure_ascii=False) + "\n")

asyncio.run(main())

```

按顺序跑 1 / 4 / 8 / 16 并发，把每轮输出记下来。你会看到类似这样的基线表：

这张表就是基线。后面任何改动的价值，都用它来衡量——注意看 16 并发那行，吞吐几乎没涨、延迟掉崖、错误率上来了，说明瓶颈要么是上游配额，要么是网关并发上限设小了。

步骤 8：写自优化闭环：滚动窗口 + 阈值规则

"自优化"不需要一上来就上强化学习，一条滚动窗口统计加几条阈值规则，就能覆盖大部分收益。

```python

~/mag-layer/data/optimize.py

import json, collections, statistics

WINDOW = 200 # 每个路由看最近 200 条

BUDGET = { # 各路由的 p95 延迟预算（毫秒）

"inline_completion": 1500,

"explain_or_test": 8000,

"multi_file_refactor": 30000,

}

def p95(xs):

xs = sorted(xs)

return xs[min(len(xs) - 1, int(len(xs) * 0.95))] if xs else 0.0

buckets = collections.defaultdict(list)

for line in open("logs/requests.jsonl", encoding="utf-8"):

r = json.loads(line)

if r.get("route"):

buckets[r["route"]].append(r)

for route, rows in buckets.items():

rows = rows[-WINDOW:]

lat = p95([r["latency_ms"] for r in rows])

err = sum(1 for r in rows if not r["ok"]) / len(rows)

budget = BUDGET.get(route)

if budget is None:

continue

if lat > budget * 1.2 or err > 0.02:

print(f"[降级候选] {route}: p95={lat:.0f}ms(预算{budget}) err={err:.1%} → 降低并发上限或切更稳上游")

elif lat < budget * 0.5 and err < 0.005:

print(f"[升级候选] {route}: p95={lat:.0f}ms(预算{budget}) err={err:.1%} → 可试更便宜的模型，跑 A/B")

else:

print(f"[保持] {route}: p95={lat:.0f}ms err={err:.1%}")

```

把 logs/requests.jsonl 换成网关真实落盘的日志（字段名按你的日志格式改一行即可）。

规则背后的逻辑：

• 降级候选：某条路由最近明显慢了或开始报错，先把它的并发上限调低，或者切到更稳的上游。这一步通常是立竿见影的。
• 升级候选：某条路由长期远离预算，说明这个任务用小模型就够了，可以拿一部分流量做 A/B，看质量指标（单测通过率、人工返工率、diff 采纳率）有没有下降。
• 质量门禁：成本优化必须挂质量指标。延迟和成本再好看，如果返工率涨了 5%，整体反而是亏的。所以 A/B 的对照指标要用"任务最终是否被采纳"，而不是单纯的成功率。

步骤 9：灰度、对比与回滚

路由表就是代码，用 git 管起来。每次改动走同一条路径：

1. 新建分支改 magnitude.yaml，写清改了什么、预期收益是什么。

2. 在 staging 用同一套压测集跑一遍，和基线表对比。

3. 线上按 5% → 20% → 50% 的比例灰度（网关侧按请求头或仓库 ID 分流）。

4. 观察一个完整的业务周期（编码任务通常看一天就够），确认质量指标没掉。

5. 固化，记录这一版的路由表 commit。

回滚就是 git revert 加一次热加载。建议网关开启配置文件热加载，避免每次改动都重启——重启会打断正在飞行的长请求。

验证部署是否成功

按顺序执行下面四条，全部通过就说明这层推理编排立起来了。

1. 存活检查

```bash

curl -sS -o /dev/null -w '%{http_code}\n' http://127.0.0.1:8080/healthz

```

预期输出：200。

2. 路由命中检查

```bash

curl -sS -D - http://127.0.0.1:8080/v1/chat/completions \

-H "Authorization: Bearer sk-local-test" \

-H "Content-Type: application/json" \

-H "x-task-type: completion" \

-H "x-files-touched: 1" \

-d '{"model":"auto","messages":[{"role":"user","content":"写一个判断字符串是否为空的函数"}]}'

```

预期：响应体里有正常的 choices 内容；响应头里能看到 x-route: inline_completion。把 x-task-type 换成 refactor、x-files-touched 换成 3 再跑一次，x-route 应变成 multi_file_refactor。

3. 指标可读

```bash

curl -sS http://127.0.0.1:8080/metrics | grep -E 'requests_total|latency' | head -20

```

预期：能看到按路由或按上游维度拆分的请求计数和延迟直方图，且计数随着前面的 curl 请求在增长。

4. 压测结果可复现

```bash

python data/bench.py # 并发依次改为 1、4、8、16

```

预期：每个并发档位都能跑完，成功率在可接受范围，并生成 logs/bench_c*.jsonl。同一份配置连跑两次，p95 波动一般在 15% 以内；波动远超这个数，说明上游本身不稳定，需要先固定上游再谈优化。

常见报错与解决

1. {"error":{"message":"invalid_api_key"}}，HTTP 401

原因：客户端把上游的 Key 直接发给了网关，或者网关容器里没注入 UPSTREAM_KEY_* 环境变量，转发时上游鉴权失败。这两种 401 长得一样，要分开看。

解决：

```bash

先确认网关进程能看到 Key（容器内执行用 docker exec）

printenv | grep UPSTREAM_KEY

再确认客户端用的是网关自己的 Key

curl -sS -o /dev/null -w '%{http_code}\n' http://127.0.0.1:8080/v1/chat/completions \

-H "Authorization: Bearer sk-local-test" -H "Content-Type: application/json" \

-d '{"model":"auto","messages":[{"role":"user","content":"ping"}]}'

```

如果 printenv 为空，重启服务时补上环境变量；如果客户端返回 401，检查 Agent 侧是不是漏配了 base_url。

2. HTTP 429 rate_limit_exceeded，或压测到高并发时错误率陡增

原因：网关并发上限设得比上游配额高，或者压测脚本里 CONCURRENCY 加得太快。

解决：把 concurrency 调低到上游配额以内，并保留重试退避：

```yaml

concurrency:

small-coder: 8

defaults:

max_retries: 3

retry_on: [429, 500, 502, 503]

```

同时确认压测脚本设置了信号量（本文脚本里的 asyncio.Semaphore），不要让 asyncio.gather 把上千个任务一次性打出去。

3. context_length_exceeded 或 maximum context length is N tokens，HTTP 400

原因：路由的匹配条件只看了 task_type，没看 context_tokens，一个超大仓库的上下文被塞给了窗口较小的模型。

解决：给每条路由补上上下文上限，并在网关侧开启截断：

```yaml

routes:

• name: inline_completion

match:

task_type: completion

context_tokens: {lte: 4000} # 关键：加窗口约束

budget:

max_context_tokens: 128000 # 全局兜底

```

配合 Agent 侧做上下文裁剪：只带相关文件和签名，不要整仓塞进去。

4. httpx.ConnectError: [Errno 111] Connection refused

原因：网关没起来，或容器端口没映射出来，或监听地址写成了 127.0.0.1（容器内监听回环，宿主机连不上）。

解决：

```bash

ss -lntp | grep 8080 # 宿主机上看端口在不在

docker ps --format '{{.Names}}\t{{.Ports}}' # 容器方式看端口映射

```

listen 要写成 0.0.0.0:8080，容器启动加 -p 8080:8080。

5. 启动时报 yaml.scanner.ScannerError: mapping values are not allowed here

原因：YAML 缩进不一致，或者中文冒号/全角空格混进去了。从聊天窗口复制配置最容易踩这个坑。

解决：

```bash

python -c "import yaml;yaml.safe_load(open('config/magnitude.yaml'));print('yaml ok')"

grep -nP '[\x{3000}\x{FF1A}]' config/magnitude.yaml # 找全角空格与全角冒号

```

6. x-route 一直返回 fallback 的路由名

原因：请求头没带上，或者值的拼写和路由表里的不一致（比如写了 Completion 而配置是 completion）。匹配是区分大小写的。

解决：

```bash

用 -v 打印实际发出的请求头

curl -v http://127.0.0.1:8080/v1/chat/completions -H "x-task-type: completion" ...

```

统一在 Agent 侧的包装函数里定义任务类型的枚举，不要在各处手写字符串。

后续维护

配置与版本管理：路由表、压测脚本、优化脚本放同一个仓库，路由表改动走 PR。每次上线记录三项——改动内容、压测对比结果、回滚 commit。数据集的脱敏规则也一起进仓库，方便审计。

备份：需要定期备份的是路由表、压测数据集、以及网关的请求日志。日志按天切割并设保留期（比如 30 天），原始 prompt 如果落盘必须脱敏并限制访问权限。指标数据的保留期在 Prometheus 侧配置，长于日志即可。数据库有的话按常规做全量加增量。

升级：升级网关或上游模型前，固定依赖版本（容器镜像用 digest 固定，Python 依赖用锁文件）。升级流程固定为：staging 跑一遍完整压测集 → 和基线表逐项对比 → 通过再上生产。模型侧也一样，上游模型静默更新是常见事故源，建议在响应头或日志里记录上游返回的模型标识，出问题时能对上号。

日志与监控：日志统一成结构化 JSON，至少包含 request_id、route、upstream、model、input_tokens、output_tokens、latency_ms、ttft_ms、retry_count、status。监控告警建议盯这几个：

• 各路由的 p95 延迟，超过预算即告警；
• 上游 429 占比，超过 1% 说明并发上限需要回调；
• fallback 命中率，持续上升说明路由匹配条件该更新了；
• 单任务成本额度消耗速度，按天对比，异常上涨通常意味着某类任务被路由到了强模型；
• 质量指标（diff 采纳率、单测通过率），这是唯一能证明"优化"没走偏的信号。

定期重放：每周用同一套压测集跑一次回归，结果和上周对比，形成一条长期的延迟与成本曲线。路由表每次变更前必须跑这套回归。上游模型有更新时，也用它来确认行为没有变化。