单Agent搞不定复杂任务这句话我一开始是真不服。直到我在一个真实的调研报告生成任务上连续翻车两周才承认问题不在模型而在架构。后来我把流程改成双Agent用LangGraph做状态编排、Milvus做外部向量记忆才算把这套流程彻底跑稳。这篇就围绕这条实战路径展开先从单Agent的翻车点说起再聊双Agent架构怎么设计然后是LangGraph和Milvus的落地细节、实测对比以及那些不跑一遍根本发现不了的坑。1. 先从一次翻车说起单Agent在处理复杂任务时的真实场景先交代一下任务背景。我有一个内部小工具需要根据用户提供的一句话产品方向自动产出一份包含竞品分析、目标人群画像、渠道投放建议的完整调研报告。比如用户输入智能水杯系统要走完检索资料、拆分维度、逐段分析、最后汇总成文的全流程。最初版本用的是很多人熟悉的单Agent模式一个大模型实例配上搜索、向量检索等若干工具让模型自己在ReAct循环里规划、调用、观察、再规划。1.1 一个看起来很简单的多步骤任务单Agent版的第一版Demo很快跑通了看起来每一步都合理模型先说要搜索智能水杯 主要品牌然后搜了接着说要查市场规模也搜了最后开始写报告。但一旦把任务从跑通变成稳定跑对问题就全出来了。我记得有次实测运行到第7轮时模型已经忘了最初要求的是竞品分析开始自顾自写智能水杯的功能定义还有一次它卡在搜索和重试的循环里先后调了11次工具消耗了大量token最后输出的段落里居然还混着原始搜索结果里的广告文案。这不是模型能力不行而是单Agent这种一个角色干所有事的模式在多步骤长任务里天然有结构性缺陷。这类任务不像Chat问答每一步都有明确依赖得先确定分析维度才能决定搜什么得先搜到资料才能写竞品对比得先写完对比才能总结投放建议。步骤之间有严格顺序任何一个环节跑偏后面全崩。1.2 单Agent模式的三个致命伤跑完一轮又一轮之后我把单Agent的失败模式归纳成了三类这三类问题后来也成为我下决心改造架构的直接原因。第一是上下文膨胀与注意力漂移。单Agent的所有中间思考、工具返回结果、历史错误都要堆在同一个对话里。任务越长历史越臃肿模型对最初指令的关注度就越低。典型表现是写着写着偏题或把中间检索的原始片段当结论输出。我当时的实测里第10轮之后生成的文本有将近三分之一是早期搜索结果的复读。第二是工具调用与推理交错导致的死循环。ReAct循环里模型需要自己判断现在该调用工具还是该直接回答。这个判断在高难度任务中经常失败导致它反复调用同一个工具、反复拿到同样的结果、反复重试。我碰到过最夸张的一次连续搜索同一个关键词5遍每次返回结果一样它还在说根据新的搜索结果显示明显陷入了循环却没有机制打断它。第三是难以注入确定性约束。复杂任务往往需要固定流程比如先拆解维度再逐项检索最后统一成稿。这种流程对单体Agent来说只能靠Prompt软约束。但Prompt写短了约束不住写长了模型会忽略后面部分。你很难在这个模式里硬性规定第2步做完之前不允许进入第3步。模型自由发挥的空间太大确定性就无从谈起。阶段小结单Agent不是不能跑而是它更适合单跳任务——一次检索、一次回答、一次生成。多步骤、强依赖、长输出的任务必须要从架构层面把角色拆开让不同环节各司其职。这也是双Agent方案出现的直接动机。2. 双Agent架构设计把想和做分开既然单Agent的问题在于角色负载过重那最直接的办法就是拆角色。我把整个流程拆成两个Agent一个叫Planner规划者一个叫Executor执行者。前者负责想后者负责做。Milvus则作为两者的外部记忆底座负责存储和召回历史经验让每一步决策都站在过往任务肩膀上而不是每次都从零开始。2.1 为什么是双Agent而不是更多Agent有人会问既然拆开好为什么不拆成三四个Agent比如再搞个专门的检索Agent、写作Agent我的经验是多Agent的协调成本是指数级上升的。每多一个角色就意味着多一层通信协议、多一份上下文传递、多N种互相等对方结果的阻塞风险。Agent之间对齐、传递中间产物、确认状态每一步都在消耗token和时间。拆到双Agent是复杂度下降和协调开销增加之间的平衡点。以这个调研报告任务为例拆成三四个Agent的收益很有限。检索和写作都只是执行动作完全可以归Executor管而真正需要判断下一步做什么的决策权收归Planner就好。两个角色分工干净协作简单效果提升却非常明显。2.2 分工边界Planner主控加Executor执行两个角色必须边界清晰否则还是变相单Agent。Planner只负责拆解目标。接收用户需求后生成一个结构化的任务清单每个任务项包含目标、依赖、完成标准。它还负责在每步执行完后检查结果质量决定继续、重试还是收尾。Planner自己不去调用搜索工具也不去拼接报告文本。Executor只负责执行单个动作。拿到Planner下发的任务项去调用外部工具包括搜索引擎、Milvus记忆检索、文本生成等然后把结果写回状态。Executor不做全局规划只关注当前这一步。两者的协作协议也很简单我用JSON格式传任务项和结果状态。Planner下发{ task_id: task_001, goal: 收集智能水杯主要竞品及产品定位, dependencies: [], output_format: markdown, max_retry: 2 }Executor执行完回传{ task_id: task_001, status: success, summary: 共识别出5个主要竞品..., raw_data: 原始资料拼接..., memory_key: milvus_8f3a... }这套协议的核心价值在于Planner不用关心原始数据有多长Executor不用操心整体目标是什么。每轮协作的上下文开销都固定下来了。2.3 为什么Executor需要Milvus一开始Executor是纯无状态的每次执行都从空白开始搜到什么用什么。后来发现一个问题——同一个方向的任务跑第二次时它还是会绕弯路还是会把早期犯过的错误再犯一遍。比如第一次执行时Executor走了先搜资讯、再搜竞品、再搜渠道的顺序效果很好第二次遇到相似任务它却先搜了渠道再返回来补竞品浪费了好几轮。这说明什么说明Agent缺记忆。对话上下文里的记忆是短时的、会溢出的真正需要的是把任务特征和成功路径固化下来的长期记忆。Milvus在这里的作用就是把每次成功执行的任务描述、步骤、结果摘要向量化存储下次遇到相似任务时通过语义检索把这条前车之鉴捞出来直接注入Executor的Prompt。这套设计后来被证明是整个方案里性价比最高的改进。3. LangGraph状态图落地把Agent协作变成可视化编排角色和协议定了之后下一步就是把它变成代码。我当时调研了几个方案最终选择LangGraph理由很直接它把Agent协作表达成一张状态图节点是任务动作边是流转条件每个节点的输入输出都走统一状态对象。这让谁先执行、什么条件下重试、什么条件下结束变得完全显式不再依赖模型自觉。3.1 为什么选LangGraph而不是自己写一个while循环有人可能觉得双Agent嘛用两个Prompt加一个for循环交替调用不就完了我最初也是这么想的但很快发现几个痛点没有状态快照Agent跑到一半崩溃了所有中间结果全丢只能从头再来。循环逻辑写死在代码里想加一个质量检查不合格就重试的边要改一大段控制流。打印Debug信息极其痛苦你不知道当前在哪一步、上一步发生了什么、为什么走到了这里。LangGraph把这些东西都变成了图的属性。你定义好State、Node、Edge它负责调度和执行还能自动保存Graph状态快照断点恢复、人工介入都方便很多。调试时直接看状态流转记录比看一堆杂乱日志清晰得多。3.2 核心状态设计与图结构状态是整个图的数据中枢我定义成下面这样from typing import TypedDict, List, Dict, Optional class AgentState(TypedDict): user_query: str # 用户原始需求 plan: List[Dict] # 任务清单由Planner生成 current_step: int # 当前正在执行的任务下标 task_context: Dict[str, Dict] # 每个任务的执行结果存放处 memory_hits: List[Dict] # 从Milvus召回的历史经验 final_report: Optional[str] # 最终交付的调研报告 status: str # planning / executing / completed / failed所有节点函数统一接收state、返回state的部分更新。这样每个节点都只关心自己负责的那一块数据不用维护全局变量。完整图结构如下from langgraph.graph import StateGraph, END graph StateGraph(AgentState) graph.add_node(planner, planner_node) graph.add_node(executor, executor_node) graph.add_node(quality_check, quality_check_node) graph.set_entry_point(planner) # Planner生成计划后决定是进入执行还是直接收尾 graph.add_conditional_edges( planner, decide_next_step, {execute: executor, finished: END} ) # 执行完一个任务项进入质量检查 graph.add_edge(executor, quality_check) # 质量检查通过就交给Planner推进下一步不通过就回Executor重试 graph.add_conditional_edges( quality_check, review_result, {next: planner, retry: executor, accept: END} ) app graph.compile()这里有三个节点和两条条件边。实际运行时会发现这个图结构看起来简单但它完整覆盖了规划-执行-质检-推进这个闭环。这也是LangGraph带来的核心价值流程本身是一张可视化图哪里会循环、哪里会结束一眼就能看懂而不是埋在一堆if-else里。3.3 三个节点的实现细节Planner节点的第一职责是生成任务清单。我用一个专门的结构化Prompt让模型输出JSON格式的步骤列表每项包含目标、依赖和产物说明def planner_node(state: AgentState): # 首次进入生成执行计划 if not state.get(plan): plan build_plan(state[user_query]) return {plan: plan, current_step: 0, status: executing} # 后续进入判断是否所有任务项都已执行完 if state[current_step] len(state[plan]): report build_final_report(state) return {final_report: report, status: completed} return {}build_plan内部就是一次带JSON格式约束的模型调用要求只输出任务列表不做实际执行。这是整个架构的关键约束Planner永远只说做什么不说怎么做。Executor节点负责执行单步任务。它在动手前会先去Milvus检索相似历史经验再把经验连同任务项一起交给模型最后调用工具拿回结果def executor_node(state: AgentState): step state[plan][state[current_step]] task_key step[task_id] # 1. 从Milvus检索相似任务的历史成功路径 memory retrieve_memory(state[user_query], step[goal], top_k3) # 2. 调用搜索工具获取本步骤原始资料 raw_data web_search(step[goal]) # 3. 把当前结果存入task_context new_entry { goal: step[goal], raw_data: raw_data, memory: memory, retry_count: 0 } updated dict(state[task_context]) updated[task_key] new_entry return {task_context: updated}Executor的Prompt里会把memory拼进去明确提示以下是历史相似任务的执行经验请注意规避其中提到的坑。这一步看着不起眼但正是因为有了Milvus第二次及以后执行相同领域的任务时质量会肉眼可见地提升。质量检查节点是双Agent方案里很容易被人忽略的一个角色。没有它节点之间只是机械地执行完就推进错误会被一路带到最终报告。加上它之后每次执行完一个任务项都会校验结果里是否包含关键信息、是否偏离目标def quality_check_node(state: AgentState): step state[plan][state[current_step]] task_key step[task_id] current state[task_context].get(task_key, {}) # 简单规则结果为空、或明显不含目标关键词则判定不合格 if not contains_key_info(current.get(raw_data, ), step[goal]): # 记录重试次数防止无限重试 current[retry_count] current.get(retry_count, 0) 1 updated dict(state[task_context]) updated[task_key] current return {task_context: updated} # 合格任务完成后推进current_step return {current_step: state[current_step] 1}条件边的路由函数则根据当前状态决定往哪走def decide_next_step(state: AgentState): if state[status] completed: return finished return execute def review_result(state: AgentState): step state[plan][state[current_step]] current state[task_context].get(step[task_id], {}) if current.get(retry_count, 0) 3: return accept # 重试次数太多先接受避免死循环 if current.get(retry_count, 0) 0 and current[retry_count] 2: return retry return next这套实现跑起来后我最直观的感受是问题定位变得极其简单。哪个任务项质量不行看状态就知道哪个环节循环了看边就知道。单Agent时代那种黑盒里乱跑的失控感彻底消失了。4. Milvus向量记忆系统让Agent在复杂任务中不失忆LangGraph解决了流程问题Milvus解决的是经验问题。这一节单独展开因为它值得展开——如果只是双Agent协作效果提升可能只有30%加上Milvus的长期记忆后提升才真正到了质变级别。4.1 为什么需要一套独立于对话的外部记忆Agent的上下文窗口再大也是有限资源。你可以把历史经验塞在对话里但问题是经验会占据大量context导致当前任务的可用空间变小而且语义上过去的成功步骤和当前任务描述混在一起模型容易混淆。更好的做法是把记忆存到外部数据库用向量相似度来召回这正是Milvus擅长的场景。我把Milvus定位成Agent的长期工作记忆。每次任务完成后把这条任务的特征描述、执行步骤、结果摘要存为一个记忆单元下次执行新任务时先按语义相似度捞回最相关的几条历史记忆。这样Agent不是每次从零开始推理而是站在历史经验上做增量决策。4.2 记忆写入管线分段、向量化、入库记忆写入发生在整个任务成功结束之后。流程是三步清洗、向量化、upsert。清洗阶段我会从最终计划和结果里抽取出三个字段task_description当前任务最核心的一句话描述比如智能水杯竞品分析报告。solution_steps完整执行步骤JSON是这次任务的成功路径。result_summary最终报告的前2000字摘要让Agent知道这条路最终产出长什么样。然后调用Embedding模型把task_description转成向量最后写入Milvus。代码示意import time from pymilvus import MilvusClient, DataType client MilvusClient(urihttp://localhost:19530) # 建议显式定义schema避免动态字段在频繁写入时出现类型不一致 schema client.create_schema(auto_idFalse, enable_dynamic_fieldFalse) schema.add_field(field_nameid, datatypeDataType.INT64, is_primaryTrue) schema.add_field(field_nametask_type, datatypeDataType.VARCHAR, max_length128) schema.add_field(field_nametask_description, datatypeDataType.VARCHAR, max_length1024) schema.add_field(field_namesolution_steps, datatypeDataType.VARCHAR, max_length8192) schema.add_field(field_nameresult_summary, datatypeDataType.VARCHAR, max_length4096) schema.add_field(field_nameembedding, datatypeDataType.FLOAT_VECTOR, dim1024) client.create_collection( collection_nameagent_memory, schemaschema, metric_typeIP, consistency_levelBounded ) def save_task_memory(user_query, plan, final_report): client.upsert( collection_nameagent_memory, data[{ id: int(time.time() * 1000), task_type: 调研报告, task_description: user_query, solution_steps: json.dumps(plan, ensure_asciiFalse), result_summary: final_report[:2000], embedding: embed_text(user_query) }] )注意dim1024必须和Embedding模型输出维度一致。如果用不同的向量模型要么改dim要么重新建collection。这个坑我在后面会细说。4.3 记忆召回相似任务的前车之鉴召回流程嵌入Executor执行前。Executor拿到一个任务项后先把整体用户需求当前步骤目标拼成一个查询句子做Embedding然后去Milvus里搜top_k条相似记忆def retrieve_memory(user_query, step_goal, top_k3): query_text f{user_query} {step_goal} query_embedding embed_text(query_text) result client.search( collection_nameagent_memory, data[query_embedding], limittop_k, output_fields[task_description, solution_steps, result_summary], search_params{metric_type: IP, params: {nprobe: 16}} ) hits [] for item in result[0]: entity item[entity] hits.append({ score: item[distance], task_description: entity[task_description], solution_steps: entity[solution_steps], result_summary: entity[result_summary] }) return hits召回结果不会原封不动全塞给模型而是经过一道轻量压缩只保留与当前步骤目标重叠度最高的步骤序列并按score排序拼进Prompt。例如以下是从记忆库中召回的历史经验供参考注意规避其中的失败教训 1. (相似度0.87) 任务方向智能水杯竞品分析成功步骤先拆维度-再竞品-再渠道... 2. (相似度0.82) 任务方向智能手环竞品分析成功步骤先市场规模-再头部品牌...这个过程最大的价值在于冷启动后的持续进化。第一个任务没有历史记忆Agent纯靠探索第二个任务开始就有了可参考的路径第三个、第四个任务记忆库越来越厚成功率自然越来越高。我实测里同一领域的第三个任务比第一个任务少用了接近三分之一的执行轮次。5. 实测数据与避坑经验双Agent到底值不值最后说结论之外的东西。我拿同一套调研报告生成任务集分别用单Agent方案和双AgentLangGraphMilvus方案跑了20个任务每个任务的产品方向都不重复但有部分相似领域保证测试公平。5.1 同一个任务集两套方案的效果差距结果对比看下面这张表数据是这轮实测的相对值不同模型、不同Prompt词可能会变但趋势是一致的。指标单Agent方案双Agent LangGraph Milvus任务成功率最终报告可交付35%80%平均执行轮次18.611.4平均token消耗单次任务万12.38.1报告结构达标率章节完整50%90%同领域第二次执行轮次下降无记忆不适用下降约28%这里最打动我的不是成功率翻倍而是token消耗反而降了。很多人以为架构复杂了成本肯定更高。实际恰恰相反单Agent大量token浪费在偏题、重复搜索、无效推理上双Agent把每一步限定在固定任务项上Milvus又减少了探索成本总消耗反而更省。5.2 折腾过程中踩过的几个坑这套方案不是一次跑通的中间踩的坑可以单独写一篇。挑几个最典型的第一个坑LangGraph节点里忘了return状态静默丢失。LangGraph的State是快照式的节点必须把要修改的字段显式return出来。有几次我在Executor里直接改了传入的state字典以为会生效结果节点执行完改动全没了。排查半天才发现是返回对象里少写了字段。我的经验是每个节点函数结尾一定要显式return所有需要变更的字段哪怕没变更的部分也要明确返回。第二个坑Milvus检索结果塞爆了Prompt。刚开始我把top_k设成10把每条记忆的solution_steps完整拼进Prompt结果超长导致模型反而忽略关键信息。后来把top_k压到3每条记忆只截取与当前步骤最相关的部分效果立刻好转。外部记忆不是越多越好你要喂给Agent的是被筛选过的经验不是原始数据堆。第三个坑Executor陷入无限重试。质量检查逻辑如果只看结果里有没有关键词很容易让Executor反复重试也达不到标准。后来加了重试上限超过三次就强制接受当前结果宁可把这个任务项标记为低置信度也不要卡死在循环里。这个思路后来也被证明正确复杂任务里某个步骤不完美是常态整体交付比局部完美更重要。第四个坑Milvus的一致性级别。这个容易被忽略。如果collection的一致性级别设成Strong写入后立刻能查到但写入性能会打折如果设成Eventually刚写入的记忆很可能检索不到。对于Agent记忆这种场景我推荐Bounded或Session级别结合具体一致性需求选择既保证绝大多数情况能即时读到又不牺牲太多写入性能。5.3 这套架构后续还能怎么扩展双Agent只是起点这套架构的可扩展性其实比想象中高。我个人目前在往两个方向深化一是记忆的衰减与合并。现在Milvus里的记忆只会不断堆积时间久了会有大量相似记忆彼此干扰。后续可以按时间戳做衰减把过时的执行路径降权还可以把多条相似记忆合并成一条通用路径减少检索噪声。二是加入第三个校验Agent。这里说的第三个Agent不是加重协调负担而是只负责最终报告的人工模拟审核不参与中间流程。它独立存在可以避免写报告的人自己检查自己的盲区。用LangGraph加一个节点就能接进去状态图也完全兼容。最后再分享一个个人体会。踩过几次坑之后我最大的感触是Agent应用的工程难点早就不是模型能干什么而是怎么把一个个能干的动作约束成一条可靠的流程。双Agent、LangGraph、Milvus这三个东西恰好分别回答了谁来做按什么顺序做用过什么经验做这三个问题。如果你也在被类似的长任务Agent搞得头疼不妨照这个思路先搭一版哪怕不用LangGraph用自己写的有界状态机也一样先把角色拆开把记忆留下再谈复杂优化。
单Agent搞不定复杂任务这句话我一开始是真不服。直到我在一个真实的调研报告生成任务上连续翻车两周才承认问题不在模型而在架构。后来我把流程改成双Agent用LangGraph做状态编排、Milvus做外部向量记忆才算把这套流程彻底跑稳。这篇就围绕这条实战路径展开先从单Agent的翻车点说起再聊双Agent架构怎么设计然后是LangGraph和Milvus的落地细节、实测对比以及那些不跑一遍根本发现不了的坑。1. 先从一次翻车说起单Agent在处理复杂任务时的真实场景先交代一下任务背景。我有一个内部小工具需要根据用户提供的一句话产品方向自动产出一份包含竞品分析、目标人群画像、渠道投放建议的完整调研报告。比如用户输入智能水杯系统要走完检索资料、拆分维度、逐段分析、最后汇总成文的全流程。最初版本用的是很多人熟悉的单Agent模式一个大模型实例配上搜索、向量检索等若干工具让模型自己在ReAct循环里规划、调用、观察、再规划。1.1 一个看起来很简单的多步骤任务单Agent版的第一版Demo很快跑通了看起来每一步都合理模型先说要搜索智能水杯 主要品牌然后搜了接着说要查市场规模也搜了最后开始写报告。但一旦把任务从跑通变成稳定跑对问题就全出来了。我记得有次实测运行到第7轮时模型已经忘了最初要求的是竞品分析开始自顾自写智能水杯的功能定义还有一次它卡在搜索和重试的循环里先后调了11次工具消耗了大量token最后输出的段落里居然还混着原始搜索结果里的广告文案。这不是模型能力不行而是单Agent这种一个角色干所有事的模式在多步骤长任务里天然有结构性缺陷。这类任务不像Chat问答每一步都有明确依赖得先确定分析维度才能决定搜什么得先搜到资料才能写竞品对比得先写完对比才能总结投放建议。步骤之间有严格顺序任何一个环节跑偏后面全崩。1.2 单Agent模式的三个致命伤跑完一轮又一轮之后我把单Agent的失败模式归纳成了三类这三类问题后来也成为我下决心改造架构的直接原因。第一是上下文膨胀与注意力漂移。单Agent的所有中间思考、工具返回结果、历史错误都要堆在同一个对话里。任务越长历史越臃肿模型对最初指令的关注度就越低。典型表现是写着写着偏题或把中间检索的原始片段当结论输出。我当时的实测里第10轮之后生成的文本有将近三分之一是早期搜索结果的复读。第二是工具调用与推理交错导致的死循环。ReAct循环里模型需要自己判断现在该调用工具还是该直接回答。这个判断在高难度任务中经常失败导致它反复调用同一个工具、反复拿到同样的结果、反复重试。我碰到过最夸张的一次连续搜索同一个关键词5遍每次返回结果一样它还在说根据新的搜索结果显示明显陷入了循环却没有机制打断它。第三是难以注入确定性约束。复杂任务往往需要固定流程比如先拆解维度再逐项检索最后统一成稿。这种流程对单体Agent来说只能靠Prompt软约束。但Prompt写短了约束不住写长了模型会忽略后面部分。你很难在这个模式里硬性规定第2步做完之前不允许进入第3步。模型自由发挥的空间太大确定性就无从谈起。阶段小结单Agent不是不能跑而是它更适合单跳任务——一次检索、一次回答、一次生成。多步骤、强依赖、长输出的任务必须要从架构层面把角色拆开让不同环节各司其职。这也是双Agent方案出现的直接动机。2. 双Agent架构设计把想和做分开既然单Agent的问题在于角色负载过重那最直接的办法就是拆角色。我把整个流程拆成两个Agent一个叫Planner规划者一个叫Executor执行者。前者负责想后者负责做。Milvus则作为两者的外部记忆底座负责存储和召回历史经验让每一步决策都站在过往任务肩膀上而不是每次都从零开始。2.1 为什么是双Agent而不是更多Agent有人会问既然拆开好为什么不拆成三四个Agent比如再搞个专门的检索Agent、写作Agent我的经验是多Agent的协调成本是指数级上升的。每多一个角色就意味着多一层通信协议、多一份上下文传递、多N种互相等对方结果的阻塞风险。Agent之间对齐、传递中间产物、确认状态每一步都在消耗token和时间。拆到双Agent是复杂度下降和协调开销增加之间的平衡点。以这个调研报告任务为例拆成三四个Agent的收益很有限。检索和写作都只是执行动作完全可以归Executor管而真正需要判断下一步做什么的决策权收归Planner就好。两个角色分工干净协作简单效果提升却非常明显。2.2 分工边界Planner主控加Executor执行两个角色必须边界清晰否则还是变相单Agent。Planner只负责拆解目标。接收用户需求后生成一个结构化的任务清单每个任务项包含目标、依赖、完成标准。它还负责在每步执行完后检查结果质量决定继续、重试还是收尾。Planner自己不去调用搜索工具也不去拼接报告文本。Executor只负责执行单个动作。拿到Planner下发的任务项去调用外部工具包括搜索引擎、Milvus记忆检索、文本生成等然后把结果写回状态。Executor不做全局规划只关注当前这一步。两者的协作协议也很简单我用JSON格式传任务项和结果状态。Planner下发{ task_id: task_001, goal: 收集智能水杯主要竞品及产品定位, dependencies: [], output_format: markdown, max_retry: 2 }Executor执行完回传{ task_id: task_001, status: success, summary: 共识别出5个主要竞品..., raw_data: 原始资料拼接..., memory_key: milvus_8f3a... }这套协议的核心价值在于Planner不用关心原始数据有多长Executor不用操心整体目标是什么。每轮协作的上下文开销都固定下来了。2.3 为什么Executor需要Milvus一开始Executor是纯无状态的每次执行都从空白开始搜到什么用什么。后来发现一个问题——同一个方向的任务跑第二次时它还是会绕弯路还是会把早期犯过的错误再犯一遍。比如第一次执行时Executor走了先搜资讯、再搜竞品、再搜渠道的顺序效果很好第二次遇到相似任务它却先搜了渠道再返回来补竞品浪费了好几轮。这说明什么说明Agent缺记忆。对话上下文里的记忆是短时的、会溢出的真正需要的是把任务特征和成功路径固化下来的长期记忆。Milvus在这里的作用就是把每次成功执行的任务描述、步骤、结果摘要向量化存储下次遇到相似任务时通过语义检索把这条前车之鉴捞出来直接注入Executor的Prompt。这套设计后来被证明是整个方案里性价比最高的改进。3. LangGraph状态图落地把Agent协作变成可视化编排角色和协议定了之后下一步就是把它变成代码。我当时调研了几个方案最终选择LangGraph理由很直接它把Agent协作表达成一张状态图节点是任务动作边是流转条件每个节点的输入输出都走统一状态对象。这让谁先执行、什么条件下重试、什么条件下结束变得完全显式不再依赖模型自觉。3.1 为什么选LangGraph而不是自己写一个while循环有人可能觉得双Agent嘛用两个Prompt加一个for循环交替调用不就完了我最初也是这么想的但很快发现几个痛点没有状态快照Agent跑到一半崩溃了所有中间结果全丢只能从头再来。循环逻辑写死在代码里想加一个质量检查不合格就重试的边要改一大段控制流。打印Debug信息极其痛苦你不知道当前在哪一步、上一步发生了什么、为什么走到了这里。LangGraph把这些东西都变成了图的属性。你定义好State、Node、Edge它负责调度和执行还能自动保存Graph状态快照断点恢复、人工介入都方便很多。调试时直接看状态流转记录比看一堆杂乱日志清晰得多。3.2 核心状态设计与图结构状态是整个图的数据中枢我定义成下面这样from typing import TypedDict, List, Dict, Optional class AgentState(TypedDict): user_query: str # 用户原始需求 plan: List[Dict] # 任务清单由Planner生成 current_step: int # 当前正在执行的任务下标 task_context: Dict[str, Dict] # 每个任务的执行结果存放处 memory_hits: List[Dict] # 从Milvus召回的历史经验 final_report: Optional[str] # 最终交付的调研报告 status: str # planning / executing / completed / failed所有节点函数统一接收state、返回state的部分更新。这样每个节点都只关心自己负责的那一块数据不用维护全局变量。完整图结构如下from langgraph.graph import StateGraph, END graph StateGraph(AgentState) graph.add_node(planner, planner_node) graph.add_node(executor, executor_node) graph.add_node(quality_check, quality_check_node) graph.set_entry_point(planner) # Planner生成计划后决定是进入执行还是直接收尾 graph.add_conditional_edges( planner, decide_next_step, {execute: executor, finished: END} ) # 执行完一个任务项进入质量检查 graph.add_edge(executor, quality_check) # 质量检查通过就交给Planner推进下一步不通过就回Executor重试 graph.add_conditional_edges( quality_check, review_result, {next: planner, retry: executor, accept: END} ) app graph.compile()这里有三个节点和两条条件边。实际运行时会发现这个图结构看起来简单但它完整覆盖了规划-执行-质检-推进这个闭环。这也是LangGraph带来的核心价值流程本身是一张可视化图哪里会循环、哪里会结束一眼就能看懂而不是埋在一堆if-else里。3.3 三个节点的实现细节Planner节点的第一职责是生成任务清单。我用一个专门的结构化Prompt让模型输出JSON格式的步骤列表每项包含目标、依赖和产物说明def planner_node(state: AgentState): # 首次进入生成执行计划 if not state.get(plan): plan build_plan(state[user_query]) return {plan: plan, current_step: 0, status: executing} # 后续进入判断是否所有任务项都已执行完 if state[current_step] len(state[plan]): report build_final_report(state) return {final_report: report, status: completed} return {}build_plan内部就是一次带JSON格式约束的模型调用要求只输出任务列表不做实际执行。这是整个架构的关键约束Planner永远只说做什么不说怎么做。Executor节点负责执行单步任务。它在动手前会先去Milvus检索相似历史经验再把经验连同任务项一起交给模型最后调用工具拿回结果def executor_node(state: AgentState): step state[plan][state[current_step]] task_key step[task_id] # 1. 从Milvus检索相似任务的历史成功路径 memory retrieve_memory(state[user_query], step[goal], top_k3) # 2. 调用搜索工具获取本步骤原始资料 raw_data web_search(step[goal]) # 3. 把当前结果存入task_context new_entry { goal: step[goal], raw_data: raw_data, memory: memory, retry_count: 0 } updated dict(state[task_context]) updated[task_key] new_entry return {task_context: updated}Executor的Prompt里会把memory拼进去明确提示以下是历史相似任务的执行经验请注意规避其中提到的坑。这一步看着不起眼但正是因为有了Milvus第二次及以后执行相同领域的任务时质量会肉眼可见地提升。质量检查节点是双Agent方案里很容易被人忽略的一个角色。没有它节点之间只是机械地执行完就推进错误会被一路带到最终报告。加上它之后每次执行完一个任务项都会校验结果里是否包含关键信息、是否偏离目标def quality_check_node(state: AgentState): step state[plan][state[current_step]] task_key step[task_id] current state[task_context].get(task_key, {}) # 简单规则结果为空、或明显不含目标关键词则判定不合格 if not contains_key_info(current.get(raw_data, ), step[goal]): # 记录重试次数防止无限重试 current[retry_count] current.get(retry_count, 0) 1 updated dict(state[task_context]) updated[task_key] current return {task_context: updated} # 合格任务完成后推进current_step return {current_step: state[current_step] 1}条件边的路由函数则根据当前状态决定往哪走def decide_next_step(state: AgentState): if state[status] completed: return finished return execute def review_result(state: AgentState): step state[plan][state[current_step]] current state[task_context].get(step[task_id], {}) if current.get(retry_count, 0) 3: return accept # 重试次数太多先接受避免死循环 if current.get(retry_count, 0) 0 and current[retry_count] 2: return retry return next这套实现跑起来后我最直观的感受是问题定位变得极其简单。哪个任务项质量不行看状态就知道哪个环节循环了看边就知道。单Agent时代那种黑盒里乱跑的失控感彻底消失了。4. Milvus向量记忆系统让Agent在复杂任务中不失忆LangGraph解决了流程问题Milvus解决的是经验问题。这一节单独展开因为它值得展开——如果只是双Agent协作效果提升可能只有30%加上Milvus的长期记忆后提升才真正到了质变级别。4.1 为什么需要一套独立于对话的外部记忆Agent的上下文窗口再大也是有限资源。你可以把历史经验塞在对话里但问题是经验会占据大量context导致当前任务的可用空间变小而且语义上过去的成功步骤和当前任务描述混在一起模型容易混淆。更好的做法是把记忆存到外部数据库用向量相似度来召回这正是Milvus擅长的场景。我把Milvus定位成Agent的长期工作记忆。每次任务完成后把这条任务的特征描述、执行步骤、结果摘要存为一个记忆单元下次执行新任务时先按语义相似度捞回最相关的几条历史记忆。这样Agent不是每次从零开始推理而是站在历史经验上做增量决策。4.2 记忆写入管线分段、向量化、入库记忆写入发生在整个任务成功结束之后。流程是三步清洗、向量化、upsert。清洗阶段我会从最终计划和结果里抽取出三个字段task_description当前任务最核心的一句话描述比如智能水杯竞品分析报告。solution_steps完整执行步骤JSON是这次任务的成功路径。result_summary最终报告的前2000字摘要让Agent知道这条路最终产出长什么样。然后调用Embedding模型把task_description转成向量最后写入Milvus。代码示意import time from pymilvus import MilvusClient, DataType client MilvusClient(urihttp://localhost:19530) # 建议显式定义schema避免动态字段在频繁写入时出现类型不一致 schema client.create_schema(auto_idFalse, enable_dynamic_fieldFalse) schema.add_field(field_nameid, datatypeDataType.INT64, is_primaryTrue) schema.add_field(field_nametask_type, datatypeDataType.VARCHAR, max_length128) schema.add_field(field_nametask_description, datatypeDataType.VARCHAR, max_length1024) schema.add_field(field_namesolution_steps, datatypeDataType.VARCHAR, max_length8192) schema.add_field(field_nameresult_summary, datatypeDataType.VARCHAR, max_length4096) schema.add_field(field_nameembedding, datatypeDataType.FLOAT_VECTOR, dim1024) client.create_collection( collection_nameagent_memory, schemaschema, metric_typeIP, consistency_levelBounded ) def save_task_memory(user_query, plan, final_report): client.upsert( collection_nameagent_memory, data[{ id: int(time.time() * 1000), task_type: 调研报告, task_description: user_query, solution_steps: json.dumps(plan, ensure_asciiFalse), result_summary: final_report[:2000], embedding: embed_text(user_query) }] )注意dim1024必须和Embedding模型输出维度一致。如果用不同的向量模型要么改dim要么重新建collection。这个坑我在后面会细说。4.3 记忆召回相似任务的前车之鉴召回流程嵌入Executor执行前。Executor拿到一个任务项后先把整体用户需求当前步骤目标拼成一个查询句子做Embedding然后去Milvus里搜top_k条相似记忆def retrieve_memory(user_query, step_goal, top_k3): query_text f{user_query} {step_goal} query_embedding embed_text(query_text) result client.search( collection_nameagent_memory, data[query_embedding], limittop_k, output_fields[task_description, solution_steps, result_summary], search_params{metric_type: IP, params: {nprobe: 16}} ) hits [] for item in result[0]: entity item[entity] hits.append({ score: item[distance], task_description: entity[task_description], solution_steps: entity[solution_steps], result_summary: entity[result_summary] }) return hits召回结果不会原封不动全塞给模型而是经过一道轻量压缩只保留与当前步骤目标重叠度最高的步骤序列并按score排序拼进Prompt。例如以下是从记忆库中召回的历史经验供参考注意规避其中的失败教训 1. (相似度0.87) 任务方向智能水杯竞品分析成功步骤先拆维度-再竞品-再渠道... 2. (相似度0.82) 任务方向智能手环竞品分析成功步骤先市场规模-再头部品牌...这个过程最大的价值在于冷启动后的持续进化。第一个任务没有历史记忆Agent纯靠探索第二个任务开始就有了可参考的路径第三个、第四个任务记忆库越来越厚成功率自然越来越高。我实测里同一领域的第三个任务比第一个任务少用了接近三分之一的执行轮次。5. 实测数据与避坑经验双Agent到底值不值最后说结论之外的东西。我拿同一套调研报告生成任务集分别用单Agent方案和双AgentLangGraphMilvus方案跑了20个任务每个任务的产品方向都不重复但有部分相似领域保证测试公平。5.1 同一个任务集两套方案的效果差距结果对比看下面这张表数据是这轮实测的相对值不同模型、不同Prompt词可能会变但趋势是一致的。指标单Agent方案双Agent LangGraph Milvus任务成功率最终报告可交付35%80%平均执行轮次18.611.4平均token消耗单次任务万12.38.1报告结构达标率章节完整50%90%同领域第二次执行轮次下降无记忆不适用下降约28%这里最打动我的不是成功率翻倍而是token消耗反而降了。很多人以为架构复杂了成本肯定更高。实际恰恰相反单Agent大量token浪费在偏题、重复搜索、无效推理上双Agent把每一步限定在固定任务项上Milvus又减少了探索成本总消耗反而更省。5.2 折腾过程中踩过的几个坑这套方案不是一次跑通的中间踩的坑可以单独写一篇。挑几个最典型的第一个坑LangGraph节点里忘了return状态静默丢失。LangGraph的State是快照式的节点必须把要修改的字段显式return出来。有几次我在Executor里直接改了传入的state字典以为会生效结果节点执行完改动全没了。排查半天才发现是返回对象里少写了字段。我的经验是每个节点函数结尾一定要显式return所有需要变更的字段哪怕没变更的部分也要明确返回。第二个坑Milvus检索结果塞爆了Prompt。刚开始我把top_k设成10把每条记忆的solution_steps完整拼进Prompt结果超长导致模型反而忽略关键信息。后来把top_k压到3每条记忆只截取与当前步骤最相关的部分效果立刻好转。外部记忆不是越多越好你要喂给Agent的是被筛选过的经验不是原始数据堆。第三个坑Executor陷入无限重试。质量检查逻辑如果只看结果里有没有关键词很容易让Executor反复重试也达不到标准。后来加了重试上限超过三次就强制接受当前结果宁可把这个任务项标记为低置信度也不要卡死在循环里。这个思路后来也被证明正确复杂任务里某个步骤不完美是常态整体交付比局部完美更重要。第四个坑Milvus的一致性级别。这个容易被忽略。如果collection的一致性级别设成Strong写入后立刻能查到但写入性能会打折如果设成Eventually刚写入的记忆很可能检索不到。对于Agent记忆这种场景我推荐Bounded或Session级别结合具体一致性需求选择既保证绝大多数情况能即时读到又不牺牲太多写入性能。5.3 这套架构后续还能怎么扩展双Agent只是起点这套架构的可扩展性其实比想象中高。我个人目前在往两个方向深化一是记忆的衰减与合并。现在Milvus里的记忆只会不断堆积时间久了会有大量相似记忆彼此干扰。后续可以按时间戳做衰减把过时的执行路径降权还可以把多条相似记忆合并成一条通用路径减少检索噪声。二是加入第三个校验Agent。这里说的第三个Agent不是加重协调负担而是只负责最终报告的人工模拟审核不参与中间流程。它独立存在可以避免写报告的人自己检查自己的盲区。用LangGraph加一个节点就能接进去状态图也完全兼容。最后再分享一个个人体会。踩过几次坑之后我最大的感触是Agent应用的工程难点早就不是模型能干什么而是怎么把一个个能干的动作约束成一条可靠的流程。双Agent、LangGraph、Milvus这三个东西恰好分别回答了谁来做按什么顺序做用过什么经验做这三个问题。如果你也在被类似的长任务Agent搞得头疼不妨照这个思路先搭一版哪怕不用LangGraph用自己写的有界状态机也一样先把角色拆开把记忆留下再谈复杂优化。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
