基于LlamaIndex的多智能体手机客服系统：RAG检索与Agent编排实战

1. 项目缘起与整体架构设计

手机客服这个场景，做过的人都知道，它跟通用问答机器人完全是两码事。用户问“这个机型支持多少瓦快充”，你答错一个数字，售后成本立刻飙升；用户说“我要办合约机”，你路由错了业务线，一单就黄了。我接手这个模拟项目的时候，核心诉求就三条：机型参数必须准、品牌知识必须专、下单流程必须通。而这三条恰好对应了三种不同的能力——单机器人问答、品牌专家路由、多智能体协作。LlamaIndex 这套框架最吸引我的地方，就是它能把 RAG 检索和 Agent 编排揉在同一个体系里，不用在多个框架之间来回倒腾。

先说整体思路。我把整个系统拆成了三层：数据层、检索层、编排层。数据层是一份结构化的手机机型数据集，包含品牌、型号、芯片、屏幕、电池、快充功率、摄像头参数、价格区间、合约方案等字段。检索层用 LlamaIndex 的 VectorStoreIndex 做语义检索，同时挂了一个 SummaryIndex 做精确字段查询。编排层是重点——我用 AgentRunner 作为总入口，底下挂三个子 Agent：参数问答 Agent、品牌专家 Agent、下单协作 Agent。用户的问题进来，先由路由 Agent 判断意图，再分发给对应的子 Agent 处理。

为什么这么设计？因为单一 RAG 链路在处理“帮我对比一下 A 品牌旗舰和 B 品牌旗舰的影像能力”这种问题时，检索回来的 chunk 往往是碎片化的，模型拼出来的答案容易张冠李戴。而引入品牌专家 Agent 之后，每个品牌专家只负责自己品牌的知识库，检索范围收窄，准确率明显提升。下单协作 Agent 则是一个典型的工具调用场景——它需要查库存、算价格、生成订单号，这些都不是纯文本检索能搞定的，必须走 Function Calling。

提示：多智能体不是越多越好。我一开始设计了五个 Agent，结果路由准确率反而下降，因为意图边界模糊了。后来砍到三个，每个 Agent 的职责用一句话能说清楚，路由准确率从 72% 拉到了 91%。

整个项目的技术栈选型如下表所示：

这个选型逻辑的核心就一句话：数据不出域、延迟可控、编排灵活。手机机型参数属于商业敏感信息，走外部 API 做 embedding 我不放心；客服场景用户等三秒就挂电话，本地推理虽然单次质量略逊于超大模型，但胜在稳定和快。

2. 手机机型数据集的结构化处理与索引构建

数据集是整个系统的地基。我拿到的原始数据是一份 CSV，大概两千多行，每行一个机型，字段包括品牌、型号、发布时间、芯片型号、屏幕尺寸、分辨率、刷新率、电池容量、快充功率、后置摄像头配置、前置摄像头配置、内存组合、存储组合、官方售价、合约价、库存状态。这份数据看着规整，但直接灌进向量库效果很差，因为用户不会用字段名提问。

2.1 从结构化字段到自然语言 chunk 的转换策略

我的做法是：每个机型生成多条自然语言描述，而不是把整行 CSV 拼成一段文本。具体来说，一个机型会生成以下几类 chunk：

• 参数概览 chunk：把核心参数用一段话串起来，比如“某品牌某型号搭载某芯片，配备 6.7 英寸 120Hz 屏幕，电池容量 5000mAh，支持 120W 快充，后置三摄主摄 5000 万像素”。
• 卖点 chunk：针对影像、续航、性能等维度分别生成描述，方便用户问“这个手机拍照怎么样”时精准命中。
• 合约方案 chunk：把合约价、月费、合约期、包含流量等信息单独成块。
• 对比 chunk：同品牌同价位段的机型生成对比描述，方便处理“A 和 B 哪个更值得买”这类问题。

为什么要拆这么细？因为向量检索的本质是语义相似度匹配。用户问“快充多少瓦”，如果 chunk 里是一整行 CSV 拼接的文本，快充功率这个信息被淹没在大量字段中，相似度会被稀释。拆成参数概览 chunk 之后，快充功率在 chunk 中的权重相对提高，召回准确率肉眼可见地上升。

# 机型 chunk 生成的核心逻辑示意 def build_machine_chunks(row): chunks = [] # 参数概览 overview = f"{row['brand']}{row['model']}搭载{row['chip']}芯片，" overview += f"屏幕{row['screen_size']}英寸{row['refresh_rate']}Hz，" overview += f"电池{row['battery']}mAh，支持{row['fast_charge']}W快充。" chunks.append({"text": overview, "type": "overview", "model": row['model']}) # 影像卖点 camera = f"{row['brand']}{row['model']}后置{row['rear_camera']}，" camera += f"前置{row['front_camera']}，支持{row['camera_feature']}。" chunks.append({"text": camera, "type": "camera", "model": row['model']}) # 合约方案 contract = f"{row['brand']}{row['model']}合约价{row['contract_price']}元，" contract += f"月费{row['monthly_fee']}元，合约期{row['contract_period']}个月。" chunks.append({"text": contract, "type": "contract", "model": row['model']}) return chunks

这段逻辑看着简单，但有几个坑我踩过。第一个坑是数字格式不统一，原始数据里快充功率有的写“120W”，有的写“120瓦”，有的写“120”，不统一的话检索时用户问“120W 快充”可能匹配不到写“120瓦”的那条。我的处理是在 chunk 生成阶段统一转成“120W”格式，同时在 metadata 里保留原始值。第二个坑是型号别名，用户可能说“某品牌 15 Pro”也可能说“某品牌十五 Pro”，我在 chunk 里同时写入数字和中文数字两种表述。

2.2 索引构建与 metadata 设计

LlamaIndex 的 VectorStoreIndex 默认会把 chunk 的文本做 embedding，但 metadata 不会自动参与检索。我的做法是把关键字段写进 metadata，然后在检索时用 MetadataFilters 做预过滤。比如用户明确说了“某品牌”，我就先按 brand 字段过滤，再在过滤后的子集里做向量检索。这样既提高了准确率，又减少了检索范围，延迟也降下来了。

metadata 的设计如下：

构建索引的时候，我用的是VectorStoreIndex.from_documents，同时挂了一个SummaryIndex用于精确字段查询。为什么还要 SummaryIndex？因为有些问题是精确匹配，比如“某型号的电池容量是多少”，这种问题用向量检索反而容易召回相似但不相关的机型，而 SummaryIndex 配合关键词过滤能直接定位到目标 chunk。

注意：SummaryIndex 在 LlamaIndex 里默认是全量遍历，数据量大的时候延迟很高。我的优化是先用 metadata 过滤缩小范围，再走 SummaryIndex，实测延迟从 800ms 降到了 120ms 左右。

3. 单机器人问答链路的搭建与调优

单机器人问答是整个系统的基础能力，也是用户最常触发的路径。这条链路的目标很明确：用户问机型参数，机器人答得准、答得快、答得全。听起来简单，但要做到“准”和“全”之间的平衡，调优花了我不少时间。

3.1 检索策略：Top-K 与相似度阈值的取舍

LlamaIndex 默认的similarity_top_k是 2，这个值在通用场景下够用，但在机型参数场景下明显不够。用户问“这个手机的影像配置”，可能涉及主摄、超广角、长焦、防抖、视频规格等多个 chunk，Top-2 根本覆盖不全。我把 Top-K 调到了 6，同时加了一个相似度阈值 0.75，低于这个阈值的 chunk 直接丢弃。

为什么是 6 和 0.75？这是我用一批测试问题跑出来的。Top-K 从 2 加到 6，召回率从 61% 提升到 89%；加到 8 的时候召回率只提升了 2 个百分点，但延迟增加了 40%。0.75 的阈值是我观察相似度分布后定的，低于 0.75 的 chunk 基本都是跨品牌或跨价位的噪声。这两个参数不是拍脑袋定的，建议你也用自己的数据跑一遍网格搜索。

from llama_index.core import VectorStoreIndex, QueryBundle from llama_index.core.retrievers import VectorIndexRetriever from llama_index.core.postprocessor import SimilarityPostprocessor retriever = VectorIndexRetriever( index=index, similarity_top_k=6, ) postprocessor = SimilarityPostprocessor(similarity_cutoff=0.75) query_bundle = QueryBundle(query_str="这个手机拍照怎么样") nodes = retriever.retrieve(query_bundle) nodes = postprocessor.postprocess_nodes(nodes)

3.2 回答生成：Prompt 模板的设计要点

检索回来的 chunk 怎么喂给模型，直接决定了回答质量。我试过三种 Prompt 模板：

第一种是直接拼接，把所有 chunk 文本拼在一起让模型总结。问题是 chunk 之间有重复信息，模型容易啰嗦，而且不同机型的参数混在一起时容易串台。

第二种是带来源标注的拼接，每个 chunk 前面加上“来源：某品牌某型号”。这样模型知道每段话对应哪个机型，串台问题明显改善。

第三种是结构化注入，把 chunk 按类型分组，参数类放一组、影像类放一组、合约类放一组，每组前面加小标题。这种效果最好，但 Prompt 长度也最长。

最终我选了第二种和第三种结合：按类型分组 + 来源标注。Prompt 模板大致如下：

你是一个手机客服助手。请根据以下检索到的机型信息回答用户问题。 如果信息不足以回答，请明确告知用户，不要编造参数。 检索信息： {context_str} 用户问题：{query_str} 回答要求： 1. 参数必须与检索信息一致，不得修改数字。 2. 如果涉及多个机型对比，请分点说明。 3. 如果用户问的是合约方案，请同时说明合约期和月费。

这个模板里最关键的是第一条“不得修改数字”。我实测发现，如果不强调这一点，模型有时候会把“120W”写成“120瓦”或者“100W”，虽然语义相近，但在客服场景下这是事故。

3.3 单机器人链路的性能实测

我用 200 条真实客服问题做了测试，覆盖参数查询、对比、合约咨询三类意图。结果如下：

参数准确率没到 100%，主要失分在跨品牌对比问题上。比如用户问“A 品牌旗舰和 B 品牌旗舰哪个快充更快”，单机器人链路检索回来的 chunk 可能只覆盖了一个品牌，导致对比不完整。这个问题在后面引入品牌专家 Agent 之后才得到解决。

实操心得：单机器人链路不要追求万能。它的定位就是“快问快答”，复杂问题应该路由给专家 Agent。我一开始想让单机器人搞定所有问题，结果 Prompt 越写越长，延迟越来越高，准确率反而下降。后来明确边界之后，整体体验反而好了。

4. 品牌专家路由机制的设计与实现

品牌专家路由解决的是“专”的问题。手机行业有个特点：每个品牌的参数体系、卖点话术、合约方案都不一样。A 品牌强调影像，B 品牌强调续航，C 品牌强调性价比。用一个通用机器人回答所有品牌的问题，话术不专业，用户一眼就能看出来。

4.1 路由意图识别：从关键词到语义匹配

路由的第一步是判断用户问题属于哪个品牌。我试过两种方案：

方案一：关键词匹配。维护一个品牌别名词典，用户问题里出现某个品牌名或别名，就路由到对应专家。这个方案快，但覆盖不全。用户说“那个主打拍照的牌子”，关键词匹配就歇菜了。

方案二：语义路由。用 embedding 把用户问题向量化，跟每个品牌的描述向量做相似度比较，取最高分。这个方案覆盖全，但有个问题：用户问“哪个牌子拍照好”，语义上跟多个品牌都相似，路由结果不稳定。

最终我用的是混合方案：先走关键词匹配，命中则直接路由；未命中则走语义路由，但加一个置信度阈值，低于阈值则路由到通用 Agent 而不是强行分给某个品牌专家。这个逻辑用 LlamaIndex 的RouterRetriever或者自定义Selector都能实现，我用的是自定义 Selector，因为需要嵌入业务规则。

class BrandRouterSelector(Selector): def __init__(self, brand_experts, brand_keywords, embed_model): self.brand_experts = brand_experts self.brand_keywords = brand_keywords self.embed_model = embed_model def _select(self, choices, query_bundle): query = query_bundle.query_str # 关键词优先 for brand, keywords in self.brand_keywords.items(): if any(kw in query for kw in keywords): return [brand] # 语义兜底 query_emb = self.embed_model.get_text_embedding(query) best_brand, best_score = None, 0 for brand, expert in self.brand_experts.items(): score = cosine_similarity(query_emb, expert.description_emb) if score > best_score: best_brand, best_score = brand, score if best_score > 0.82: return [best_brand] return ["general"]

这个 Selector 的核心逻辑就三层：关键词命中直接返回、语义相似度高于阈值返回对应品牌、否则返回通用。阈值 0.82 是我调出来的，低于这个值强行路由到品牌专家，回答质量反而不如通用 Agent。

4.2 品牌专家的知识库隔离

每个品牌专家 Agent 挂载的是该品牌专属的索引。为什么要隔离？因为如果所有品牌共用一个索引，检索时跨品牌噪声很难避免。隔离之后，A 品牌专家检索时只会在 A 品牌的 chunk 里找，准确率提升非常明显。

隔离的实现方式有两种：物理隔离和逻辑隔离。物理隔离是每个品牌建一个独立的 VectorStoreIndex，逻辑隔离是一个索引但用 metadata 过滤。我选的是物理隔离，因为机型数据量不大，每个品牌几千个 chunk，内存完全扛得住，而且物理隔离的检索延迟更低。

品牌专家的 Prompt 也做了定制。比如影像旗舰品牌的专家，Prompt 里会强调“重点说明影像配置和拍摄场景”；性价比品牌的专家，Prompt 里会强调“重点说明价格优势和合约方案”。这样每个专家的话术风格跟品牌调性一致，用户感知更专业。

4.3 路由准确率的实测与优化

路由准确率我用 300 条测试问题做了评估，结果如下：

混合方案比纯语义提升了 12 个百分点，主要贡献来自关键词优先策略。用户明确说了品牌名的时候，关键词匹配几乎不会错，而语义匹配反而可能因为表述差异路由到错误品牌。

避坑技巧：品牌别名要维护全。我一开始只写了品牌全称，结果用户说“某品牌”的简称时路由失败。后来我把常见简称、英文名、甚至用户常说的错别字都加进了别名词典，路由准确率又提升了 5 个百分点。

5. 多智能体协作下单全流程拆解

下单是客服场景里最复杂的环节，因为它涉及多轮对话、工具调用、状态管理。用户说“我要办这个机型的合约”，背后需要：确认机型、确认合约方案、查库存、算价格、生成订单、确认支付方式。这一串操作单靠一个 Agent 很难搞定，所以我用了多智能体协作。

5.1 协作架构：Orchestrator 与 Worker 的分工

我的协作架构是一个 Orchestrator Agent 带三个 Worker Agent：

• Orchestrator：负责理解用户意图、拆解任务、调度 Worker、汇总结果。
• Inventory Worker：负责查库存，输入机型型号，输出库存状态和预计到货时间。
• Pricing Worker：负责算价格，输入机型型号和合约方案，输出合约价、月费、总成本。
• Order Worker：负责生成订单，输入用户信息和机型信息，输出订单号和确认信息。

Orchestrator 本身不查数据，它只做调度。这样做的好处是每个 Worker 的职责单一，Prompt 可以写得很聚焦，工具调用的参数校验也更容易做。

from llama_index.core.agent import AgentRunner from llama_index.core.tools import FunctionTool # 定义 Worker 工具 inventory_tool = FunctionTool.from_defaults( fn=check_inventory, name="check_inventory", description="查询指定机型的库存状态，输入机型型号，返回库存数量和到货时间" ) pricing_tool = FunctionTool.from_defaults( fn=calculate_price, name="calculate_price", description="计算合约方案价格，输入机型型号和合约期，返回合约价、月费和总成本" ) order_tool = FunctionTool.from_defaults( fn=create_order, name="create_order", description="生成订单，输入用户ID、机型型号、合约方案，返回订单号" ) # Orchestrator 挂载所有工具 orchestrator = AgentRunner.from_llm( tools=[inventory_tool, pricing_tool, order_tool], llm=llm, system_prompt="你是一个手机客服下单助手。用户要办合约时，先查库存，再算价格，最后生成订单。每一步都要跟用户确认。" )

5.2 多轮对话的状态管理

下单流程天然是多轮的。用户说“我要办合约”，Orchestrator 需要追问“哪个机型”“哪个合约期”“你的用户 ID 是多少”。这些信息需要在多轮对话中保持，LlamaIndex 的ChatMemoryBuffer可以做到，但默认的 memory 是全局的，多个用户并发时会串。

我的做法是每个会话一个 memory 实例，用 session_id 做隔离。同时，我在 memory 里额外存了一个order_state字典，记录当前下单进度：已确认机型、已确认合约方案、已查库存、已算价格、已生成订单。Orchestrator 每轮对话前先读order_state，知道当前进行到哪一步，避免重复询问。

class OrderSession: def __init__(self, session_id): self.session_id = session_id self.memory = ChatMemoryBuffer.from_defaults(token_limit=3000) self.order_state = { "model": None, "contract_period": None, "inventory_checked": False, "price_calculated": False, "order_created": False, "order_id": None }

这个order_state是下单流程的核心。没有它，Orchestrator 每轮都要从对话历史里重新推断进度，容易出错。有了它，每一步的状态都是显式的，逻辑清晰很多。

5.3 工具调用的参数校验与异常处理

工具调用最容易出问题的地方是参数。用户说“我要办 24 个月的合约”，Orchestrator 提取出contract_period=24，但 Pricing Worker 可能只支持 12、24、36 三个档位，24 是合法的，但如果用户说“18 个月”，就会报错。

我的处理是在工具函数内部做参数校验，不合法时返回明确的错误信息，让 Orchestrator 转述给用户。比如：

def calculate_price(model: str, contract_period: int) -> dict: valid_periods = [12, 24, 36] if contract_period not in valid_periods: return { "error": f"合约期 {contract_period} 个月暂不支持", "valid_options": valid_periods } # 正常计算逻辑 ...

这样做的好处是 Orchestrator 不需要预判所有异常，工具自己会兜底。Orchestrator 收到 error 之后，把 valid_options 转述给用户，用户重新选择即可。

实操心得：工具函数的返回值一定要结构化。我一开始让工具返回纯文本，Orchestrator 解析起来经常出错。后来改成返回 dict，Orchestrator 直接读字段，稳定性大幅提升。

5.4 下单全流程的实测记录

我用模拟用户跑了 50 单完整下单流程，成功率 88%。失败的 12% 主要卡在两个地方：一是用户中途改主意换机型，Orchestrator 没有正确重置 order_state；二是库存查询超时，工具没有做重试。

针对第一个问题，我在 Orchestrator 的 Prompt 里加了一条规则：“如果用户更换机型，必须重置 order_state 中所有已确认字段”。针对第二个问题，我给库存查询工具加了两次重试，超时时间从 3 秒调到 5 秒。优化之后成功率拉到了 96%。

6. 常见问题排查与避坑经验实录

这个项目从搭建到跑通，我踩的坑不算少。挑几个最有代表性的分享出来，希望能帮你少走弯路。

6.1 检索召回不准的排查思路

召回不准是最常见的问题。我的排查顺序是：先看 chunk 质量，再看 embedding 模型，最后看检索参数。

chunk 质量的问题通常是切分太粗或太细。太粗的话，一个 chunk 里混了多个机型的信息，检索时相似度被稀释；太细的话，一个完整参数被拆成好几块，模型拼不起来。我的经验是：一个 chunk 只讲一个机型的一个维度，比如“某型号的影像配置”是一个 chunk，“某型号的续航配置”是另一个 chunk。

embedding 模型的问题通常是领域不匹配。通用 embedding 模型对手机参数这种专业术语的区分度不够，比如“快充”和“闪充”在通用模型里可能很相似，但在手机行业里是两个不同的概念。如果条件允许，用领域数据微调一下 embedding 模型，效果提升很明显。

检索参数的问题就是 Top-K 和阈值。这两个参数没有万能值，必须用自己的数据调。我的建议是先用一批标注好的测试问题，跑一遍不同参数组合，画一条召回率-延迟曲线，选拐点。

6.2 多智能体路由错乱的修复

路由错乱的表现是：用户问 A 品牌的问题，被路由到了 B 品牌专家。排查的时候先看路由日志，确认是关键词没命中还是语义相似度算错了。

关键词没命中通常是别名词典不全。我的做法是定期从用户问题里挖掘新出现的品牌表述，补充到词典里。语义相似度算错通常是品牌描述向量不够区分。每个品牌的描述文本要突出该品牌的独特卖点，不要写得太泛。

还有一个隐蔽的问题是路由阈值设得太低。阈值低的时候，用户问一个模糊问题，语义相似度可能对多个品牌都超过阈值，路由结果就不稳定。我的经验是阈值宁高勿低，高于阈值才路由到品牌专家，低于阈值走通用 Agent，这样至少不会答错品牌。

6.3 工具调用失败的兜底策略

工具调用失败的原因很多：参数格式不对、外部服务超时、返回值解析失败。我的兜底策略是三层防护：

第一层是参数校验，在工具函数入口做，不合法直接返回错误信息。第二层是重试机制，对外部服务调用加两次重试，指数退避。第三层是降级方案，如果工具彻底失败，Orchestrator 转述一个友好的错误提示，并建议用户稍后重试或转人工。

import time def with_retry(fn, max_retries=2, backoff=1.0): def wrapper(*args, **kwargs): for attempt in range(max_retries + 1): try: return fn(*args, **kwargs) except Exception as e: if attempt == max_retries: return {"error": f"服务暂时不可用：{str(e)}"} time.sleep(backoff * (2 ** attempt)) return wrapper

这个重试装饰器我挂在了所有外部调用的工具上，实测下来能挽回大部分偶发失败。

6.4 常见问题速查表

最后分享一个小技巧：整个系统上线前，我用一批“刁钻问题”做了一轮压力测试，比如“某品牌某型号的 120W 快充在合约机里是不是也支持”“我要办合约但是我的用户 ID 忘了怎么办”。这些问题暴露了不少边界情况，提前修掉比上线后被用户发现要好得多。

这个项目后续还可以扩展的方向有几个：一是把机型数据集换成真实业务数据，验证规模化之后的性能表现；二是引入用户画像，让品牌专家的话术根据用户偏好动态调整；三是把下单流程跟真实的订单系统对接，做端到端联调。不过那是下一步的事了，当前这套架构跑通单机器人问答、品牌专家路由、多智能体协作下单三条链路，已经能满足大部分手机客服场景的需求。