用 Decision-1 做快决策层：意图分诊与工具路由

很多智能体跑得慢、花得贵，问题不在主模型，而在于每一条"今天天气怎么样"都要走一遍完整的大模型规划循环。这一层快决策的意义，就是把廉价、模式化、可以枚举的请求在主循环之前拦下来，只把真正需要推理的请求交给主模型。

这篇能做出什么

跟着做完，你会得到一个可以挂在任何智能体前面的 decide() 决策层，它在主循环之前先跑一次小模型调用，返回一个结构化结果：

• intent：意图标签，比如 faq、query_data、change_config、chitchat、complaint
• route：下一步走哪条路，比如 direct_answer、tool:kb_search、planner、clarify、human
• confidence：0 到 1 的置信度

主循环拿到这个结果之后，按路由分派。简单请求直接命中模板或工具，复杂请求才进主模型，置信度低就先反问用户，工具失败再回落到主模型并带上错误上下文。

同时你会拿到一套可复用的测量脚本，输出三组数字：分诊层的延迟分位（p50 / p95）、每次调用的 token 用量、以及路由准确率。这三组数字决定了这个决策层到底值不值得留在线上。

前置条件清单

• 一个可以调用的 Decision-1 入口（Azure AI Foundry、或任何兼容 OpenAI Chat Completions 协议的服务，部署名与调用方式以官方文档当前版本为准）
• Python 3.10 及以上，安装 openai SDK
• 一个已有的智能体主循环，哪怕只有几十行伪代码也能接
• 一份 50 到 200 条的小评测集，字段是 text 和人工标注的 gold_route
• 三个环境变量：DECISION_ENDPOINT、DECISION_API_KEY、DECISION_DEPLOYMENT

评测集是整套东西的地基。没有它，你只能靠感觉说"好像变快了"，而感觉在优化这件事上不可靠。评测集从哪来？从真实线上日志里抽样最省事，先抽 50 条，两个人独立标注一遍，不一致的地方讨论到一致。

分步骤

第 1 步：先定 schema，别急着写提示词

分诊层的输出必须是"选择题"，不能是"作文题"。标签集合控制在 8 个以内，超过之后模型的混淆率会明显上升，而且你也维护不过来。

```json

{

"intent": "faq",

"route": "direct_answer",

"confidence": 0.86,

"reason": "问题与帮助中心的固定条目匹配"

}

```

路由值建议分成四类语义：

• direct_answer：有现成模板或缓存可以答
• tool:xxx：需要调一个明确的外部工具
• planner：需要主模型推理、多步规划
• clarify / human：信息不足或需要人工

第 2 步：写一个最小调用封装

先跑通一次调用，别一上来就接进主循环。

```python

decision.py

import json, os, time

from openai import OpenAI

client = OpenAI(

base_url=os.environ["DECISION_ENDPOINT"],

api_key=os.environ["DECISION_API_KEY"],

timeout=float(os.environ.get("DECISION_TIMEOUT", "1.2")),

max_retries=0, # 分诊层自己控重试，不让 SDK 悄悄拖长时间

)

MODEL = os.environ["DECISION_DEPLOYMENT"]

```

max_retries=0 这一行值得单独说。SDK 默认会在超时后自动重试，如果分诊层本来只允许 400 毫秒，自动重试三次就变成了 1.2 秒，比直接走主循环还慢。超时和重试策略必须由你自己掌握。

第 3 步：把决策写成系统提示词里的对照表

提示词的目标不是让模型"聪明"，而是让它"稳定地做选择题"。三条规矩：给枚举清单、给两三个正负样例、明确说不知道就选 planner。

```python

SYSTEM = """你是一个请求分诊器。只输出 JSON，不要输出任何解释文本。

标签定义：

• intent: faq | query_data | change_config | chitchat | complaint | unknown
• route: direct_answer | tool:kb_search | tool:db_query | planner | clarify | human

判断规则：

1. 用户问的是产品帮助文档里的固定问题 → intent=faq, route=direct_answer

2. 用户要求查询具体数值/记录 → intent=query_data, route=tool:db_query

3. 用户要求执行有副作用的操作（改配置、发消息、删数据）→ intent=change_config, route=planner

4. 信息不足以判断 → route=clarify，confidence 不超过 0.5

5. 任何你不确定的场景 → route=planner，不要猜

confidence 是你对 route 判断的把握，0 到 1 的小数。

示例：

输入 {"user_input": "怎么改密码"} → {"intent":"faq","route":"direct_answer","confidence":0.92}

输入 {"user_input": "把生产环境的超时改成 30 秒"} → {"intent":"change_config","route":"planner","confidence":0.88}

输入 {"user_input": "帮我看看"} → {"intent":"unknown","route":"clarify","confidence":0.30}

输出 JSON 字段：intent, route, confidence, reason（不超过 20 字）"""

```

注意第 5 条：把"不确定就走 planner"写进提示词。这是 fail-open 的思路，宁可多花一次主模型的调用，也不要让分诊层自作主张走进错误分支。

第 4 步：实现 decide 函数并强制校验

```python

ALLOWED_ROUTES = {

"direct_answer", "tool:kb_search", "tool:db_query",

"planner", "clarify", "human",

}

def _valid(d):

return (

isinstance(d, dict)

and d.get("route") in ALLOWED_ROUTES

and isinstance(d.get("confidence"), (int, float))

and 0.0 <= float(d["confidence"]) <= 1.0

)

def decide(text: str, ctx: dict | None = None, tools: list | None = None) -> dict:

payload = {

"user_input": text[:600],

"recent_summary": (ctx or {}).get("recent_summary", "")[:300],

"available_tools": tools or ["kb_search", "db_query"],

}

t0 = time.perf_counter()

try:

r = client.chat.completions.create(

model=MODEL,

temperature=0,

max_tokens=120,

messages=[

{"role": "system", "content": SYSTEM},

{"role": "user", "content": json.dumps(payload, ensure_ascii=False)},

],

response_format={"type": "json_object"},

)

d = json.loads(r.choices[0].message.content)

if not _valid(d):

raise ValueError(f"schema invalid: {d}")

d["_latency_ms"] = (time.perf_counter() - t0) * 1000

usage = getattr(r, "usage", None)

d["_prompt_tokens"] = getattr(usage, "prompt_tokens", 0) or 0

d["_completion_tokens"] = getattr(usage, "completion_tokens", 0) or 0

return d

except Exception as e:

fail-open：任何异常都交给主模型

return {

"intent": "unknown", "route": "planner", "confidence": 0.0,

"_error": str(e)[:200],

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

"_prompt_tokens": 0, "_completion_tokens": 0,

}

```

response_format 的具体取值取决于服务端的支持情况，支持结构化输出的用 JSON Schema 约束更好，不支持的退回 json_object 加上自己手动校验。以官方文档当前版本为准。

上下文只喂两样东西：用户输入和一段压缩过的历史摘要。把整个对话历史塞进去，prompt token 会涨，延迟也会涨，而分诊准确率通常不会跟着涨。

第 5 步：插进主循环

```python

def handle_turn(user_input: str, session: dict) -> str:

d = decide(user_input, session)

route = d["route"]

conf = d["confidence"]

if route == "clarify" or conf < 0.6:

return ask_clarifying_question(user_input)

if route == "direct_answer":

answer = lookup_template(user_input) # 缓存或模板库

if answer:

return answer

route = "planner" # 模板没命中，退回主循环

if route.startswith("tool:"):

tool_name = route.split(":", 1)[1]

try:

return call_tool(tool_name, user_input, session)

except Exception as e:

session["last_error"] = str(e) # 带错误上下文回落

route = "planner"

if route == "human":

return handoff_to_human(user_input, session)

return main_agent(user_input, session) # 大模型主循环

```

三个兜底点都在这里：模板没命中退回 planner，工具报错带着错误信息退回 planner，置信度不足先反问。主循环永远是可用的最后一道兜底，这条链路不能断。

第 6 步：把路由表外置成配置

路由规则会频繁变，写死在代码里每次都要发版。

```yaml

routes.yaml

thresholds:

min_confidence: 0.6

decide_timeout_ms: 400

routes:

direct_answer:

handler: template_lookup

on_miss: planner

"tool:kb_search":

tool: kb_search

on_error: planner

"tool:db_query":

tool: db_query

require_permission: [analyst, admin]

on_error: planner

planner:

handler: main_agent

clarify:

handler: ask_clarifying_question

human:

handler: handoff

```

require_permission 这类字段提醒一件事：分诊层只负责分诊，不负责授权。权限校验必须在工具执行之前单独做一遍，不能让模型输出的路由结果绕过权限体系。

第 7 步：跑实测，拿到三组数字

```python

bench.py

import json, statistics

from decision import decide

cases = [json.loads(l) for l in open("eval_set.jsonl", encoding="utf-8")]

rows = []

for c in cases:

d = decide(c["text"])

rows.append({

"text": c["text"],

"gold": c["gold_route"],

"pred": d["route"],

"hit": d["route"] == c["gold_route"],

"latency_ms": d["_latency_ms"],

"pt": d["_prompt_tokens"],

"ct": d["_completion_tokens"],

})

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

n = len(lat)

print("样本数:", n)

print("p50 延迟(ms):", round(lat[n // 2], 1))

print("p95 延迟(ms):", round(lat[min(n - 1, int(n * 0.95))], 1))

print("平均 prompt tokens:", round(sum(r["pt"] for r in rows) / n, 1))

print("平均 completion tokens:", round(sum(r["ct"] for r in rows) / n, 1))

acc = sum(r["hit"] for r in rows) / n

print("路由准确率:", f"{acc:.1%}")

混淆矩阵：看错在哪两类之间

from collections import Counter

conf = Counter((r["gold"], r["pred"]) for r in rows if not r["hit"])

for (g, p), c in conf.most_common(5):

print(f" 应为 {g} → 判为 {p} : {c} 次")

```

下面这张表用来示范记录格式。数值取自一次本地小样本运行（N=120，单区域，普通办公网络），不同评测集、不同网络、不同并发下结果会完全不同，请以你自己跑出来的数字为准。

从这组示例数字能读出三件事。

第一，分诊层的准确率低于主模型是正常的。它的价值不在准确率，而在于用约 1/8 的延迟和约 1/8 的 token 量，把大部分请求挡在主循环之外。示例里 120 条请求有 63 条被 direct_answer 或 tool: 短路，这 63 条的平均端到端耗时从约 2.1 秒降到约 0.24 秒。

第二，成本要按短路率算。假设分诊层每次约 200 token，主模型每次约 1600 token，短路率 52%，那么总 token 量大约降到原来的 48% 加上分诊自身的开销。具体单价以官方页面为准，把 token 数乘上去就是你的账单变化。

第三，准确率要拆开看。整体 92% 看起来很漂亮，但混淆矩阵往往显示错误集中在 clarify 和 planner 之间——这两类错判的代价很小，因为它们最终都会走到主模型。真正要警惕的是 direct_answer 和 planner 之间的错判：把需要操作的请求判成 direct_answer，用户会得到一个答非所问的模板回复。

常见坑与排错

输出不是合法 JSON。 先用 response_format 约束，再加一层自己的 _valid() 校验，校验失败直接 fail-open 走 planner，不要重试超过一次。重试两次以上的收益远低于它带来的延迟抖动。

分诊层越写越大。 提示词从 200 字涨到 2000 字，标签从 5 个涨到 20 个，延迟和成本就回到了主模型的量级。控制在 8 个标签、两三个样例以内，多出来的规则应该放进 routes.yaml 而不是提示词。

超时设置比主循环还长。 分诊层的超时时间应该明显小于主模型的首 token 时间。如果主模型 1 秒开始出字，分诊层给 400 到 800 毫秒是合理区间；设成 5 秒，快决策就变成了慢决策。

把分诊结果当成事实。 分诊层输出的永远是"建议路由"，权限校验、参数校验、副作用确认必须在执行层重新做。特别是 change_config 这类有副作用的意图，建议强制走 planner 并加二次确认。

缓存键不完整。 用 (归一化后的输入, 租户ID, 权限等级) 做键，别只用输入文本。否则 A 用户的问题会命中 B 用户的答案，这是安全事故级别的坑。

评测集是"自己出题自己答"。 用真实日志抽样，标注时两个人独立做，分歧点单独列出来讨论。用自己写的样例测自己调的提示词，准确率必然虚高。

统计口径混乱。 把分诊层耗时、工具耗时、主模型耗时分开记。只记总耗时会让你看不出是分诊层拖慢了还是工具拖慢了，优化的时候会找错方向。

并发下的长尾。 p50 好看不代表体验好，p95 才是用户投诉的来源。压测时至少打到线上峰值的两倍 QPS，观察 p95 和错误率的变化。

下一步建议

先别急着用分诊结果做拦截。上线第一周跑影子模式：分诊层照常调用、照常记录，但路由结果不生效，所有请求仍然走原来的主循环。一周之后对比分诊结果和实际执行路径，你会看到真实的偏差分布，再决定哪些路由可以放心短路。

第二件事是加漂移检测。用户的问法会随季节、随产品改版变化，评测集上的 92% 不代表三个月后还是 92%。每周从线上随机抽 30 条重新标注，跑一遍 bench.py，把准确率画成时间序列。

第三件事是扩到多轮。当前的分诊只看单轮输入加一段历史摘要，多轮场景下"那把它改成 60 秒"这种指代明确的请求容易被判成 clarify。常见的做法是把上一轮的分诊结果一并作为输入，让模型知道当前处在哪个话题里。

最后，等路由表稳定了，再考虑把分诊层从"提示词工程"升级成"小模型微调"。有几千条高质量标注之后，微调版本通常能同时拿到更低的延迟和更高的准确率，但这属于下一阶段的优化，先把这一层跑通、量准、稳住，比什么都重要。