企业官网建设流程全解析

简介：基于Dify与LangGraph构建企业级多智能体系统的完整代码包，面向具备一定AI开发基础、希望以低代码可视化方式结合专业编排框架完成多智能体项目落地的开发者。代码包围绕Dify平台与LangGraph编排框架的技术融合展开，包含Python核心脚本、前端交互页面及配套配置，共4个文件，压缩包仅11KB，轻量且结构清晰，便于直接阅读和二次改造。目前已有243人学习下载。通过对照工程代码，可深入理解持久化执行、人机交互、完整内存系统等关键特性的工程实现，掌握在Dify中集成LangGraph工作流的具体写法；其中对话分析系统示例，清晰展示了智能体如何按预设工作流自主执行与相互协作，可作为同类业务场景的起步模板，帮助缩短开发周期，提升编排灵活性与系统可靠性。

1. 为什么 Dify + LangGraph 是多数团队落地多智能体系统的现实组合

把 Dify 和 LangGraph 放进同一个工程，很多人第一反应是“重复造轮子”：Dify 能可视化编排工作流，LangGraph 也能画图，到底谁听谁的？但做过多智能体落地的人会告诉你，多智能体系统真正难的不是画流程图，而是状态流转、路由决策、循环终止和工具调用这些代码层细节，可视化平台很难给你足够的控制力。反过来，LangGraph 把协作逻辑写得再清楚，它也补不上知识库、用户会话管理、应用发布和版本迭代这层产品化能力。

所以我见到的多数落地组合是：LangGraph 负责“多智能体怎么协作”，Dify 负责“用户怎么用上这套协作”，两者用 HTTP 服务连接。这样做的好处是，你既能用代码精确控制每个智能体的行为，又能用 Dify 快速做出可运营的前端应用。

这篇文章直接把一条能跑通的最小链路拆开：从 LangGraph 状态图怎么写，到 Dify 里怎么注册自定义工具，再到生产环境最容易翻车的几个位置。内容偏实操，适合已经了解基本概念、准备动手的开发者。

2. 架构分工：LangGraph 管编排、Dify 管产品层，先定边界再写代码

2.1 先分清楚：Dify 的智能体节点和 LangGraph 的智能体编排不是一回事

Dify 的 Agent 节点，本质上还是一个“单智能体 + 工具调用”的封装：它有一个主提示词，模型根据提示决定要不要调工具，调完把结果带回继续生成。你当然可以在工作流里放多个 Agent 节点，让它们一个接一个执行，但这种方式更接近“流水线”，不是真正意义上的多智能体协作。

LangGraph 里的编排单位是图：每个节点是一个函数，每条边是执行顺序，节点之间共享一个状态对象。你可以让 A 智能体先做意图判断，再动态决定下一步是走 B 还是走 C；可以在 B 和 C 之间做条件跳转，也可以让 A 拿到结果后回来收敛。这个能力是 Dify 工作流里最别扭的部分，因为工作流编辑器的分支逻辑是静态的，分支条件在画图时就定死了，而智能体协作里的路由往往要等模型输出之后才能决定。

所以我的切分原则很简单：凡是“下一步走哪个智能体”由模型推理结果决定的，放在 LangGraph；凡是“用户请求进来之后要做什么、要展示什么”这类固定流程，放在 Dify。这类边界先划清楚，后面几乎不会因为责任不明而翻车。

2.2 最小可行架构：一个路由节点、两个执行节点、一个收敛节点

我一般会用四个角色来组一套最小多智能体系统：路由智能体负责分诊，执行智能体负责干活，收敛节点负责把结果整理成最终回答，还有一个兜底的执行智能体负责处理不需要检索的请求。路由智能体的模型用小而快的，比如gpt-4o-mini，温度设成 0，只输出一个预设的路由标识；执行智能体才接搜索、代码执行这类工具；收敛节点用稍高温度，负责把检索结果或工具输出改写成用户能直接看的回复。

为什么强调“路由不干活”？因为一旦路由节点既要判断又要回答，它的输出格式就会不稳定。你让它返回一个 JSON，它可能在 JSON 里夹带“好的，让我帮您……”这类话，下游解析时就报错。把路由做成一个只输出枚举值的节点，是控制多智能体稳定性的第一道关卡。

执行节点允许有多个：检索型、代码型、数据库型。每个执行节点只做自己职责内的一件事，不要硬塞进一个“万能”智能体。这里的取舍是 token 成本：拆得越细越容易定位问题，但整体调用次数会变多，费用跟着涨。一般中等复杂度的系统，配置两个到三个执行节点就够了。

2.3 数据流设计：用户请求从 Dify 进来后，经过哪些环节再返回

把边界划分写在前面，但实际系统还要考虑数据从哪里进来、往哪里去。我通常让用户消息先进 Dify 应用，Dify 负责把用户输入、历史会话、知识库检索结果拼装好，然后调用 LangGraph 服务。LangGraph 内部完成多智能体协作后，把结果返回给 Dify，由 Dify 拼接引用来源和格式化文本。

这里有一个经常被忽略的关键点：LangGraph 服务不应该自己去 Dify 拉历史消息，也不应该重复做 Dify 已经做过的知识库检索。数据流应该单向流动，Dify 把需要的东西都塞进请求体，LangGraph 只处理塞进来的内容。这样才能保证两边职责不重叠：知识库检索、权限管理和前端展示归 Dify，状态流转和工具编排归 LangGraph。

我自己习惯在 Dify 工作流里增设一个“前置检索”节点，把知识库结果放到系统提示词的上下文里，再和用户消息一起传给 LangGraph。LangGraph 里的执行节点如果需要外部事实，就直接从上下文取，而不是再次触发知识库工具。这样既快又省 token，也能避免同一个问题被两套知识库检索逻辑各处理一次，产生不一致的回答。

3. 用 LangGraph 写最小多智能体图：状态定义、条件路由与可运行代码

3.1 状态定义：多个智能体之间靠一个字典通信

LangGraph 里，多个智能体之间不直接互相调用。每个节点读到一个状态对象，返回一个字典，框架自动把返回内容合并进状态，下一个节点再读到的就是更新后的完整状态。这个数据契约，通常用TypedDict定义。我一般会在状态里放四类字段：对话消息、路由决策、执行结果和最终回复。先定义状态，再写节点，顺序一定不能反。

from typing import TypedDict class AgentState(TypedDict): messages: list # 对话消息历史，所有节点可见 route: str # 路由决策，由路由节点写入 research_answer: str # 检索智能体的原始输出 final_answer: str # 经过收敛节点整理后的回答

用TypedDict而不是普通类的原因，是 LangGraph 的节点合并是基于字典的。每个节点函数接收一个state，返回一个普通字典，框架用update语义合并字段。messages是特殊字段，LangGraph 会按追加方式更新，其余字段则是覆盖式写入。这一点务必记住：如果某个字段在多个节点里都有写入，后执行的节点会覆盖先执行的节点，而不是拼接。

状态字段并不是越多越好。字段太多会让每个节点的输入参数变得不可控，调试时你根本分不清某个字段是谁写的。我的习惯是，只有需要跨节点传递的数据才放进状态，节点内部临时变量一律留在函数作用域里。

3.2 路由节点和执行节点：用函数定义智能体行为

下面是一套能直接跑的最小编排代码。路由节点负责判断要不要走检索，执行节点负责调用搜索工具，收敛节点负责把结果整理成答案，还有一个直接回答节点处理不需要检索的场景。这里为了演示，搜索工具用了通用搜索封装，你实际项目里换成自己公司的搜索接口或内部 API 即可。

from langchain_openai import ChatOpenAI from langchain_community.tools import DuckDuckGoSearchRun def route_node(state: AgentState) -> dict: """路由节点：判断当前问题是否需要实时检索，只输出 research 或 direct。""" llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) last_msg = state["messages"][-1].content prompt = ( f"判断以下问题是否需要实时外部检索。\n" f"问题：{last_msg}\n" f"如果问题涉及最新信息、特定事件、实时数据，回复 research；" f"否则回复 direct。" ) decision = llm.invoke(prompt).content.strip().lower() if decision not in ("research", "direct"): decision = "direct" return {"route": decision} def research_node(state: AgentState) -> dict: """执行节点：调用搜索工具，把原始材料写入状态。""" search_tool = DuckDuckGoSearchRun() query = state["messages"][-1].content raw = search_tool.invoke(query) return {"research_answer": raw} def answer_node(state: AgentState) -> dict: """收敛节点：把检索材料整理成对用户友好的回答。""" llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3) resp = llm.invoke( f"根据以下材料回答用户问题。\n材料：{state['research_answer']}\n" f"用户问题：{state['messages'][-1].content}\n" f"回答要简洁，不写过程。" ) return {"final_answer": resp.content} def direct_answer_node(state: AgentState) -> dict: """直接回答节点：不检索，仅靠模型自身知识回答。""" llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3) resp = llm.invoke(state["messages"][-1].content) return {"final_answer": resp.content}

这段代码有三个关键点需要注意。第一，路由节点的temperature=0，保证同一个问题每次得到的路由判断一致，避免“时而是 research、时而是 direct”的玄学现象。第二，DuckDuckGoSearchRun在部分网络环境下会被限流，如果执行时报速率限制，把它替换成你自己的搜索 API 封装即可，节点逻辑不用变。第三，收敛节点和直接回答节点虽然功能相似，但输入来源不同，一个读research_answer，一个直接读消息，因此不能合并成一个节点，否则会互相覆盖状态。

3.3 条件边与图编译：让路由决定执行路径

节点是函数，图则负责把这些函数串起来。这里用add_conditional_edges实现动态路由：执行到路由节点后，读取状态里的route字段，根据它的值决定进入哪个节点。这个思路和 Dify 工作流里的条件分支完全不同，LangGraph 的分支条件是运行时才确定的。

from langgraph.graph import StateGraph, START, END graph_builder = StateGraph(AgentState) graph_builder.add_node("route", route_node) graph_builder.add_node("research", research_node) graph_builder.add_node("answer", answer_node) graph_builder.add_node("direct_answer", direct_answer_node) graph_builder.add_edge(START, "route") graph_builder.add_conditional_edges( "route", lambda state: state["route"], # 读取路由字段 { "research": "research", # 需要检索时走执行节点 "direct": "direct_answer", # 不需要检索时直接回答 }, ) graph_builder.add_edge("research", "answer") graph_builder.add_edge("answer", END) graph_builder.add_edge("direct_answer", END) graph = graph_builder.compile()

编译完成后可以立即测试。第一次请求带一个需要实时信息的问题，验证它是否走了“检索+收敛”链路；第二次带一个常识问题，验证是否走了直接回答链路。这样就能确认路由没有死绑在某一条分支上。

from langchain_core.messages import HumanMessage result = graph.invoke({ "messages": [HumanMessage(content="帮我查一下某地今天的实时气温")], }) print(result["final_answer"])

这里提醒一点：add_conditional_edges映射表的键必须覆盖路由节点可能返回的所有值。路由节点我已经加了兜底逻辑，任何非预期输出都会归为direct，所以映射表是完整的。如果你在别的项目里省略了兜底，映射表又没有对应项，LangGraph 运行时会直接抛异常，而且报错信息不太直观，排查起来很费时间。

另外要注意recursion_limit。默认情况下，图运行超过一定步数会强制终止并报错，这是防止智能体死循环的一层保护。我在演示代码里保留了默认值，实际生产环境建议在invoke时主动设置，比如graph.invoke(data, {"recursion_limit": 25})，一旦两个节点来回调用超过 25 次，服务会主动中断而不是无限消耗 token。

4. 把 LangGraph 服务接进 Dify：OpenAPI 工具注册与三个关键参数

4.1 用 FastAPI 把编译好的图封装成 HTTP 接口

LangGraph 图本身是 Python 对象，Dify 无法直接加载它，所以需要一层薄薄的 HTTP 封装。我用 FastAPI 写这个服务，把graph.invoke暴露成一个 POST 接口，请求体里带上消息列表和会话标识。注意这里的thread_id目前只是透传占位，真正用到它要等第 6 章的 Checkpointer，但提前留好字段能省掉后面改接口的麻烦。

# server.py from typing import List, Optional from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn from graph import graph # 上一步编译好的 LangGraph 图 class AgentRequest(BaseModel): messages: List[dict] # 消息列表，最后一条是用户输入 thread_id: Optional[str] = None # 会话标识，预留用于状态恢复 class AgentResponse(BaseModel): final_answer: str route: str app = FastAPI(title="langgraph-agent-service") @app.post("/agent/run", response_model=AgentResponse) def run_agent(req: AgentRequest): try: result = graph.invoke({"messages": req.messages}) return AgentResponse( final_answer=result.get("final_answer", ""), route=result.get("route", ""), ) except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc @app.get("/healthz") def healthz(): return {"status": "ok"} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)

服务启动前，先用最小依赖把环境跑通：

pip install langgraph langchain-openai langchain-community fastapi uvicorn uvicorn server:app --host 0.0.0.0 --port 8000

启动后先用curl或接口调试工具打一次/agent/run，确认服务能返回final_answer再继续后面的 Dify 配置。我见过不少人在这一步直接跳到 Dify，结果发现服务根本起不来，回头排查半天，其实只是依赖版本冲突。先把服务在命令行里调通，后面所有问题都好定位。

4.2 在 Dify 注册自定义工具：OpenAPI Schema 与端点匹配

Dify 的自定义工具支持通过 OpenAPI schema 接入 HTTP 服务。我一般先准备好 schema 文件，再把服务地址填进去注册。传给 Dify 的 schema 只需要描述清楚接口的请求和响应结构，Dify 会根据operationId自动生成工具方法名，并把 schema 里定义的字段暴露成工具参数。

我这里用的 schema 是最小化的，只保留核心字段：

openapi: 3.0.0 info: title: LangGraph Multi-Agent Service version: 1.0.0 servers: - url: http://127.0.0.1:8000 paths: /agent/run: post: operationId: runAgent summary: 多智能体协作推理 requestBody: required: true content: application/json: schema: type: object properties: messages: type: array description: 消息列表，最后一条是用户当前输入 items: type: object thread_id: type: string description: 会话标识，用于多轮状态恢复 responses: '200': description: 成功 content: application/json: schema: type: object properties: final_answer: type: string route: type: string

在 Dify 自定义工具管理里，把这段 schema 粘进去，然后配置服务地址。这里最常踩的坑是servers.url：schema 里的 URL 必须能被 Dify 服务端访问。如果你的 Dify 是 Docker 部署，127.0.0.1指向的是 Dify 容器内部，不是运行 LangGraph 服务的宿主机，这时要改成宿主机 IP 或容器网络里可解析的主机名。

工具注册好之后，Dify 会把它当成一个可拖拽的工具节点。你需要手动设置请求参数来源：messages一般取“会话中的用户问题”，thread_id取会话 ID。记住：Dify 不会自动把消息列表传给你，必须显式映射，映射错误的话接口会收到空数组，服务端报错。

4.3 三个关键参数：超时、上下文截断、重试次数

接入完成后，系统能跑，但离“可用”还差三个参数的调优。这三个参数直接决定用户体验和账单数字，值得单独展开。

第一是超时。多智能体链路不是单次模型调用，它包含路由调用、工具调用、收敛调用，整体耗时通常远高于单次 LLM 请求。Dify 工具节点的默认超时偏短，我一般会调到 60 秒以上。如果 Dify 侧没有暴露超时配置，就在工作流节点的配置里找；实在改不了，就把 LangGraph 服务内部的模型换成更快的型号，压缩单次调用时间。

第二是上下文截断。LangGraph 服务收到的messages会随会话变长而膨胀。如果每个节点都把全部历史消息塞给语言模型，第十轮对话时输入就已经超出窗口。我在服务端加了一层截断逻辑，只保留最近 5 轮消息和当前问题，再把 Dify 知识库检索的结果拼进上下文。这样既保留多轮上下文，又控制输入长度。具体保留几轮，取决于任务复杂度，我项目里的经验是 5 到 8 轮比较平衡。

第三是重试次数。很多人在 Dify 工具节点看到失败就点重试，这是最烧钱的操作。多智能体链路内部已经对模型和工具做了多次调用，它的失败率是多个环节的叠加；重试一次等于把整个链路再跑一遍，成本翻倍。我的做法是：业务性失败（比如路由结果不合理）直接返回给用户，网络类失败才允许重试一次。控制手段是在 FastAPI 封装层里对异常类型做区分，而不是在 Dify 侧统一重试。

4.4 Dify 工作流里编排多个工具结果

当 LangGraph 服务作为 Dify 的工具节点工作后，你还可以让它和 Dify 工作流里的其他节点协作。我常用的做法是串行两段：先做知识库检索，再调用多智能体服务。Dify 知识库节点把检索出来的段落拼进一个变量，这个变量随请求一起传给 LangGraph 服务，最终回答就有了知识库引用来源。这个过程 LangGraph 侧完全不用改动，它只是把 Dify 传来的上下文当作外部材料。

有一点要注意：不要在 Dify 侧再放一个大模型节点去二次总结工具结果。LangGraph 服务已经返回了final_answer，Dify 直接把这个字段放进消息即可。二次总结不仅多花一次模型调用，还可能把final_answer里的关键信息改丢。我在早期项目里就犯过这个错误，结果“最终回答”被二次总结得面目全非，排查了很久才发现问题出在工作流多了个多余的 LLM 节点。

另外，如果工作流里有多条工具调用路径，结束节点建议把final_answer完整透传出去，不要拼接。有些情况下final_answer本身就包含了检索材料的引用，拼接会让文本重复，用户看到两遍同样的内容，体验很差。

5. 多智能体系统的避坑清单：状态串线、死循环、token 失控 5 连排

这一章是血泪经验集，每一条都是我在实际调试中真实踩过的。

5.1 会话串线：上一个用户的问题出现在下一个人的回答里

现象：两个用户同时使用系统，A 用户问的问题，出现在 B 用户的回答上下文里；更糟的是，A 用户上一轮的检索材料被 B 用户看到。

原因：早期版本我为了“性能好”，把中间结果存在模块级全局变量里，以为每次请求会覆盖。但 FastAPI 是并发处理请求的，两个请求同时进来时，全局变量被交叉读写，状态就串了。LangGraph 本身不会主动跨请求保存状态，串线一定是外部存储或全局变量造成的。

解决：保证每个请求的状态都从graph.invoke的参数开始，不在任何位置持有跨请求的中间变量。现在我的封装层里，每个请求的对象都限于函数作用域内，不用模块级 dict 缓存任何查询结果。如果确实需要跨请求共享，用支持并发读写的存储，并且带上会话唯一键，而不是简单地在内存里放一个字典。

5.2 路由死循环：两个智能体互相踢皮球

现象：调用不返回，日志显示请求在 route 节点和 research 节点之间来回跳；token 消耗飙升，最终超时才被中断。

原因：某个节点返回了意外的route值，条件边把它又送回 route 节点，形成循环。比如 research 节点最后返回了包含"route": "research"的字段，于是图又从 route 开始走。这种问题在节点多、字段多的时候极难发现。

解决：三个办法一起用。第一，每个非路由节点绝不写route字段，路由字段只允许路由节点读写；第二，在invoke时设置recursion_limit，比如 25，达到上限强制终止并抛出异常，让问题暴露而不是让账单爆炸；第三，在条件边映射表里加一个兜底目标，任何未知值都落到直接回答节点。做好这三步，死循环基本能被拦在正式环境之外。

5.3 工具调用风暴：一次搜索失败导致五轮重试

现象：用户问一个需要实时数据的问题，外部搜索接口报错，智能体自动重试，重试失败后换了个工具再试，最后账单显示这一条问题消耗了上万 token。

原因：搜索工具内部把超时、限流、返回空结果三种错误全部当作可重试错误；加上智能体编排层默认允许多次调用工具，错误触发后不断重试同一操作。

解决：在工具封装里区分错误类型。超时类和网络抖动类可以重试一次，业务类错误（比如搜索无结果）直接返回空字符串。更重要的是在编排层限制工具调用次数：create_agent的配置里设置最大迭代上限，这样即使工具反复报错，整个链路也会在有限步内结束。重试不是越多越好，多智能体系统里每次重试都是叠加成本。

5.4 路由不稳定：同一句话这次检索、下次直接回答

现象：输入完全一样的问题，第一次走了 research 分支，第二次走了 direct 分支，用户两次拿到的答案风格和质量完全不同。

原因：路由节点用的语言模型温度没设 0，或者提示词里给了模型太多自由发挥空间。模型在多智能体路由场景下并不稳定，温度稍高一点，输出就会出现波动。这个现象特别像“模型玄学”，但根因其实是参数问题。

解决：路由节点的temperature=0，同时把输出严格限制为一组枚举值，比如research和direct。如果模型输出不在枚举范围内，直接默认走direct分支。我甚至会在路由节点加一层模式匹配，用正则把模型输出里的有效词提取出来，再判断路由，而不是直接比较整段文本。

5.5 上下文膨胀：聊到第十轮，token 输入超出模型窗口

现象：前几轮回答正常，会话持续一段时间后开始报“上下文超限”，或者响应变得非常慢，费用明显上涨。

原因：messages 字段无限制累积。LangGraph 状态里的 messages 每轮追加，FastAPI 接口每次把全部消息传给语言模型。对话轮次一多，输入很快撑满窗口。很多模型会自动截断最前面的内容，导致用户最早的需求被遗忘，回答质量断崖式下跌。

解决：在 FastAPI 封装层做消息压缩，只保留最近 N 轮的消息。更稳妥的做法是抽取出每次的用户意图和关键实体，形成会话摘要，用摘要加最近消息的方式传给模型。我的做法是先保留最近 5 轮，会话超过 5 轮时用摘要替换更早的消息。这样既保留长期记忆，又不无限膨胀。注意：摘要是离线更新还是实时更新，要提前设计好，否则每轮都重新摘要反而是另一种浪费。

6. 进阶技巧：用 Checkpointer 恢复多轮会话状态

多智能体系统的多轮交互，和普通聊天机器人有个本质区别：它不仅需要记住“用户说过什么”，还要记住“上一次运行到哪个节点、哪个智能体产出了什么结果”。比如用户第一轮说“帮我查某地气温”，第二轮说“顺便看看适不适合跑步”，系统如果只记得第一轮的消息文本，不知道第一轮已经走过了 research 分支，就会把同一套检索流程重复跑一遍。

LangGraph 的 Checkpointer 机制能解决这个问题。给编译好的图传入一个检查点存储器，再把调用时的配置带上thread_id，LangGraph 会把这个线程的状态保存下来；下次用同一个线程 ID 调用时，图自动从上次结束的位置恢复，而不是从初始状态重新开始。这样多轮会话里已经跑过的工具结果、已经完成的路由决策都会被保留，后续节点可以直接复用。

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = graph_builder.compile(checkpointer=checkpointer) # 第一次调用，保存状态 graph.invoke( {"messages": [HumanMessage(content="帮我查某地今天气温")]}, config={"configurable": {"thread_id": "conv-001"}}, ) # 第二次调用，恢复状态 result = graph.invoke( {"messages": [HumanMessage(content="顺便看看适不适合跑步")]}, config={"configurable": {"thread_id": "conv-001"}}, ) print(result["final_answer"])

代码里有两个重要细节。第一，MemorySaver是内存级的检查点，开发测试足够，但服务重启或多实例部署时状态会丢，生产环境要换成支持持久化的检查点存储，接口用法基本一致，把MemorySaver换成对应实现即可。第二，thread_id不要直接用用户的conversation_id原值，最好加上应用标识和用户标识拼成复合键，防止不同用户之间状态串线。我在第 4 章提到预留的thread_id字段，到这里就真正派上用场了。

接入 Dify 的方式也很直接：把 Dify 工作流里的会话 ID 映射到thread_id传给 LangGraph 服务。注意 Dify 的会话 ID 生命周期和智能体服务的会话状态生命周期要一致，如果 Dify 侧新开了一个会话，旧的thread_id对应的状态就没有必要再保留。我现在的习惯是：先在小范围验证路由稳定，再开 Checkpointer，最后才接入生产；这样出问题时每一步都找得到责任人。希望这篇文章能把你的多智能体系统从“能跑”带到“能上线”，帮到你。

本文还有配套的精品资源，点击获取