前不久我在一个手机电商客服方向的实验项目里,用同一份手机机型数据集,把 LlamaIndex RAG 到多智能体 Agent 的完整链路跑了一遍。这个链路分三层:第一层是单机器人问答,用户问“3000 元拍照好的手机推荐哪款”,系统从产品库里检索出参数并给出带依据的回答;第二层是品牌专家路由,用户提到“极昼”或“云途”时,问题会被自动分发给对应品牌的专家引擎;第三层是多智能体协作下单,用户咨询完说“帮我下单”,产品专家、库存专员、下单专员三个 Agent 接力完成订单创建。整个过程用的数据只有一份:一百多行的手机产品表。
为什么拿手机机型数据而不是某本书或新闻语料来写教程?因为手机客服这个场景把 RAG 和 Agent 要面对的问题都凑齐了——既有结构化参数(价格、电池、摄像头),又有非结构化表达(“通勤用够不够”“拍照怎么样”),最后还要落到真实业务动作(查库存、下订单)。把这三层学完,你基本就能把同样的套路迁移到家电、理财、租赁等任何垂直客服场景。下面我按从简单到复杂的顺序,把每一层的实现思路、代码和踩过的坑完整写出来。
1. 为什么拿“手机机型数据集”当 RAG 与 Agent 的练手主线
1.1 一个数据集,正好覆盖问答、路由、协作三层复杂度
很多 RAG 教程喜欢用 PDF 或网页文章做素材,但那种数据有一个问题:检索到就基本算成功,因为答案就藏在片段里。真实客服场景完全不是这样,用户不会按教科书提问,也不会老老实实只问单个品牌。手机机型数据有一个优点:它天然就是“表格里的结构化知识”,但它要服务的却是“口语化的非结构化问题”。
我用的演示数据集长这样,品牌和型号都做了脱敏处理,统一换成化名:极昼、云途、山岚。
| 品牌 | 型号 | 上市年份 | 价格 | 主摄 | 电池 | 主要特点 |
|---|---|---|---|---|---|---|
| 极昼 | A15 Pro | 2024 | 5999 | 1 英寸大底主摄 | 4800mAh | 影像旗舰、金属机身 |
| 云途 | M10 | 2024 | 3999 | 5000 万像素主摄 | 5500mAh | 长续航、商务外观 |
| 山岚 | B8 | 2023 | 1999 | 6400 万像素主摄 | 5000mAh | 高性价比、游戏优化 |
这个表总共几十行,覆盖三个品牌、不同价位段。第一层单机器人问答,用整张表灌索引;第二层品牌路由,按 brand 字段把索引拆成三份,让每个品牌有自己的专家引擎;第三层下单,模型名和价格字段直接变成订单参数的来源。同一份数据三遍使用,代码量不大,但每一遍都在解决上一遍暴露的新问题。
1.2 LlamaIndex 在整套链路里到底扮演什么角色
LlamaIndex 不是数据库,也不是大模型,它是数据和 Agent 之间的一层编排框架。你可以把文档、切分规则、向量索引、查询引擎、工具、Agent 全部定义在一个项目里,它负责把这些组件串起来。我最看重的一点是:同一批文档,既能喂给向量索引做基础问答,也能封装成工具给 Agent 调用,不需要为不同阶段重写数据管道。
需要说明的是,LlamaIndex 的 API 版本迭代比较快,尤其是 Agent 部分。我下面的代码基于相对新的写法:单机器人问答用 VectorStoreIndex 和 query engine,多智能体用 AgentWorkflow 和 FunctionAgent。如果你装的是老版本,类名和参数名可能不同,但核心思路是一致的,照着思路换成对应 API 即可。
2. 第一层:单机器人问答——先把 RAG 的底子打稳
2.1 数据建模:把产品参数表翻成“人话文档”
很多人拿到表格第一反应就是直接把 CSV 灌进索引,我试过,效果很差。原因很简单:Embedding 模型是在自然语言句子上训练的,你给它键值对,它很难理解“5999 元对应的是哪一款、这款主打什么”。更好的做法是先把每一行手机参数拼成一段读起来像商品描述的文本,再作为 Document 交给索引。
from llama_index.core import Document from llama_index.core.node_parser import SentenceSplitter def build_documents(df): docs = [] for _, row in df.iterrows(): text = ( f"{row['brand']} {row['model']},{row['year']}年发布," f"上市价 {int(row['price'])} 元。" f"屏幕:{row['screen']};处理器:{row['cpu']};" f"主摄:{row['camera_main']};长焦:{row['camera_tele']};" f"电池:{row['battery']};充电:{row['charge']};" f"重量:{row['weight']};主要特点:{row['features']}。" ) docs.append(Document( text=text, metadata={ "brand": row["brand"], "model": row["model"], "price": float(row["price"]), "year": int(row["year"]), }, )) return docs这段代码里有两件事很重要。第一,正文 text 必须是完整的自然语言描述,这决定了检索能不能命中。第二,metadata 保留 brand、price、year,这样后面做路由和过滤时可以直接用,不用重新解析文本。数据建模阶段多花 10 分钟,后面能省很多麻烦。
2.2 构建索引与查询引擎
数据文档准备好之后,剩下的事情就简单了。切分、向量化、建索引,LlamaIndex 都可以在几行代码里完成。
from llama_index.core import VectorStoreIndex # splitter:对结构化产品表,chunk 不需要太大。 # 一条手机数据通常 200 字左右,chunk_size=512 够用, # 既能保证一条记录尽量在一个节点里,又不会切碎完整信息。 splitter = SentenceSplitter(chunk_size=512, chunk_overlap=0) nodes = splitter.get_nodes_from_documents(docs) # embed_model 换成你常用的中文 embedding 即可, # 本地开源模型或云端接口都行,只要 LlamaIndex 能调用。 index = VectorStoreIndex(nodes, embed_model=embed_model) query_engine = index.as_query_engine(similarity_top_k=4, llm=llm) resp = query_engine.query("预算 3500 左右,拍照好一点的手机推荐哪款?") print(resp)这里有一个容易踩的坑:不要以为 top_k 越大越好。对于手机参数表,每个节点代表一条完整机型信息,取 top 3 到 5 就够了。取太多会把不相关的机型也拉进来,LLM 生成答案时反而容易被无关参数干扰。另外,chunk_size 没必要设得特别大,因为你希望一个节点就是一个“完整机型”,而不是把多台手机拼进一个节点,否则检索到节点后模型要自己分辨是哪个品牌,错误率会明显上升。
查询输出大概是这样的:“根据目前产品信息,3500 元价位段推荐云途 M10,售价 3999 元,5000 万像素主摄,电池 5500mAh,主打商务长续航;如果预算下探到 3000 元,可以考虑山岚 B8……”只要检索到的节点是对的,稍微调一下提示词就能得到稳定格式。
2.3 单机器人的能力边界:哪些问题它回答不了
基础问答版本跑通后,我先不急着往上加复杂功能,而是拿真实客服问题去压它。这一压就发现了三个问题。
第一个问题是张冠李戴。用户问“极昼 A15 和云途 M10 哪个拍照好”,单机器人可能会把极昼的参数安到云途头上,因为两个品牌的数据都在同一个索引里,语义相近的片段相互干扰。第二个问题是品牌偏好没有被尊重。用户明确说“我只想看极昼”,系统却可能因为语义相近把云途的产品推出来。第三个问题最致命:用户问完“有货吗”之后,紧跟着说“帮我下单”,单机器人完全没法执行,因为它只有检索能力,没有调用业务系统的能力。
这三个问题对应三个升级方向:品牌路由、多品牌对比、Agent 工具调用。我先做了品牌路由,因为这是最容易立刻改善问答效果的一步。
3. 第二层:品牌专家路由——让问题自动转给对应品牌
3.1 为什么要按品牌拆索引,而不是全塞进一个索引
把三个品牌的数据塞进同一个向量索引,确实最省事。但客服场景天然是分品牌的,极昼用户关心影像,云途用户关心续航,山岚用户关心性价比,话术和促销政策也可能完全不同。全量索引的检索结果往往是“三个品牌各取几条”,回答时容易出现品牌信息串味。
按品牌拆索引后,每个品牌都有自己的知识库和查询引擎:极昼引擎只回答极昼的问题,云途引擎只回答云途的问题。这样路由一旦判断准确,答案的纯度会高很多,而且每个品牌后续可以独立挂接自己的促销接口、售后政策,互不影响。
brand_engines = { "极昼": index_ji_as_engine, "云途": index_yun_as_engine, "山岚": index_shan_as_engine, }这里的 index_ji、index_yun、index_shan 是怎么来的?就是从最开始那份数据里按 brand 字段筛选后各自构建出来的,代码和 2.2 完全一致,只是数据换成了对应品牌的行。一份数据集这时候开始展现出复用的价值。
3.2 两种路由实现:查询引擎路由与函数工具路由
路由最简单的实现是用 LlamaIndex 的 RouterQueryEngine。它本质上是一个“路由器”,根据用户问题自动选择最合适的一个查询引擎。
from llama_index.core.query_engine import RouterQueryEngine from llama_index.core.tools import QueryEngineTool router_engine = RouterQueryEngine( query_engine_tools=[ QueryEngineTool.from_defaults( brand_engines["极昼"], description="查询极昼品牌手机的型号、参数、价格与特点", ), QueryEngineTool.from_defaults( brand_engines["云途"], description="查询云途品牌手机的型号、参数、价格与特点", ), QueryEngineTool.from_defaults( brand_engines["山岚"], description="查询山岚品牌手机的型号、参数、价格与特点", ), ] ) resp = router_engine.query("云途 M10 电池多少毫安?")这种做法的好处是代码量少,适合只想先看到效果的场景。但它的短板也很明显:路由器本身是黑盒,你不好控制“跨品牌问题”和“品牌未识别问题”的走向。所以我更推荐第二种做法:把路由逻辑做成一个函数工具,交给 Agent 使用。这样路由判断的细节可以自己控制,而且后续升级到多智能体时,这个工具可以直接复用。
def ask_brand_expert(question: str, brand: str) -> str: """查询指定品牌手机的型号、参数、价格等事实问题。""" if brand not in brand_engines: return f"当前没有品牌[{brand}]的知识库,请向用户说明无法回答。" return str(brand_engines[brand].query(question))再加上一个品牌识别函数,先用规则匹配品牌关键词,匹配不到就返回 unknown:
BRAND_LIST = ["极昼", "云途", "山岚"] def detect_brand(question: str) -> str: for brand in BRAND_LIST: if brand in question: return brand return "unknown"规则匹配的好处是不消耗大模型调用,几乎零延迟。真实场景里的品牌名可能被写错或省略,这时候再用 LLM 做二次识别补漏。我的经验是:能用规则就用规则,规则覆盖不到的才交给模型,这样路由稳定性反而更高。
3.3 路由兜底:未知品牌、跨品牌、多品牌怎么处理
路由做完后,测试问题集里冒出三类特别容易翻车的情况。
第一类是用户没提品牌,只说“3000 元以下推荐一台”。这不该路由到某个品牌专家,而应该走全量索引做综合推荐。我的做法是:检测到 unknown 时,调用一个默认的全量引擎兜底,不做品牌切分。第二类是跨品牌对比,用户问“极昼和云途谁续航好”。如果硬路由给其中一个品牌,答案就会偏颇。正确的做法是在系统提示词里要求路由 Agent 发现多品牌意图时,先分别查询两个品牌的资料,再整理成对比答案。第三类是用户用别名提问,比如“主打影像的极昼旗舰”,问题里没有出现完整品牌名。这时规则匹配会失败,需要依靠 LLM 补充识别。
兜底策略不是越复杂越好,关键是给“失败”留一个出口。我在路由函数里永远保留一个“未识别”分支,宁可回答“请稍等,我来查一下综合资料”,也绝不硬猜品牌。硬猜一次,用户信任度就掉一次。
4. 第三层:多智能体协作下单——从“能回答”到“能办事”
4.1 三个角色:产品专家、库存专员、下单专员
到了这一步,系统要从“回答问题”升级成“完成任务”。用户提出“帮我查云途 M10 有没有货,有货就下单”时,涉及三类知识:机型信息、库存状态、下单动作。如果所有逻辑都塞在一个 Agent 里,提示词会越来越长,工具一多就容易互相干扰。所以我把职责拆成三个角色,每个角色只负责一件事。
| Agent | 职责 | 拥有的工具 | 不负责的事 |
|---|---|---|---|
| 产品专家 | 回答机型、参数、价格、优缺点 | 品牌知识检索 | 库存、下单 |
| 库存专员 | 查询某机型在某城市是否有货、预计发货时间 | 库存查询 | 推荐机型、下单 |
| 下单专员 | 创建订单、确认收货信息 | 下单工具 | 机型推荐、库存 |
这样拆的好处是,每个 Agent 的 system prompt 可以写得非常收敛。产品专家不需要知道订单表结构,下单专员也不需要理解摄像头像素。哪个环节出了问题,定位时一眼就能看出是哪个角色的责任。
4.2 工具封装:让 Agent 拥有可以调用的“手”
多智能体的基础是工具。工具的名字、参数、说明都是写给 Agent 看的,所以 docstring 必须说清楚“这个工具能做什么、参数要求什么、什么时候不能用”。
def search_knowledge(question: str, brand: str) -> str: """从对应品牌知识库检索机型、参数、价格、特点。 只能回答知识库中存在的信息,不存在时如实说明。""" if brand not in brand_engines: return f"没有品牌[{brand}]的知识库。" return str(brand_engines[brand].query(question)) def check_stock(model: str, city: str) -> dict: """查询指定城市某型号是否有现货,返回库存状态。 注意:model 必须是完整型号名,例如'云途 M10'。""" stock_table = load_stock_table() rec = stock_table.get((model, city)) if rec is None: return {"in_stock": False, "eta_days": None} return {"in_stock": rec["count"] > 0, "eta_days": rec["eta_days"]} def place_order(customer_name: str, model: str, city: str, address: str) -> dict: """创建订单。调用前必须得到用户明确确认,否则禁止调用。""" order_id = create_order_record(customer_name, model, city, address) return {"order_id": order_id, "status": "created", "model": model}一个问题值得注意:工具返回格式尽量统一,我用 dict,并且把是否成功、失败原因都放进返回里。千万不要让工具在失败时抛异常,Agent 一旦遇到异常容易自己编造结果。我把返回改为包含 in_stock 与 eta_days 这样的结构,Agent 拿到后能直接组织自然语言回复,不需要二次猜。
4.3 先跑通总控 Agent,再升级为多智能体 Workflow
如果一上来就上三个 Agent,出了问题你都不知道找谁。我的做法是先写一个总控 Agent,把三个工具全部挂在一个 Agent 下跑通流程,确认工具本身可靠后,再拆成独立角色。两个阶段我都写出来。
总控 Agent 版本:
from llama_index.core.agent.workflow import FunctionAgent from llama_index.core.tools import FunctionTool knowledge_tool = FunctionTool.from_defaults(fn=search_knowledge) stock_tool = FunctionTool.from_defaults(fn=check_stock) order_tool = FunctionTool.from_defaults(fn=place_order) controller = FunctionAgent( name="客服总控", description="手机客服总控,负责回答产品问题并协助用户下单。", system_prompt=( "你是手机客服总控。回答必须引用检索到的真实参数;" "用户未明确表达下单意向前,禁止调用下单工具;" "订单必须包含客户姓名、机型、城市、地址,缺一不可。" ), tools=[knowledge_tool, stock_tool, order_tool], )跑通后,再拆成三个独立 Agent,用 AgentWorkflow 编排。核心变化是每个 Agent 只拿自己的工具,通过移交机制把任务交给下一个角色。
from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent product_expert = FunctionAgent( name="产品专家", description="负责回答手机参数、价格、拍照、续航等知识问题。", system_prompt="只回答知识库能查到的事实,不承诺折扣,不下单。", tools=[knowledge_tool], can_handoff_to=["库存专员", "下单专员"], # 不同版本参数名可能不同,以官方文档为准 ) stock_specialist = FunctionAgent( name="库存专员", description="负责查询指定城市和型号的库存情况。", system_prompt="只查询库存,不报价不下单,查完结果交给其他角色。", tools=[stock_tool], can_handoff_to=["下单专员"], ) order_specialist = FunctionAgent( name="下单专员", description="负责创建订单,确认收货信息。", system_prompt="用户未确认前禁止下单。下单前复述订单信息请用户确认。", tools=[order_tool], ) workflow = AgentWorkflow( agents=[product_expert, stock_specialist, order_specialist], root_agent=product_expert, timeout=120, )协作流程跑起来后大致是:用户说“云途 M10 有货吗,有就下单”——产品专家先确认机型,把查询库存的意图移交给库存专员;库存专员返回“有货,预计 2 天送达”;产品专家告知用户;用户回复“确认下单”;下单专员校验用户姓名和地址后调用 place_order,返回订单号。整个过程每个角色只干自己那件事,上下文清晰很多。
4.4 下单场景的安全边界:确认、幂等、信息脱敏
Agent 一旦能触达下单接口,安全就是第一优先级。我在这个项目里加了三条硬规则。
第一,下单前必须用户明确确认。我在 system prompt 里反复强调“用户未确认前禁止调用下单工具”,同时在工具函数里也写清楚这个前提。有些用户只是问“能不能下单”并不是真的想下,Agent 不能自作主张。
第二,订单接口要做幂等。如果 Agent 因为网络抖动重试了一次,不应该产生两笔订单。我在 create_order_record 里用 customer_name + model + city + address 的哈希做订单去重键,重复提交返回同一个订单号。
第三,日志脱敏。Agent 在对话过程中会拿到用户姓名、电话、地址,调试日志里绝不能原样打印。我在工具返回和日志记录里把地址数据截断成“市 + 区 + 小区前两个字”,电话号码只保留后四位。这个习惯即使项目只是在 Demo 阶段也需要养成,后期迁移生产时能少踩很多合规的坑。
5. 全流程复盘:评估方法、检索调优与多智能体 Debug 心得
5.1 用一小组问题集检验三层链路
整套链路搭完之后,最关键的步骤是评估。我先准备了一份 20 条问题的小测试集,覆盖三类场景:单品牌参数查询、跨品牌对比、完整下单流程。每次改完代码,就跑一遍这 20 条,记录下面几个指标。
| 测试类型 | 关键指标 | 我遇到的主要问题 |
|---|---|---|
| 单品牌参数查询 | 回答中的型号、参数是否正确 | 检索到错误品牌节点 |
| 跨品牌对比 | 是否分别引用两个品牌真实信息 | 只路由到单一品牌 |
| 下单流程 | 是否产生正确订单号 | 用户未确认就尝试下单 |
| 未知问题 | 是否诚实说不知道 | Agent 编造不存在的机型 |
这个测试集不大,但非常有效。它帮我发现了一个很隐蔽的问题:有两条下单测试里,用户地址被工具调用时丢了一个字段,Agent 没有主动追问,而是默默用“未知地址”直接下单成功。后来我在 system prompt 里强制要求“创建订单前必须逐项确认姓名、地址、数量”,并在 place_order 内部增加参数校验,缺字段就返回错误,这个 bug 才彻底堵住。
5.2 检索质量最容易被忽略的三个细节
第一是 top_k 的选取。刚才在单机器人部分说过,top 3 到 5 比较好,但对路由后的品牌索引,每个索引数据量变小,top_k 可以降到 3,减少噪音。第二是元数据过滤。用户给出预算区间时,不要只靠向量相似度去猜,而是先用 metadata 里的 price 字段过滤一遍,再进向量检索,准确率会高出不少。第三是切块粒度。对于产品表这类“一条记录一个完整答案”的数据,千万别用默认的长文本切分,否则一条手机数据可能被切成两半,导致回答缺失电池或摄像头参数。
很多搜索质量问题的根源其实都在 data modeling,而不是模型本身。检索结果错了,换再强的 LLM 也只能把错的内容编得更流畅。
5.3 多智能体跑起来之后怎么调试
多智能体调试和普通 RAG 调试完全不一样。RAG 出问题,你看检索节点和生成答案就行;多智能体出问题,你还得看每个 Agent 理解了什么、调用了什么工具、把任务移交给了谁。我的经验有三个。
第一,开启完整事件日志。AgentWorkflow 支持事件回调,把每一步的输入输出都打出来。第二,给 workflow 设置 timeout,我设了 120 秒,超过就终止,避免 Agent 在工具调用里死循环。第三,限制重试次数。Agent 调用工具失败时最多重试一次,否则它可能反复尝试,白白消耗 token 和时间。
还有一个排查技巧:如果 Agent 回答里出现某个参数,但你知道检索结果里没有这个信息,大概率是 LLM 幻觉,而不是 Agent 逻辑问题。这时候不要急着改 prompt,先去看它实际检索到的节点内容,往往问题出在 embedding 模型对专业词汇的相似度区分不够,或者切块把关键数字切丢了。
整个链路跑下来,我最深的体会是:多智能体的难点不在于“让多个 Agent 聊天”,而在于把数据边界、工具边界、职责边界划清楚。产品表一份,路由一层,工具一层,Agent 一层,每一层之间都靠清晰的接口连接。你加一个品牌,只需要往数据里加几行,往 brand_engines 里加一个引擎;你加一个业务动作,只需要写一个新的 FunctionTool。这种结构化的演进方式,比一开始就堆复杂架构要省心得多。
最后再分享一个小技巧:在项目初期,我会把每个版本的测试结果记录在一个表格里,备注是哪一次改动导致指标上升或下降。多智能体系统的不确定性比普通 RAG 高很多,没有记录,你很难判断到底是 prompt 的问题、工具的问题还是数据的问题。拿同一份手机机型数据,从单机器人问答做到品牌专家路由,再到多智能体协作下单,这套路径值得每个打算深入 Agent 应用的人亲手跑一遍。