系列：AgentScope 2.0 学习笔记 · 第 007 篇
难度：进阶
适合谁：工具和 RAG 都会了、但不知道「怎么合体成一个客服」的开发者
前置：建议先读 004《工具调用入门》、005/006《RAG 检索入门/进阶》
阅读收获：把工具和 RAG 合体，拼出一个能上岗的客服 Agent
运行环境：WSL Ubuntu-24.04 + Python 3.11+ + AgentScope 2.x

---

你有没有见过这种「半吊子」客服机器人：你问它「积雪草精华多少钱」，它答得飞快、价格库存全对；你接着问「我泛红用什么好」，它就卡壳了——因为它只会查价格，不会给建议。反过来也有一种：护肤知识背得滚瓜烂熟，你一问「有货吗」，它只能含含糊糊说「应该有的吧」。

这两种机器人的毛病是同一个：只握了一把剑。 真实用户不会按你的功能分类来提问，他一会儿问价格、一会儿问建议，有时候两个一起问。这一篇，就是把前六篇攒的两把剑——工具和 RAG——合体，拼出一个真正能上岗的客服。

---

一、前六篇，我们攒了两把剑

回顾一下我们的家底：

• 004 篇，给 Agent 装上了「手」——工具，能查价格、查库存（结构化数据）；
• 005/006 篇，给 Agent 建起了「知识库」——RAG，能理解「补水」就是「保湿」（非结构化知识）。

但一个真实的客服，不会只用其中一样。用户一会儿问「积雪草精华多少钱」（要查价格），一会儿问「我泛红用什么好」（要查知识）。只会查价格的 Agent，接不住护肤咨询；只会背知识的 Agent，报不出实时库存。

真正的客服，需要两把剑都握在手里。这一篇，我们把它们合体。

---

二、两把剑的分工：结构化 vs 非结构化

先想清楚一件事：工具和 RAG 到底各管什么？

	工具（004）	RAG（005/006）
查什么	结构化数据	非结构化知识
例子	价格、库存、肤质字段	产品描述、成分、适用场景
怎么匹配	字段精确匹配	语义相似匹配
典型问题	「多少钱？有货吗？」	「我泛红用什么？」

一句话概括分工：工具管「数字」（价格、库存这些能存进数据库字段的），RAG 管「话」（描述、成分、建议这些一段段文字）。

用户问「多少钱」，是问数字 → 工具；用户问「用什么好」，是问建议 → RAG。这两类问题在真实客服里交错出现，所以 Agent 得两个都会。

给你一个简单的判断口诀：用户的话里有明确的「产品名 + 数字诉求」（多少钱、有没有货、还剩几件），走工具；用户的话是「症状 + 求建议」（我干、我泛红、我油皮，用什么好），走 RAG。 记住这个，你写客服 Agent 时就知道该给用户配哪把剑了。

这个口诀别看简单，它是整个客服系统设计的地基：问题分类对了，架构就顺了；分类错了，后面全是补丁。 很多客服 Agent 做不好，不是模型不行，是一开始就没分清楚「哪些问题该查库、哪些该查知识」。

---

三、双剑合璧的架构

合体的方式很自然，就一句话：

Agent 带上工具，同时每次回答前先用 RAG 把相关知识喂给它。

具体是这样运转的：

1. 用户提问；
2. 先走 RAG：把问题向量化，检索出最相关的产品知识，塞进 prompt；
3. Agent 回答时——
   • 用户问价格/库存 → 它主动调工具查（因为工具在它的 toolkit 里）；
   • 用户问护肤建议 → 它基于注入的知识答（因为知识已经在 prompt 里）；
4. 两者结合的问题（「推荐个产品，顺便说价格」）→ 先用知识选产品，再调工具查价格。

这样，一个 Agent 就同时具备了「会查」（工具）和「会答」（RAG）两种能力。

这里有一个新手容易踩的坑：别把 RAG 也做成一个工具。 有人会把「检索」也包成一个 retrieve_tool，让 Agent 自己决定「要不要检索」。结果往往是——模型有时忘了检索，直接凭记忆回答，幻觉又回来了。更稳的做法是本文这种：RAG 在 Agent 外部、每次强制注入，工具在 Agent 内部、由它自主调用。前者保证「知识一定喂到」，后者保证「查价灵活」。

这个坑值得多说一句。把 RAG 包成工具，看起来「更优雅」（都是让模型自己决定嘛），但实测下来就是会翻车——因为「要不要检索」这个决定，模型经常做错：它对自己的记忆过于自信，觉得「这个我知道」，于是跳过检索直接答，幻觉就回来了。凡是「不能赌」的能力，就别交给模型自己判断，用架构强制保证。 这也是这一篇最重要的一个架构判断。

---

四、运行环境（同 005/006，多一个 Embedding 的 Key）

• 系统：WSL Ubuntu-24.04
• Python：3.11+，虚拟环境 venv
• 框架：AgentScope 2.x
• API Key：DEEPSEEK_API_KEY（对话）+ BAILIAN_API_KEY（Embedding 走百炼）

wsl -d Ubuntu-24.04
cd /2026_Study/agentscope/
source venv/bin/activate

export DEEPSEEK_API_KEY="sk-xxxx"   # 对话
export BAILIAN_API_KEY="sk-xxxx"    # Embedding

python 007_customer_service.py

---

五、完整代码（单文件自包含）

保存为 007_customer_service.py。三类问题演示双剑合璧：查价格（工具）/ 查知识（RAG）/ 两者结合。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AgentScope 2.0 · 双剑合璧——工具 + RAG，做一个会查会答的客服
  工具（QueryProductTool）：查结构化数据——价格、库存、适合肤质（字段精确）；
  RAG（Embedding 检索）：查非结构化知识——产品描述、成分、适用场景（语义匹配）。
运行环境：WSL Ubuntu-24.04 + Python 3.11+ + AgentScope 2.x
运行命令：python 007_customer_service.py
"""

import os
import gc
import math
import asyncio
import httpx
import warnings

warnings.filterwarnings("ignore", category=DeprecationWarning)

from agentscope.agent import Agent
from agentscope.message import Msg, TextBlock
from agentscope.model import DeepSeekChatModel
from agentscope.credential import DeepSeekCredential, DashScopeCredential
from agentscope.embedding import DashScopeEmbeddingModel
from agentscope.tool import ToolBase, ToolChunk, Toolkit
from agentscope.permission import PermissionDecision, PermissionBehavior


# 1. 敏辰产品库（结构化字段：价格/库存/肤质 + 非结构化描述）
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": "镇静舒缓，敏感肌可用，晒后修护"},
]


# 2. 剑一：工具（查结构化数据——价格 / 库存 / 肤质）
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)])


# 3. 剑二：RAG（查非结构化知识——描述 / 成分 / 场景）
def cosine_sim(a: list, b: list) -> float:
    """余弦相似度：两个向量夹角余弦，越接近 1 越相似。"""
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(y * y for y in b))
    return dot / (na * nb) if na and nb else 0.0


def _to_vector(emb) -> list[float]:
    if isinstance(emb, list):
        return emb
    if hasattr(emb, "vector"):
        return emb.vector
    return list(emb)


async def embed_vectors(model, texts: list[str]) -> list[list[float]]:
    """批量文本转向量（Embedding 是可调用对象，见 005 篇）。"""
    resp = await model(texts)
    return [_to_vector(e) for e in resp.embeddings]


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]


# 4. 模型构建
def build_embedding_model():
    """百炼文本嵌入模型（dimensions 必填，见 005 篇）。"""
    return DashScopeEmbeddingModel(
        credential=DashScopeCredential(api_key=os.environ["BAILIAN_API_KEY"]),
        model="text-embedding-v4",
        dimensions=1024,
    )


def build_chat_model():
    """对话模型（DeepSeek，stream=False 干净退出）。"""
    return DeepSeekChatModel(
        credential=DeepSeekCredential(api_key=os.environ["DEEPSEEK_API_KEY"]),
        model="deepseek-v4-flash",
        stream=False,
    )


def extract_text(response) -> str:
    for block in response.content:
        if hasattr(block, "type") and block.type == "text":
            return block.text
    return ""


async def cleanup(chat_model):
    """收尾：关闭对话模型连接池 + 兜底清理 embedding 残留。"""
    closed = 0
    client = getattr(chat_model, "client", None)
    if client is not None:
        try:
            await client.close()
            closed += 1
        except Exception:
            pass
    for obj in gc.get_objects():
        if isinstance(obj, httpx.AsyncClient) and not getattr(obj, "is_closed", False):
            try:
                await obj.aclose()
                closed += 1
            except Exception:
                pass
    if closed:
        print(f"\n🧹 已关闭 {closed} 个 HTTP 连接池")


# 5. 客服 Agent：工具 + RAG 双剑合璧
async def demo_customer_service(chat_model, emb, product_vecs) -> None:
    print("=" * 60)
    print("  双剑合璧：工具 + RAG 客服")
    print("=" * 60)

    agent = Agent(
        name="小帮",
        model=chat_model,
        system_prompt=(
            "你是「敏辰」护肤顾问「小帮」。\n"
            "你可以使用工具 query_product 查询产品的价格、库存、肤质。\n"
            "当用户问到具体产品的价格、有没有货时，请先调用工具查询。\n"
            "我会在每条消息里附上「参考知识」，请基于它回答护肤建议，不要编造。"
        ),
        toolkit=Toolkit(tools=[QueryProductTool()]),
    )

    cases = [
        "积雪草舒缓精华多少钱？有货吗？",
        "我皮肤容易泛红，有什么能修复屏障的？",
        "我皮肤干还起皮，帮我推荐个产品，顺便说下价格。",
    ]

    for q in cases:
        q_vec = (await embed_vectors(emb, [q]))[0]
        hits = retrieve(product_vecs, q_vec, k=2)
        ctx = "\n".join(f"[{i+1}] {p['name']}：{p['desc']}，适合{p['skin']}" for i, (_, p) in enumerate(hits))

        user_content = (
            f"【参考知识】\n{ctx}\n\n"
            f"用户问题：{q}\n\n"
            f"请严格基于【参考知识】回答护肤建议；"
            f"若用户问价格/库存，用工具查询后再回答；知识库未覆盖的请礼貌说明，严禁编造。"
        )

        resp = await agent.reply(
            Msg(name="user", content=[{"type": "text", "text": user_content}], role="user")
        )
        print(f"\n  👤 用户：{q}")
        print(f"  🤖 小帮：{extract_text(resp)}")

    print("\n  💡 观察：问价格→调工具；问护肤→用知识；两者结合→工具+RAG 一起上")


def check_keys():
    need = ["DEEPSEEK_API_KEY", "BAILIAN_API_KEY"]
    missing = [k for k in need if not os.environ.get(k)]
    if missing:
        print("❌ 缺少环境变量：" + ", ".join(missing))
        return False
    return True


async def main():
    if not check_keys():
        return

    chat_model = build_chat_model()
    emb = build_embedding_model()

    try:
        product_texts = [f"{p['name']}：{p['desc']}，适合{p['skin']}" for p in PRODUCTS]
        product_vecs = await embed_vectors(emb, product_texts)
        print(f"  ✅ 产品库已向量化：{len(product_vecs)} 条，维度 {len(product_vecs[0])}\n")

        await demo_customer_service(chat_model, emb, product_vecs)

        print("\n" + "=" * 60)
        print("  ✅ 双剑合璧客服完成（工具 + RAG）")
        print("=" * 60)
    finally:
        await cleanup(chat_model)
        del chat_model
        gc.collect()


if __name__ == "__main__":
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", DeprecationWarning)
        warnings.simplefilter("ignore", FutureWarning)
        warnings.simplefilter("ignore", PendingDeprecationWarning)
        asyncio.run(main())

---

六、代码拆解：三句话看懂合体

① 工具在 toolkit，知识在 prompt。 这是合体的关键动作——Agent 的 toolkit 参数里放了 QueryProductTool（剑一），而 RAG 检索出的知识则拼进每一条 user 消息（剑二）。两条路各走各的，互不干扰。

② 分工靠「指令」而不是「代码判断」。 你注意到没有，我们没有写 if 用户问价格 then 调工具 这种代码。分工是用 prompt 描述出来的：system_prompt 里说「问价格就调工具」，user 消息里说「基于参考知识回答」。模型自己会判断——这正是大模型和传统 if-else 的分水岭。

③ 两剑合璧的典型场景，是第三个 case。 「推荐个产品，顺便说价格」——Agent 先用 RAG 注入的知识选出「玻尿酸保湿霜」（因为用户说「干还起皮」），再主动调工具查它的价格。知识负责「选哪个」，工具负责「查详情」，两个能力接力完成一次服务。

④ 价格为什么必须走工具，而不是写进知识库？ 因为价格是实时变动的——今天 268、明天可能 258。如果写进知识库，你就得天天重新向量化；而工具每次都是实时查的，改数据库一行的值就生效。这正是「结构化数据用工具、非结构化知识用 RAG」这条分工线的深层原因：会变的数字，交给工具实时查；相对稳定的话术，交给 RAG 向量化。

---

七、运行结果（真实输出）

下面是本文件在 WSL Ubuntu-24.04 里的真实运行结果。三类问题，三种打法，全都踩在了点子上。

Case 1（查价格 → 工具） —— 问「多少钱有货吗」，Agent 调工具报出精确数据：

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

Case 2（查知识 → RAG） —— 问「泛红修复屏障」，Agent 基于检索知识推荐，不查价格：

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

Case 3（两者结合 → 工具 + RAG） —— 问「推荐个产品顺便说价格」，两剑接力：

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

最后一行 🧹 已关闭 1 个 HTTP 连接池，程序干净退出、零收尾报错。

最值得盯的是 Case 3：它把双剑合璧演得明明白白——

• 「选哪个」靠 RAG 知识：用户说「干还起皮」，语义检索锁定了「玻尿酸保湿霜」（深层补水）；
• 「报价格」靠工具：接着调用 query_product 查出 268 元、库存 50 件。

两个能力接力完成一次服务，中间没有写一行 if 用户说干 then 推荐保湿霜 的判断——全是大模型自己读懂了意图。这就是「会查会答」的真客服。

---

八、总结与下一步

这一篇，你把前六篇攒的两把剑，合成了一个能上岗的客服：

工具（004）= 查数字（价格/库存）
RAG（005/006）= 查话术（描述/成分）
双剑合璧（本篇）= 会查会答的真客服

核心就一句：工具和 RAG 不是二选一，而是分工协作——工具管结构化、RAG 管非结构化，Agent 按用户的问题自己判断用哪个、或两个一起用。

回头看，这个客服已经悄悄长成了一个小而全的系统：会多轮记忆（002）、有人设守红线（003）、能查价格库存（004）、懂护肤知识（005）、会调 Top-K 防幻觉（006）、还把查和答合体（本篇）。接下来的事，就是把它从一个「单兵」扩成一支「队伍」。

但到这里，我们的客服还是个「单打独斗」的 Agent。真实业务里，导购、售后、投诉是不同类型的活儿，让一个 Agent 全包会顾此失彼。下一篇《AgentScope 2.0 学习笔记：一分为多——意图路由与多 Agent 协作》，我们把一个 Agent 拆成一队：一个「路由器」判断用户意图，再分发给「导购 Agent」「售后 Agent」各司其职。敬请期待。

---

九、写在最后：从「会查会答」到「能上岗」

写完这一篇，系列已经走完了「一个客服 Agent 从 0 到能干活」的完整链路。回头看，其实每一篇解决的都是真实上线时的拦路虎：记忆是会话的根基、人设是合规的门、工具是数据的入口、RAG 是知识的入口、Top-K 是质量的阀——缺任何一个，这个客服都上不了线。

如果你跟着系列一路写到这里，手头应该已经有一个「麻雀虽小五脏俱全」的客服原型了。但「原型」和「上岗」之间，还差着不少东西：性能、并发、日志、评测、降级……这些是另一个维度的活儿。

如果你不是想「学着玩」，而是真想交付一个能上线的客服系统——不管是给自家业务做，还是给客户做——欢迎来评论区讨论。这套系列讲的是「积木怎么搭」，而「怎么搭得又稳又省」，是要对着具体业务一个个细节抠的。下一篇「多 Agent」见。

---

本文为「AgentScope 2.0 学习笔记」系列第 007 篇，代码已通过 py_compile 语法校验，运行环境见第四节。