【意图识别】规则、语义路由和 LLM 三级兜底的一次实验
目录
给金融线索业务做一个意图分类器。输入一句话，判断它是 5 个意图里的哪一个，或者不属于任何一个（unknown）：
线索提报、线索全生命周期跟踪反馈、线索分析、产品准入分析、产品知识问答。
单靠一种方案都有明显短板：正则写不全所有说法，语义路由分不清近义意图，直接上大模型又慢又贵。所以用了三级级联，便宜的先跑，跑不动的交给下一级。
三级结构
flowchart TD
A["用户输入"] --> B{"L1 规则匹配"}
B -->|"命中"| C["直接返回意图"]
B -->|"未命中"| D{"L2 语义路由<br/>相似度 ≥ 阈值?"}
D -->|"命中"| E["返回意图"]
D -->|"未命中 (name=None)"| F{"L3 大模型兜底"}
F -->|"识别出意图"| G["返回意图"]
F -->|"无法判断"| H["返回 unknown"]
L1：正则
只写有把握的 pattern，命中就出，不命中不猜：
RULES = {
"线索提报": r"提报(一条|一下|流程)|录入系统|新建一条线索|登记一下",
"线索全生命周期跟踪反馈": r"到哪一步|上次跟进|卡在哪个环节|有人跟进吗|反馈",
}这层最容易出的问题是一个词写太宽。最初”线索提报”里有个裸的 提报，结果”上周提报的那条线索，后续有人跟进吗？“（应该是跟踪反馈）被它截胡了。收窄成 提报(一条|一下|流程) 就修好了。
L2：semantic-router
用 semantic-router 把每个意图的校准说法编码成向量，输入进来算相似度，返回最接近的意图。关键不在算相似度，在 fit()：
router = IndexedRouter(encoder=FastEmbedEncoder(name="jinaai/jina-embeddings-v2-base-zh"),
routes=[Route(name=k, utterances=v) for k, v in CALIB.items()])
router.fit(X=FIT_X, y=FIT_Y)
router.confidence_threshold = 0.5FIT_Y 里那些 None 是给路由器看”哪些句子不属于任何意图”的，教它学会拒识。没有这一步，它会对任何输入都硬塞一个意图名，L3 就永远轮不到了。
L3：LLM + 结构化输出
用 with_structured_output 直接拿 Pydantic 对象，省掉”让模型只输出意图名、再用正则兜底解析”那套：
class Intent(BaseModel):
intent: str = Field(description="用户意图")
confidence: float = Field(description="置信度 0.0 ~ 1.0")
model = init_chat_model("deepseek-flash", model_provider="deepseek",
extra_body={"thinking": {"type": "disabled"}})\
.with_structured_output(Intent, method="json_schema", strict=True)测试集要分层
最开始我拿 35 条测试跑，全对，还挺高兴。后来发现不对：30 条域内样本里 29 条直接被正则命中，L2 只被验证了 1 条。也就是说 L2 的真实水平根本没测出来，那个 100% 是虚的。
测试集得按”这一条该由哪一层接住”来分：
graph TD
T["测试集 62 条"] --> A["L1 层样本 29 条<br/>规则可命中"]
T --> B["L2 层样本 25 条<br/>绕规则·语义可辨"]
T --> C["L3 层样本 1 条<br/>绕规则·绕语义"]
T --> D["OOS 样本 7 条<br/>域外·应拒识"]
- L1 层：常规说法，规则能命中。
- L2 层：故意绕开所有正则、但语义清楚的口语变体，比如”新谈了个客户，帮我把他存进线索库”（提报）、“这客户到底有没有戏？“（分析）。
- L3 层：连语义路由都匹配不上（相似度低于阈值）的间接说法，逼它走到大模型。
- OOS：域外输入，测拒识。
三类数据不能混：建路由的 utterances、校准集、测试集各是各的。拿测试集去建路由，等于先看答案再考试。
flowchart LR
U["建路由 utterances<br/>(CALIB)"] --> R["Router"]
F2["校准集<br/>(FIT_X / FIT_Y)"] --> R
T2["测试集<br/>(62 条)"] -.->|"只评估<br/>不参与构建"| R
踩过的坑
graph LR
A["bge-small<br/>9 错"] -->|"换 jina"| B["jina-base-zh<br/>4 错"]
B -->|"补 CALIB 5 句"| C["5 错"]
B -->|"补 CALIB 2 句"| D["6 错"]
B -.->|"回退原 CALIB"| E["4 错"]
embedding 模型才是 L2 的瓶颈
bge-small-zh-v1.5（512 维）分不清这几个近义意图，更麻烦的是它从不说”不知道”。对口语化的陌生说法也给出 0.8+ 的高置信度，然后硬塞一个错误意图。结果两件事都坏：近义意图高置信度误判，同时 L3 根本触达不了——因为 L2 从来不拒识。
换成 jina-embeddings-v2-base-zh（768 维）后，错误从 9 掉到 4，L3 才第一次真正接到域内样本。
补校准句是打地鼠
jina 会把”分析""跟踪”的句子系统性判成”准入”。第一反应是给弱势意图补校准句。结果：
- 补 5 句 → 5 错（治好 2 条，引入 3 条新错）
- 补 2 句 → 6 错
- 回退原始 3 句/意图 → 4 错
原因是 router.fit() 是全局阈值优化，不是线性叠加，补一句会挪动整个决策边界，经常治一个引一个。
LLM 兜底的 prompt 也得给定义
L3 最早的 prompt 只列了 5 个意图名，没定义，还写”无法判断输出 unknown 不要猜”。结果”还款期限是多久”这种明显的问题，模型因为对不上抽象标签就保守地回了 unknown。改成每个意图给一句话定义 + 一个示例：
- 产品知识问答：询问产品本身的属性，如额度、利率、期限、材料、流程。例：这个产品利率是多少结果
| 层级 | 拦截 | 层内准确率 |
|---|---|---|
| L1 规则 | 29 (46.8%) | 100% |
| L2 语义路由 | 25 (40.3%) | 88.0% |
| L3 大模型 | 8 (12.9%) | 87.5% |
总准确率 93.55%（62 条，4 错），OOS 拒识 100%。混淆矩阵几乎是对角线，剩下的 4 个错集中在两处：3 条被判成”产品准入分析”（jina 对”准入”有偏向），1 条”产品知识问答”漏进 unknown。
看指标别只看总数，要看错在哪一层、往哪个方向偏。混淆矩阵比一个百分比有用。
生产上注意这几点
- 测试集按层级分，明确每层该接住哪些样本，不然会拿到虚高数字。
- 建路由 utterances、校准集、测试集分开，别混。
- 近义意图多的场景，先换更强的 embedding（或上 reranker），再谈调阈值、补校准句。
fit()用 held-out 查询，别用存储的 utterances 本身——它们和自己匹配约等于 1.0，会让阈值学偏。- LLM 兜底的 prompt 给定义 + 示例，光列标签名加一句”不要猜”会让模型过度保守。
- 误分类（判错）和误拒（该接没接/该拒没拒）是两种问题，修法不一样。
- 每层都打日志，记下输入、命中结果、置信度。级联系统出问题没日志只能靠猜。
- API key 走环境变量，别硬编码进源码，迟早会进 git 历史。
结论
L1 负责稳，L2 决定准不准，L3 兜底。系统准确率基本卡在 L2 的 embedding 区分度上，这个靠调阈值、补校准句救不回来。
这次实验真正有用的不是 93.55% 这个数，是两件事：测试集怎么设计才敢信指标，以及每一层的瓶颈在哪。知道了这两点，后面换 embedding、上 reranker、换更强的 LLM，都清楚该往哪使劲。
依赖
semantic-router[fastembed] # 语义路由
fastembed # 本地 embedding，不用调 API
langchain # init_chat_model + with_structured_output
pydantic # 输出 schema附录：完整代码
# 三级级联：规则 → 语义路由 → LLM 兜底
import csv, re, os, logging
from collections import Counter, defaultdict
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-7s %(message)s", datefmt="%H:%M:%S")
log = logging.getLogger("cascade")
from langchain.chat_models import init_chat_model
from pydantic import BaseModel, Field
INTENTS = ["线索提报", "线索全生命周期跟踪反馈", "线索分析", "产品准入分析", "产品知识问答"]
PROMPT_V2 = """你是意图分类器。从以下5个意图中选一个并输出置信度，无法归入任一类时输出 unknown、置信度0。
- 线索提报：把新客户/新线索录入系统、登记建档。例：帮我把这个客户登记一下
- 线索全生命周期跟踪反馈：查询线索的进展、状态、谁在跟进、有无新回复。例：这条线索到哪一步了
- 线索分析：评估线索质量、成交概率、给线索排序。例：这条线索值不值得跟
- 产品准入分析：判断客户或行业是否符合准入资格。例：这个客户能不能做我们的产品
- 产品知识问答：询问产品本身的属性，如额度、利率、期限、材料、流程。例：这个产品利率是多少
只输出意图名或 unknown，不要解释。"""
os.environ["DEEPSEEK_API_KEY"] = os.environ.get("DEEPSEEK_API_KEY", "") # 从环境变量读，别硬编码
class Intent(BaseModel):
intent: str = Field(description="用户意图")
confidence: float = Field(description="置信度 0.0 ~ 1.0")
model = init_chat_model(
model="deepseek-flash",
model_provider="deepseek",
extra_body={"thinking": {"type": "disabled"}},
).with_structured_output(Intent, method="json_schema", strict=True)
RULES = {
"线索提报": r"提报(一条|一下|流程)|录入系统|新建一条线索|登记一下",
"线索全生命周期跟踪反馈": r"到哪一步|现在.*状态|上次跟进|卡在哪个环节|有人跟进吗|反馈",
"线索分析": r"分析.*(质量|线索)|转化可能性|哪个更优先|跟丢的原因|梳理下规律|总结",
"产品准入分析": r"准入|能不能做|符不符合.*条件|能不能进",
"产品知识问答": r"额度是多少|利率|申请材料|办理流程|有什么区别|支持哪些行业",
}
def rule_level(text):
for name, pat in RULES.items():
if re.search(pat, text):
log.info("L1 rule 命中 | %s -> %s | pattern=%s", text, name, pat)
return name, pat
log.info("L1 rule 未命中 | %s", text)
return None, None
from semantic_router import Route, IndexedRouter
from semantic_router.encoders import FastEmbedEncoder
CALIB = {
"线索提报": ["请帮我把这个客户提交上去", "新谈了一个客户，需要录入", "销售给了一个客户资料，帮忙登记"],
"线索全生命周期跟踪反馈": ["这条线索进展到哪里了", "帮我看下这条线索的最新动态", "这条线索最近有没有人处理"],
"线索分析": ["评估一下这条线索靠不靠谱", "这批线索哪个最值得优先跟", "帮我复盘这条线索为什么没成"],
"产品准入分析": ["这个客户做我们的产品合规吗", "这家企业能不能准入", "查一下这个行业的准入要求"],
"产品知识问答": ["这个产品最高能贷多少", "申请这个产品要准备什么", "这两个产品有什么不同"],
}
encoder = FastEmbedEncoder(name="jinaai/jina-embeddings-v2-base-zh")
router = IndexedRouter(encoder=encoder, routes=[Route(name=k, utterances=v) for k, v in CALIB.items()])
FIT_X = [u for v in CALIB.values() for u in v] + ["今天深圳天气怎么样？", "帮我请个假", "中午吃什么好"]
FIT_Y = [k for k, v in CALIB.items() for _ in v] + [None, None, None]
router.fit(X=FIT_X, y=FIT_Y)
router.confidence_threshold = 0.5
def router_level(text):
choice = router(text)
if choice is None or getattr(choice, "name", None) is None:
log.info("L2 router 拒识(落L3) | %s", text)
return None, choice
log.info("L2 router 命中 | %s -> %s | conf=%.3f", text, choice.name, getattr(choice, "confidence", 0.0))
return choice.name, choice
def small_model_level(text):
prompt = PROMPT_V2.format(text=text)
intent = model.invoke(input=prompt, temperature=0.1)
log.info("L3 small 输出 | %s -> %s | conf=%.2f", text, intent.intent, intent.confidence)
raw_text = str(intent.intent).strip()
if raw_text in INTENTS or raw_text == "unknown":
return raw_text, raw_text
for name in INTENTS:
if name in raw_text:
return name, raw_text
return "unknown", raw_text
def report_metrics(results):
INTENT_ORDER = INTENTS + ["unknown"]
N = len(results)
ok = sum(1 for r in results if r[5])
log.info("===== 指标汇总 =====")
log.info("总览 | 样本 %d | 正确 %d | 错误 %d | 准确率 %.2f%%", N, ok, N - ok, ok / N * 100)
wrong = [r for r in results if not r[5]]
if wrong:
log.info("错误样例 %d 条:", len(wrong))
for r in wrong:
log.info(" #%s gold=%s -> %s | 层级=%s | %s", r[0], r[2], r[4], r[3], r[1])
lvl = Counter(r[3] for r in results)
prev = N
for L in ("L1_rule", "L2_router", "L3_small"):
n = lvl[L]
log.info("拦截 %s | %d 条 | 占总样本 %.1f%% | 占上级未命中 %.1f%%", L, n, n / N * 100, n / prev * 100)
prev -= n
for L in ("L1_rule", "L2_router", "L3_small"):
sub = [r for r in results if r[3] == L]
ok_l = sum(1 for r in sub if r[5])
log.info("层准确率 %s | %d/%d = %.1f%%", L, ok_l, len(sub), ok_l / len(sub) * 100)
for g in INTENT_ORDER:
sub = [r for r in results if r[2] == g]
ok_g = sum(1 for r in sub if r[5])
log.info("意图准确率 %s | %d/%d = %.1f%%", g, ok_g, len(sub), ok_g / len(sub) * 100)
oos = [r for r in results if r[2] == "unknown"]
oos_ok = sum(1 for r in oos if r[5])
mis_reject = [(r[0], r[2]) for r in results if r[2] != "unknown" and r[4] == "unknown"]
log.info("拒识 | OOS %d/%d = %.1f%% | 域内误拒 %s", oos_ok, len(oos), oos_ok / len(oos) * 100, mis_reject)
mat = defaultdict(Counter)
for _, _, g, _, p, _ in results:
mat[g][p] += 1
log.info("混淆矩阵 (gold\\预测):")
log.info(" " + "".join(f"{p[:4]:>7s}" for p in INTENT_ORDER))
for g in INTENT_ORDER:
log.info("%-6s %s", g[:4], "".join(f"{mat[g][p]:>7d}" for p in INTENT_ORDER))
TEST = [(int(r["id"]), r["text"], r["gold"]) for r in csv.DictReader(open("week1-utterances.csv", encoding="utf-8"))]
results = []
with open("week1_cascade.csv", "w", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w.writerow(["id", "text", "gold", "solved_at", "rule", "router", "router_raw", "small", "small_raw", "correct"])
for i, text, gold in TEST:
rule, _ = rule_level(text)
if rule:
w.writerow([i, text, gold, "L1_rule", rule, "", "", "", "", rule == gold])
results.append((i, text, gold, "L1_rule", rule, rule == gold))
continue
rname, rraw = router_level(text)
if rname:
w.writerow([i, text, gold, "L2_router", "", rname, str(rraw), "", "", rname == gold])
results.append((i, text, gold, "L2_router", rname, rname == gold))
continue
sname, sraw = small_model_level(text)
w.writerow([i, text, gold, "L3_small", "", "", "", sname, str(sraw), sname == gold])
results.append((i, text, gold, "L3_small", sname, sname == gold))
report_metrics(results)