简介面向企业级AI应用开发者这套代码演示了Dify与LangGraph在构建多智能体系统时的融合方式。Dify提供可视化开发界面LangGraph负责复杂工作流编排两者结合能有效降低智能体构建门槛同时提升系统的开发效率、编排灵活性与可靠性。压缩包内共有4个文件以Python脚本、HTML页面、Inscode配置和Git忽略文件为主包含可运行的对话分析系统示例完整代码仅11KB。目前已有208人学习。通过阅读项目代码可重点掌握Dify平台与LangGraph框架的接口对接方法理解持久化执行、人机交互、完整内存系统等核心机制在多智能体协同中的实际运用适合希望借助低代码平台快速落地复杂AI应用的开发者参考。Dify LangGraph 构建多智能体系统从架构设计到代码落地的完整记录先把结论说在前面Dify 负责把多智能体的“门面”和流程串起来LangGraph 负责把真正复杂的路由、循环、状态转移落到代码里。这套组合非常适合那些“既要可视化编排又要保持代码级可控”的落地场景。如果你正在纠结“Dify 到底能不能做 Agent 编排”“LangGraph 是不是太重了”这篇分享应该能给你一个清晰的答案。这篇文章不是纯理论科普我会拿一套实际跑通的多智能体系统来拆解主管 Agent 分流、条件路由、循环反思、子图复用、并行分支以及怎么和 Dify 工作流对接。建议阅读顺序是从头到尾代码段可以直接复制改但更重要的是理解“为什么这么设计”。1. 为什么要把 Dify 和 LangGraph 放在一起两者的边界与定位1.1 Dify 擅长什么LangGraph 又擅长什么Dify 是一个开源的 LLM 应用开发平台内置了知识库、工作流、Agent、模型管理和可视化的日志面板。它的强项是“把线性的业务逻辑快速搭起来”你可以在画布上拖一个知识检索节点接一个 LLM 节点再挂一个条件分支一个 RAG 问答应用就出来了。Dify 的 Agent 功能也支持工具调用但它的 Agent 编排本质上是“预设节点 工具调用”一旦遇到复杂的循环、嵌套子流程、需要精确控制状态迁移的场景画布上的连线会变得很难维护。LangGraph 是 LangChain 团队推出的图状态编排框架。它的核心思想是把一次对话或一次任务执行定义成一个状态机节点是处理逻辑边是状态迁移规则。正因为这种“图”结构LangGraph 可以天然表达循环、条件分支、并行执行、子图嵌套等能力。说白了Dify 适合做“流程可视化”LangGraph 适合做“逻辑精确化”。很多人会问Dify 不是也有工作流吗为什么还要绕一圈去写 LangGraph我实际用下来的体会是当你的业务只需要“单轮检索 生成回答”时Dify 工作流完全够用。但当你要做一个真正意义上的多智能体系统——比如一个主管 Agent 要把任务拆给多个专业 Agent还要根据执行结果决定是继续追问还是结束——用 Dify 画布来表达这种“动态循环”是很痛苦的。而 LangGraph 用几行conditional_edge就能写清楚。1.2 这套组合的真实适用场景结合我最近做的一个项目这套组合最适合以下三类场景第一客服工单分诊系统。用户进来先由主管 Agent 理解意图然后分发给售后、技术支持、财务等子 Agent每个子 Agent 可能还要调用各自的工具或知识库最后主管汇总答案。第二代码生成与审查系统。一个 Agent 负责写代码另一个负责 Review写完了不满意就循环回去改最多循环 N 次。这种“反馈闭环”用 Dify 工作流的普通分支很难写但在 LangGraph 里就是一个带终止条件的循环边。第三需要人工审核的自动化流程。比如内容审核先由多 Agent 对内容进行初筛再交给 Dify 工作流里的人工审核节点做保底。Dify 有现成的人工审核组件LangGraph 内部处理复杂判断两者互补。2. 多智能体系统整体设计先定角色再定协议2.1 多智能体的三种主流分工模式在我见过的多智能体架构里分工模式基本逃不出三类管道模式、主管-工人模式、网络模式。管道模式适合流水线场景每个 Agent 处理完一个阶段就传给下一个比如“意图识别 - 检索 - 总结”。网络模式最灵活Agent 之间互相通信但调试成本极高一般创业团队不建议碰。这套系统我选的是主管-工人模式Supervisor-Worker这也是多数生产级多智能体系统的首选。主管 Agent 不直接干活它负责理解用户需求、拆解任务、决定把任务派给哪个工人 Agent、以及判断结果是否满足要求。工人 Agent 各司其职比如“代码生成 Agent”“知识库问答 Agent”“外部工具调用 Agent”。这种模式的好处是任务流转路径清晰每个 Agent 的职责边界明确出问题容易排查。坏处是主管 Agent 容易成为瓶颈——如果它判断错误整个任务就偏了。所以我会在主管 Agent 的提示词里强调“不确定时优先询问用户不要自作主张”。2.2 我们的设计主管技能 Agent知识库具体到这次实现我设计了这样一套结构一个主管 AgentSupervisor三个工人 Agent——代码生成 Agent、信息检索 Agent、外部工具 Agent。另外挂了一个 Dify 知识库用于领域问答。任务流转规则是用户请求进来后主管先做意图识别如果涉及代码生成就走代码 Agent涉及具体领域问答就走知识库检索涉及外部数据比如查询天气、查汇率就走外部工具 Agent。工人 Agent 执行完毕后把结果回报给主管主管判断是否需要补充或调整不需要就返回最终答案。这套设计里有一个关键点每个工人 Agent 都是相对独立的节点可以在 LangGraph 里单独测试。我特别建议你先把每个 Agent 单独跑通再组装成图否则出了问题你根本不知道是哪个环节的问题。3. LangGraph 核心代码状态、路由、循环与子图3.1 环境准备与状态定义先安装依赖。我用的 Python 版本是 3.11LangGraph 当前版本支持得很稳定。pip install langgraph langchain langchain-openaiLangGraph 的核心是状态定义。状态就是一个 TypedDict它会在图的每个节点之间传递。我的状态定义如下from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[List, add_messages] # 对话消息流 current_agent: str # 当前应该交给哪个 Agent task: str # 拆分后的子任务 result: str # 工人 Agent 的执行结果 retry_count: int # 循环重试次数这里有几个设计细节值得一提。messages字段使用了add_messages注解这是 LangGraph 的一个巧妙的机制。普通的字典字段在节点间传递时会直接覆盖但加了add_messages之后新节点返回的消息会自动追加到历史列表里。这非常利于多 Agent 场景下的对话历史累积——主管能看到所有工人 Agent 的中间输出。retry_count是给循环用的。后面我会用条件边判断如果工人 Agent 执行结果不合格且重试次数少于 3就重新进入工人节点否则强制结束。这就是多智能体系统里“有限次反思循环”的标准写法。3.2 条件路由与分支控制conditional_edge先定义几个节点函数。每个节点的输入是整个 AgentState输出是部分状态的更新。来看主管节点和工人节点# 主管 Agent 节点 async def supervisor_node(state: AgentState): prompt f 你是多智能体系统的主管请根据用户请求选择处理路径 - code: 用户需要写代码或调试代码 - search: 用户需要知识库问答或信息检索 - tool: 用户需要调用外部工具天气、汇率等 用户请求{state[messages][-1].content} 只输出一个词code / search / tool response await llm.ainvoke(prompt) next_agent response.content.strip().lower() return {current_agent: next_agent} # 工人 Agent代码生成 async def code_agent_node(state: AgentState): prompt f你是资深工程师请完成以下任务并输出可运行代码{state[task]} response await llm.ainvoke(prompt) return {result: response.content} # 工人 Agent信息检索简化版 async def search_agent_node(state: AgentState): prompt f请基于知识库回答{state[task]} response await llm.ainvoke(prompt) return {result: response.content}接下来是条件路由的核心conditional_edge。它接收一个路由函数该函数根据当前状态返回下一个节点的名称。def route_after_supervisor(state: AgentState): agent_map { code: code_agent, search: search_agent, tool: tool_agent, } if state[current_agent] in agent_map: return agent_map[state[current_agent]] return END # 无法识别时直接结束避免死循环 # 构建图 graph StateGraph(AgentState) graph.add_node(supervisor, supervisor_node) graph.add_node(code_agent, code_agent_node) graph.add_node(search_agent, search_agent_node) graph.add_node(tool_agent, tool_agent_node) graph.add_edge(START, supervisor) graph.add_conditional_edges(supervisor, route_after_supervisor) graph.add_edge(code_agent, supervisor) # 工人执行完回到主管 graph.add_edge(search_agent, supervisor) graph.add_edge(tool_agent, supervisor) app graph.compile()这段代码的逻辑是用户请求进入主管主管输出一个 agent 类型条件边根据这个类型决定走到哪个工人节点工人执行完把结果写回result字段然后回到主管。这个图已经具备最基本的“主管分发 - 工人干活 - 主管再判断”的闭环。如果你看到这里觉得有点抽象可以这么理解conditional_edge就像是一个 switch-case 语句但它的判断条件是整个图的状态而不是某个局部变量。这在实际工程里太重要了——因为判断结果往往依赖前一个节点的输出而不是提前写死的。3.3 循环与终止机制有限次反思上面那个图有个问题工人执行完回到主管主管可能无论如何都不满意于是又派给工人来回无限循环。这个问题在真实系统里非常常见。解决方法是引入计数器用add_condition判断是否达到最大重试次数。我在状态里已经定义了retry_count。修改主管节点的逻辑async def supervisor_node(state: AgentState): last_result state.get(result, ) # 如果已有执行结果判断是否满足要求 if last_result: # 这里可以接入一个“质量评估”LLM 调用简化起见只做长度判断 if len(last_result) 50 or state.get(retry_count, 0) 2: return {messages: [{role: assistant, content: last_result}]} # 否则继续拆分任务 prompt f... 根据用户请求判断应该走哪个 Agent ... response await llm.ainvoke(prompt) return { current_agent: response.content.strip().lower(), retry_count: state.get(retry_count, 0) 1, }然后在图里加一个条件边当主管决定结束或者重试次数超过阈值时直接走到 ENDdef should_continue(state: AgentState): # 如果已经产生了最终答案结束 if final_answer in state: return END # 如果重试次数过多强制结束 if state.get(retry_count, 0) 3: return END return supervisor graph.add_conditional_edges(supervisor, should_continue, { supervisor: supervisor, # 继续循环 END: END, # 结束 })这里我踩过一个坑如果不在should_continue里加retry_count判断线上环境会经常出现“同一句话反复调用接口到超时”的问题。这种循环不是 bug但它会导致大量费用消耗。所以我的建议是给每一个循环边都加一个显式的终止条件并且把终止条件放在条件边函数最前面判断。3.4 子图与并行分支等基础闭环跑通之后你会需要更高阶的能力子图和并行。子图的价值在于“模块化复用”——同一套逻辑可以在不同的父图节点里被多次引用。并行分支的价值在于“提速”——比如在处理一个复杂问题时让代码 Agent 和检索 Agent 同时开工最后汇总。子图的写法很简单。先构建一个子图然后在父图里用add_node注册# 构建子图 subgraph StateGraph(AgentState) subgraph.add_node(search_agent, search_agent_node) subgraph.add_edge(search_agent, END) # 在父图里引用 graph.add_node(search_subgraph, subgraph.compile())并行分支需要借助Send对象。在 LangGraph 里Send允许你从同一个节点出发携带不同的状态副本去执行多个下游节点from langgraph.types import Send def spawn_parallel_tasks(state: AgentState): tasks state[task].split(|) # 按竖线切分成多个子任务 return [ Send(code_agent, {task: t.strip(), retry_count: 0}) for t in tasks ] graph.add_conditional_edges(supervisor, spawn_parallel_tasks)这样主管拆出 3 个子任务就会并发启动 3 个 code_agent 执行最终结果通过messages合并回来。这个模式在处理“批量生成 汇总”类需求时特别有用。4. Dify 侧编排把 LangGraph 接入工作流4.1 思路Dify 管入口LangGraph 管逻辑很多人的误区是把 LangGraph 当成一个“新的 Agent 框架”来替代 Dify但其实两者不是替代关系。LangGraph 是一个可嵌入的编排引擎不具备任何前端交互、模型管理、用户权限能力。Dify 恰好补齐了这些。我这个项目里的整体架构是用户在 Dify 前端对话Dify 的工作流节点负责接收用户输入然后通过一个“HTTP 请求”节点把请求转发给 LangGraph 服务LangGraph 完成多智能体调度后把结果返回给 DifyDify 再做后续处理比如知识库混合检索、人工审核、日志存储。4.2 HTTP 工具节点把 LangGraph 发布为接口首先要把 LangGraph 应用发布成一个可调用的 API。LangGraph 官方推荐的方式是使用langgraph serve或者langgraph dev它会自动生成一个带 Swagger 文档的服务。我的做法是写一个独立的 FastAPI 服务来包装from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): message: str session_id: str default class ChatResponse(BaseModel): result: str session_id: str app.post(/agent/chat, response_modelChatResponse) async def chat(req: ChatRequest): config {configurable: {thread_id: req.session_id}} final_state await graph.ainvoke( {messages: [{role: user, content: req.message}]}, configconfig ) return ChatResponse( resultfinal_state[messages][-1].content, session_idreq.session_id )把这个服务启动在 8000 端口后就可以在 Dify 工作流里添加一个 HTTP 请求节点配置如下请求方式POSTURLhttp://localhost:8000/agent/chatHeaderContent-Type: application/jsonBody{message: {{#sys.query#}}, session_id: {{#conversation.id#}}}Dify 内部有大量的变量引用语法sys.query是当前用户输入conversation.id是会话 ID。通过session_idLangGraph 的thread_id可以记住同一个会话的多轮状态——这是多智能体能正常对话的关键。这里有个重要的经验Dify 的 HTTP 节点默认超时时间是 60 秒但多智能体系统因为有多个 LLM 调用串联很容易超时。我第一次部署时就是在这里卡了很久。解决办法是去 Dify 环境变量里调整HTTP_REQUEST_NODE_MAX_CONNECT_TIMEOUT和HTTP_REQUEST_NODE_MAX_READ_TIMEOUT或者把 LangGraph 服务放到内网缩短网络延迟。生产环境建议超时时间至少给到 120 秒。5. 从代码到上线启动、联动与常见问题5.1 LangGraph 服务启动与自测代码写完后我习惯先用命令行模式做一轮自测确认逻辑没问题再启动服务。uv run langgraph devlanggraph dev会启动一个带有交互式 Playground 的开发服务器你可以在浏览器里发送测试消息并查看每一步的状态流转。这个功能对排查“主管路由走错了”“循环终止条件没生效”这一类问题帮助极大。启动后控制台会输出一个 localhost 地址通常带上/playground后缀。测试的时候我会特意构造几类输入代码问题、知识库问题、外部工具问题、以及一个模棱两可的问题。前三个用来验证路由是否正确最后一个用来验证“主管不确定时是否会让用户补充信息”——这恰恰是最容易出问题的点。5.2 Dify 部署要点docker composeDify 的部署其实没什么玄学官方提供了 docker compose 一键启动脚本。我的建议是不要直接拉最新版而是去 Releasess 页面找一个稳定版本然后 clone 对应的docker目录来启动。git clone https://github.com/langgenius/dify.git cd dify/docker cp .env.example .env docker compose up -d这里最容易遇到的一个坑是新版 Dify 引入了多租户概念环境变量ENABLE_MULTI_TENANCY默认值在不同版本之间可能不同。如果你需要多租户要在.env里显式设置如果不需要保持默认单租户就好。启动完成后访问http://localhost/install进行初始化。然后创建应用 - 选择工作流类型 - 添加 HTTP 节点 - 填写 LangGraph 服务的地址。首次打通之后建议先在“预览”面板里测试一轮完整对话。5.3 常见问题排查表我把这个项目踩到的坑整理成了一张表按出现频率排序问题现象可能原因解决方案LangGraph 启动失败提示启动失败代码2依赖版本冲突或端口被占用检查 Python 版本是否 3.10lsof -i:8000查端口占用优先用uv run langgraph dev统一依赖环境Dify HTTP 节点请求超时多智能体链路太长超过默认 60 秒增大 Dify 环境变量中的 HTTP 超时配置对 LangGraph 做流式输出先返回“正在处理”主管路由总是选错 Agent提示词不够结构化或 LLM 温度过高在提示词里给出明确的“只输出一个词”约束把温度调到 0.2 以下必要时用 few-shot 示例循环反思退不出来条件边缺少终止条件或重试计数未更新检查retry_count是否在主管节点里累加在should_continue里先判断终止条件再判断路由子 Agent 返回内容丢失状态字段被覆盖而非追加messages用Annotated[List, add_messages]注解其他需要累加的字段也要做类似处理Dify 知识库和 LangGraph 检索结果不一致两边用了不同的检索策略或 embedding 模型尽量统一 embedding 模型如果必须双路检索加一个合并节点做去重除了这张表还有一个容易被忽略的问题Dify 升级后工作流节点配置可能发生不兼容。我的经验是生产环境不要随意在线升级 Dify至少先在测试环境跑通再切。热词里反复出现的“更新dify”“dify在线升级windows”本质上都是在问“升级过程中会不会影响现有应用”——答案是会。所以升级前一定要备份数据库和 docker volume。6. 把外部服务集成到多智能体系统里的正确姿势拆解完架构和核心路由还有一个高频问题要单独拿出来讲怎么把“看似无关的外部能力”接进这套系统。比如项目里经常提到的“把小龙虾或者爱马仕集成到多智能体系统中”换成工程语言就是如何把一个外部 API 或自定义服务挂载为多智能体的一个可用工具。LangGraph 里最标准的做法是用tool节点。先定义一个普通函数用tool装饰器包装然后在某个工人 Agent 节点里把它作为工具列表传给 LLMfrom langchain_core.tools import tool tool def query_crayfish_price(region: str) - str: 查询指定地区的小龙虾市场价格单位元/斤。 # 这里实际上是调用一个外部价格查询 API return requests.get(fhttps://api.example.com/crayfish?region{region}).json()[price] tool def query_hermes_stock(model: str) - str: 查询指定型号爱马仕商品的购买可行性。 # 调用库存系统 return 有货 if model in [Birkin 25, Kelly 28] else 无货定义好工具后在工人节点里通过bind_tools绑定from langchain_openai import ChatOpenAI llm_with_tools ChatOpenAI(modelgpt-4o-mini).bind_tools( [query_crayfish_price, query_hermes_stock] ) async def tool_agent_node(state: AgentState): response await llm_with_tools.ainvoke([ {role: system, content: 你是一个工具调用助手根据用户问题调用合适的工具。}, {role: user, content: state[task]} ]) # 这里的 response.tool_calls 会包含结构化的调用参数 for tool_call in response.tool_calls: result tool_map[tool_call[name]].invoke(tool_call[args]) state[result] f\n{tool_call[name]}: {result} return {result: state[result]}这时候主管路由如果判断走tool系统就会调用真正的价格/库存接口并把结果返回给用户。这个模式可以推广到任何第三方服务查快递、查天气、下单、甚至调用内部业务系统。社区里常说的“把某某集成到多智能体”本质上都是在做这一步。要注意的点有两个第一工具函数最好加上region、model这类参数说明LLM 才能准确抽取参数第二tool_calls返回的参数是字符串格式接入真实 API 之前务必做类型校验和异常捕获。我在测试时就遇到过 LLM 把“小龙虾”识别成region龙虾的离谱情况幸好有 try/except 兜底。7. 最后再分享几点实战体会说句实在话Dify LangGraph 这套组合真正难的不是 API 怎么调而是“人和人之间怎么配合”。这套系统跑通之后我有几个非常明显的感受第一把质量评估做成一个显式节点。不要只在主管的提示词里写“请判断结果是否满意”而是专门用一次 LLM 调用对工人结果打分把分数写进状态。这样出了问题你能看到分数低在哪而不是黑盒一句“不满意”。同理所有节点执行完都要能打印出关键的中间状态。第二用会话级线程隔离状态。LangGraph 的thread_id是天然的多会话隔离方案但如果你要在 Dify 里做多用户隔离一定要把session_id映射到每个用户会话的组合 ID 上。否则两个用户共用同一个 context会出现互相串话的诡异问题。第三控制成本从限制循环次数开始。多智能体系统最大的成本隐患不是模型贵而是循环不终止。一次看似简单的提问可能因为主管不满意导致同一个任务被反复执行三四次。建议在系统上线前专门对循环场景做压测把retry_count上限设到合理值三个 Agent 以内上限 3 次是多数场景可接受的平衡点。最后再分享一个小技巧开发调试时不要直接调大模型先给主管和工人节点都接一个“假 LLM”用固定返回值测试图的逻辑是否正确。等图完全稳定了再换真模型。这样一来路由、循环、并行这些框架层面的 bug 和模型回答质量的问题就彻底分离了排查效率能提升一倍以上。这套东西后续还可以往“记忆持久化”“多租户隔离”“流式输出”方向继续扩展但先把上面的基础架构吃透后面都是水到渠成的事。本文还有配套的精品资源点击获取 Dify + LangGraph 构建多智能体系统：架构设计与代码落地
简介面向企业级AI应用开发者这套代码演示了Dify与LangGraph在构建多智能体系统时的融合方式。Dify提供可视化开发界面LangGraph负责复杂工作流编排两者结合能有效降低智能体构建门槛同时提升系统的开发效率、编排灵活性与可靠性。压缩包内共有4个文件以Python脚本、HTML页面、Inscode配置和Git忽略文件为主包含可运行的对话分析系统示例完整代码仅11KB。目前已有208人学习。通过阅读项目代码可重点掌握Dify平台与LangGraph框架的接口对接方法理解持久化执行、人机交互、完整内存系统等核心机制在多智能体协同中的实际运用适合希望借助低代码平台快速落地复杂AI应用的开发者参考。Dify LangGraph 构建多智能体系统从架构设计到代码落地的完整记录先把结论说在前面Dify 负责把多智能体的“门面”和流程串起来LangGraph 负责把真正复杂的路由、循环、状态转移落到代码里。这套组合非常适合那些“既要可视化编排又要保持代码级可控”的落地场景。如果你正在纠结“Dify 到底能不能做 Agent 编排”“LangGraph 是不是太重了”这篇分享应该能给你一个清晰的答案。这篇文章不是纯理论科普我会拿一套实际跑通的多智能体系统来拆解主管 Agent 分流、条件路由、循环反思、子图复用、并行分支以及怎么和 Dify 工作流对接。建议阅读顺序是从头到尾代码段可以直接复制改但更重要的是理解“为什么这么设计”。1. 为什么要把 Dify 和 LangGraph 放在一起两者的边界与定位1.1 Dify 擅长什么LangGraph 又擅长什么Dify 是一个开源的 LLM 应用开发平台内置了知识库、工作流、Agent、模型管理和可视化的日志面板。它的强项是“把线性的业务逻辑快速搭起来”你可以在画布上拖一个知识检索节点接一个 LLM 节点再挂一个条件分支一个 RAG 问答应用就出来了。Dify 的 Agent 功能也支持工具调用但它的 Agent 编排本质上是“预设节点 工具调用”一旦遇到复杂的循环、嵌套子流程、需要精确控制状态迁移的场景画布上的连线会变得很难维护。LangGraph 是 LangChain 团队推出的图状态编排框架。它的核心思想是把一次对话或一次任务执行定义成一个状态机节点是处理逻辑边是状态迁移规则。正因为这种“图”结构LangGraph 可以天然表达循环、条件分支、并行执行、子图嵌套等能力。说白了Dify 适合做“流程可视化”LangGraph 适合做“逻辑精确化”。很多人会问Dify 不是也有工作流吗为什么还要绕一圈去写 LangGraph我实际用下来的体会是当你的业务只需要“单轮检索 生成回答”时Dify 工作流完全够用。但当你要做一个真正意义上的多智能体系统——比如一个主管 Agent 要把任务拆给多个专业 Agent还要根据执行结果决定是继续追问还是结束——用 Dify 画布来表达这种“动态循环”是很痛苦的。而 LangGraph 用几行conditional_edge就能写清楚。1.2 这套组合的真实适用场景结合我最近做的一个项目这套组合最适合以下三类场景第一客服工单分诊系统。用户进来先由主管 Agent 理解意图然后分发给售后、技术支持、财务等子 Agent每个子 Agent 可能还要调用各自的工具或知识库最后主管汇总答案。第二代码生成与审查系统。一个 Agent 负责写代码另一个负责 Review写完了不满意就循环回去改最多循环 N 次。这种“反馈闭环”用 Dify 工作流的普通分支很难写但在 LangGraph 里就是一个带终止条件的循环边。第三需要人工审核的自动化流程。比如内容审核先由多 Agent 对内容进行初筛再交给 Dify 工作流里的人工审核节点做保底。Dify 有现成的人工审核组件LangGraph 内部处理复杂判断两者互补。2. 多智能体系统整体设计先定角色再定协议2.1 多智能体的三种主流分工模式在我见过的多智能体架构里分工模式基本逃不出三类管道模式、主管-工人模式、网络模式。管道模式适合流水线场景每个 Agent 处理完一个阶段就传给下一个比如“意图识别 - 检索 - 总结”。网络模式最灵活Agent 之间互相通信但调试成本极高一般创业团队不建议碰。这套系统我选的是主管-工人模式Supervisor-Worker这也是多数生产级多智能体系统的首选。主管 Agent 不直接干活它负责理解用户需求、拆解任务、决定把任务派给哪个工人 Agent、以及判断结果是否满足要求。工人 Agent 各司其职比如“代码生成 Agent”“知识库问答 Agent”“外部工具调用 Agent”。这种模式的好处是任务流转路径清晰每个 Agent 的职责边界明确出问题容易排查。坏处是主管 Agent 容易成为瓶颈——如果它判断错误整个任务就偏了。所以我会在主管 Agent 的提示词里强调“不确定时优先询问用户不要自作主张”。2.2 我们的设计主管技能 Agent知识库具体到这次实现我设计了这样一套结构一个主管 AgentSupervisor三个工人 Agent——代码生成 Agent、信息检索 Agent、外部工具 Agent。另外挂了一个 Dify 知识库用于领域问答。任务流转规则是用户请求进来后主管先做意图识别如果涉及代码生成就走代码 Agent涉及具体领域问答就走知识库检索涉及外部数据比如查询天气、查汇率就走外部工具 Agent。工人 Agent 执行完毕后把结果回报给主管主管判断是否需要补充或调整不需要就返回最终答案。这套设计里有一个关键点每个工人 Agent 都是相对独立的节点可以在 LangGraph 里单独测试。我特别建议你先把每个 Agent 单独跑通再组装成图否则出了问题你根本不知道是哪个环节的问题。3. LangGraph 核心代码状态、路由、循环与子图3.1 环境准备与状态定义先安装依赖。我用的 Python 版本是 3.11LangGraph 当前版本支持得很稳定。pip install langgraph langchain langchain-openaiLangGraph 的核心是状态定义。状态就是一个 TypedDict它会在图的每个节点之间传递。我的状态定义如下from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[List, add_messages] # 对话消息流 current_agent: str # 当前应该交给哪个 Agent task: str # 拆分后的子任务 result: str # 工人 Agent 的执行结果 retry_count: int # 循环重试次数这里有几个设计细节值得一提。messages字段使用了add_messages注解这是 LangGraph 的一个巧妙的机制。普通的字典字段在节点间传递时会直接覆盖但加了add_messages之后新节点返回的消息会自动追加到历史列表里。这非常利于多 Agent 场景下的对话历史累积——主管能看到所有工人 Agent 的中间输出。retry_count是给循环用的。后面我会用条件边判断如果工人 Agent 执行结果不合格且重试次数少于 3就重新进入工人节点否则强制结束。这就是多智能体系统里“有限次反思循环”的标准写法。3.2 条件路由与分支控制conditional_edge先定义几个节点函数。每个节点的输入是整个 AgentState输出是部分状态的更新。来看主管节点和工人节点# 主管 Agent 节点 async def supervisor_node(state: AgentState): prompt f 你是多智能体系统的主管请根据用户请求选择处理路径 - code: 用户需要写代码或调试代码 - search: 用户需要知识库问答或信息检索 - tool: 用户需要调用外部工具天气、汇率等 用户请求{state[messages][-1].content} 只输出一个词code / search / tool response await llm.ainvoke(prompt) next_agent response.content.strip().lower() return {current_agent: next_agent} # 工人 Agent代码生成 async def code_agent_node(state: AgentState): prompt f你是资深工程师请完成以下任务并输出可运行代码{state[task]} response await llm.ainvoke(prompt) return {result: response.content} # 工人 Agent信息检索简化版 async def search_agent_node(state: AgentState): prompt f请基于知识库回答{state[task]} response await llm.ainvoke(prompt) return {result: response.content}接下来是条件路由的核心conditional_edge。它接收一个路由函数该函数根据当前状态返回下一个节点的名称。def route_after_supervisor(state: AgentState): agent_map { code: code_agent, search: search_agent, tool: tool_agent, } if state[current_agent] in agent_map: return agent_map[state[current_agent]] return END # 无法识别时直接结束避免死循环 # 构建图 graph StateGraph(AgentState) graph.add_node(supervisor, supervisor_node) graph.add_node(code_agent, code_agent_node) graph.add_node(search_agent, search_agent_node) graph.add_node(tool_agent, tool_agent_node) graph.add_edge(START, supervisor) graph.add_conditional_edges(supervisor, route_after_supervisor) graph.add_edge(code_agent, supervisor) # 工人执行完回到主管 graph.add_edge(search_agent, supervisor) graph.add_edge(tool_agent, supervisor) app graph.compile()这段代码的逻辑是用户请求进入主管主管输出一个 agent 类型条件边根据这个类型决定走到哪个工人节点工人执行完把结果写回result字段然后回到主管。这个图已经具备最基本的“主管分发 - 工人干活 - 主管再判断”的闭环。如果你看到这里觉得有点抽象可以这么理解conditional_edge就像是一个 switch-case 语句但它的判断条件是整个图的状态而不是某个局部变量。这在实际工程里太重要了——因为判断结果往往依赖前一个节点的输出而不是提前写死的。3.3 循环与终止机制有限次反思上面那个图有个问题工人执行完回到主管主管可能无论如何都不满意于是又派给工人来回无限循环。这个问题在真实系统里非常常见。解决方法是引入计数器用add_condition判断是否达到最大重试次数。我在状态里已经定义了retry_count。修改主管节点的逻辑async def supervisor_node(state: AgentState): last_result state.get(result, ) # 如果已有执行结果判断是否满足要求 if last_result: # 这里可以接入一个“质量评估”LLM 调用简化起见只做长度判断 if len(last_result) 50 or state.get(retry_count, 0) 2: return {messages: [{role: assistant, content: last_result}]} # 否则继续拆分任务 prompt f... 根据用户请求判断应该走哪个 Agent ... response await llm.ainvoke(prompt) return { current_agent: response.content.strip().lower(), retry_count: state.get(retry_count, 0) 1, }然后在图里加一个条件边当主管决定结束或者重试次数超过阈值时直接走到 ENDdef should_continue(state: AgentState): # 如果已经产生了最终答案结束 if final_answer in state: return END # 如果重试次数过多强制结束 if state.get(retry_count, 0) 3: return END return supervisor graph.add_conditional_edges(supervisor, should_continue, { supervisor: supervisor, # 继续循环 END: END, # 结束 })这里我踩过一个坑如果不在should_continue里加retry_count判断线上环境会经常出现“同一句话反复调用接口到超时”的问题。这种循环不是 bug但它会导致大量费用消耗。所以我的建议是给每一个循环边都加一个显式的终止条件并且把终止条件放在条件边函数最前面判断。3.4 子图与并行分支等基础闭环跑通之后你会需要更高阶的能力子图和并行。子图的价值在于“模块化复用”——同一套逻辑可以在不同的父图节点里被多次引用。并行分支的价值在于“提速”——比如在处理一个复杂问题时让代码 Agent 和检索 Agent 同时开工最后汇总。子图的写法很简单。先构建一个子图然后在父图里用add_node注册# 构建子图 subgraph StateGraph(AgentState) subgraph.add_node(search_agent, search_agent_node) subgraph.add_edge(search_agent, END) # 在父图里引用 graph.add_node(search_subgraph, subgraph.compile())并行分支需要借助Send对象。在 LangGraph 里Send允许你从同一个节点出发携带不同的状态副本去执行多个下游节点from langgraph.types import Send def spawn_parallel_tasks(state: AgentState): tasks state[task].split(|) # 按竖线切分成多个子任务 return [ Send(code_agent, {task: t.strip(), retry_count: 0}) for t in tasks ] graph.add_conditional_edges(supervisor, spawn_parallel_tasks)这样主管拆出 3 个子任务就会并发启动 3 个 code_agent 执行最终结果通过messages合并回来。这个模式在处理“批量生成 汇总”类需求时特别有用。4. Dify 侧编排把 LangGraph 接入工作流4.1 思路Dify 管入口LangGraph 管逻辑很多人的误区是把 LangGraph 当成一个“新的 Agent 框架”来替代 Dify但其实两者不是替代关系。LangGraph 是一个可嵌入的编排引擎不具备任何前端交互、模型管理、用户权限能力。Dify 恰好补齐了这些。我这个项目里的整体架构是用户在 Dify 前端对话Dify 的工作流节点负责接收用户输入然后通过一个“HTTP 请求”节点把请求转发给 LangGraph 服务LangGraph 完成多智能体调度后把结果返回给 DifyDify 再做后续处理比如知识库混合检索、人工审核、日志存储。4.2 HTTP 工具节点把 LangGraph 发布为接口首先要把 LangGraph 应用发布成一个可调用的 API。LangGraph 官方推荐的方式是使用langgraph serve或者langgraph dev它会自动生成一个带 Swagger 文档的服务。我的做法是写一个独立的 FastAPI 服务来包装from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): message: str session_id: str default class ChatResponse(BaseModel): result: str session_id: str app.post(/agent/chat, response_modelChatResponse) async def chat(req: ChatRequest): config {configurable: {thread_id: req.session_id}} final_state await graph.ainvoke( {messages: [{role: user, content: req.message}]}, configconfig ) return ChatResponse( resultfinal_state[messages][-1].content, session_idreq.session_id )把这个服务启动在 8000 端口后就可以在 Dify 工作流里添加一个 HTTP 请求节点配置如下请求方式POSTURLhttp://localhost:8000/agent/chatHeaderContent-Type: application/jsonBody{message: {{#sys.query#}}, session_id: {{#conversation.id#}}}Dify 内部有大量的变量引用语法sys.query是当前用户输入conversation.id是会话 ID。通过session_idLangGraph 的thread_id可以记住同一个会话的多轮状态——这是多智能体能正常对话的关键。这里有个重要的经验Dify 的 HTTP 节点默认超时时间是 60 秒但多智能体系统因为有多个 LLM 调用串联很容易超时。我第一次部署时就是在这里卡了很久。解决办法是去 Dify 环境变量里调整HTTP_REQUEST_NODE_MAX_CONNECT_TIMEOUT和HTTP_REQUEST_NODE_MAX_READ_TIMEOUT或者把 LangGraph 服务放到内网缩短网络延迟。生产环境建议超时时间至少给到 120 秒。5. 从代码到上线启动、联动与常见问题5.1 LangGraph 服务启动与自测代码写完后我习惯先用命令行模式做一轮自测确认逻辑没问题再启动服务。uv run langgraph devlanggraph dev会启动一个带有交互式 Playground 的开发服务器你可以在浏览器里发送测试消息并查看每一步的状态流转。这个功能对排查“主管路由走错了”“循环终止条件没生效”这一类问题帮助极大。启动后控制台会输出一个 localhost 地址通常带上/playground后缀。测试的时候我会特意构造几类输入代码问题、知识库问题、外部工具问题、以及一个模棱两可的问题。前三个用来验证路由是否正确最后一个用来验证“主管不确定时是否会让用户补充信息”——这恰恰是最容易出问题的点。5.2 Dify 部署要点docker composeDify 的部署其实没什么玄学官方提供了 docker compose 一键启动脚本。我的建议是不要直接拉最新版而是去 Releasess 页面找一个稳定版本然后 clone 对应的docker目录来启动。git clone https://github.com/langgenius/dify.git cd dify/docker cp .env.example .env docker compose up -d这里最容易遇到的一个坑是新版 Dify 引入了多租户概念环境变量ENABLE_MULTI_TENANCY默认值在不同版本之间可能不同。如果你需要多租户要在.env里显式设置如果不需要保持默认单租户就好。启动完成后访问http://localhost/install进行初始化。然后创建应用 - 选择工作流类型 - 添加 HTTP 节点 - 填写 LangGraph 服务的地址。首次打通之后建议先在“预览”面板里测试一轮完整对话。5.3 常见问题排查表我把这个项目踩到的坑整理成了一张表按出现频率排序问题现象可能原因解决方案LangGraph 启动失败提示启动失败代码2依赖版本冲突或端口被占用检查 Python 版本是否 3.10lsof -i:8000查端口占用优先用uv run langgraph dev统一依赖环境Dify HTTP 节点请求超时多智能体链路太长超过默认 60 秒增大 Dify 环境变量中的 HTTP 超时配置对 LangGraph 做流式输出先返回“正在处理”主管路由总是选错 Agent提示词不够结构化或 LLM 温度过高在提示词里给出明确的“只输出一个词”约束把温度调到 0.2 以下必要时用 few-shot 示例循环反思退不出来条件边缺少终止条件或重试计数未更新检查retry_count是否在主管节点里累加在should_continue里先判断终止条件再判断路由子 Agent 返回内容丢失状态字段被覆盖而非追加messages用Annotated[List, add_messages]注解其他需要累加的字段也要做类似处理Dify 知识库和 LangGraph 检索结果不一致两边用了不同的检索策略或 embedding 模型尽量统一 embedding 模型如果必须双路检索加一个合并节点做去重除了这张表还有一个容易被忽略的问题Dify 升级后工作流节点配置可能发生不兼容。我的经验是生产环境不要随意在线升级 Dify至少先在测试环境跑通再切。热词里反复出现的“更新dify”“dify在线升级windows”本质上都是在问“升级过程中会不会影响现有应用”——答案是会。所以升级前一定要备份数据库和 docker volume。6. 把外部服务集成到多智能体系统里的正确姿势拆解完架构和核心路由还有一个高频问题要单独拿出来讲怎么把“看似无关的外部能力”接进这套系统。比如项目里经常提到的“把小龙虾或者爱马仕集成到多智能体系统中”换成工程语言就是如何把一个外部 API 或自定义服务挂载为多智能体的一个可用工具。LangGraph 里最标准的做法是用tool节点。先定义一个普通函数用tool装饰器包装然后在某个工人 Agent 节点里把它作为工具列表传给 LLMfrom langchain_core.tools import tool tool def query_crayfish_price(region: str) - str: 查询指定地区的小龙虾市场价格单位元/斤。 # 这里实际上是调用一个外部价格查询 API return requests.get(fhttps://api.example.com/crayfish?region{region}).json()[price] tool def query_hermes_stock(model: str) - str: 查询指定型号爱马仕商品的购买可行性。 # 调用库存系统 return 有货 if model in [Birkin 25, Kelly 28] else 无货定义好工具后在工人节点里通过bind_tools绑定from langchain_openai import ChatOpenAI llm_with_tools ChatOpenAI(modelgpt-4o-mini).bind_tools( [query_crayfish_price, query_hermes_stock] ) async def tool_agent_node(state: AgentState): response await llm_with_tools.ainvoke([ {role: system, content: 你是一个工具调用助手根据用户问题调用合适的工具。}, {role: user, content: state[task]} ]) # 这里的 response.tool_calls 会包含结构化的调用参数 for tool_call in response.tool_calls: result tool_map[tool_call[name]].invoke(tool_call[args]) state[result] f\n{tool_call[name]}: {result} return {result: state[result]}这时候主管路由如果判断走tool系统就会调用真正的价格/库存接口并把结果返回给用户。这个模式可以推广到任何第三方服务查快递、查天气、下单、甚至调用内部业务系统。社区里常说的“把某某集成到多智能体”本质上都是在做这一步。要注意的点有两个第一工具函数最好加上region、model这类参数说明LLM 才能准确抽取参数第二tool_calls返回的参数是字符串格式接入真实 API 之前务必做类型校验和异常捕获。我在测试时就遇到过 LLM 把“小龙虾”识别成region龙虾的离谱情况幸好有 try/except 兜底。7. 最后再分享几点实战体会说句实在话Dify LangGraph 这套组合真正难的不是 API 怎么调而是“人和人之间怎么配合”。这套系统跑通之后我有几个非常明显的感受第一把质量评估做成一个显式节点。不要只在主管的提示词里写“请判断结果是否满意”而是专门用一次 LLM 调用对工人结果打分把分数写进状态。这样出了问题你能看到分数低在哪而不是黑盒一句“不满意”。同理所有节点执行完都要能打印出关键的中间状态。第二用会话级线程隔离状态。LangGraph 的thread_id是天然的多会话隔离方案但如果你要在 Dify 里做多用户隔离一定要把session_id映射到每个用户会话的组合 ID 上。否则两个用户共用同一个 context会出现互相串话的诡异问题。第三控制成本从限制循环次数开始。多智能体系统最大的成本隐患不是模型贵而是循环不终止。一次看似简单的提问可能因为主管不满意导致同一个任务被反复执行三四次。建议在系统上线前专门对循环场景做压测把retry_count上限设到合理值三个 Agent 以内上限 3 次是多数场景可接受的平衡点。最后再分享一个小技巧开发调试时不要直接调大模型先给主管和工人节点都接一个“假 LLM”用固定返回值测试图的逻辑是否正确。等图完全稳定了再换真模型。这样一来路由、循环、并行这些框架层面的 bug 和模型回答质量的问题就彻底分离了排查效率能提升一倍以上。这套东西后续还可以往“记忆持久化”“多租户隔离”“流式输出”方向继续扩展但先把上面的基础架构吃透后面都是水到渠成的事。本文还有配套的精品资源点击获取