1. 双Agent架构为什么单个Agent撑不住复杂任务先聊一个我踩坑踩出来的认知很多人上来就怼“单Agent也能做啊加个ReAct循环不就行了”这话对一半。单Agent在任务链路短、状态简单的时候确实够用但一旦涉及多轮工具调用、条件分支、上下文累积问题就暴露得很彻底——记忆混乱、工具调用序列失控、错误恢复几乎为零。我第一次做复杂任务Agent时用的就是单Agent方案需求其实不复杂“根据用户提问调用搜索工具找到相关资料再调用分析工具做总结最后输出答案”。听着简单对吧真正跑起来就发现灾难了第一轮搜索返回了5条结果Agent把它们全塞进上下文。第二轮调用分析工具时上下文已经有8000多token模型开始“选择性失忆”。一旦中间某一步工具返回格式异常整个链路全崩从头再来。调了十几次prompt还是治标不治本。后来我把架构改成双Agent协作问题迎刃而解。简单说就是一个Agent负责“想”一个Agent负责“做”。两个Agent各司其职通过共享状态和消息通道协同工作整个系统的稳定性和可维护性直接上了一个台阶。用生活类比解释一下单Agent就像一个全能店员既当收银员又当仓库管理员还当客服忙起来必然出错。双Agent则是分工明确的团队——一个前台接待拆解需求一个后台执行具体操作前台把需求清晰传递给后台后台把结果回传给前台。这就是本篇要展开的核心思想。2. 方案选型为什么是LangGraph Milvus这对组合2.1 LangGraph选型理由市面上的Agent框架不少我为什么选LangGraph而不是其他方案三个字可控性。LangGraph的核心卖点是“图结构的状态机”你可以在设计阶段就明确Agent的状态流转路径而不是完全交给模型自由发挥。我之前也试过纯LangChain的方式它的问题是线性链太死板没法很好地表达条件分支和循环而完全自研状态机工作量又太大维护成本极高。LangGraph提供了几个关键能力节点Node一个节点就是一个处理逻辑比如“规划节点”“执行节点”。边Edge节点之间的流转关系支持条件边通过条件函数决定走哪个分支。状态State整个图的共享状态是所有节点之间的信息传递载体。检查点Checkpoint自动保存状态快照任务中断后可以恢复续跑。这些特性刚好命中复杂Agent的需求。设计模式层面我采用的是Supervisor模式一个Supervisor Agent主管负责拆解任务和调度一个Worker Agent执行者负责具体工具调用。在LangGraph里这就是两个节点加上条件边。2.2 Milvus在架构中的角色很多人在做Agent知识库时习惯用传统关系型数据库或ES但我在这个项目里明确选择了Milvus向量数据库原因有四语义检索Agent需要的是“根据问题意思找相关内容”传统数据库的关键词匹配显然不够。高并发实时性Milvus基于Milvus的标量向量混合检索能力能在毫秒级返回TopK相似内容。动态Schema知识库的字段结构可以灵活调整不用像关系库那样提前定死表结构。开源生态好周边工具链成熟和LangChain/LangGraph都有官方集成接入成本低。打个比方LangGraph是大脑中枢Milvus就是记忆存储区。Agent处理任务时需要快速检索相关“记忆片段”就是通过Milvus完成的。2.3 整体架构总览先给一个整体视图方便你先建立全局认识用户提问 → 进入LangGraph的入口节点Supervisor Agent主管分析问题意图拆解子任务判断是否需要检索知识库如果需要检索 → 执行子任务调用Milvus进行语义检索取回TopK相关片段Worker Agent执行者基于检索结果执行具体操作如调用工具、生成答案结果回传给Supervisor Agent进行最终校验与汇总输出实际实现中我把它设计成了四个核心节点意图识别节点、检索规划节点、工具调用节点、结果综合节点。其中前两个属于Supervisor Agent后两个属于Worker Agent中间用条件边连接。3. 环境准备与Milvus向量库的构建3.1 环境依赖安装这里直接给出我在项目的依赖清单用的Python 3.10pip install langgraph langchain langchain-openai pymilvus如果需要做文本向量化再加一个嵌入模型库我用的是HuggingFace的sentence-transformers安装方式pip install sentence-transformers3.2 启动Milvus服务Milvus目前支持Docker部署和Milvus Lite本地模式。我们的项目用的是Milvus Lite方便开发和测试一条命令启动from pymilvus import MilvusClient client MilvusClient(milvus_demo.db)如果你用Docker部署则是另一个流程这里不展开。Milvus Lite的好处是零配置对个人项目和Demo阶段特别友好正式环境可以无缝切换Docker或集群版。3.3 设计Collection结构在实际项目中我设计的是这样一个Collection字段名类型说明idINT64主键自增contentVARCHAR知识片段的原文vectorFLOAT_VECTOR文本嵌入向量维度768sourceVARCHAR来源标记timestampINT64入库时间创建Collection的代码from pymilvus import DataType schema client.create_schema(auto_idTrue, enable_dynamic_fieldTrue) schema.add_field(field_nameid, datatypeDataType.INT64, is_primaryTrue) schema.add_field(field_namecontent, datatypeDataType.VARCHAR, max_length2048) schema.add_field(field_namevector, datatypeDataType.FLOAT_VECTOR, dim768) schema.add_field(field_namesource, datatypeDataType.VARCHAR, max_length128) schema.add_field(field_nametimestamp, datatypeDataType.INT64) index_params client.prepare_index_params() index_params.add_index(field_namevector, index_typeAUTOINDEX, metric_typeCOSINE) client.create_collection(collection_nameagent_knowledge, schemaschema, index_paramsindex_params)注意向量维度必须和你的嵌入模型输出维度严格一致不一致会直接报错。我用的模型是768维如果你是其他模型先确认维度再建表。3.4 知识入库拆分与向量化这里有个重要细节不能整个文档直接塞进向量库需要先拆分。拆分策略我踩过不少坑后面会单独讲。简单版本from sentence_transformers import SentenceTransformer model SentenceTransformer(shibing624/text2vec-base-chinese) # 假设knowledge_chunks是拆分好的文本片段列表 chunks [Milvus是一个向量数据库系统, LangGraph是构建Agent的框架, ...] embeddings model.encode(chunks).tolist() data [ {content: text, vector: vec, source: manual, timestamp: 1710000000} for text, vec in zip(chunks, embeddings) ] client.insert(collection_nameagent_knowledge, datadata)实践心得入库前先做一次简单数据清洗比如去除HTML标签、特殊符号、空白行。脏数据会影响检索质量这是最容易被忽略却最影响结果的一环。4. LangGraph双Agent的完整实现4.1 定义全局状态LangGraph里状态是图流转的核心。我先定义了一个总的状态类型from typing import TypedDict, List class AgentState(TypedDict): question: str intent: str need_knowledge: bool context_chunks: List[str] tool_result: str answer: str error: str这个状态贯穿全程每个节点按需读取或更新状态字段。实际项目中状态字段可能会更多但核心设计原则是节点尽量轻状态尽量全——也就是每个节点只负责一小块逻辑所有上下文信息放在状态里避免重复计算。4.2 Supervisor Agent节点意图识别第一个节点是意图识别也就是判断用户到底想问什么、需不需要走知识库检索。from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage llm ChatOpenAI(modelgpt-4o, temperature0) def intent_recognition_node(state: AgentState): prompt f 你是任务主管请分析用户的提问意图。 用户问题{state[question]} 输出格式要求 1. 判断是否需要检索知识库是/否 2. 用一句话概括任务类型 response llm.invoke([HumanMessage(contentprompt)]) # 简单解析响应这里省略复杂解析代码 state[intent] 需要知识检索 state[need_knowledge] 是 in response.content return state这里有个设计讲究意图识别节点只做判断不做具体检索。把“想”和“做”分离的好处是后续如果要加新的知识源或新的工具只需要改“做”的部分主管Agent完全不用动。4.3 条件边决定后续走向在LangGraph中条件边通过一个函数决定流转到哪个节点def route_after_intent(state: AgentState): if state[need_knowledge]: return retrieval_node else: return direct_answer_node这就是双Agent架构的核心优势——主管Agent负责路由决策而不是让一个Agent既做判断又做执行避免上下文混乱。4.4 Worker Agent节点Milvus检索检索节点由Worker Agent承担负责调用Milvus并取回相关内容def retrieval_node(state: AgentState): question_embedding model.encode([state[question]])[0].tolist() result client.search( collection_nameagent_knowledge, data[question_embedding], limit3, output_fields[content], search_params{metric_type: COSINE, params: {}}, ) chunks [hit[entity][content] for hit in result[0]] state[context_chunks] chunks return state这里注意一个细节我限制返回3条结果然后在Prompt里明确要求模型只能基于这些内容回答超出范围的宁可说“不知道”。这样能大幅减少幻觉。4.5 工具调用节点与结果生成工具调用节点是Worker Agent的另一部分根据检索结果组合成Prompt交给LLM生成最终答案def tool_call_node(state: AgentState): context \n\n.join(state[context_chunks]) prompt f 你是执行者请基于以下知识内容回答用户问题。 知识内容 {context} 用户问题{state[question]} 要求 - 只能基于给定知识回答不要编造 - 如果知识中不含相关信息明确说“暂未找到相关信息” response llm.invoke([HumanMessage(contentprompt)]) state[tool_result] response.content return state4.6 组装LangGraph图最后把这些节点组装成图from langgraph.graph import StateGraph, START, END graph StateGraph(AgentState) graph.add_node(intent_recognition, intent_recognition_node) graph.add_node(retrieval, retrieval_node) graph.add_node(generate, tool_call_node) graph.add_edge(START, intent_recognition) graph.add_conditional_edges(intent_recognition, route_after_intent, { retrieval_node: retrieval, direct_answer_node: generate }) graph.add_edge(retrieval, generate) graph.add_edge(generate, END) app graph.compile()到这里一个完整的双Agent应用就跑起来了。调用方式也很简单result app.invoke({question: 什么是向量数据库}) print(result[answer])5. 实操过程中的关键细节与参数调优5.1 文本拆分的粒度选择这是我最想强调的坑。刚开始我把知识库文本切成长度300字符的chunk检索效果很差经常找到内容“沾边但不精准”。后来改为按语义段落切分效果明显提升。我最终的拆分逻辑是按章节标题划分一级结构每个章节内再按空行和句号切分单条chunk长度控制在200-400字符之间相邻chunk保留少量重叠约50字符防止信息断裂。为什么这么做向量检索本质是“找最相似的语义片段”片段太小则信息不完整片段太大则语义被稀释嵌入向量不聚焦。200-400字符是实测下来性价比最高的区间。5.2 向量检索的TopK与相似度阈值两个参数直接影响结果质量TopK我一般设置3-5。太少了可能遗漏关键信息太多了噪声多影响LLM生成质量。相似度阈值Milvus支持返回相似度分数我实践中设定低于0.65的结果视为“不相关”直接过滤掉。这个值可以根据你的数据调如果检索效果普遍很好可以上调到0.7如果数据噪声大可以降到0.6。filtered_results [ hit for hit in result[0] if hit[distance] 0.65 ]5.3 Prompt设计经验双Agent架构里两个角色的Prompt定位完全不同主管Agent的Prompt要强调“判断优先不要执行”避免它越界执行Agent的Prompt要强调“严格基于资料不要自由发挥”避免幻觉。另外可以做一个小技巧在执行Agent的Prompt中加入“如果知识内容与问题不相关请直接说明”这能在很大程度上提高Response的可信度和准确性。实际业务中用户高质量提问的概率远低于预期这个兜底逻辑非常实用。5.4 LangGraph的调试技巧LangGraph给了一个可视化调试工具langgraph.dev但我实际用得最多的是手动打印状态from langgraph.graph import StateGraph def trace_node(node_name): def wrapper(state): print(f--- 进入节点: {node_name}) print(f当前状态: {state}) return state return wrapper在正式调用之前先在每个节点上加打印看状态流转是否符合预期。这一步能帮你省下大量排查时间。6. 常见问题与排查实录6.1 LangGraph节点重复执行现象同一个节点执行了好几次导致一次问答调用了多次LLM。原因条件边写错了路径图出现了循环边。LangGraph本身支持循环但如果循环条件没控制好就会卡死或重复。排查方法用上面的trace_node把每个节点进出情况打出来马上能看出来哪里多了路径。6.2 Milvus检索结果为空现象search返回的结果列表为空但collection里明明有数据。原因最常见的问题是collection没有被加载到内存。解决client.load_collection(collection_nameagent_knowledge)另一个原因提问的embedding维度与collection定义不匹配这会在查询时就报错不会静默失败。6.3 LLM输出内容超出知识范围现象基于知识库的问答模型还是会把常识甚至编造内容混进来。原因模型本身的知识注入无法“关闭”Prompt里的限制只能尽量约束。解决我用了两层保障。第一层是Prompt强约束第二层是答案后校验——用一个小模型判断答案中的关键信息是否能在检索结果中找到出处找不到就标记为“低置信度答案”。这个方法实测能把幻觉率降一半以上。6.4 最终答案质量不稳定同一问题多次回答内容差异很大。这个问题的根源是LLM的采样随机性和检索命中的细微差异。解决把LLM的temperature调低到0或不高于0.2同时固定检索返回顺序把相似度最高的结果放在最前面。这两步做完稳定性能提升非常多。6.5 状态字段越积越多LangGraph的State会随着节点流转而积累如果每个节点都往state里塞辅助信息最终state会越来越大。建议只保留下游节点真正需要的字段临时字段用完即删。我实际项目中State里还加了cache字段但设计清空逻辑防止不必要的token浪费。7. 性能优化与线上部署经验7.1 耗时瓶颈分析一个完整问答链路的耗时分布大致是环节耗时占比意图识别LLM调用15%Milvus检索5%答案生成LLM调用70%其它开销10%优化思路大多数场景不需要把两段LLM调用串行化。如果用户提问明显是知识查询类可以直接跳过意图识别走检索生成。我后来在入口加了一个“快速拦截”规则用极小的模型或正则先做一次预判把简单问题直接分流到短链路整个响应时间能缩短一半。7.2 向量索引与缓存Milvus自带AUTOINDEX已经比较高效但如果数据量大到百万级建议关注索引类型。我测试下来IVF_FLAT在精度和速度上表现均衡HNSW在高维数据上延迟更低但索引构建更耗时。另外强烈推荐加一层查询缓存把历史问题和检索结果存到Redis里相同问题直接命中缓存不再走LLM和Milvus。线上约30%的重复查询可以从缓存直接返回。7.3 部署注意事项我最终部署用的是容器化方案LangGraph API服务用的是FastAPI封装Milvus用的是独立服务。部署时的关键点Milvus的CPU和内存分配要预留检索峰值余量LLM服务或调用API要做超时和重试封装整个LangGraph应用做成无状态服务便于横向扩展。具体封装代码在这里不展开了但这一层在实际项目里很重要——否则demo跑得好好的一上线就各种超时和内存爆掉。8. 双Agent架构的边界与扩展方向双Agent并不是万能的。任务极简单时双Agent反而增加延迟和token消耗。判断要不要用双Agent我建议看三点任务链路是否超过3步是否同时涉及“知识检索”和“工具操作”是否需要问题拆解与结果汇总。以上三点命中两条就可以考虑双Agent。如果觉得双Agent还不够用可以进一步扩展为多Agent比如专门负责长文档处理的Research Agent专门负责表格处理的Analysis Agent等。LangGraph天然支持这种多Agent编排你只需要把图节点扩展状态类型扩展逻辑基本是复用。我自己在这个项目之后又迭代了一个带人工审核环节的版本在最终答案生成后如果置信度低自动发起人工复核任务。这其实已经走向了“Human-in-the-loop”方向LangGraph的interrupt机制完全可以实现这一点感兴趣的话可以沿着这个思路继续深入。回到个人实操体会双Agent架构最关键的不是框架选型而是职责边界的设计。主管Agent管“做什么”执行Agent管“怎么做”这条线画清楚了后面所有问题都好解决。至于Milvus的选型对于知识密集型Agent场景向量库几乎是刚需。两者结合这套方案已经足够覆盖绝大多数复杂任务Agent的落地需求。
)
