★ FEATURED ARTICLE 最近把一个客服工单自动处理系统从单Agent改成了双Agent协作核心是用LangGraph把“主管”和“执行”两条Agent链路编排成一张可控制的执行图再用Milvus给Agent挂上长期记忆和领域知识检索。实测下来复杂工单的一次处理成功率从不到60%拉到了85%以上多轮任务不再频繁丢上下文工具调用也不会动不动就断在中间。这篇就是这次硬核实战的完整复盘包括单Agent为什么会翻车、双Agent节点怎么设计、LangGraph状态流怎么组织、Milvus建集合和索引的参数怎么定以及跑通之后踩到的一堆坑。适合已经在做Agent、但一遇到多步骤任务调度就发怵的开发者也适合想弄清楚Agent编排框架和向量数据库到底怎么配合的人。1. 单Agent为什么会翻车瓶颈不在模型在架构单个Agent处理简单问答、单次工具调用基本没什么问题。可一旦任务变成“理解用户意图、查知识库、更新业务数据、生成回复”这种多步骤、多分支、还要根据中间结果反复决策的复杂任务单Agent就开始捉襟见肘。我试过把工单分类、相似历史查询、回复生成全塞进一个Agent里表面上看prompt写得挺全实际跑起来问题接二连三。1.1 上下文被一次对话吃满早期信息逐渐失真单Agent最直接的问题是所有中间结果都会拼进上下文中。第一个工具返回了2000字的技术文档第二个工具又返回800字的历史工单再往后模型还要记住用户的原始问题上下文越滚越胖。虽然现在各家模型窗口越做越大但大窗口不等于能有效利用模型对上下文中间部分的注意力天然会衰减。我实际观察过当一个Agent的上下文超过一定长度后它经常会“忘记”最开始的约束条件比如用户明确说过不要用短信通知执行到最后它反倒调了短信通道。知识检索类的中间结果尤其占地方。你检索回来5段文档每段几百字这些内容对最终决策有用的可能就一两句但全部堆在上下文里既浪费token又干扰判断。双Agent方案里至少可以把“检索”和“决策”分开检索阶段拿到信息后先做结构化筛选主管只看到提炼后的结论。1.2 工具调用只能串行推进一根筋走到黑单Agent在工具编排上也比较僵硬。它本质上靠模型自己“想起来”下一步该调什么工具中间只要有一次工具返回了意外格式后续步骤就开始放飞。比如我让单Agent先查工单表格、然后判断是否需要人工介入、最后更新工单状态一旦查到的工单不存在模型就有可能跳过更新步骤直接给用户回复“办好了”。这种错误在业务系统里是很致命的。还有一类问题是单Agent很难处理“分支回退”。任务做到一半发现前置条件不满足正确的做法应该是回到上一个节点补充信息或者换一条路径重新执行。单Agent在纯文本流里做这种回退只能靠prompt维持状态一复杂就乱。1.3 状态不可见出了问题没法定责单Agent调试起来非常痛苦。它内部就是一个黑盒某一步错了你是该改prompt、调工具参数、还是优化检索结果没有清晰的分层这些问题全都搅在一起。我在开发阶段最耗时的不是写功能而是猜Agent到底在哪一步、因为什么原因走的弯路。后来换成双Agent主管和执行各管一段日志打出来哪一步错了一目了然。那是不是所有任务都必须上双Agent不是。单Agent处理简单任务完全够用。但当任务具备“需要领域知识检索、需要多工具协同、需要根据中间结果动态决策”这三个特征时双Agent带来的收益会非常明显。下面这张表是我在项目里做的对比对比维度单Agent双Agent主管执行上下文占用所有中间结果都堆在一个上下文里主管只接收提炼后的结果占用可控工具失败容错一个工具异常容易拖垮整条链路执行节点失败后主管可以重新调度任务状态管理依赖模型隐式记忆容易漂移状态显式存储在图中每步可查分支与回退靠prompt硬撑不稳定用条件边显式控制路由逻辑透明调试排错黑盒只能靠猜节点独立日志清晰可单节点重放扩展成本加一个新能力要重写prompt加一个新工具/新执行节点即可主管prompt微调2. 整体架构拆解LangGraph和Milvus在双Agent里各自负责什么很多项目卡在“知道该用双Agent但不知道怎么落地”。这里的关键是选对编排框架和记忆存储。我这次选LangGraph做Agent流程编排选Milvus做向量检索两者一个管“流程怎么走”一个管“知识怎么存”配合起来正好补上单Agent最虚的两块短板。2.1 LangGraph本质是一张会执行的状态机LangGraph的核心概念其实不复杂它把Agent的每一步操作都建模成“图里的一个节点”节点之间的跳转关系就是“边”。你定义一个共享的状态对象每个节点读状态、处理、改写状态然后通过边把控制权交给下一个节点。比传统链式调用强的地方在于它支持条件边也就是节点可以根据当前状态自己决定下一步走哪个分支甚至可以回到之前的节点重新处理。你可以把这套机制理解成一个项目组状态对象是共享的工作台所有成员往工作台上放文件、取文件主管Agent是项目经理负责看工作台上的信息、分派活执行Agent是干活的工程师只在被分到具体任务时才动手。LangGraph就是那张项目流程图谁干完活、下一步到谁图说了算而不是靠项目经理脑子里临时想。这个项目里我建了一个很小的图四个核心节点入口节点负责接收用户问题主管节点负责拆解任务、调用模型做决策执行节点负责调用具体工具并返回结果归档节点负责把关键信息写入Milvus。核心的循环逻辑是入口 → 主管 → 执行 → 主管 → 执行……直到主管判断任务已经完成跳到归档节点收尾。2.2 Milvus不是“装样子的向量库”它是Agent的记忆层那Milvus在这套架构里到底解决什么问题直观上是给Agent做领域知识检索但更准确的定位是“记忆层”。单Agent没有长期记忆每次对话都是从零开始哪怕同一个工单类型它已经处理过一百遍下一次还是要重新推理。把历史处理经验、领域知识库、业务规则都向量化存进MilvusAgent在每次决策前先查一遍相关内容等于把过往经验直接拼接到当前决策里。我选了Milvus而不是轻量级方案是因为实际业务里知识库的量级不会停留在几千条。涵盖历史工单、产品文档、业务FAQ的数据量往十万级以上走时向量召回性能和索引能力就会拉开差距。Milvus比较成熟支持多种索引类型、标量字段过滤、动态Schema我还能按时间、来源这类业务字段做过滤检索维度更灵活。另外它是独立服务不占用Agent进程的内存和LangGraph放在一起一个管计算一个管存储职责很清楚。2.3 本次项目的完整数据流用一个真实场景串一下用户提交一条工单“我的订单已经付款三天还没发货想申请退款”。任务进来后入口节点先做预处理把工单文本和用户ID写入状态对象。主管节点拿到状态先从Milvus里检索相似历史工单把历史处理方案作为参考然后调LLM判断意图这个工单属于退款申请需要走退款流程。主管在状态里写入“执行计划”路由到执行节点。执行节点做三件事查订单数据库确认订单状态、查退款规则确认是否允许退款、调用退款接口。每一步的结果都写回状态对象。主管收到执行结果后再检索一遍相关规则确认所有条件都满足生成最终回复把本次工单的处理摘要写入Milvus最后结束。这套流程里LangGraph保证了“谁先谁后、失败了怎么办”Milvus保证了“每次都站在历史经验上做决策”两个Agent节点各司其职单Agent最头疼的多步调度和记忆缺失问题就同时被解决掉了。3. 环境准备、Schema设计与Embedding管线我直接说结论双Agent能不能跑通一半看代码另一半看数据准备。很多项目代码写完了卡在Milvus集合设计不合理、向量维度对不上、索引参数乱填导致检索结果差得离谱。这节先把底子打好。3.1 依赖安装与版本选择项目用Python实现核心依赖是这些pip install langgraph pymilvus sentence-transformers三个库各管一摊langgraph负责Agent图编排pymilvus负责和Milvus通信sentence-transformers负责把文本转成向量。实际项目中还会有你自己业务侧的数据库驱动和LLM SDK这里先不展开。版本上我的建议是“不要追最新用稳定版”。LangGraph的API迭代速度比较快小版本之间的函数签名偶尔会变pymilvus也同理。遇到问题先锁定版本代码能跑通再考虑升级。我第一次装的时候直接拿了最新版结果一个条件节点的写法在文档里还是旧样式排查了很久。锁定大版本之后再装后续复现也方便。3.2 Milvus Collection设计与索引参数Collection是Milvus里的核心概念类比关系型数据库的表。建Collection之前先把字段想清楚。我这次用的Schema包含五个字段from pymilvus import ( CollectionSchema, FieldSchema, DataType, connections, Collection ) connections.connect(host127.0.0.1, port19530) fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namesource, dtypeDataType.VARCHAR, max_length128), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length2048), FieldSchema(namecreate_time, dtypeDataType.VARCHAR, max_length32), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim768), ] schema CollectionSchema( fieldsfields, descriptionAgent long-term memory and domain knowledge ) collection Collection(nameagent_memory, schemaschema)这里有几个“为什么”要讲清楚。embedding维度定768是因为我用的开源中文Embedding模型输出的向量维度就是768必须先确认模型维度再定Schema反过来就会报错。source和create_time两个标量字段从一开始就预留好是为了后面做分区过滤和按时间筛选不然跑到一半想按业务维度筛选数据还得重建集合。索引参数我这么配index_params { index_type: HNSW, metric_type: COSINE, params: {M: 16, efConstruction: 200} } collection.create_index(field_nameembedding, index_paramsindex_params) collection.load()文本场景下度量方式选COSINE比L2直观因为文本向量的绝对长度受句子长度影响余弦相似度只关心方向。HNSW是图索引检索速度和召回率均衡适合数据量中等、QPS要求不极端的场景。M控制每个节点的最大连接数M越大召回越准、内存占用越大efConstruction控制建索引时的候选集大小越大索引质量越高、建索引越慢。M16、efConstruction200是一个比较稳的起点后面按数据量调。3.3 Embedding与写入注意事项Embedding模型我选了中文效果比较好的开源向量模型句子直接编码成768维向量。核心注意点第一写入Milvus的向量必须和查询时用同一个模型生成混用模型会让检索结果完全失真。第二单条文本过短时向量语义会不清晰我写入前做了简单清洗工单文本至少保留完整句子不把碎片化关键词直接入库。第三批量写入时控制batch size我习惯每批200条左右太大容易把内存打满太小写入效率低。一个写入示例from pymilvus import utility texts [订单付款后超过48小时未发货可以申请退款, 退款申请需要用户确认商品未使用] embeddings embedding_model.encode(texts).tolist() data [ {source: refund_policy, text: texts[0], create_time: 2025-01-01, embedding: embeddings[0]}, {source: refund_policy, text: texts[1], create_time: 2025-01-01, embedding: embeddings[1]}, ] collection.insert(data) collection.flush()flush操作强烈建议每次批量写入后做一次确保数据落盘否则紧接着就去search偶尔会查不到刚插入的数据。这个坑我踩过后面排查章节再细说。4. 双Agent核心代码实战这一节写核心代码逻辑。为了不牵扯具体业务我把工具做了抽象重点展示LangGraph的编排思路和Milvus的集成方式。完整的图有四个节点代码结构大致如下。4.1 状态结构与工具集定义LangGraph里的状态对象推荐用TypedDict定义字段越明确后续调试越省心。from typing import TypedDict, List, Optional class AgentState(TypedDict): user_request: str user_id: str intent: str plan: List[str] tool_results: dict final_answer: str finished: booluser_request是用户的原始输入intent是主管分析出来的意图plan是执行计划tool_results用来存放工具返回结果。每个节点都只修改自己负责的字段职责不交叉。工具集可以统一封装成一个字典执行Agent按名字调用def search_order_db(order_id: str) - dict: # 模拟查询订单数据库 return {status: paid, days_since_payment: 3} def search_refund_policy(product_type: str) - dict: # 从Milvus检索退款规则 query_embedding embedding_model.encode([product_type]).tolist() results collection.search( dataquery_embedding, anns_fieldembedding, param{metric_type: COSINE, params: {ef: 64}}, limit3, output_fields[text, source] ) return [{text: hit.entity.get(text), score: hit.score} for hit in results[0]] def create_refund_order(user_id: str, order_id: str) - dict: # 创建退款单 return {refund_id: RF20250101, status: created} tools { search_order_db: search_order_db, search_refund_policy: search_refund_policy, create_refund_order: create_refund_order, }工具返回的必须是dict最好能直接JSON序列化这样主管节点拿到结果后可以直接塞进上下文不用再额外清洗。工具函数里埋日志执行到哪一步、返回了什么之后排查全靠这些日志。4.2 主管Agent拆解调度主管节点是整张图的大脑它做的事情是读取状态里的用户请求从Milvus检索相似历史经验把请求拆解成步骤然后用LLM做一次决策决策结果必须是结构化的JSON里面带next_action字段是继续执行、还是结束。def coordinator_node(state: AgentState) - AgentState: # 1. 从Milvus检索历史相似工单 history search_similar_history(state[user_request]) # 2. 构造给LLM的决策上下文 prompt f 你是工单处理主管。用户请求{state[user_request]} 历史参考{history} 请输出JSON决策 - intent: 用户意图 - plan: 执行步骤列表 - next_action: execute 或 finish - final_answer: 如果finish直接给最终回复 decision llm.chat(prompt, response_formatjson) state[intent] decision[intent] state[plan] decision[plan] state[final_answer] decision.get(final_answer, ) state[finished] decision[next_action] finish return state为什么一定要让LLM输出JSON而不是自由文本因为后面条件边要判断next_action的值如果是自由文本模型很容易输出“好的现在去执行吧”或者“下面我打算……”字符串匹配的稳定性会很差。JSON强制让模型在固定字段里选择配合response_format参数基本可以保证结构可以稳定解析。这里有一个我吃过亏的细节如果LLM输出JSON时夹带了markdown代码块标记解析会失败所以prompt里要明确写“只输出JSON不要带任何多余内容”解析层也要做好兜底。4.3 执行Agent干活并写回结果执行节点不负责决策只负责执行plan里的动作调用对应的工具然后把结果写回状态。def worker_node(state: AgentState) - AgentState: results {} for step in state[plan]: if step in tools: # 这里根据plan里的参数调用工具 results[step] tools[step](**state.get(tool_params, {})) state[tool_results] results return state实际业务里工具参数可能比较复杂我的做法是让主管在拆解任务时把每个步骤所需参数也一并输出到plan里比如plan可以是[{“action”: “search_order_db”, “args”: {“order_id”: “123456”}}]这种结构执行节点按顺序遍历执行。参数解析要做好校验缺少必填字段时返回明确的错误信息不要直接抛异常打断整个图。执行节点里还会做一个小动作把工具结果做摘要后写回状态。例如检索退款规则返回了3段文本执行节点把它们压缩成“规则允许退款条件是商品未使用”而不是把原文堆给主管。这一步对控制上下文膨胀非常关键双Agent省上下文的优势就是从这里体现出来的。4.4 条件边让Agent自己决定下一步LangGraph里实现“主管-执行”循环核心就在条件边。主管节点每次执行完根据状态里的finished字段决定往哪里走from langgraph.graph import StateGraph, END def should_continue(state: AgentState) - str: if state[finished]: return archive return worker graph StateGraph(AgentState) graph.add_node(coordinator, coordinator_node) graph.add_node(worker, worker_node) graph.add_node(archive, archive_node) graph.set_entry_point(coordinator) graph.add_conditional_edges( coordinator, should_continue, {archive: archive, worker: worker} ) graph.add_edge(worker, coordinator) graph.add_edge(archive, END)这张图的执行逻辑是从coordinator进入如果主管说还没干完就去worker执行工具worker干完再回到coordinator让主管基于最新工具结果做下一轮决策如果主管说干完了就进入archive节点做记忆归档然后结束。这个“回到主管”的设计是双Agent的精髓。单Agent是“一条路走到黑”多轮工具调用全靠模型自己串双Agent把“决策”和“执行”拆成循环每一次工具执行完决策逻辑都重新评估一遍。这种模式非常适合任务步骤不确定、可能需要中途改变方案的场景。4.5 编译执行图定义好之后编译调用app graph.compile() result app.invoke({ user_request: 订单付款三天没发货想申请退款, user_id: user_1001, intent: , plan: [], tool_results: {}, final_answer: , finished: False, }) print(result[final_answer])实际跑一次大概会经过coordinator → worker → coordinator → worker → coordinator → archive。前两个coordinator负责决策和验证最后一次判断完成然后归档。跑通后建议用app.stream()替代invoke()stream可以逐步吐出每个节点的状态调试时能看到每一步状态变化效果拔群。5. 跑通之后的参数调优与记忆管理图能跑起来只是第一步效果好不好全看参数和记忆策略。这节分享几个影响最大的调试点。5.1 向量检索参数怎么调我先说结论top_k不要贪多相似度阈值要结合业务调索引参数要看数据量。top_k决定Agent每次决策前参考多少条历史结果。我初始设置是5跑了一段时间发现主管的prompt里塞了5段历史经验后模型反而容易“参考过头”把不相关的历史也当成依据。后来压到3准确率反而上去了。top_k的调整思路是历史经验是参考资料不是唯一决策依据给3条典型case足够给太多等于引入噪音。search_params { metric_type: COSINE, params: {ef: 64} } results collection.search( dataquery_embedding, anns_fieldembedding, paramsearch_params, limit3, exprsource refund_policy, output_fields[text, source] )这里的expr是对标量字段的过滤表达式我按业务来源过滤后检索结果相关性明显提升。如果只做纯向量检索查出来的结果可能领域不对口再加上业务过滤条件命中率会上一个台阶。相似度阈值是个需要反复试的值。我在实验环境里统计过一批已知答案的查询正确命中的分数集中在0.72以上错误命中的分数一般在0.5以下。所以把0.6作为“可接受最低相似度”低于这个值的结果直接丢弃避免模型拿低质量历史当参考。这个阈值每个业务不一样建议你先跑一批标注数据画出分数分布再定阈值不要拍脑袋。5.2 长期记忆写入策略Milvus不能光用来查还要会写。什么时候写、写什么内容直接决定记忆库的质量。我的策略是任务结束后生成的摘要才写入Milvus中间过程的临时状态不写。每个任务结束时会把用户请求、最终回复、关键业务字段组合成一条简短的文本摘要同时保留source和create_time字段。写入前一定要做去重。同一个用户重复咨询类似问题如果每次都无脑插入Milvus里会堆满重复数据召回时就会把相似但无效的结果排到前面。我的做法是插入前先做一次向量查询如果相似度超过0.95就判断为重复记录跳过本次写入或者用upsert更新原有记录。另外历史数据要设置合理的保留周期我这里保留90天再久远的数据对当前决策的参考价值不大还浪费存储和检索时间。5.3 LangGraph执行参数LangGraph默认递归深度限制是25双Agent循环几次没问题但遇到任务步骤特别多、反复调整方案的场景可能跑着跑着就报超出深度限制。遇到这个报错先别急着调大参数先检查是不是条件边逻辑有问题导致死循环然后才考虑调大recursion_limitapp graph.compile() result app.invoke(initial_state, config{recursion_limit: 40})给节点加超时控制也很有用。工具调用如果依赖外部服务一旦服务卡住整个图都会被拖住。我在执行节点外层加了超时包裹单个工具超过20秒就返回超时错误让主管重新决策。相比让整个任务挂死这个错误反馈更有价值。还有一个容易被忽略的点状态爆炸。如果工具执行结果每次都原样存进状态几十轮循环下来状态对象会越来越大。我前面的worker_node已经做了摘要压缩另外还要养成习惯状态字段里只保留本轮循环需要的信息用不上的中间结果统一清理。6. 实战中的坑与排查记录最后这部分是这次实战最花时间的部分把踩过的坑整理成速查表。遇到类似问题可以先对号入座。6.1 Milvus启动与连接问题最开始的坑是Milvus容器启动失败。本地开发机配置一般容器起不来排查发现是内存约束没设置好。Milvus单机模式下依赖的组件比较多内存不够就直接Out Of Memory。如果你的机器配置不太高建议把数据目录和内存限制调整小一点或者直接用官方提供的轻量模式做开发和联调生产环境再上独立部署。还有个很低级的坑collection建完以后忘了load()。Milvus里建索引和查询之间是有状态的collection必须load之后才能search。我一开始写完插入代码立刻search结果返回不了数据还以为是写入有问题实际是没load。flush之后记得确认collection状态是Loaded。6.2 召回质量与向量维度问题有一次检索结果全是乱码查了半天发现是Embedding模型加载错误生成向量的维度不是768而是别的维度插入时报错很明显。但更隐蔽的问题是查询时误用了另一个Embedding模型Milvus因为维度相同不会报错但语义空间完全对不上召回来的结果稀烂。后来我把模型加载封装成单例整个进程只用一个Embedding实例才根治这个问题。召回为空的情况也要优先检查是不是用了不同的向量模型。排除了这个问题再看索引参数COSINE检索时向量要不要归一化Milvus会在内部处理但如果你在生成向量后自己又做了一遍归一化而历史数据没做归一化分数分布会乱掉。建议统一不在代码里做额外归一化让Milvus按metric_type统一处理。6.3 图执行与工具链问题死循环是双Agent图最怕的问题。表现是任务一直反复执行最终报recursion_limit超出。我排查后发现是条件边判断逻辑写反了主管返回next_action是“finish”时should_continue函数却返回了“worker”导致永远在循环。这类问题建议在should_continue函数里加日志打印每次判断结果和关键字段肉眼就能看出哪一步路由错了。工具返回格式不统一也是个高频问题。有的工具返回列表有的返回字典主管节点解析时就需要兼容多种格式。我给所有工具规定了统一的返回结构必须包含success和data两个字段success表示工具调用是否成功data是业务数据。执行节点先检查success字段为False就直接把错误信息交给主管重决策不再继续执行后面的计划。现象可能原因排查思路插入成功但查询不到数据未flush或collection未load检查flush状态确认collection是Loaded检索结果相关性差查询与写入用了不同Embedding模型检查向量来源确保全链路统一模型图执行死循环条件边路由逻辑写反在判断函数里打印状态检查next_action工具异常导致整体失败工具返回结构不统一统一返回successdata结构增加底层容错LLM输出JSON解析失败prompt没约束模型输出多余文本prompt强制“只输出JSON”解析层做兜底主管上下文过大工具结果没有摘要压缩执行节点返回前先压缩信息保留关键结论最后说两句这次把单Agent改成双Agent最直接的变化不是代码变复杂了而是思路变清晰了。原来所有职责都压在一个Agent身上它既要理解业务、又要调度工具、还要记住历史模型能力再强也容易被信息流冲昏头。拆成主管和执行之后每个Agent的职责变得明确LangGraph负责把两者的协作流程固化成图Milvus负责让每次决策都有据可循整个系统的可控性一下子上来了。不过我也有个实在的体会双Agent不是银弹。如果你的任务只是简单问答单Agent加上一个检索步骤就够了硬上双Agent反而增加复杂度和延迟。但如果你手里的任务正好是我开头说的那种“需要领域知识检索、需要多工具协同、需要根据中间结果动态决策”的情况这套架构值得一试。后续我还打算在主管节点里增加一个独立的质检Agent专门对最终回复做合规检查相当于在决策链路上再加一道保险。这样的扩展在LangGraph的图结构里改起来并不复杂这大概就是编排型架构比单个Prompt更值得投入的地方。
阅读完成 · 觉得有帮助？
