Strands Decider 2B：给智能体配决策小模型

这篇能做出什么

读完之后，你会得到一个能跑起来的小系统：一个 2B 级别的决策模型在本机（或一台小规格机器）上提供 OpenAI 兼容接口，智能体在把请求交给主模型之前，先问它一句"这活该走哪条路"，或者"这 30 条候选里留哪 8 条"。主模型只负责它真正擅长的事：写长文案、做多步推理、生成最终答复。

具体交付物有五样：

1. 一个 decide() 函数，输入一段文本，输出经过校验的 JSON，解析失败率可控。

2. 一个路由分支，把用户请求分到 search / database / calculator / direct / human 几条路上。

3. 一个筛选器，把候选列表从几十条压到几条，再交给主模型。

4. 一个对照脚本，量出小模型和主模型在延迟、token 消耗上的差别。

5. 一套兜底策略：置信度低、格式不合法、超时，一律升级给主模型。

用一个贯穿例子说明场景：一个客服智能体每天接 20 万次请求，其中大约七成是"查订单状态""改收货地址""问退换货政策"这类归属很明确的请求。这些请求的判断依据只有几个关键词和一张意图表，用主模型做判断，等于每次都用大炮打蚊子。

前置条件清单

• 一台能跑推理的机器。2B 级别模型经量化后，消费级显卡甚至纯 CPU 也能跑，只是并发能力有限；具体显存需求看模型卡说明。
• Python 3.10 以上（具体下限以官方文档当前版本为准）。
• 一个 OpenAI 兼容的推理服务，任选其一：vLLM、Ollama、llama.cpp、TGI、SGLang。本文示例用 OpenAI 兼容的 /v1/chat/completions 路径，换实现时只改 base_url。
• 一个 Agent 编排框架。本文以 Strands Agents 为例，它通过"模型提供方"接口接入模型，具体类名与参数以官方文档当前版本为准；用别的框架同样能套这套思路。
• 主模型的 API key（走云端）。
• 可选但强烈建议：一份 50～200 条的决策评估集，每条包含"输入 + 正确选项"。

模型名称、许可协议、量化版本这类信息，不要凭记忆写进代码，去模型卡或仓库首页确认当前版本。

步骤一：先划清哪些决策该交给小模型

不要一上来就跑模型。先拿一天的真实请求日志，把"需要判断"的地方标出来，然后按三条标准筛：

典型的小决策：意图路由、语言识别、是否要检索、候选去重、结果排序、敏感词初筛、要不要转人工、把口语化的地址归一化成字段。

典型的"看着小其实不小"：判断用户是否在表达不满、判断两段文本是否语义矛盾、判断一段代码有没有安全漏洞。这几类要么需要世界知识，要么需要细粒度推理，小模型容易给出看似合理但错的答案。

步骤二：把决策模型跑起来

以 vLLM 为例，启动一个 OpenAI 兼容服务：

```bash

python -m vllm.entrypoints.openai.api_server \

--model <MODEL_ID_OR_LOCAL_PATH> \

--served-model-name decider \

--port 8000 \

--max-model-len 4096 \

--gpu-memory-utilization 0.85

```

如果机器上没有 GPU，用 Ollama 更省事：

```bash

ollama pull <MODEL_TAG>

ollama serve

默认监听 http://127.0.0.1:11434，OpenAI 兼容路径为 /v1

```

起来之后先做一次连通性检查，别等到业务代码里才发现端口不对：

```bash

curl -s http://127.0.0.1:8000/v1/models | head -c 400

```

返回里能看到 decider 这个 id，就说明服务正常。

步骤三：写一个"只会输出 JSON"的决策函数

决策模型和生成模型最大的区别是：它的输出必须能被程序消费。所以第一件事是定 schema，第二件事是强制约束解码。

```python

decider_client.py

import json

import os

import time

from openai import OpenAI

BASE_URL = os.environ.get("DECIDER_BASE_URL", "http://127.0.0.1:8000/v1")

MODEL = os.environ.get("DECIDER_MODEL", "decider")

client = OpenAI(base_url=BASE_URL, api_key="EMPTY", timeout=3.0)

ROUTE_SCHEMA = {

"type": "object",

"properties": {

"route": {

"type": "string",

"enum": ["search", "database", "calculator", "direct", "human"],

},

"confidence": {"type": "number", "minimum": 0, "maximum": 1},

"reason": {"type": "string", "maxLength": 60},

},

"required": ["route", "confidence", "reason"],

"additionalProperties": False,

}

SYSTEM_PROMPT = (

"你是一个请求路由器。根据用户消息选择唯一的 route。"

"规则：涉及订单号、物流、账户数据的查询选 database；"

"涉及政策、条款、说明文档的选 search；"

"需要算数的选 calculator；"

"闲聊寒暄、简单确认选 direct；"

"包含投诉、威胁、要求人工的选 human。"

"confidence 表示你对判断的把握程度，0 到 1 之间。"

"reason 用不超过 20 个汉字说明理由。只输出 JSON。"

)

def decide_route(text: str) -> dict:

resp = client.chat.completions.create(

model=MODEL,

messages=[

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

{"role": "user", "content": text},

],

temperature=0,

max_tokens=128,

response_format={"type": "json_schema", "json_schema": {

"name": "route_decision",

"schema": ROUTE_SCHEMA,

}},

)

raw = resp.choices[0].message.content

data = json.loads(raw)

无论服务端有没有强约束，客户端都要再校验一次

assert data["route"] in ROUTE_SCHEMA["properties"]["route"]["enum"]

return data

```

如果服务端不支持 response_format 里的 json_schema，改用服务端自己的约束解码参数：vLLM 的 extra_body={"guided_json": ROUTE_SCHEMA}、Ollama 的 format=ROUTE_SCHEMA，具体字段名以各自官方文档为准。原理一样：在采样阶段就把非法 token 屏蔽掉，而不是生成完再靠正则去捞。

步骤四：把决策接进编排里

关键点是：小模型不做生成，只做判断；主模型不浪费算力在判断上。下面是 Strands Agents 风格的编排骨架：

```python

agent_flow.py

from decider_client import decide_route

from strands import Agent # 具体导入路径以官方文档为准

main_agent = Agent(system_prompt="你是客服助手，回答要简洁、有依据。")

def handle(user_input: str) -> str:

d = decide_route(user_input)

if d["confidence"] < 0.6:

兜底：不确定就交给主模型自己决定

return str(main_agent(user_input))

route = d["route"]

if route == "database":

rows = query_order_db(user_input) # 你自己的工具

return str(main_agent(f"根据以下订单数据回答：\n{rows}\n\n用户问题：{user_input}"))

if route == "search":

docs = search_knowledge_base(user_input, top_k=5)

return str(main_agent(f"根据以下资料回答：\n{docs}\n\n用户问题：{user_input}"))

if route == "calculator":

return str(main_agent(f"请先调用计算工具再回答：{user_input}"))

if route == "human":

return escalate_to_human(user_input)

return str(main_agent(user_input))

```

这样一次"查订单"请求，主模型只被调用一次，且拿到的已经是干净的上下文，不需要先思考"该不该查数据库"。

筛选场景同理，把列表切片喂给小模型，让它打分：

```python

FILTER_SCHEMA = {

"type": "object",

"properties": {

"keep_ids": {"type": "array", "items": {"type": "string"}, "maxItems": 8},

"drop_reason": {"type": "string", "maxLength": 80},

},

"required": ["keep_ids", "drop_reason"],

"additionalProperties": False,

}

```

注意列表长度控制在小模型AI 词典：上下文窗口">上下文窗口内。30 条超了就先按规则粗筛到 15 条，再交给它。别指望 2B 模型在 200 条候选里挑出最优解。

步骤五：加缓存和兜底

决策请求有个特点：输入高度重复。"我的快递到哪了"和"快递怎么还没到"会映射到同一个 route。做一层归一化缓存，命中率往往不低。

```python

import hashlib

from functools import lru_cache

def norm(text: str) -> str:

return "".join(text.split()).lower()

@lru_cache(maxsize=10000)

def cached_route(text: str) -> str:

import json as _json

return _json.dumps(decide_route(text), ensure_ascii=False, sort_keys=True)

def decide_route_safe(text: str) -> dict:

import json as _json

try:

return _json.loads(cached_route(norm(text)))

except Exception:

任何异常：超时、解析失败、服务挂了

return {"route": "direct", "confidence": 0.0, "reason": "fallback"}

```

置信度阈值不要拍脑袋定。拿评估集跑一遍，看看在哪个阈值下"升级率"和"错误率"的乘积最小。0.6 只是个起点，不同模型、不同 prompt 差别很大。

步骤六：做成本与延迟对照

这才是决定"到底值不值得拆"的依据。写个对照脚本，固定同一批 100 条真实请求，分别用两种方式跑：

```python

import time, statistics

def bench(fn, cases, warmup=5):

for c in cases[:warmup]:

fn(c) # 预热，避开首次加载

lat = []

for c in cases:

t0 = time.perf_counter()

fn(c)

lat.append((time.perf_counter() - t0) * 1000)

lat.sort()

return {

"p50_ms": lat[len(lat) // 2],

"p95_ms": lat[int(len(lat) * 0.95)],

"avg_ms": statistics.mean(lat),

}

print("small:", bench(decide_route, cases))

print("main :", bench(lambda c: main_agent(c), cases))

```

成本侧，把每次调用的输入 token 数和输出 token 数记下来，再按各家官方定价页面的当前价格换算。要算的不只是 API 费用，还有：

• 小模型常驻占用的机器成本（按月摊到每次调用）。
• 因为多了一次网络往返而增加的端到端延迟。
• 因为分类错误导致的返工成本——这条常常被忽略，却往往是决定性的。

一个务实的判断：如果决策类调用占总调用量的比例低于某个水平，或者主模型本身的延迟已经在可接受范围，那拆出来的收益有限，不如先不拆。拆分的价值来自"高频 + 低复杂度 + 主模型调用贵"这三件事同时成立。

常见坑与排错

输出解析失败。最常见的原因是 prompt 里说了"只输出 JSON"，但没在解码层约束。自然语言指令对 2B 模型来说不够硬。解决办法是上约束解码，客户端再校验一遍。

同一输入结果飘。把 temperature 设成 0，并且尽量固定 prompt 顺序。如果服务端支持 seed，也一并固定。决策系统的可复现性比多样性重要得多。

枚举值写成中文或近义词。模型可能返回"查询数据库"而不是 database。枚举值一律用英文小写下划线，中文映射放在代码里做。

上下文超了不看。小模型窗口小，把 50 条候选直接塞进去，要么被截断要么报错。做法是先规则粗筛，或分批打分再合并。

只看平均延迟，不看长尾。路由的意义在于削掉主模型的调用，如果小模型的 p95 比主模型的 p50 还高，整体体验反而变差。至少看 p95，最好看 p99。

服务冷启动被算进性能里。模型首次加载可能花几十秒。压测前先预热，生产环境保持常驻，别做按需拉起。

模型悄悄换了版本。把模型 id 和量化版本写进配置并记录到日志，模型升级后必须重跑评估集，别让行为变化无声无息地流到线上。

没有评估集。没有评估集就无法回答"小模型能不能替代"。从真实日志里抽 100 条，人工标好正确选项，存成 JSONL，这是整个方案里投入产出比很高的一件事。

把需要推理的决策也交给它。2B 模型在"从固定选项里选一个"上表现通常够用，在"判断这段话是否构成投诉"上就不一定。发现某类决策准确率上不去，就把它移回主模型，不要硬扛。

下一步建议

1. 先给现有决策点建评估集和回归测试，再谈替换。没有测试的重构是赌博。

2. 加一层决策日志，记录输入、输出、置信度、是否兜底、最终是否被用户接受。这份日志既能调阈值，也能当微调数据。

3. 用日志做 A/B：一半流量走小模型路由，一半走主模型自决，对比任务完成率和端到端延迟。

4. 如果小模型在某几类决策上始终差一点，可以拿主模型的决策记录做监督微调，把这几类补上。

5. 把多个决策合并成一次调用。比如"路由 + 是否检索 + 语言"用一个 prompt 一次输出，能省掉往返开销。

6. 最后再考虑扩到更多决策类型，比如把结果排序、摘要长度控制这类判断也迁过来。