单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_k=3) # 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(uri="http://localhost:19530") # 建议显式定义schema，避免动态字段在频繁写入时出现类型不一致 schema = client.create_schema(auto_id=False, enable_dynamic_field=False) schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True) schema.add_field(field_name="task_type", datatype=DataType.VARCHAR, max_length=128) schema.add_field(field_name="task_description", datatype=DataType.VARCHAR, max_length=1024) schema.add_field(field_name="solution_steps", datatype=DataType.VARCHAR, max_length=8192) schema.add_field(field_name="result_summary", datatype=DataType.VARCHAR, max_length=4096) schema.add_field(field_name="embedding", datatype=DataType.FLOAT_VECTOR, dim=1024) client.create_collection( collection_name="agent_memory", schema=schema, metric_type="IP", consistency_level="Bounded" ) def save_task_memory(user_query, plan, final_report): client.upsert( collection_name="agent_memory", data=[{ "id": int(time.time() * 1000), "task_type": "调研报告", "task_description": user_query, "solution_steps": json.dumps(plan, ensure_ascii=False), "result_summary": final_report[:2000], "embedding": embed_text(user_query) }] )

注意dim=1024必须和Embedding模型输出维度一致。如果用不同的向量模型，要么改dim，要么重新建collection。这个坑我在后面会细说。

4.3 记忆召回：相似任务的"前车之鉴"

召回流程嵌入Executor执行前。Executor拿到一个任务项后，先把"整体用户需求+当前步骤目标"拼成一个查询句子，做Embedding，然后去Milvus里搜top_k条相似记忆：

def retrieve_memory(user_query, step_goal, top_k=3): query_text = f"{user_query} {step_goal}" query_embedding = embed_text(query_text) result = client.search( collection_name="agent_memory", data=[query_embedding], limit=top_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方案和双Agent+LangGraph+Milvus方案跑了20个任务，每个任务的产品方向都不重复但有部分相似领域，保证测试公平。

5.1 同一个任务集，两套方案的效果差距

结果对比看下面这张表，数据是这轮实测的相对值，不同模型、不同Prompt词可能会变，但趋势是一致的。

这里最打动我的不是成功率翻倍，而是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，用自己写的有界状态机也一样，先把角色拆开，把记忆留下，再谈复杂优化。