1. 项目概述：为什么一份手机数据集能跑通从单点问答到多智能体协作的完整链路？
最近在某实验室带一个RAG+Agent的实战训练营，学员普遍卡在一个认知盲区：总以为“多智能体”是高不可攀的架构设计，得先搭好消息总线、定义好Agent通信协议、再搞状态同步——结果三个月还在画UML图。直到我们用一份只有327行JSON的国产手机机型数据集（含品牌、发布时间、屏幕尺寸、主摄参数、电池容量、起售价、是否支持5G等12个字段），在不到4小时里，跑通了从“用户问‘哪款华为手机拍照最好’”这种单轮问答，到“用户说‘我要买一台适合拍夜景的折叠屏，预算6000以内，今天下单能发货吗’”，系统自动拆解为品牌专家路由→影像能力比对→形态筛选→库存与履约校验→生成订单摘要的全流程。这不是Demo，是真实可部署的轻量级服务。
核心关键词就三个：LlamaIndex RAG、手机客服场景、多智能体Agent协作。它不碰大模型微调，不依赖私有GPU集群，全程用LlamaIndex 0.10.x + OpenAI兼容接口（实测本地Ollama跑Qwen2-7B也稳），所有代码跑在一台16GB内存的MacBook Pro上。关键在于——我们把“多智能体”的复杂性，全部压在了数据结构设计和任务分解逻辑上，而不是基础设施上。比如“品牌专家路由”这个环节，根本没用LangChain的RouterChain，而是用LlamaIndex的VectorStoreIndex对品牌知识做向量化，再让LLM基于query embedding的相似度打分，选Top1品牌节点；而“协作下单”也不是靠Agent间发消息，而是用一个共享的OrderContext字典对象，每个Agent只读写自己负责的字段（影像专家填best_camera_model，形态专家填form_factor，履约专家填shipping_available_today），最后由Coordinator Agent聚合输出。这种设计，让整个系统像乐高一样可插拔：明天要加个“环保材料顾问”，只要新增一个Agent类，实现analyze()方法往OrderContext里写eco_certification字段就行，不用动其他模块。
适合谁学？三类人最受益：一是刚学完LangChain基础、想落地真实业务场景的开发者，这篇教程会告诉你RAG怎么从“查文档”升级为“驱动业务动作”；二是客服系统产品经理，你能看清一个“智能导购”背后真正的决策链条，不是简单调API，而是如何把模糊需求拆解成可验证的原子条件；三是中小厂商的技术负责人，你们没有资源养一支AI工程团队，但这份方案证明：用结构化数据+轻量RAG+明确分工的Agent，就能做出比传统规则引擎更灵活、比纯大模型更可控的客服系统。接下来所有内容，都围绕这份手机数据集展开——它不是玩具数据，而是我们刻意设计的“最小可行业务域”：字段覆盖用户决策关键点，数据量小到能人工核对每条记录，但又足够支撑多层推理。现在，我们开始拆解这个系统的骨架。
2. 整体架构设计：为什么放弃“标准Agent框架”，选择“数据流驱动”的极简协作模式？
2.1 标准Agent框架的三大隐性成本，我们在第一轮POC就踩了坑
刚开始，我们按教科书走：用LangChain的AgentExecutor+Tool体系，给每个Agent配独立的LLM实例和工具集。结果三天调试下来，发现三个致命问题：
提示：第一个问题是上下文爆炸。当用户问“华为Mate60 Pro和小米14 Ultra哪个更适合拍星空？”，Router Agent先判断品牌，再传给影像专家，影像专家调用工具查参数，再传给对比Agent……每跳一次，就得把前面所有历史、工具返回、中间结论塞进新prompt。327行数据集的embedding向量才2KB，但一个5轮交互的prompt轻松突破120KB，Ollama直接OOM。
提示：第二个问题是状态漂移。影像专家说“Mate60 Pro主摄1英寸传感器”，形态专家却说“它不是折叠屏”，但没人校验这两条信息是否来自同一数据源。我们发现，不同Agent用的其实是同一份JSON，但各自解析时用了不同key名（有的用
main_camera_sensor，有的用camera_primary_size），导致最终订单里camera_sensor字段为空。
提示：第三个问题是故障不可追溯。当订单生成失败，日志里只有一串“Agent execution failed”，根本不知道是路由错了品牌，还是影像专家误判了长焦能力，还是履约接口超时。传统框架把错误封装在
AgentExecutor.run()里，你只能看到“Exception in chain”，看不到具体哪个环节、哪条数据出了问题。
所以第二版，我们彻底转向“数据流驱动”：不设Agent运行时，不搞消息总线，所有Agent都是纯函数，输入是UserQuery和SharedContext，输出是更新后的SharedContext。整个流程变成一条清晰的数据管道：
UserQuery → RouterAgent → SharedContext(brand='huawei') → CameraExpert → SharedContext(brand='huawei', camera_rank=1) → FormFactorExpert → SharedContext(brand='huawei', camera_rank=1, form_factor='foldable') → FulfillmentExpert → SharedContext(brand='huawei', camera_rank=1, form_factor='foldable', shipping_available=True) → Coordinator → FinalOrderSummary这个设计的核心优势是：每个环节的输入输出完全可序列化、可断点调试、可单元测试。比如测试CameraExpert，你直接构造一个SharedContext(brand='huawei')，喂给它，断言它是否正确设置了camera_rank和best_model字段。不需要启动LLM，不需要mock网络请求。
2.2 LlamaIndex RAG在这里不是“检索器”，而是“领域知识编译器”
很多人把LlamaIndex当搜索引擎用，建完index就调query_engine.query()。但在本项目里，我们把它用成了“静态知识编译器”。具体操作分三步：
预编译阶段：加载手机数据集后，不直接建VectorStoreIndex，而是先用Pandas做清洗——统一
price字段为整数（去掉“¥”和“起”字），标准化release_date为YYYY-MM-DD格式，把5g_support布尔值从字符串"yes"转为True/False。这步耗时2秒，但避免了90%的LLM幻觉。索引构建阶段：我们建了三个独立索引，而非一个大索引：
BrandIndex：只索引brand和model字段，用于品牌路由；CameraIndex：索引model、main_camera_sensor、ultra_wide_aperture、night_mode_rating（人工标注的1-5分）；FulfillmentIndex：索引model、stock_status、shipping_days、warranty_months。
这样做的理由很实在：品牌路由需要毫秒级响应，用
BrandIndex查"xiaomi"比在全量索引里筛快3倍；而影像分析需要细粒度参数，CameraIndex的embedding向量只包含相关字段，噪声更少。查询增强阶段：
query_engine.query()返回结果后，我们不直接喂给LLM，而是用正则提取关键数值。比如query_engine.query("华为哪款手机夜景最强？")返回一段文本：“华为Mate60 Pro，夜景评分4.8分，主摄1英寸传感器”，我们用re.search(r'夜景评分(\d\.\d)分', text)抽取出4.8，存入SharedContext.night_mode_score。这步看似多余，实则是把LLM的“语言理解”能力，降维成“结构化数据提取”能力，准确率从82%提升到99.3%（实测100次query，仅1次因标点异常漏匹配）。
这种用法，让LlamaIndex从“黑盒检索器”变成了“白盒数据管道”，每个环节的输入输出都是确定的。当你发现订单里night_mode_score为空，直接查CameraIndex的query日志，就知道是原始数据里night_mode_rating字段缺失，而不是怪LLM没理解“夜景”这个词。
2.3 多智能体协作的本质，是“字段所有权”的精确划分
所谓“多智能体”，在本项目里就是四个Python类：RouterAgent、CameraExpert、FormFactorExpert、FulfillmentExpert。它们之间零耦合，不import彼此，不调用彼此方法，只通过一个约定好的SharedContext对象通信。这个对象长这样：
class SharedContext: def __init__(self): self.user_query: str = "" self.brand: Optional[str] = None # RouterAgent写 self.camera_rank: Optional[int] = None # CameraExpert写 self.best_camera_model: Optional[str] = None # CameraExpert写 self.form_factor: Optional[str] = None # FormFactorExpert写（'foldable', 'slate', 'flip'） self.shipping_available: Optional[bool] = None # FulfillmentExpert写 self.shipping_days: Optional[int] = None # FulfillmentExpert写 self.order_summary: Optional[str] = None # Coordinator写关键设计原则是：每个字段有且只有一个Agent有写权限。brand只能由RouterAgent设置，camera_rank只能由CameraExpert计算。如果FormFactorExpert试图修改brand字段，代码会抛出AttributeError（我们重写了__setattr__做字段锁）。这种设计带来两个好处：
第一，调试成本归零。当form_factor为空，你只看FormFactorExpert.analyze()方法就行，不用翻10个文件找谁改了它。
第二，扩展性极强。要加“价格顾问”，只需新增字段budget_friendly: bool，写一个PriceExpert类，在analyze()里根据SharedContext.brand和user_query里的“预算”关键词，查FulfillmentIndex的price字段，设置budget_friendly=True/False。整个过程不涉及任何架构调整。
我们刻意避免使用LangGraph或AutoGen这类框架，就是因为它们的“图编排”抽象，在小规模业务中反而增加了心智负担。当你只有4个Agent时，“顺序调用+共享状态”比“定义节点+配置边+处理循环”更直观、更易维护。这就像写SQL：SELECT * FROM phones WHERE brand='huawei' AND night_mode_rating > 4，比写MapReduce作业清晰多了。
3. 核心模块实现：从数据准备到订单生成的逐行代码解析
3.1 手机数据集的设计哲学：327行JSON如何承载完整决策链？
这份数据集不是随便爬的，而是按“用户决策树”反向设计的。我们访谈了23位真实购机用户，总结出他们决策的6个关键节点：品牌偏好→预算范围→核心功能（影像/游戏/续航）→形态偏好（直板/折叠/翻盖）→交付时效→售后保障。数据集字段严格对应这些节点：
| 字段名 | 类型 | 示例值 | 设计意图 |
|---|---|---|---|
brand | str | "huawei" | 品牌路由的唯一键，小写无空格 |
model | str | "Mate60 Pro" | 所有Agent引用机型的统一ID |
price | int | 6999 | 用于价格顾问和预算校验 |
main_camera_sensor | str | "1-inch" | 影像专家的核心参数 |
night_mode_rating | float | 4.8 | 人工标注，避免LLM主观评分 |
form_factor | str | "foldable" | 形态专家的分类依据（'foldable','slate','flip'） |
battery_capacity_mah | int | 5000 | 续航专家的输入 |
shipping_days | int | 1 | 履约专家的交付承诺 |
warranty_months | int | 24 | 售后专家的保障依据 |
特别说明night_mode_rating字段：我们没让LLM自己评，而是请3位手机摄影师，用同一组夜景样张，在暗光环境下对327款机型主摄进行盲评，取平均分。为什么这么做？因为LLM对“夜景强”的理解是模糊的（可能认为像素高=夜景强），而真实用户要的是“拍出来不糊、噪点少、色彩准”。这个人工标注字段，让CameraExpert的输出准确率从76%提升到94%。
数据加载代码极简：
import json import pandas as pd def load_phone_dataset() -> pd.DataFrame: with open("phones.json", "r", encoding="utf-8") as f: data = json.load(f) df = pd.DataFrame(data) # 强制类型转换，避免后续LLM解析失败 df["price"] = df["price"].astype(int) df["night_mode_rating"] = df["night_mode_rating"].astype(float) df["battery_capacity_mah"] = df["battery_capacity_mah"].astype(int) return df # 实测：327行数据，pandas加载+类型转换耗时0.012秒这份数据集的价值，不在于“大”，而在于“准”和“全”。它让每个Agent的决策都有明确的数据锚点，而不是在LLM的幻觉里打转。
3.2 RouterAgent：用向量相似度代替关键词匹配的品牌路由
品牌路由看似简单，但传统关键词匹配（如if "华为" in query: brand="huawei"）在真实客服场景中会崩盘。用户说“国货之光”、“鸿蒙系统”、“麒麟芯片”，关键词匹配全失效。我们的方案是：用LlamaIndex的VectorStoreIndex对品牌知识做向量化，让LLM自己学“什么是华为”。
具体步骤：
构建品牌知识库：不是直接用
brand字段，而是为每个品牌生成一段100字以内的描述，包含其技术标签和用户心智词。例如华为的描述是：“华为，中国科技企业，以鸿蒙操作系统、麒麟芯片、XMAGE影像技术著称，用户常关联‘国产高端’、‘信号强’、‘商务旗舰’”。创建BrandIndex：
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader from llama_index.core.node_parser import SentenceSplitter from llama_index.embeddings.huggingface import HuggingFaceEmbedding # 加载品牌描述文本 brand_docs = [ Document(text="华为，中国科技企业，以鸿蒙操作系统..."), Document(text="小米，互联网科技公司，以MIUI系统、徕卡影像..."), # ... 其他品牌 ] # 使用bge-small-zh-v1.5中文嵌入模型（128MB，Ollama可跑） embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-zh-v1.5") index = VectorStoreIndex.from_documents( brand_docs, embed_model=embed_model, transformations=[SentenceSplitter(chunk_size=128)] )- 路由逻辑：
RouterAgent.analyze()接收user_query，用同一embed_model生成query向量，然后在BrandIndex中找最相似的品牌描述：
def analyze(self, user_query: str) -> str: # 生成query embedding query_embedding = self.embed_model.get_text_embedding(user_query) # 在BrandIndex中搜索最相似的品牌描述 retriever = index.as_retriever(similarity_top_k=1) nodes = retriever.retrieve(query_embedding) if not nodes: return "unknown" # 提取品牌名：从"华为，中国科技企业..."中提取"华为" brand_name = nodes[0].text.split("，")[0] return brand_name.lower().replace(" ", "")实测效果：对100个含蓄表达的query（如“想买个用鸿蒙系统的手机”、“信号最好的国产机”、“拍月亮的手机”），路由准确率达92%，远超关键词匹配的63%。关键是，这个模型可以增量更新——新出一个品牌，只需加一段描述，重新index.insert()，不用改任何代码。
3.3 CameraExpert：从参数到体验的语义映射
影像专家的任务不是罗列参数，而是回答“哪款最适合拍夜景？”。难点在于：参数（如f/1.4光圈）和体验（“夜景明亮”）之间没有直接映射。我们的解法是：用LlamaIndex的SubQuestionQueryEngine做多跳推理。
CameraExpert不直接查night_mode_rating，而是把用户query拆成子问题：
- 用户要什么？→ 提取核心需求（“夜景”、“星空”、“暗光”）
- 哪些参数影响这个需求？→ 查知识库：“影响夜景成像的关键参数是光圈大小、传感器尺寸、夜景算法评分”
- 当前品牌下，哪些机型满足？→ 用
CameraIndex查brand匹配的机型，按night_mode_rating排序
代码实现：
from llama_index.core.query_engine import SubQuestionQueryEngine def analyze(self, shared_context: SharedContext) -> None: # Step1: 提取需求关键词 demand_prompt = f"从以下用户问题中，提取1个最相关的摄影场景关键词，只输出关键词，不要解释。问题：{shared_context.user_query}" demand_keyword = self.llm.complete(demand_prompt).text.strip() # Step2: 查询该场景的关键参数（预置知识） knowledge_map = { "夜景": ["night_mode_rating", "main_camera_sensor", "ultra_wide_aperture"], "人像": ["front_camera_mp", "bokeh_effect_rating"], "远景": ["telephoto_focal_length", "periscope_zoom"] } params = knowledge_map.get(demand_keyword, ["night_mode_rating"]) # Step3: 用SubQuestionQueryEngine查参数 query_engine = SubQuestionQueryEngine.from_defaults( query_engine_tools=[ QueryEngineTool.from_defaults( query_engine=self.camera_index.as_query_engine(), metadata=ToolMetadata(name="camera_db", description="手机影像参数数据库") ) ] ) sub_questions = [f"华为品牌下，{param}值最高的机型是？" for param in params] responses = query_engine.query("\n".join(sub_questions)) # Step4: 解析响应，取交集机型 best_models = [] for resp in responses: model_match = re.search(r"机型是(.+?)。", resp.response) if model_match: best_models.append(model_match.group(1).strip()) # 取交集，得到综合最优机型 from collections import Counter most_common = Counter(best_models).most_common(1) if most_common: shared_context.best_camera_model = most_common[0][0] # 同时查该机型的night_mode_rating rating = self._get_rating_by_model(shared_context.best_camera_model, "night_mode_rating") shared_context.night_mode_score = rating这个设计的精妙之处在于：它把LLM的“推理”能力，限定在“子问题生成”和“结果聚合”两个环节，而参数查询、排序、交集计算全部由代码完成。这样既利用了LLM的语义理解，又规避了它的幻觉风险。实测中，对“拍星空”这种长尾需求，准确率比单次query高37%。
3.4 FormFactorExpert与FulfillmentExpert：用结构化查询替代LLM自由发挥
形态专家和履约专家的任务最“枯燥”，但也最容错。用户说“我要折叠屏”，FormFactorExpert只需查form_factor == 'foldable'；用户问“今天能发货吗”，FulfillmentExpert只需查shipping_days <= 1。但我们发现，直接让LLM执行这种布尔查询，准确率只有68%（LLM常把shipping_days=2误判为“能发货”）。
解决方案：把LLM降级为“查询生成器”，把执行交给Pandas。
def analyze(self, shared_context: SharedContext) -> None: # 让LLM生成Pandas查询字符串，不是直接回答 query_prompt = f""" 用户问题：{shared_context.user_query} 请生成一个pandas DataFrame查询字符串，用于筛选手机数据。 要求： 1. 只用df[...]语法，不要用for循环 2. 字段名必须是：form_factor, shipping_days, stock_status 3. 输出格式：df[条件1 & 条件2] 示例：用户问“折叠屏”，输出：df[df['form_factor'] == 'foldable'] """ pandas_query = self.llm.complete(query_prompt).text.strip() # 安全执行查询（白名单校验） if not self._is_safe_pandas_query(pandas_query): raise ValueError("Unsafe pandas query detected") # 执行查询，取第一条 result_df = self.phone_df.eval(pandas_query) if not result_df.empty: shared_context.form_factor = result_df.iloc[0]["form_factor"] shared_context.shipping_available = result_df.iloc[0]["shipping_days"] <= 1 shared_context.shipping_days = result_df.iloc[0]["shipping_days"]_is_safe_pandas_query()做白名单校验：只允许df[...]、==、<=、&、字段名、字符串和数字。这样，LLM再怎么胡说，也执行不了恶意代码。实测100次查询，安全校验拦截了7次无效输入（如LLM生成df.drop()），但所有有效查询100%准确执行。
3.5 Coordinator：用模板填充替代LLM自由创作的订单摘要
最后一步，Coordinator把所有字段拼成自然语言订单摘要。很多教程让LLM“根据上下文生成摘要”，结果输出千奇百怪：“您选择了华为Mate60 Pro，它很棒！”——这根本不能用。
我们的方案：预定义模板，用字段值填充。
ORDER_TEMPLATES = { "full": "已为您匹配到最佳机型：{best_camera_model}（{brand}品牌），夜景评分{night_mode_score}分，{form_factor}形态，预计{shipping_days}天内发货。", "short": "{best_camera_model}，{brand}，{night_mode_score}分，{shipping_days}天发货。" } def generate_summary(self, shared_context: SharedContext) -> str: # 检查必填字段是否齐全 required_fields = ["best_camera_model", "brand", "night_mode_score", "shipping_days"] missing = [f for f in required_fields if getattr(shared_context, f) is None] if missing: return f"订单生成失败：缺少字段{missing}" # 选择模板并填充 template = ORDER_TEMPLATES["full"] if len(shared_context.user_query) > 20 else ORDER_TEMPLATES["short"] try: summary = template.format(**shared_context.__dict__) return summary except KeyError as e: return f"模板填充失败：字段{e}未设置"这个设计的好处是：摘要格式100%可控，字段缺失立刻报错，运维人员一眼看出哪个Agent没干活。上线后，订单摘要的NPS（用户满意度）从72分升到94分，因为用户终于看到了确定的机型、分数、发货时间，而不是“可能”“大概”“推荐”。
4. 实操避坑指南：那些文档里不会写的血泪教训
4.1 LlamaIndex版本陷阱：0.10.x的as_query_engine()和0.9.x的query()行为完全不同
这是我们在第三天凌晨2点发现的致命坑。旧教程全用index.query("xxx")，但升级到LlamaIndex 0.10.x后，query()方法被废弃，必须用index.as_query_engine().query("xxx")。表面只是方法名变，实际行为差异巨大：
- 0.9.x的
query()：返回Response对象，.response属性是纯文本，.source_nodes是原始chunk。 - 0.10.x的
as_query_engine().query()：返回StreamingResponse，默认是流式输出，.response是generator，必须用str(response)才能拿到完整文本。
我们一开始没注意，CameraExpert里直接写：
# 错误！0.10.x下response是generator，.response是<generator object> result = self.camera_index.as_query_engine().query("华为夜景最强机型") print(result.response) # 输出：<generator object ...>结果result.response被当成字符串传给正则，re.search()永远返回None，所有影像分析失败。修复方案：
# 正确：强制转为字符串 result = self.camera_index.as_query_engine().query("华为夜景最强机型") full_text = str(result) # 或 result.get_response()更坑的是，这个bug在本地Ollama上不报错（因为流式输出很快），但部署到云服务器时，网络延迟导致generator卡住，整个Agent挂死。建议：所有as_query_engine().query()调用后，立即加str()或.get_response()，并在日志里打印len(str(result))，确保返回了预期长度的文本。
4.2 中文嵌入模型的选择：别迷信“SOTA”，小模型才是生产力
项目初期，我们试了bge-large-zh-v1.5（1.2GB），向量质量确实好，但Ollama在MacBook上加载要47秒，每次重启服务都要等半分钟。后来换成bge-small-zh-v1.5（128MB），加载只要3秒，而品牌路由准确率只降了1.2%（92.1%→90.9%）。
关键洞察：在结构化数据场景，嵌入模型的“绝对精度”不如“相对稳定性”重要。bge-small对“鸿蒙”和“HarmonyOS”的向量距离，和bge-large几乎一致，但它的推理速度是后者的8倍。我们做了AB测试：用bge-small跑1000次路由，平均耗时83ms；bge-large是672ms。对客服系统，83ms意味着QPS（每秒查询数）从12提升到120，这才是真实业务价值。
所以选型口诀：数据量<1万行，字段<20个，优先选small模型；只有当你需要跨语言检索（如中英混合query），才考虑large。
4.3 SharedContext的并发安全：多用户请求下的字段污染
上线压力测试时，我们发现一个诡异现象：用户A问“华为拍照”，返回Mate60 Pro；用户B同时问“小米续航”，返回的却是“华为Mate60 Pro，电池5000mAh”。查日志，SharedContext对象被两个请求共用了。
根源在于：我们最初把SharedContext做成全局单例。修复方案很简单：每个请求创建独立实例。
# 错误：全局单例 context = SharedContext() def handle_request(query: str): context.user_query = query router_agent.analyze(context) # ... 其他Agent return coordinator.generate_summary(context) # 正确：请求级实例 def handle_request(query: str): context = SharedContext() # 每次新建 context.user_query = query router_agent.analyze(context) # ... 其他Agent return coordinator.generate_summary(context)但要注意：SharedContext里不能存大对象（如整个DataFrame），否则内存暴涨。我们把phone_df作为Agent的成员变量，SharedContext只存轻量字段。实测单实例内存占用从12MB降到0.3MB。
4.4 LLM提示词的“防抖”设计：如何让输出格式100%稳定？
RouterAgent曾出现过这样的输出：“华为（中国品牌）”，导致brand字段存了"华为（中国品牌）"，后续所有查询都失败。原因是LLM喜欢加括号解释。
解决方案：在提示词末尾加“输出格式强制约束”。
router_prompt = f""" 你是一个手机品牌路由专家。请从用户问题中识别出最可能的品牌。 要求： 1. 只输出品牌名，小写字母，无空格，无标点，无解释 2. 例如：“华为Mate60 Pro拍照怎么样？” → 输出：huawei 3. “想买个用鸿蒙系统的手机” → 输出：huawei 4. 如果无法确定，输出：unknown 用户问题：{user_query} 输出： """重点是第1条和第4条。我们还加了后处理：
def clean_brand_output(raw: str) -> str: # 移除所有非字母数字字符，转小写 cleaned = re.sub(r"[^a-zA-Z0-9]", "", raw).lower() # 白名单校验 valid_brands = ["huawei", "xiaomi", "apple", "samsung", "oppo", "vivo"] return cleaned if cleaned in valid_brands else "unknown"这套组合拳，让品牌输出准确率从89%提升到100%。记住：LLM不是神，它是工具，你要用代码给它画牢笼。
4.5 数据集版本管理：如何避免“线上跑着V1，开发改着V2”的灾难
我们吃过亏：开发环境用phones_v2.json（新加了eco_certification字段），但生产环境还是phones_v1.json，结果PriceExpert一查eco_certification就报KeyError，整个服务雪崩。
现在强制执行：数据集文件名带版本号，代码里硬编码版本。
# config.py PHONE_DATASET_VERSION = "v2" PHONE_DATASET_PATH = f"phones_{PHONE_DATASET_VERSION}.json" # load_phone_dataset()函数里加校验 def load_phone_dataset() -> pd.DataFrame: path = PHONE_DATASET_PATH if not os.path.exists(path): raise FileNotFoundError(f"Dataset {path} not found! Check version.") # ... 加载逻辑每次数据更新，必须：
- 改
PHONE_DATASET_VERSION常量； - 提交新数据文件；
- CI流水线跑
pytest，检查所有Agent能否用新数据正常运行。
这个流程让数据变更从“风险操作”变成“可审计事件”。上线半年，零数据相关故障。
5. 场景延展与工程化建议：从教程到生产系统的最后一公里
5.1 如何把这份教程变成可交付的产品？三个必须做的工程化动作
这份教程的代码，离生产环境还有三步：
第一步：增加健康检查端点。别等用户投诉才知服务挂了。加一个/health接口，返回各模块状态：
@app.get("/health") def health_check(): return { "status": "ok", "components": { "router_agent": _test_router(), "camera_index": _test_camera_index(), "shared_context": "ready" }, "dataset_version": PHONE_DATASET_VERSION }_test_router()不是调LLM，而是查BrandIndex能否返回至少一个品牌。这样，K8s的liveness probe能秒级发现索引损坏。
第二步：为每个Agent加超时熔断。LLM调用可能卡死，必须设timeout：
from tenacity import retry, stop_after_attempt, wait_fixed @retry(stop=stop_after_attempt(3), wait=wait_fixed(2)) def safe_analyze(self, context: SharedContext): # 原analyze逻辑 pass三次失败后，返回兜底值（如brand="unknown"），保证服务不雪崩。
第三步：日志结构化。别再用print()，用structlog记录每个Agent的输入输出：
import structlog logger = structlog.get_logger() def analyze(self, context: SharedContext): logger.info("router_agent_start", user_query=context.user_query) brand = self._route(context.user_query) logger.info("router_agent_end", brand=brand, user_query=context.user_query) return brand这样，ELK里能直接查“所有brand="unknown"的请求”，快速定位问题。
5.2 业务场景的自然延伸：从“下单”到“售后”的闭环设计
这份教程止于订单生成，但真实客服要闭环。我们已在某客户项目中落地了两个延伸：
售后咨询Agent：当用户问“我的Mate60 Pro屏幕碎了，保修吗？”，它查warranty_months和purchase_date（用户提供的购买日期），用FulfillmentIndex查该机型的warranty_policy字段（如“屏幕碎裂不在保修范围，但可享8折维修”），生成明确答复。
竞品对比Agent：用户说“iPhone15和Mate60 Pro哪个好？”，它不自己评，而是并行查两个model的night_mode_rating、battery_capacity_mah、price，生成对比表格。关键点：所有对比字段必须来自同一数据源，避免“苹果官网参数 vs 华为论坛爆料”的混乱。
这两个延伸，都没新增架构，只是加了两个Agent类和几个字段。这印证了我们最初的设计哲学：用数据定义能力边界，用Agent定义职责边界。
5.3 给技术决策者的务实建议：什么时候该用，什么时候不该用？
最后分享一个血泪经验：别为了用RAG而用RAG，别为了多Agent而多Agent。
该用的场景：
- 你的数据是结构化的（如本项目的JSON），且字段能覆盖用户决策点；
- 你的业务流程可拆解为明确步骤（路由→分析→筛选→确认）；
- 你有明确的“成功标准”（如订单摘要必须含机型、分数、发货时间）。
不该用的场景：
- 数据是纯PDF扫描件，OCR后全是乱码——先做数据清洗，再谈R
