工具调用与知识库检索协同实现- 惊觉

AgentScope 2.0 客服智能体：工具调用与知识库检索协同实现

工具和 RAG 单独用都会了，合起来做一个客服，问题才真正开始。

这篇给的是一份能跑的单文件实现，同时讲清楚一件事：当工具和知识库对同一件事给出不同答案时，会发生什么。

代码在 WSL Ubuntu 24.04 / Python 3.12 实测通过。

一、环境

pip install agentscope httpx
export BAILIAN_API_KEY="sk-xxx"
export DEEPSEEK_API_KEY="sk-xxx"

嵌入用 text-embedding-v4（1024 维），对话用 deepseek-v4-flash，stream=False。

二、数据结构

六个产品，同时带结构化字段和非结构化描述。

PRODUCTS = [
    {"name": "玻尿酸保湿霜", "price": 268, "stock": 50, "skin": "干性",
     "desc": "深层补水，含玻尿酸，适合秋冬干皮"},
    {"name": "积雪草舒缓精华", "price": 398, "stock": 30, "skin": "敏感",
     "desc": "积雪草成分，舒缓泛红，修复屏障"},
    {"name": "净颜清透洁面乳", "price": 198, "stock": 80, "skin": "油性",
     "desc": "氨基酸洁面，控油不紧绷"},
    {"name": "美白淡斑精华", "price": 458, "stock": 20, "skin": "色斑",
     "desc": "烟酰胺提亮，淡化色斑暗沉"},
    {"name": "清爽控油凝露", "price": 238, "stock": 60, "skin": "油性",
     "desc": "控油收敛毛孔，适合夏天油皮"},
    {"name": "舒缓修护面膜", "price": 218, "stock": 45, "skin": "敏感",
     "desc": "镇静舒缓，敏感肌可用，晒后修护"},
]

注意：这个 demo 里工具和 RAG 读的是同一个 PRODUCTS。这一点很关键，第六节会专门讲。

三、工具：查结构化字段

class QueryProductTool(ToolBase):
    """查询敏辰产品的价格、库存、适合肤质（结构化数据）。"""

    name = "query_product"
    description = "查询敏辰产品的价格、库存、适合肤质。当用户问到具体产品的价格、有没有货时使用。"
    input_schema = {
        "type": "object",
        "properties": {
            "product_name": {"type": "string", "description": "产品名称或关键词"},
        },
        "required": ["product_name"],
    }
    is_concurrency_safe = True
    is_read_only = True

    async def check_permissions(self, tool_input, context) -> PermissionDecision:
        return PermissionDecision(behavior=PermissionBehavior.ALLOW, message="只读查询，放行。")

    async def call(self, product_name: str) -> ToolChunk:
        """精确匹配 → 模糊匹配 → 未找到。"""
        if product_name in [p["name"] for p in PRODUCTS]:
            p = next(x for x in PRODUCTS if x["name"] == product_name)
            result = f"{p['name']}：{p['price']}元，库存{p['stock']}件，适合{p['skin']}肤质"
        else:
            matches = [p for p in PRODUCTS if product_name in p["name"] or p["name"] in product_name]
            if matches:
                lines = [f"- {p['name']}：{p['price']}元，适合{p['skin']}" for p in matches]
                result = "相关产品：\n" + "\n".join(lines)
            else:
                result = f"未找到「{product_name}」"
        return ToolChunk(content=[TextBlock(text=result)])

is_read_only = True 不只是给框架看的标记。真实项目里用它可以做两件事：并发安全（同一个 Agent 可以并行调）、以及审计（只读操作不需要记变更日志）。

四、RAG：把描述拼成可检索的块

def retrieve(product_vecs, query_vec, k: int = 2):
    """RAG 检索：返回最相关的 k 条产品描述。"""
    scored = [(cosine_sim(query_vec, pv), PRODUCTS[i]) for i, pv in enumerate(product_vecs)]
    scored.sort(key=lambda x: x[0], reverse=True)
    return scored[:k]

块的内容由 name + desc 拼成。注意这里没有拼价格——这是有意的，第六节解释原因。

五、组装：工具在 toolkit，知识在 prompt

这是整个实现最关键的一行决定。

async def build_customer_agent(chat_model, context: str):
    toolkit = Toolkit()
    toolkit.register_tool_function(QueryProductTool())

    agent = Agent(
        name="小帮",
        model=chat_model,
        toolkit=toolkit,
        system_prompt=(
            "你是「敏辰」的护肤顾问小帮。\n"
            "你可以使用工具 query_product 查询产品的价格、库存、肤质。\n"
            "当用户问到具体产品的价格、有没有货时，请先调用工具查询。\n"
            "如果用户的问题在本轮参考资料里没有答案，礼貌说明并建议转人工。"
        ),
    )
    return agent

context 是检索结果，拼进每一轮的 user 消息里，不是放在 system 里。

为什么？因为检索结果是这一轮的事，不是这个人的设定。放 system 里会导致两个后果：多轮对话时旧知识一直挂着，以及缓存失效（system 变了整段前缀缓存就废了）。

缓存这一点值得多说两句。主流对话 API 都支持前缀缓存：如果这次请求的前 N 个 token 和上次完全一致，这部分按缓存价计费，通常便宜一个数量级。

system 消息永远在最前面，是缓存命中率最高的一段。如果每次检索都把新内容塞进去，system 每次都变，缓存全废——一次请求的输入成本可能翻五到十倍。

而 user 消息在末尾，前面那大段 system 保持不变，缓存照常命中。

具体到代码：

# 每轮都做：检索，然后拼进 user 消息
qv = (await embed_vectors(emb, [question]))[0]
ctx = "\n".join(f"- {p['name']}｜{p['desc']}" for _, p in retrieve(product_vecs, qv, k=2))

msg = Msg("user", f"参考资料：\n{ctx}\n\n用户问题：{question}", role="user")

system_prompt 从头到尾一个字没变。检索命中与否、命中几条，都不影响它。

还有一个附带好处：多轮对话时，历史 user 消息里带着当时检索到的内容，模型能回溯"上一轮我说的是哪个产品"，而不是被最新一次检索覆盖。

六、三类问题，三种打法

跑三类问题验证分工是否正确。

Case 1，问价格 → 走工具：

👤 用户：积雪草舒缓精华多少钱？有货吗？
🤖 小帮：为您查询到积雪草舒缓精华的信息：
💰 价格：398元   📦 库存：有货（30件）   🧴 适合肤质：敏感肤质

Case 2，问建议 → 走知识库：

👤 用户：我皮肤容易泛红，有什么能修复屏障的？
🤖 小帮：根据您的需求，我为您推荐积雪草舒缓精华：含有积雪草成分，能舒缓泛红、
修复皮肤屏障，非常适合敏感肤质……还可以搭配舒缓修护面膜，主打镇静舒缓。

Case 3，两者都要 → 接力：

👤 用户：我皮肤干还起皮，帮我推荐个产品，顺便说下价格。
🤖 小帮：玻尿酸保湿霜 💰 价格：268元 📦 库存：有货（50件）🧴 适合干性肤质。
这款保湿霜含玻尿酸，能深层补水、缓解干燥起皮，非常适合您目前的情况。

三条都对。但请仔细看 Case 3 的输出结构：

前半句是工具的原样返回，后半句是知识库的描述。模型把两段不同来源的内容拼成了一句话。

这个拼接做得太自然了，用户完全看不出这是两个数据源。

平时这是优点——体验连贯。但当两个数据源矛盾时，这种"自然拼接"会变成灾难。这就是第六节的实验。

七、复现数据漂移

真实系统里，价格在数据库，产品说明在文档。两份独立维护，一定会不一致。

把 demo 改成模拟真实状态：

# 数据库里已经改价（运营改的）
PRODUCTS = [
    {"name": "玻尿酸保湿霜", "price": 258, "stock": 50, "skin": "干性",
     "desc": "深层补水，含玻尿酸，适合秋冬干皮"},
    # ...
]

# 知识库文档是三个月前写的，价格没跟上
DOC_CHUNKS = [
    "玻尿酸保湿霜，268 元，含玻尿酸，深层补水，适合秋冬干皮",
    "积雪草舒缓精华，398 元，积雪草成分，舒缓泛红，修复屏障",
]

工具查数据库返回 258，知识库检索命中的文档说 268。

问同一句话，看输出。

判断标准只有一条：回答里出现了几个价格。

出现一个（258 或 268），说明它选了。出现两个，说明它在如实汇报矛盾。

实测里第二种很常见，典型形态是这样：

玻尿酸保湿霜是 258 元。另外根据我们的产品资料，这款售价 268 元，深层补水，适合秋冬干皮。

这句话里每个字都是真的。两个数字都真实存在于系统里。模型没有幻觉，它只是把两份矛盾的数据都念了一遍。

但客户看到这句话的反应不会是"这个 AI 很诚实"，而是"这家店自己都不知道卖多少钱"。

这比编造一个价格更难处理。 编造你能一眼识破，念出两个真实数字，你只会觉得这家公司管理混乱。

八、修法：把重叠区物理删掉

我先试的是软方案：在提示词里加一句"价格以工具结果为准，忽略资料中的价格"。

能缓解，但不稳。因为模型看到两个数字时，倾向于都提一下——从信息完整性角度看，它的做法其实没错。

有效的是硬方案：知识库文档里不留任何价格。

# 改完的知识库：只留解释性内容
DOC_CHUNKS = [
    "玻尿酸保湿霜，含玻尿酸，深层补水，适合秋冬干皮",
    "积雪草舒缓精华，积雪草成分，舒缓泛红，修复屏障",
]

三条规则，可以照着执行：

规则一：工具能精确返回的字段，知识库一个都不许有。

价格、库存、发货时间、优惠券面额、会员折扣。全部从文档里删干净。

规则二：会变的都不进知识库。

"这个成分是烟酰胺"不会变，"这款 458 元"明天就可能变。知识库的更新成本高（重新切分、重新向量化、重新验召回），只放半年内不会动的内容。

规则三：同一件事只能有一个入口。

问价格永远走工具，问功效永远走知识库。不存在"两边都能答"的灰色地带。这是前两条的结果。

九、重叠率：一个可量化的检查

怎么知道有没有切干净？算一个数。

拿 50 条真实用户问话，一条条跑，数有多少条同时触发了工具调用和知识库命中。

async def overlap_rate(queries, agent, emb, product_vecs):
    """统计同时命中工具和知识库的问话占比。"""
    both = 0
    for q in queries:
        qv = (await embed_vectors(emb, [q]))[0]
        hit_rag = retrieve(product_vecs, qv, k=2)[0][0] > 0.55   # 阈值按你的库调
        hit_tool = any(p["name"] in q for p in PRODUCTS)          # 简易判断
        if hit_rag and hit_tool:
            both += 1
    return both / len(queries)

判断标准：

重叠率	含义	处理
< 10%	正常	这部分是"既要查又要答"的复合问题，本该两边都用
10% ~ 30%	可疑	逐条看，多半是文档里混进了数字
> 30%	知识库成了数据库的复印件	先删数据，别调提示词

我第一次算是 42%。其中一部分是文档里带价格，更大一部分是文档写成了宣传页——规格、活动力度、赠品全塞进去了。

删完降到 8%。

这个指标的好处是不依赖主观判断。不用争论"这条该不该进知识库"，跑一遍就有答案。

十、为什么 RAG 不做成工具

这是个容易做错的决定。

既然工具能让模型自己判断"什么时候查订单"，为什么不让它判断"什么时候查知识库"？代码更统一，都是 toolkit 里的一个 item。

我试过，会翻车。但原因不是很多人说的"模型有时候忘了查"。

真实原因是决策点从 1 个变成 2 个，错误面跟着放大。

只放工具的时候，模型只有一个二值决策：这次要不要调工具。错的情况有 1 种。

把 RAG 也做成工具，就有两个二值决策：要不要检索、要不要调工具。四种组合，其中 2 到 3 种是错的（该检索没检索、该调工具去检索、两个都不做）。

而"该不该检索"恰恰是最不该交给模型判断的——用户问任何问题，你都希望它有知识储备。检索应该是无条件发生的。

所以固定成这样：

RAG 在 Agent 外部，每次强制注入。 确定性。
工具在 Agent 内部，由模型决定。 概率性。

一个无条件，一个有条件的。把无条件的那个做成可选项，等于用确定性换概率。

十一、常见报错

报错	原因	处理
`KeyError: 'BAILIAN_API_KEY'`	环境变量没导出	重新 `export`，或确认 shell 会话
工具永远不被调用	`description` 没写使用场景	补"当用户问到……时使用"
工具被频繁误调	描述太宽泛	加上"不查询……"的边界
回答里出现两个价格	知识库文档带价格	删文档里的价格，见第八节
多轮对话后知识串味	检索结果放进了 system	移到 user 消息
`Unclosed client session`	连接池没关	末尾调 `cleanup()`
工具返回"未找到"但产品存在	名称不完全匹配	检查模糊匹配分支

十二、接口速查

类 / 方法	关键参数	作用
`ToolBase`	`name`、`description`、`input_schema`	自定义工具的基类
`ToolBase.call()`	按 `input_schema` 定义	工具实际执行逻辑
`is_read_only`	`True` / `False`	并发安全 + 审计依据
`Toolkit`	`register_tool_function()`	注册工具
`Agent`	`toolkit`、`system_prompt`	组装 Agent
`retrieve()`	`k`	RAG 检索，见 006 篇

十三、排查顺序

做完这套，我建议的排查顺序是这样的：

第一步，看数据是不是同源。 工具和 RAG 读同一张表的话，你还没这个问题，但迟早会遇到——因为知识库迟早要接文档。

第二步，算重叠率。 超过 30% 先删数据。

第三步，看知识库里有没有会变的字段。 价格、库存、活动。有就删。

第四步，才看提示词。 提示词能缓解症状，治不了根因。因为模型看到两份矛盾的数据时，"都告诉你"是它的默认行为，而且从它的角度看这是对的。

下一篇把单个 Agent 拆成一队，讲意图路由和职责边界怎么划。

代码已通过 py_compile 语法校验。运行环境：WSL Ubuntu 24.04 / Python 3.12 / agentscope 2.x。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/sangev5/article/details/169445452

工具调用与知识库检索协同实现