企业官网建设流程全解析

最近把一个客服工单自动处理系统从单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带来的收益会非常明显。下面这张表是我在项目里做的对比：

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没有长期记忆，每次对话都是从零开始，哪怕同一个工单类型它已经处理过一百遍，下一次还是要重新推理。把历史处理经验、领域知识库、业务规则都向量化存进Milvus，Agent在每次决策前先查一遍相关内容，等于把过往经验直接拼接到当前决策里。

我选了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(host="127.0.0.1", port="19530") fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=128), FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=2048), FieldSchema(name="create_time", dtype=DataType.VARCHAR, max_length=32), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768), ] schema = CollectionSchema( fields=fields, description="Agent long-term memory and domain knowledge" ) collection = Collection(name="agent_memory", schema=schema)

这里有几个“为什么”要讲清楚。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_name="embedding", index_params=index_params) collection.load()

文本场景下度量方式选COSINE比L2直观，因为文本向量的绝对长度受句子长度影响，余弦相似度只关心方向。HNSW是图索引，检索速度和召回率均衡，适合数据量中等、QPS要求不极端的场景。M控制每个节点的最大连接数，M越大召回越准、内存占用越大；efConstruction控制建索引时的候选集大小，越大索引质量越高、建索引越慢。M=16、efConstruction=200是一个比较稳的起点，后面按数据量调。

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: bool

user_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( data=query_embedding, anns_field="embedding", param={"metric_type": "COSINE", "params": {"ef": 64}}, limit=3, 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_format="json") 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( data=query_embedding, anns_field="embedding", param=search_params, limit=3, expr='source == "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_limit：

app = 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就直接把错误信息交给主管重决策，不再继续执行后面的计划。

最后说两句

这次把单Agent改成双Agent，最直接的变化不是代码变复杂了，而是思路变清晰了。原来所有职责都压在一个Agent身上，它既要理解业务、又要调度工具、还要记住历史，模型能力再强也容易被信息流冲昏头。拆成主管和执行之后，每个Agent的职责变得明确，LangGraph负责把两者的协作流程固化成图，Milvus负责让每次决策都有据可循，整个系统的可控性一下子上来了。

不过我也有个实在的体会：双Agent不是银弹。如果你的任务只是简单问答，单Agent加上一个检索步骤就够了，硬上双Agent反而增加复杂度和延迟。但如果你手里的任务正好是我开头说的那种“需要领域知识检索、需要多工具协同、需要根据中间结果动态决策”的情况，这套架构值得一试。后续我还打算在主管节点里增加一个独立的质检Agent，专门对最终回复做合规检查，相当于在决策链路上再加一道保险。这样的扩展，在LangGraph的图结构里改起来并不复杂，这大概就是编排型架构比单个Prompt更值得投入的地方。