这篇能做出什么
先说结果。照着这篇敲完，你会得到一个能跑起来的最小系统，包含四样东西：
1. 统一的模型适配层。同一个 chat() 函数签名，传 "big" 走重模型，传 "ember" 走 Ember。切换只改一个环境变量，业务代码不动。
2. 一个工具路由器。把工具清单（JSON Schema）和当前这一步的任务描述交给 Ember，由它决定调哪个工具、传什么参数。模型只负责「选」，不负责「想」。
3. 两段式智能体循环。重模型当规划器（planner），把用户请求拆成几个明确动作；Ember 当执行器（executor），一步一个动作地调工具。执行段不烧重模型的 token。
4. 一张自己跑出来的对比表。同一批任务，分别在「全重模型」和「重模型规划 + Ember 执行」两种配置下跑，记录端到端延迟、输入输出 token，再乘你账号里的单价，得到单任务成本。
举个具体场景，方便你对齐预期：夜里告警响了，值班同学丢给智能体一句话——「checkout-api 的 P95 延迟刚才飙到 1.2 秒，帮我看看是不是下游超时，如果确认是下游问题就重启一下」。
这件事拆开来是三个动作：查指标、搜日志、重启服务。三个动作都不需要「思考」，需要的是准确地把参数填对、按顺序执行、把结果如实带回。这种段落就是典型的「高频执行段」，也正是把模型降档到 Ember 这类轻量执行模型最划算的地方。真正需要判断的只有「要不要重启」这一步——那一步留在规划器里。
贯穿全文的一个前提：Ember 通过 OpenAI 兼容的 Chat Completions 接口对外提供服务，这是目前多数模型服务商的默认形态。如果你拿到的官方 SDK 形状不同，改动集中在第 2 步构造 client 的那几行，后面的工具路由和循环逻辑完全不用动。base_url、模型标识（model name）、是否支持 tool use / function calling、上下文长度、计费口径，一律以官方文档当前版本为准，不要凭记忆或凭本文写死。
---
前置条件清单
动手前确认这几项，缺一项后面都会卡：
- [ ] 一个能调通 Ember 的账号和 API Key，以及控制台里那个准确的模型标识字符串（
model参数用的那个，不是产品名） - [ ] 一个重模型的账号和 Key（用来做规划和兜底）
- [ ] Python 3.9 以上
- [ ]
pip install openai python-dotenv（不锁版本，装当前版本即可） - [ ] 能访问外网的环境
- [ ] 你自己的计费页：输入 token 单价、输出 token 单价。这两个数字本文不会替你猜，最后算成本时要用
- [ ] 三个能真实调用的工具函数。本文用假的监控/日志/重启函数占位，形状对上了就能替换
---
第 1 步：先跑通一次最小调用
不要一上来就写循环。先用十行代码确认「Key 对、地址对、模型标识对」。新建 .env：
```bash
.env —— 所有值都从你自己的控制台复制，不要手打
BIG_API_KEY=sk-xxxx
BIG_BASE_URL=https://your-provider.example.com/v1
BIG_MODEL=your-heavy-model-id
EMBER_API_KEY=sk-yyyy
EMBER_BASE_URL=https://your-ember-endpoint.example.com/v1
EMBER_MODEL=your-ember-model-id
```
冒烟测试 smoke.py：
```python
smoke.py
import os
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
client = OpenAI(
api_key=os.environ["EMBER_API_KEY"],
base_url=os.environ["EMBER_BASE_URL"],
)
resp = client.chat.completions.create(
model=os.environ["EMBER_MODEL"],
messages=[{"role": "user", "content": "只回复两个字：就绪"}],
temperature=0,
)
print(resp.choices[0].message.content)
print(resp.usage)
```
```bash
python smoke.py
```
看到「就绪」和一段 usage，说明链路通了。如果这一步就报 404，八成是 base_url 少了或多了 /v1；报 401 就是 Key 不对；报 model not found 就是模型标识抄错——这三种错误占了新手期的大部分时间，先在这一步排干净，别带进后面的循环里。
---
第 2 步：封一层模型适配层
后面的代码要频繁切换重模型和 Ember，把 client 和模型名收进一个字典，业务侧只认 tier 这个名字。
```python
llm.py
import os
import time
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
_CLIENTS = {
"big": OpenAI(api_key=os.environ["BIG_API_KEY"], base_url=os.environ["BIG_BASE_URL"]),
"ember": OpenAI(api_key=os.environ["EMBER_API_KEY"], base_url=os.environ["EMBER_BASE_URL"]),
}
_MODELS = {
"big": os.environ["BIG_MODEL"],
"ember": os.environ["EMBER_MODEL"],
}
def chat(tier: str, messages: list, tools: list | None = None, **kwargs) -> dict:
"""统一调用入口。返回统一结构，方便后面统一埋点。"""
payload = {"model": _MODELS[tier], "messages": messages, **kwargs}
if tools:
payload["tools"] = tools
t0 = time.perf_counter()
resp = _CLIENTS[tier].chat.completions.create(**payload)
latency_ms = (time.perf_counter() - t0) * 1000
msg = resp.choices[0].message
usage = resp.usage
return {
"tier": tier,
"text": msg.content or "",
"tool_calls": [tc.model_dump() for tc in (msg.tool_calls or [])],
"prompt_tokens": getattr(usage, "prompt_tokens", 0) or 0,
"completion_tokens": getattr(usage, "completion_tokens", 0) or 0,
"latency_ms": round(latency_ms, 1),
}
```
注意 tool_calls 统一转成了普通 dict。不同 SDK 版本返回的是对象，直接丢进 messages 里再发回去容易序列化失败，转一次省很多事。
---
第 3 步：把工具清单交给 Ember
工具描述的质量，直接决定路由准不准。写工具 schema 有三条经验：description 写「什么时候该用它」，不是「它是什么」；参数 description 给例子；别把可选参数写得太随意，模型会瞎填。
```python
tools.py
import json
def query_metrics(service: str, window: str = "5m") -> dict:
"""占位实现。替换成你真实的监控查询。"""
return {"service": service, "window": window,
"cpu": 0.82, "error_rate": 0.031, "p95_ms": 1240}
def search_logs(service: str, keyword: str, limit: int = 20) -> dict:
return {"service": service, "keyword": keyword, "hits": [
{"ts": "2025-01-01T00:00:00Z", "level": "ERROR",
"msg": f"upstream timeout calling payment-gw ({keyword})"}
][:limit]}
def restart_service(service: str) -> dict:
return {"service": service, "action": "restart", "ok": True}
REGISTRY = {
"query_metrics": query_metrics,
"search_logs": search_logs,
"restart_service": restart_service,
}
TOOL_SCHEMAS = [
{
"type": "function",
"function": {
"name": "query_metrics",
"description": "查询某个服务最近一个时间窗口的 CPU、错误率、P95 延迟。当需要判断服务是否异常、异常到什么程度时调用。",
"parameters": {
"type": "object",
"properties": {
"service": {"type": "string", "description": "服务名，例如 checkout-api"},
"window": {"type": "string", "description": "时间窗口，例如 5m、1h。默认 5m"},
},
"required": ["service"],
},
},
},
{
"type": "function",
"function": {
"name": "search_logs",
"description": "在指定服务的日志里按关键词搜索。当需要确认错误的具体原因、找到上下游是哪个依赖出问题时调用。",
"parameters": {
"type": "object",
"properties": {
"service": {"type": "string", "description": "服务名，例如 checkout-api"},
"keyword": {"type": "string", "description": "搜索关键词，例如 timeout、OOM、payment-gw"},
"limit": {"type": "integer", "description": "返回条数上限，默认 20"},
},
"required": ["service", "keyword"],
},
},
},
{
"type": "function",
"function": {
"name": "restart_service",
"description": "重启一个服务实例。属于有副作用的操作，只有在已经确认服务本身异常、且用户明确要求时才调用。",
"parameters": {
"type": "object",
"properties": {
"service": {"type": "string", "description": "服务名，例如 checkout-api"},
},
"required": ["service"],
},
},
},
]
```
工具数量建议先控制在 10 个以内。工具越多，轻量执行模型选错的概率越高——这不是 Ember 特有的问题，是所有小参数模型的共性。超过 10 个就先按业务域分组，路由两次。
---
第 4 步：写执行循环
执行器的 system prompt 只有一件事要说清楚：你只做当前这一步。不要让它规划，不要让它总结整件事，那会诱导它编造没查过的结论。
```python
executor.py
import json
from llm import chat
from tools import REGISTRY, TOOL_SCHEMAS
DANGEROUS = {"restart_service"}
EXECUTOR_SYS = (
"你是一个执行器。只完成当前这一步：根据任务描述调用合适的工具，"
"拿到工具结果后用一句话陈述事实。"
"不要规划后续步骤，不要推测工具没有返回的数据，"
"如果缺少必要参数就不要调用工具，直接说明缺什么。"
)
def run_tool(name: str, args: dict) -> dict:
fn = REGISTRY.get(name)
if fn is None:
return {"error": f"unknown tool: {name}"}
try:
return fn(**args)
except Exception as e:
return {"error": f"{type(e).__name__}: {e}"}
def execute(task: str, tier: str = "ember", max_steps: int = 6,
allow_dangerous: bool = False) -> dict:
messages = [
{"role": "system", "content": EXECUTOR_SYS},
{"role": "user", "content": task},
]
trace = []
for step in range(max_steps):
out = chat(tier, messages, tools=TOOL_SCHEMAS, temperature=0)
trace.append(out)
if not out["tool_calls"]:
return {"answer": out["text"], "trace": trace, "steps": step + 1, "error": None}
messages.append({
"role": "assistant",
"content": out["text"] or None,
"tool_calls": out["tool_calls"],
})
for tc in out["tool_calls"]:
name = tc["function"]["name"]
raw = tc["function"]["arguments"]
args = json.loads(raw) if isinstance(raw, str) else (raw or {})
if name in DANGEROUS and not allow_dangerous:
result = {"error": "该工具有副作用，需要人工确认，已拒绝执行"}
else:
result = run_tool(name, args)
messages.append({
"role": "tool",
"tool_call_id": tc["id"],
"content": json.dumps(result, ensure_ascii=False),
})
return {"answer": None, "trace": trace, "steps": max_steps, "error": "step_limit"}
```
两个细节值得单独说：
tc["function"]["arguments"] 在有的实现里是 JSON 字符串，有的直接给 dict，所以这里做了 isinstance 判断。这行兼容代码建议永久保留，换供应商时会救你一次。
tool 消息必须带 tool_call_id，并且紧跟在带 tool_calls 的 assistant 消息之后。顺序错了或者 id 对不上，接口会直接报 400。这是自研循环里出现频率最高的报错。
---
第 5 步：接成两段式智能体循环
规划器用重模型，输出严格 JSON；执行器用 Ember，一步一调。
```python
agent.py
import json
from llm import chat
from executor import execute
PLANNER_SYS = """你是规划器。把用户请求拆成 1-4 个可独立执行的小步骤。
每个步骤必须是一个明确的动作，能被下面这些工具之一完成，或是一次纯文本总结。
可用工具：query_metrics、search_logs、restart_service。
只输出 JSON，不要 markdown 围栏，不要解释。格式：
{"steps": [{"id": 1, "action": "查询 checkout-api 近 5 分钟指标", "tool_hint": "query_metrics"}]}
有副作用的重启类动作，必须单独成一步，并在 action 里写明触发条件。"""
def plan(request: str) -> tuple[list, dict]:
out = chat("big", [
{"role": "system", "content": PLANNER_SYS},
{"role": "user", "content": request},
], temperature=0)
text = out["text"].strip()
if text.startswith("```"):
text = text.split("```")[1].removeprefix("json").strip()
return json.loads(text)["steps"], out
def run_agent(request: str, allow_dangerous: bool = False) -> dict:
steps, plan_out = plan(request)
results = []
for s in steps:
task = (
f"当前这一步：{s['action']}\n"
f"用户原始请求（仅作上下文）：{request}\n"
f"请调用合适的工具完成当前这一步。"
)
r = execute(task, tier="ember", allow_dangerous=allow_dangerous)
if r["answer"] is None: # 执行器卡住或超步数，升级兜底
r = execute(task, tier="big", allow_dangerous=allow_dangerous)
results.append({"step": s, "result": r, "escalated": True})
else:
results.append({"step": s, "result": r, "escalated": False})
return {"plan": steps, "results": results, "planner": plan_out}
```
跑一下：
```python
main.py
import json
from agent import run_agent
out = run_agent(
"checkout-api 的 P95 刚才飙到 1.2 秒，帮我看看是不是下游超时，"
"如果确认是下游问题就重启一下这个服务。",
allow_dangerous=False, # 第一次跑先关掉副作用工具
)
print(json.dumps(out, ensure_ascii=False, indent=2))
```
第一次跑建议 allow_dangerous=False，让「重启」这一步被拦下来，你就能看到拒绝分支长什么样。确认链路没问题再打开。
---
第 6 步：埋点，把延迟和成本算出来
这一步是整篇的重点。降档能不能成立，取决于数据，不取决于感觉。
```python
bench.py
import time
from agent import run_agent
from executor import execute
from llm import chat
from dotenv import load_dotenv
import os
load_dotenv()
IN_PRICE = float(os.environ["EMBER_IN_PRICE"]) # 每百万输入 token 单价，从你的计费页抄
OUT_PRICE = float(os.environ["EMBER_OUT_PRICE"])
TASKS = [
"查一下 checkout-api 近 5 分钟的指标",
"在 checkout-api 日志里搜 timeout 相关的错误",
"查一下 payment-gw 的错误率",
……建议准备 15-30 条你真实会遇到的执行段任务
]
def cost(tier, p_tok, c_tok):
if tier == "big":
ip, op = float(os.environ["BIG_IN_PRICE"]), float(os.environ["BIG_OUT_PRICE"])
else:
ip, op = IN_PRICE, OUT_PRICE
return p_tok / 1_000_000 * ip + c_tok / 1_000_000 * op
def sum_tokens(trace):
return (sum(t["prompt_tokens"] for t in trace),
sum(t["completion_tokens"] for t in trace),
sum(t["latency_ms"] for t in trace))
for label, tier in [("全重模型", "big"), ("Ember 执行", "ember")]:
lat, p_tok, c_tok, spend = [], 0, 0, 0.0
for task in TASKS:
t0 = time.perf_counter()
r = execute(task, tier=tier)
lat.append((time.perf_counter() - t0) * 1000)
p, c, _ = sum_tokens(r["trace"])
p_tok += p
c_tok += c
spend += cost(tier, p, c)
lat.sort()
n = len(lat)
print(f"{label}: p50={lat[n//2]:.0f}ms p95={lat[int(n*0.95)-1]:.0f}ms "
f"tokens={p_tok}/{c_tok} 单任务成本≈{spend/n:.6f}")
```
输出会长这样（下面数字是格式示例，不是实测值，你必须自己跑）：
```
全重模型: p50=3900ms p95=7200ms tokens=41000/6200 单任务成本≈0.0XXXXX
Ember 执行: p50=1150ms p95=2300ms tokens=12000/1800 单任务成本≈0.00XXX
```
真正要看的是三个比值：
- 延迟比：p50 和 p95 都要看。执行段的收益主要来自输出短、模型小，通常 p95 改善比 p50 更明显。
- 口径比：把总 token 数除以任务数，得到「单任务 token」。如果 Ember 那一栏的 token 反而更高，说明你的 system prompt 太长或者工具 schema 太啰嗦，先精简再谈降档。
- 成本比：拿上面算出来的数除以全重模型那一栏。注意成本比不等于 token 比，因为两边的输入输出单价不同。单价永远以你自己的计费页为准。
还有一笔隐性账要记：如果 Ember 的失败率（answer is None 或升级兜底的比例）超过某个你能接受的水平，升级那部分等于双倍付费。所以 bench 里一定要统计 escalated 的条数，把它加到成本里。
---
常见坑与排错
1. tool_calls 发回去报 400
最常见。检查两点：tool_call_id 是否和 assistant 消息里的一致；tool 消息是否紧跟在对应的 assistant 消息后面。中间插了别的消息就会挂。
2. 参数解析炸在 json.loads
轻量模型偶尔会输出带尾逗号或者单引号的「近似 JSON」。加一层容错：解析失败就把原始字符串塞回 tool 消息，内容写 {"error": "参数不是合法 JSON，请重新生成"}，让它自己改一次，比直接抛异常优雅得多。
3. 模型反复调同一个工具
通常是工具返回里没有它想要的信息。检查 query_metrics 的返回结构，是不是缺少它判断所需的字段。工具返回要给「结论性字段」（error_rate、p95_ms），不要只给一堆原始时间序列。
4. 上下文越来越长，延迟越来越高
执行循环里每次都要把完整历史发一遍。控制手段：max_steps 设小（4-6 步够用）、工具返回只保留关键字段、超步数直接升级而不是继续重试。
5. 有副作用的工具被误触发
DANGEROUS 白名单是硬性闸门。重启、删除、扩缩容、发消息这类动作，默认拒绝，走人工确认。不要指望 prompt 里写一句「谨慎操作」就能拦住。
6. 降档之后「变笨」了
先分清是模型问题还是任务切分问题。执行段的任务必须是动作明确、参数可枚举、结果可验证的。像「帮我评估一下这套架构要不要拆」这种，塞给执行器一定失败——它不是执行任务，是决策任务，应该留在规划器。
7. 超时和重试
对查询类工具可以自动重试一次；对重启类工具不要自动重试。非幂等操作的重复执行比失败本身更麻烦。
8. 不同供应商的 tool use 支持粒度不一样
有的支持并行 tool calls，有的只支持单个。如果你的 Ember 端点不支持 tools 参数，退路是让它输出结构化的 JSON（{"tool": "...", "args": {...}}），你自己解析——路由逻辑一样成立，只是少了原生约束。
---
下一步建议
跑通之后，按这个顺序往下推：
第一步，扩任务集。 15 条任务跑出来的数字只能看趋势。攒到 50-100 条你真实遇到的执行段任务，覆盖查询类、搜索类、写入类，才谈得上「这条链路稳了」。
第二步，做灰度路由。 不要一次性全量切。按任务类型分流：查询类先切 Ember，写入类先留重模型。用 escalated 比例做监控指标，超过阈值自动回切。
第三步，把提示词当配置管。 执行器的 system prompt 和工具 schema 一起版本化。改 schema 就等于改模型行为，需要跟着跑一遍回归。
第四步，接可观测性。 把每次 chat() 的 tier / latency_ms / prompt_tokens / completion_tokens / tool_name 打到你的 trace 系统里。没有这份数据，后面所有「要不要再降一档」的讨论都是拍脑袋。
第五步，考虑缓存。 执行段里有大量重复查询（同一个服务、同一个时间窗口）。在 run_tool 外面套一层短 TTL 缓存，延迟和成本会同时下降——这一步的收益往往比换模型还大。
最后提醒一句：轻量执行模型的价值不在「便宜」，在「够用且快」。什么时候该降档、降到什么程度，答案永远在你自己跑出来的那张表里，不在任何人的推荐里。所有涉及版本、接口形态、计价的地方，以官方文档当前版本为准。