鸽子的回忆录

探索 AI · IoT · 嵌入式 · 架构

用 1.7B 小模型给 Agent 做意图路由

发表于 更新于 分类于 ，
本文字数： 4.8k 阅读时长 ≈ 4 分钟

之前 Agent 里每来一句话，不管三七二十一，全塞给 32B 的主模型。结果就是用户问一句”今天怎么样”，也得让大模型吭哧吭哧思考三秒，烧掉两千个 token——你说亏不亏？

说白了，大部分请求根本用不着主模型出场。要么就是寒暄两句，要么意图明明白白摆在那儿，杀鸡焉用牛刀。

后来我在前面塞了一层 1.7B 的小分类器，跑在本地 Ollama 上，专门干一件事：判断这次该走哪条路。

意图分了哪几类

按业务划了 6 类，从纯聊天到多模态病虫害诊断都覆盖到了：

1
2
3
4
5
6
7
8
9
from enum import Enum

class IntentCategory(str, Enum):
    PEST_DIAGNOSIS = "pest_diagnosis"        # 病虫害诊断（多模态）
    DEVICE_MANAGEMENT = "device_management"  # 设备管理
    DATA_ANALYSIS = "data_analysis"          # 数据分析
    AUTOMATION = "automation"                # 自动化规则
    AGRICULTURE = "agriculture"              # 农业管理
    GENERAL = "general"                      # 通用问答

分类器要的是规规矩矩的结构化输出，所以拿 Pydantic 卡一道——免得它一时兴起跟你聊起人生哲学：

1
2
3
4
5
6
7
8
from pydantic import BaseModel, Field

class RouterOutput(BaseModel):
    category: str = Field(description="意图分类")
    confidence: float = Field(default=0.5, description="置信度 0-1")
    reasoning: str = Field(default="", description="分类理由")
    requires_image: bool = Field(default=False, description="是否需要图片")
    sub_intent: str = Field(default="", description="子意图")

分类器本体

模型挑了 qwen3:1.7b，温度压到 0.1（这种活儿不需要它发挥创造力），max_tokens 给个 256 就够用，超时卡 5 秒——再长就纯属耽误事，还不如让主模型直接上：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from langchain_openai import ChatOpenAI

class IntentRouter:
    def __init__(self):
        self._router_model = ChatOpenAI(
            base_url="http://localhost:11434/v1",
            model="qwen3:1.7b",
            max_tokens=256,
            temperature=0.1,
            timeout=5.0,
        )

    async def classify(self, query: str, history: list = None) -> RoutingResult:
        try:
            return await self._classify_with_llm(query, history)
        except Exception:
            return self._keyword_fallback(query)

关键词兜底

小模型也是会抽风的——超时、吐出非法 JSON、给你一个根本不存在的类别。这些事儿要是直接报错给用户，体验立马就崩。

所以底下又垫了一层关键词匹配。逻辑笨是真的笨，但有个好处：它永远不会挂。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def _keyword_fallback(self, query: str) -> RoutingResult:
    q = query.lower()

    pest_keywords = ["病虫害", "病害", "虫害", "枯萎", "黄叶", "斑点",
                     "诊断", "识别病", "什么病", "怎么治"]
    if any(kw in q for kw in pest_keywords):
        return RoutingResult(
            category=IntentCategory.PEST_DIAGNOSIS,
            confidence=0.75,
            reasoning="关键词匹配: 病虫害相关",
        )

    auto_keywords = ["自动化", "自动", "规则", "触发", "条件", "定时",
                     "创建规则", "编排", "工作流", "联动"]
    if any(kw in q for kw in auto_keywords):
        return RoutingResult(
            category=IntentCategory.AUTOMATION,
            confidence=0.7,
            reasoning="关键词匹配: 自动化相关",
        )

    device_keywords = ["设备", "传感器", "连接", "断开", "在线", "离线", "状态"]
    if any(kw in q for kw in device_keywords):
        return RoutingResult(
            category=IntentCategory.DEVICE_MANAGEMENT,
            confidence=0.7,
            reasoning="关键词匹配: 设备管理相关",
        )

    return RoutingResult(
        category=IntentCategory.GENERAL,
        confidence=0.5,
        reasoning="未匹配到特定领域关键词",
    )

匹配顺序这事得动点脑子。举个例子，“设备联动”这句话里既有”设备”又有”联动”，但用户心里想的明明是自动化规则。要是按字母序匹配，肯定被设备这一支先吃掉，结果就是用户问”帮我做个联动”，系统跑去查设备列表给他——尴不尴尬？

所以自动化的关键词必须排在设备前面判。

工具按意图加载

分类只是手段，真正想干的事是把工具集裁到一个合理的大小。

Anthropic 之前发过一个经验值：工具一旦超过 20 个，模型选错的概率明显抬头。所以分类完了之后，按 tag 拉对应的工具子集就行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def _select_pipeline(self, routing: RoutingResult):
    registry = self._tool_registry

    if routing.category == IntentCategory.DEVICE_MANAGEMENT:
        tools = registry.get_tools_by_tags(["device", "entity"])
        system_prompt = get_prompt("device_expert")

    elif routing.category == IntentCategory.DATA_ANALYSIS:
        tools = registry.get_tools_by_tags(["data", "device"])
        system_prompt = get_prompt("data_analyst")

    elif routing.category == IntentCategory.AUTOMATION:
        tools = registry.get_tools_by_tags(["automation", "device", "entity"])
        system_prompt = get_prompt("automation_expert")

    elif routing.category == IntentCategory.AGRICULTURE:
        tools = registry.get_tools_by_tags(["agriculture", "diagnosis", "data"])
        system_prompt = get_prompt("agriculture_expert")

    else:
        tools = registry.get_tools_by_tags(["general"])
        system_prompt = get_prompt("general_assistant")

    return model, tools, system_prompt

工具注册的时候顺手打 tag，注册中心按 tag 取交集，就这么简单：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, ToolRegistration] = {}

    def register(self, tool, tags=None, description_zh="", examples=None):
        reg = ToolRegistration(
            tool=tool,
            tags=set(tags or []),
            description_zh=description_zh or tool.description,
            examples=examples or [],
        )
        self._tools[tool.name] = reg

    def get_tools_by_tags(self, tags: List[str]) -> List[BaseTool]:
        result_set = {}
        tag_set = set(tags)
        for name, reg in self._tools.items():
            if reg.tags & tag_set:
                result_set[name] = reg.tool
        return sorted(result_set.values(), key=lambda t: t.name)

跑下来什么感觉

上了这层路由以后，体感最爽的就是寒暄类的请求——基本秒回。小模型一拍板，主模型连脸都不用露。Token 账单也瘦了一圈，没具体统计，但调到自动化场景时，上下文里塞的工具描述肉眼可见地少了一大截。

最坑的地方倒不是模型本身，反而是关键词兜底的顺序——这玩意儿排错了，出来的结果能让你怀疑人生。还有个一直在我 TODO 里没动的事：小模型分类错了到底咋办？目前是无条件相信兜底，但理论上应该按 confidence 卡个线，太低就升级给主模型再判一次。

这块还在慢慢磨。

相关文章

欢迎关注我的其它发布渠道