简介面对企业级多智能体系统开发中常见的编排复杂、调试困难与状态管理不便等问题这份代码资源给出了Dify与LangGraph结合的落地示例适合具有一定AI应用开发基础、希望提升工作流控制能力的开发者研读。压缩包共4个文件含Python源码、HTML页面、inscode配置文件及gitignore文件整体约11KB体量小巧但结构完整便于快速定位关键逻辑。已有208人学习浏览属于典型的轻量型参考样例。通过其中的对话分析系统等实现可以清晰看到Dify可视化界面如何对接LangGraph的持久化执行、人机交互与内存管理能力理解多智能体协同任务和动态编排的细节从而在实际项目中减少试错成本提升系统可靠性与开发效率。1. 为什么非要把Dify和LangGraph拼在一起这两年多智能体系统被聊得满天飞但真正动手搭过的人心里都清楚多智能体最难的不是“调一个大模型接口”而是把多个具备不同职责的智能体串成一条可靠的流水线让它们在正确的时机做正确的事。我最早尝试的方案是纯LangGraph硬写代码量倒还能接受但遇到知识库问答、文档解析、工具调用这些偏工程化的能力时开发效率立刻被拖垮。后来换成纯Dify编排可视化工作流确实省事可一旦涉及复杂的动态路由、循环控制、并行分支又会被它的节点模型束缚住手脚。直到我把Dify和LangGraph组合起来用才算找到了一个比较舒服的平衡点。这套组合的核心思路并不复杂Dify负责重活累活——知识库检索、文档预处理、各种工具的封装、日常的工作流编排全都放在Dify里通过API暴露出来LangGraph负责大脑级别的调度——用一个图结构来定义多个智能体之间的关系控制消息怎么流转、什么时候走条件分支、什么时候需要人工介入。Dify的每个应用不再是孤岛而是LangGraph图中的一个节点或子图。这样既拿到了Dify的工程化能力又保住了LangGraph在编排层的灵活性。适合来读这篇文章的人我觉得有两种一种是用Dify做应用做到一定程度发现单应用已经撑不起复杂的业务逻辑想往多智能体方向升级另一种是纯LangGraph玩家被知识库、文档解析这类工程问题折磨过想知道怎么借力Dify把底层能力一次性做扎实。下面我按实际搭建顺序来写代码部分基于LangGraph 0.2.x版本Dify用的是社区版1.10整体思路在更新版本上同样适用。2. 整体架构设计谁负责调度谁负责干活2.1 一个最小可用的多智能体拓扑我先说结论不要一上来就画一个六个智能体互相调用的复杂拓扑大概率会把自己绕晕。我实际跑通的第一个版本只用了三个角色但已经能覆盖大多数业务场景。入口调度智能体Router Agent负责理解用户意图决定把请求交给哪个下游智能体。它本身不执行具体任务只做分类和路由。知识问答智能体QA Agent对接Dify的知识库应用处理“查资料、总结文档、基于私有知识回答”这类请求。工具执行智能体Tool Agent对接Dify的工作流应用处理“查天气、查订单、调外部API、执行多步骤操作”这类请求。这个拓扑的关键在于Router Agent不直接调用大模型做分类而是通过LangGraph的条件路由机制来判断。判断依据可以是结构化输出里的意图字段也可以是大模型分类结果。我比较推荐让Router Agent用大模型做意图识别输出一个固定格式的JSON然后LangGraph的conditional_edge根据JSON里的intent字段决定走向。这样即使后续新增智能体只要在这个JSON里加一个枚举值就行。2.2 消息流转与状态管理先想清楚数据怎么走LangGraph的消息流转依赖一个共享的State对象。这个State看起来简单但设计不好后面会非常难受。我第一版把所有字段都塞在一个字典里结果下游智能体经常读到上游的脏数据排查起来极其痛苦。后来我采用了一个比较规范的做法把State分成三块from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage class AgentState(TypedDict): # 1. 原始对话记录所有智能体共享只追加不修改 messages: Annotated[List[BaseMessage], add_messages] # 2. 路由信息从入口节点写入后续节点只读 intent: str confidence: float # 3. 上下文数据下游智能体运行过程中生成的中间结果 execution_context: dict用Annotated和add_messages来管理messages字段可以保证对话记录在图中的每个节点之间正确累积不会互相覆盖。intent和confidence是Router Agent写入的一次性信息execution_context则用来传递知识库召回结果、工具返回的原始数据等。这里有个经验如果你发现某个字段在多个节点里频繁被读写那说明它应该被提到State顶层而不是嵌在execution_context深处。3. 落地搭建Dify端应用编排与API暴露3.1 Dify工作流编排的关键节点先别急着写LangGraph代码Dify端的基础工作要先做好。我在这里卡过不少时间所以把关键点多说几句。Dify里创建应用时不要选“聊天助手”要选“工作流”。工作流模式才能把知识库检索、问题分类、多步工具调用清晰地编排出来。我们假设要做一个知识问答智能体那么Dify工作流至少需要这几个节点开始节点接收LangGraph传来的用户问题。这里一定要把输入变量名定好比如query后面调用API时字段名就靠它对齐。知识检索节点关联训练好的知识库设置召回策略。检索方式建议先用“向量召回”跑通之后再叠加“全文召回”的结果做RAG融合效果会稳很多。LLM节点基于检索到的上下文和用户问题生成回答。这里有一个很多人忽略的细节——给LLM节点写提示词时必须明确告诉模型“如果检索结果与问题无关直接说明未找到相关信息”。否则模型会强行根据检索片段编造答案这在多智能体链路里会被放大成灾难。结束节点输出最终回答同时可以额外输出一个references数组方便LangGraph侧记录知识来源。Dify工作流配好之后一定要先在工作流编辑页右上角点“运行”做一次测试。我看到太多人跳过这一步直接去调API结果报错了才开始怀疑是Dify配置还是LangGraph代码的问题排查成本翻倍。3.2 知识库接入与模型设置知识库接入这一步Dify自己做得足够友好但也需要注意三个实操细节。第一个是分段长度。默认的自动分段策略对长文档经常切出一些语义残废的片段导致检索召回质量很差。我建议对技术文档手动设置分段长度500-800字重叠区间50-100字。别小看这个重叠它能避免一个关键句子被拦腰切断。第二个是Embedding模型的选型。Dify社区版自带了几种Embedding方案如果只是中文场景用内置的text-embedding-ada-002或者兼容的本地Embedding模型都可以但注意索引一旦建立换了Embedding模型就得重建知识库索引。所以第一版先把模型定下来不要中途换否则所有知识库都要重新embedding非常耗时。第三个是知识库权限。同一个Dify实例可能同时服务多个应用在知识库设置里一定要限定访问该知识库的应用范围。我之前不小心把内部文档知识库设成了“所有应用可见”QA智能体在回答时甚至把不该泄露的内容也当上下文带了出来。做工具执行智能体时Dify里的HTTP请求节点是连接外部系统的关键。它支持自定义请求头、请求体也支持在节点内做简单的数据转换。我个人的建议是能放在Dify里做的数据清洗尽量放在Dify里做因为Dify的调试界面可以直观看到每个节点的输入输出定位问题比在LangGraph里反复打印日志快得多。4. LangGraph侧的编排代码把多智能体串起来4.1 依赖与项目结构Dify端应用就绪后开始写LangGraph侧的编排代码。先确认Python环境依赖我用的版本组合在本地实测没有问题pip install langgraph langchain langchain-openai dify-client requests # langgraph 0.2.x # langchain 0.2.x项目结构我习惯这样组织agent_system/ ├── main.py # 入口脚本负责编译图并启动交互 ├── router_agent.py # 路由智能体意图识别 ├── qa_agent.py # 知识问答智能体封装Dify知识库API调用 ├── tool_agent.py # 工具执行智能体封装Dify工作流API调用 ├── state.py # AgentState定义 └── config.py # Dify API地址、密钥、模型配置模块拆开的原因很简单多智能体系统不止是“图编排”每个节点的业务代码也需要独立维护。后续如果想把某个Agent替换成微服务调用只需要改动对应模块不需要动图结构。4.2 核心代码状态定义、Agent节点和条件路由先看state.py就是上一节提到的AgentState这里不再重复。再看router_agent.py它的任务是把用户问题转换成意图结构。我用LangChain的ChatPromptTemplate 结构化输出来实现这样能拿到干净的JSON字段from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import JsonOutputParser router_prompt ChatPromptTemplate.from_messages([ (system, 你是一个意图分类引擎只做一件事判断用户问题是需要知识库问答还是工具调用。 返回JSON格式包含两个字段 - intent: 必须是 qa 或 tool - reason: 一句话说明判断理由 不要输出任何多余内容。), (human, {input}) ]) class RouterAgent: def __init__(self, model_config: dict): self.llm ChatOpenAI(**model_config) self.parser JsonOutputParser() self.chain router_prompt | self.llm | self.parser def route(self, user_input: str) - dict: result self.chain.invoke({input: user_input}) return result接下来是qa_agent.py负责调用Dify的知识库应用API。Dify的API文档大家应该都见过核心就是chat-messages接口。这里要注意调用Dify API时传的参数是inputs里的字段必须是Dify工作流“开始节点”里定义过的字段名。我用的是query所以请求体里传query。import requests class QAAgent: def __init__(self, api_base: str, api_key: str, app_id: str): self.api_base api_base self.api_key api_key self.app_id app_id def run(self, user_input: str, conversation_id: str ) - str: url f{self.api_base}/v1/chat-messages headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } payload { inputs: {query: user_input}, query: user_input, response_mode: blocking, conversation_id: conversation_id, app_id: self.app_id } resp requests.post(url, headersheaders, jsonpayload, timeout120) resp.raise_for_status() data resp.json() # Dify返回结构中answer字段是最终回答 return data.get(answer, )这段代码看起来简单但有几个点值得展开说。response_mode用blocking还是streaming对多智能体链路影响很大。我建议第一版用blocking方便调试确认链路没问题后再改成streaming。conversation_id这个参数在LangGraph里的处理要慎重。多智能体场景下如果每个节点都维护自己的会话ID会导致上下文割裂。我的做法是Dify端只把它当作一次无状态请求来调用不开启多轮会话模式所有历史对话都由LangGraph的messages字段管理。这样虽然每次都会把历史消息重新发给Dify但换来的是状态逻辑的高度统一。timeout要设得足够大。知识库召回加上大模型生成慢的时候超过30秒很正常。之前我设成10秒结果系统频繁误报超时排查了很久才发现是Dify处理本身就需要时间。tool_agent.py和qa_agent.py几乎一样唯一区别是调用Dify时传入的inputs不同比如工具执行可能需要传order_id、city等附加参数。这部分代码我就不重复贴了。最后是main.py里的图编排。这一步是整个系统的中枢我把关键代码完整贴出来然后逐行解释。from langgraph.graph import StateGraph, END from state import AgentState from router_agent import RouterAgent from qa_agent import QAAgent from tool_agent import ToolAgent def create_graph(router: RouterAgent, qa: QAAgent, tool: ToolAgent): graph StateGraph(AgentState) def router_node(state: AgentState) - dict: result router.route(state[messages][-1].content) # 把意图解析结果写入state return {intent: result[intent], execution_context: {route_reason: result.get(reason, )}} def qa_node(state: AgentState) - dict: user_input state[messages][-1].content answer qa.run(user_input) # 把qa的回答追加到messages中便于后续节点或用户看到 return {messages: [{role: assistant, content: answer}]} def tool_node(state: AgentState) - dict: user_input state[messages][-1].content answer tool.run(user_input) return {messages: [{role: assistant, content: answer}]} def route_after_router(state: AgentState) - str: # 根据router_node写入的intent字段决定走哪个分支 if state.get(intent) tool: return tool return qa graph.add_node(router, router_node) graph.add_node(qa, qa_node) graph.add_node(tool, tool_node) graph.set_entry_point(router) graph.add_conditional_edges( router, route_after_router, { qa: qa, tool: tool, } ) graph.add_edge(qa, END) graph.add_edge(tool, END) return graph.compile()这段代码的核心逻辑就两条add_conditional_edges是LangGraph的灵魂它让图不再是一条直线走到黑而是可以根据节点返回的state动态选择下一个节点。route_after_router函数接收当前的state返回值必须与条件边映射字典里的key对应。我在这个函数里只判断了intent字段如果未来想增加“当置信度低于0.6时走人工审核分支”只需要在映射字典里加一个human_review并且在函数里增加判断即可。4.3 与Dify API对接的调用层看到这里细心的人可能会发现上面的LangGraph图并没有直接调用Dify而是通过QAAgent和ToolAgent这两个类来间接调用。这正是我刻意做的分层。如果未来某个Agent不再走Dify而是直接调模型或者调用内部REST服务只需要替换对应的Agent类内部实现图结构完全不用动。我再补充一个实际运行中的细节LangGraph的add_messages在合并消息时会把新消息追加到已有列表尾部。所以如果QAAgent里已经返回了answer字符串但在qa_node里没有把它包成完整消息对象LangGraph会无法正确合并。这就是为什么上面我把answer包成了{role: assistant, content: answer}格式。如果你用的是LangChain的HumanMessage、AIMessage对象同样可以只要是可序列化的BaseMessage结构就行。5. 本地跑通的完整流程与实测表现5.1 环境准备与启动步骤先把Dify跑起来。社区版1.10最常用的方式是用Docker Compose一键部署官方仓库里提供了完整的配置docker compose up -d就能拉起来。这里有一个非常容易踩的坑Dify默认占用80端口如果你的机器上有其他服务占用了80启动会直接失败。我建议在启动前先修改.env里的EXPOSE_NGINX_PORT改成8000之类的空闲端口。Dify起来之后按照官方指引创建账号、登录、创建应用。我建议把三个应用分别命名qa-agent-app、tool-agent-app。然后在“访问API”页面生成API密钥。这里的密钥要注意Dify社区版的API密钥与应用绑定不要把密钥写在代码仓库里用环境变量或config.py统一加载。LangGraph侧的启动就比较简单了。在项目目录下直接用Python运行main.py它会编译图并启动一个命令交互循环。我习惯先写一个简单的main.py入口def main(): router RouterAgent(model_config{model: gpt-4o-mini, temperature: 0}) qa QAAgent(api_baseDIFY_API_BASE, api_keyDIFY_QA_API_KEY, app_idDIFY_QA_APP_ID) tool ToolAgent(api_baseDIFY_API_BASE, api_keyDIFY_TOOL_API_KEY, app_idDIFY_TOOL_APP_ID) app create_graph(router, qa, tool) print(多智能体系统已启动输入q退出。) while True: user_input input(你: ) if user_input.strip().lower() q: break result app.invoke({ messages: [{role: user, content: user_input}], intent: , confidence: 0.0, execution_context: {} }) print(智能体:, result[messages][-1].content) if __name__ __main__: main()这里有一点值得注意app.invoke传入的初始state必须包含AgentState里定义的所有字段。如果漏了execution_contextLangGraph虽然不一定报错但节点函数里可能拿不到预期数据调试起来非常隐蔽。所以我通常会在创建initial state时把所有字段都显式写出来。5.2 实测链路与效果我本地跑通后做了几个典型测试给大家看一下实际效果。第一轮知识问答。输入“请总结一下我们知识库里关于客户投诉处理流程的主要内容”。Router Agent判断为qa意图LangGraph路由到QA AgentQA Agent调用Dify知识库应用。Dify工作流里知识检索节点召回了3段文档LLM节点生成了约200字的总结。整体耗时12秒其中大部分时间在Dify侧。返回内容引用了文档里的具体步骤标注了来源片段。第二轮工具调用。输入“帮我查一下订单10086的物流状态”。Router Agent判断为tool意图路由到Tool Agent。Tool Agent调用Dify工作流应用工作流里包含一个HTTP请求节点模拟请求外部物流API。最终返回“订单10086已签收签收时间是今天上午10:24”。整个过程耗时8秒。第三轮边界情况。输入“今天晚上吃什么”。Router Agent的判断出现了摇摆——它把这个问题也归到了qa意图QA Agent从知识库里检索不到相关内容最终回答“未找到与问题相关的信息”。这个结果虽然没有报错但用户体验不好。后来我在Router Agent的提示词里增加了一个intent枚举值general让这一类闲聊问题直接走一个默认的通用回复节点不经过知识库。效果比之前好了很多。6. 我踩过的坑和几条关键建议6.1 坑Dify工作流超时与LangGraph重试的冲突我第一次跑通之后以为链路很稳结果压测时发现当Dify工作流处理超过60秒时Dify API会返回超时错误。而LangGraph默认会对节点执行尝试重试重试时Dify端又可能再次执行同一工作流导致幂等性问题。解决办法有两个方向。一是给Dify侧的API调用设置较长的超时时间并在业务上接受慢响应二是让关键工作流在Dify端设计成异步模式用任务ID轮询结果。我的建议是第一版先用blocking模式跑通但务必在LangGraph的节点函数里做异常捕获不要把异常抛给图运行时。因为LangGraph的重试机制针对的是节点函数的异常如果你自己在节点内部捕获了Dify超时并返回一个友好提示就不会触发自动重试。6.2 坑知识库召回质量直接决定多智能体的上限多智能体链路像一条水管任何一环堵塞都会影响整体但最影响最终答案质量的往往还是知识库召回。我一开始图省事把一堆PDF直接扔进Dify知识库用默认分段策略结果回答经常漏掉关键信息。后来我做了两件事效果立竿见影把大文档按章节拆分每个章节作为一个独立文档上传而不是整本上传。在每个分段里显式标记来源例如“来源《客户投诉处理手册》第三章”。这样不仅便于追踪也方便Dify的召回结果在LLM生成时引用原文。实测下来知识库答案的准确率从刚接好的60%左右提升到了85%以上。对于一个多智能体系统来说这个提升比优化任何编排代码都来得快。6.3 几点架构建议关于Dify和LangGraph的分工我最后再啰嗦几句。不要把所有逻辑都塞进LangGraph图里。我见过有人把知识库检索也写在LangGraph节点里绕开Dify直接用向量数据库结果代码量翻倍还要自己处理文档解析、索引同步。恰恰Dify已经把这些事情做得非常成熟了该借力就借力。不要把Dify当成无状态的黑盒一直调。Dify工作流里如果涉及需要多轮对话提示词优化的业务建议还是在Dify端调整。因为工作流里每个节点的Prompt调整都会直接影响返回质量如果把这个逻辑也丢给LangGraph你最后会在一堆代码里找Prompt异常痛苦。State字段的命名要克制。多智能体系统一旦跑起来往State里加字段的诱惑非常大。每加一个字段图的复杂度不是线性增长而是指数级增长。我在实际项目中有一条原则每个字段必须回答“是谁在什么阶段写入谁在什么时候读取”这两个问题答不上来就不加。靠着这条原则我的图在后续扩展到5个智能体时依然能在一分钟内定位问题。7. 后续扩展从“多智能体串联”到“多智能体协作”最后分享一下我在这套框架上的扩展方向。很多人搭完上述系统后发现它本质上还是一条“路由—执行”的链式结构一个Agent处理完交给下一个然后结束。但真实业务中智能体之间经常需要协作QA Agent拿到知识库结果后需要Tool Agent去查一个结果Tool Agent查完后还要把新信息交回给QA Agent做二次总结。这种场景在LangGraph里并不复杂只需要增加一条边让QA节点在输出回答前先经过Tool节点形成一个小的循环。graph.add_edge(qa, tool) graph.add_edge(tool, qa)这两种实现正好对应LangGraph的add_edge和add_conditional_edges两种连接方式。前者是固定流程后者是动态判断。用这种方式可以把上面的单层路由扩展成“专家协作”模式。需要提醒的是循环图一定要设置最大循环次数或终止条件否则遇到死循环时整个程序会卡死。我通常在State里加一个iteration_count字段每经过一次协作循环加1当累计超过3次时强制跳到END节点。代码上的扩展还可以走“子图”的方式。把整个QA链路当做一个子图Tool链路当做另一个子图然后在主图里调用。子图的好处是可以复用比如将来同时存在“客服多智能体”和“运营多智能体”两套系统它们都可以共用同一个QA子图和Tool子图只是主图的调度逻辑不同。还有一个值得尝试的方向是引入“人在环路”human-in-the-loop机制。LangGraph本身支持在节点之间暂停并等待人工审批Dify也提供了对话流中的“人工接管”节点但二者结合时有一个细节LangGraph的图状态是内存态的进程重启后状态会丢失。如果想让系统在人工审批通过后继续执行需要把State持久化到Redis或数据库中对于生产级系统这是必须考虑的问题。我在这套体系上花了不少时间最终的体会是Dify和LangGraph不是竞争关系而是很好地互补。Dify帮你把底层的业务能力做得足够产品化LangGraph则让你在智能体协作逻辑上保留最大的灵活度。如果你想搭一套真正能落地的多智能体系统与其在单一框架里死磕不如试试这种混搭路线把两边各自最擅长的那一面都用起来。本文还有配套的精品资源点击获取