}

在实际工作中经常遇到这个场景，今天系统性地总结一下最佳实践和注意事项。

atom-one-light"

src="https://i-blog.csdnimg.cn/direct/f5ca931634094e90add91d365692ddc5.jpeg"

width="1664">

前言

在构建复杂的

应用时，我们常常面临一个核心挑战：如何优雅地处理多步骤、动态决策、状态依赖的执行流程。

传统的链式调用（如

LangChain

LCEL）虽然简洁高效，但在面对“需要根据中间结果动态决定下一步操作”或“多次循环检索-推理”等场景时，往往显得力不从心。

而市面上一些低代码平台（如

DIFY）虽然提供了可视化编排，却在灵活性与控制粒度上存在明显短板——尤其在需要精细干预每一步逻辑、实时反馈或并行处理多个知识源时，其“黑盒”特性反而成了瓶颈。

LangChain

生态中专为有状态、多参与者、循环迭代工作流设计的图状编排框架，LangGraph

允许开发者以节点（Node）和边（Edge）的方式显式定义执行逻辑，天然支持循环、条件跳转与状态持久化。

本文将以一个典型场景为例：用户提问后，系统先由

LLM

判断涉及哪些知识库，再对每个知识库动态执行向量检索（Top3）并生成中间响应（如“正在检索知识库

A…”），最后汇总所有结果生成最终答案。

我们将分别用纯代码、RunnableLambda/RunnableBranch

LangGraph

三种方式实现，并深入对比其可读性、扩展性与维护成本。

你会发现：当流程不再是线性，而是需要“思考-行动-再思考”的闭环时，LangGraph

不仅更香，更是不可替代的主编排引擎。

通过一个实例子来看传统编码、流程式编码、LangGraph

主流基于知识库问答的Agentic

2.先用LLM把用户的回答对着当前系统看属于哪个知识库或者哪几个知识库以list返回出来

2.1

如果用户的提问不在知识库list中或者説不属于知识库范围，那么我们用llm拟一个回答告诉用户

2.2

根据用户回答属于哪几个知识库,把用户的提问用BM25+HYDE折成3-5个关键词，循环这些关键词列表，外层知识库list一层循环内套用户提问折成的关键词列表循环

3.1

循环时由于我们不是“静RAG”，而是主流的“交互播报式RAG”，因此内部每一次循环内会进行向量检索，检索出top3，然后让llm进行中间回答类似ai

ide开发工具那种：

当外层知识库列表这一层的循环完成后，再给到llm做一次总结性回答

这种设计是主流的设计也是目前最先进的动态规划的交互式RAG，如：cursor,

qoder等AI

IDE开发工具用的都是这种方式，我们称这种RAG为Agentic

根据设计做技术选型

为了让大家非常直观的比较传统编程、流程式编码以及用LangGraph，我们就只实现1，2，2.1,

src="https://i-blog.csdnimg.cn/direct/d08962594425457ebcc54eb9ad2c9d34.png"

width="544">

传统编程式

传统编程式当然实现简单，代码写起来也很容易，下面是伪代码。

class="hljs">knowledge_repo_list=get_knowledges();

identify_result=get_identify(userInputPrompt,

knowledge_repo_list)

else用llm拟一段话术告诉用户你当前的提问不在我的服务范围

看起来不难，但总觉得有点不对劲

我们现在是用api在窜这个工作流，既然用了langchain，那么langchain里有什么更加好的办法来实现这么一个逻辑分支流程吗？

或者説编码式实现这个只是一个笨办法？

0)">传统编程式实现流程优缺点分析

现在的流程是：

1. 获取知识库列表 →

   2. LLM判断意图 →

   匹配到知识库

这是手工编排的流程，用Python代码显式控制每一步。

LangChain推荐的更优雅方案

使用

LangGraph（最推荐）

LangGraph

LangChain

1. 逻辑清晰直观：代码一看就懂，适合快速开发

2. 灵活性高：想加什么逻辑直接写Python代码

3. 调试方便：可以随时print/log

❌

1. 不够声明式：流程隐藏在代码里，不直观

2. 难以可视化：无法生成流程图

3. 状态管理混乱：变量散落在各处

4. 难以扩展：加新分支要改多处代码

5. 不符合LangChain设计理念：没有利用Runnable的组合能力

短期

来説这种实现没问题，对于简单的2-3步流程，手工编排完全够用。

中期（功能稳定后）

一般用

RunnableBranch，保持LangChain风格，代码更优雅。

长期（复杂场景）

如果后续要加：

• 多轮对话

那么我们看看什么是RunnableBranch呢？

RunnableBranch实现

其实RunnableBranch是一种声明式编程，上面那一“陀”代码没有太多本质上的改变，只是它变得更声明，当然，对于：如果用户回答属于当前知识库范围那么。

。

。

如果不属于那么。

。

。

还是要写一对if

else

_build_general_chat_chain():"""构建通用聊天的

RunnableBranch

Chain"""log.info("[RunnableChain]

RunnableBranch

RunnableLambda(_step_fetch_knowledge_repos)identify_runnable

RunnableLambda(_step_identify_intent)branch

RunnableBranch((#

0,RunnableLambda(_step_handle_matched),),#

默认分支:

未匹配或出错RunnableLambda(_step_handle_not_matched),)chain

fetch_runnable

branchlog.info("[RunnableChain]

RunnableBranch

看，这就是RunnableBranch，从代码来看有点流程的感觉了。

1. 获取知识库列表 →

   2. LLM判断意图 →

   匹配到知识库

不过这还是手工编排的流程，用Python代码显式控制每一步，只不过较声明式了。

不能只看到一种就决定设计需要多问还有没有更好的实现？

RunnableBranch不错，看着挺像工作流的，可是现在我们只是实现了一个框子，后面要实现更复杂的：

2.先用LLM把用户的回答对着当前系统看属于哪个知识库或者哪几个知识库以list返回出来

2.1

如果用户的提问不在知识库list中或者説不属于知识库范围，那么我们用llm拟一个回答告诉用户

2.2

根据用户回答属于哪几个知识库,把用户的提问用BM25+HYDE折成3-5个关键词，循环这些关键词列表，外层知识库list一层循环内套用户提问折成的关键词列表循环

3.1

循环时由于我们不是“静RAG”，而是主流的“交互播报式RAG”，因此内部每一次循环内会进行向量检索，检索出top3，然后让llm进行中间回答类似ai

ide开发工具那种：

当外层知识库列表这一层的循环完成后，再给到llm做一次总结性回答

1.

• 第一步：识别 knowledgeList（目前已经有）

• 第二步：循环每个命中的知识库：

  • 向量检索

  过程播报”（SSE：我正在检索第N个、当前查到X、准备总结…）

第三步：所有库循环结束后，再调用一次

LLM

整体是一个「多轮、多节点、有循环、有中间状态」的动态

RAG

• 多层 RunnableLambda

  while/for

优点：一切都在一个

Python

LangGraph：

• identify_knowledge → for_each_knowledge

  (loop) → vector_search → llm_partial_answer → aggregate_summary

每个节点一个函数，结构自然是“流程图级别”的，可视化（Mermaid

结论：只要引入“循环

中间播报

• RunnableBranch：

  • 本身不带循环语义，你要自己写 for / while，在

    state

  • 一旦要做“中途可中断”、“某个知识库失败是否跳过/重试”等逻辑，代码会变得非常命令式。

• LangGraph：

  • 原生支持循环、条件跳转、子图复用，可以设计类似：

    • for_each_knowledge 子图：对单个知识库运行「检索→中间回答」；

    • 外层图把 knowledgeList 展开成多次调用，或用“边+条件”控制。

• 结论：涉及多次循环

  动态决策时，LangGraph

  • RunnableBranch：

    dict

• 全部靠你自己约定

  key

  名和更新顺序，没有类型/结构上的约束。

LangGraph：

• 用 TypedDict（你现在就有 ChatState）明确：

  • 哪些字段是输入、哪些是中间、哪些是输出。

• 每个节点只关心它要读/写的那一部分，整体状态逻辑更清晰。

结论：随着状态字段数量增加，LangGraph

dict

优化建议：

如果你的项目访问量较大，建议增加缓存机制。

我们团队在优化后，接口响应时间从800ms降到了50ms，

效果非常明显。

具体的缓存策略可以根据业务场景调整。

2.4

SSE

• 两者都能做，差别在组织方式：

• Runnable

  方案：

  generate_stream()，在其中手写循环：

  • 调

  → yield SSE

  继续下一步

LangChain

Runnable

本身也可以 astream，但复杂逻辑时，你会混合“内部流式

for

LangGraph

方案：

• 工作流内部专注于“决策+状态”，router

  只关心：

更高级做法：用

LangGraph

节点回调” 思路，把“在哪个节点打什么日志

结论：SSE

实现本身两者都行，但

更有利于把“播报点”固定在节点边界，不至于散落在一堆

if/for

• 这个维度在复杂流程中很重要，即可视化调试（debug_mode、流程图）：

• RunnableBranch：

  • 全靠日志和断点，“脑内画图”。

• LangGraph：

  • 直接 visualize_workflow() 得到

    Mermaid，流程走到哪一步、被哪个条件路由了，都能对齐日志。

• 结论：后期流程一旦复杂，这一项对“理解

2.6

• 短期成本：

  Runnable

  在现有基础上继续堆逻辑，开发门槛最低。

• 引入

长期成本：

• 复杂逻辑用

3.

结合我们的最终目标，客观建议

3.1

要知道，我们的目标不是“一个简单判断分支的API”，而是一个：

RAG

• 把 LangGraph

  作为主编排，

• 把 RunnableBranch

  比“主线程用

  • 主链路：LangGraph（必选）

    • 节点划分例如：

      1. identify_knowledge（得到

         knowledgeList）

      2. iterate_knowledge（单个知识库子图：向量检索

         partial

      3. aggregate_summary（最终回答）

  • 节点内部：仍然用

    Runnable

    • 比如 iterate_knowledge 里，向量检索

      LLM

  • 对外接口仍然是

    Runnable

    形态

    • 提供统一的 execute_general_chat(user_prompt,

      session_id)，内部调用 compiled_graph.ainvoke(...)，符合“主链路是

  一句话总结

  后面要做的“多知识库循环

  动态检索

  LangGraph的实现

  核心代码

  添加节点workflow.add_node("fetch_repos",

  fetch_knowledge_repos_node)workflow.add_node("identify",

  identify_intent_node)workflow.add_node("handle_matched",

  handle_matched_node)workflow.add_node("handle_not_matched",

  设置入口点workflow.set_entry_point("fetch_repos")#

  添加边（流程连接）workflow.add_edge("fetch_repos",

  "identify")#

  添加条件边（分支路由）workflow.add_conditional_edges("identify",

  identify

  "handle_matched","handle_not_matched":

  添加结束边workflow.add_edge("handle_matched",

  END)workflow.add_edge("handle_not_matched",

  END)#

  workflow.compile()log.info("[LangGraph]

  工作流构建完成")return

  src.knowledgerepo.knowledgerepo_tool

  import

  src.utils.prompt_setting_helper

  import

  src.chat.general_chat_constants

  import

  (MODEL_NAME_CHAT,FUNCTION_NAME_USER_QUERY_IDENTIFY,FUNCTION_NAME_USER_INPUT_NOT_VALID_ANSWER,

  from

  src.chat.general_chat_prompt_helper

  import

  (format_knowledge_repo_list_for_prompt,build_prompt_template_for_query_identify,build_prompt_template_for_not_valid_answer,

  from

  src.chat.general_chat_llm_caller

  import

  Any]]:"""获取所有知识库列表（保留原有接口，供其他模块直接调用）"""log.info("[GeneralChat]

  await

  get_all_knowledge_repos()log.info(f"[GeneralChat]

  获取到

  identify_user_query_intent(user_prompt:

  str,

  Any]]]:"""识别用户查询意图（保留原有接口）"""log.info("[GeneralChat]

  开始识别用户查询意图...")#

  get_prompt_setting(model_name=MODEL_NAME_CHAT,function_name=FUNCTION_NAME_USER_QUERY_IDENTIFY,)if

  not

  prompt_config:log.error("[GeneralChat]

  未找到提示词配置:

  "f"model_name={MODEL_NAME_CHAT},

  function_name={FUNCTION_NAME_USER_QUERY_IDENTIFY}")return

  Nonesystem_role_msg

  prompt_config.get("systemRoleMsg",

  "")user_role_msg

  prompt_config.get("userRoleMsg",

  前100字:

  "f"{system_role_msg[:100]}...")#

  格式化知识库列表knowledge_repo_list_json

  format_knowledge_repo_list_for_prompt(knowledge_repos)#

  打印格式化后的知识库列表（美化输出）log.info(f"\n[GeneralChat]

  内置知识库列表（共

  个）：\n"f"{knowledge_repo_list_json}\n")#

  build_prompt_template_for_query_identify(system_role_msg=system_role_msg,user_role_msg=user_role_msg,knowledge_repo_list_json=knowledge_repo_list_json,user_prompt=user_prompt,)log.info(f"[GeneralChat]

  用户当前提示:

  get_line_settings(streaming=False)if

  not

  line_settings:log.error("[GeneralChat]

  未找到任何

  line_settings.get("master")slaver_line

  not

  slaver_line:log.error("[GeneralChat]

  主线路或备线路配置不完整")return

  call_llm_with_fallback(prompt_template=prompt_template,master_line=master_line,slaver_line=slaver_line,chain_name="UserQueryIdentify",parse_json=True,)data

  result.get("data",

  knowledge_list:log.info(f"\n[GeneralChat]

  LLM

  个知识库：\n"f"{json.dumps(knowledge_list,

  ensure_ascii=False,

  indent=2)}\n")else:log.info("[GeneralChat]

  LLM

  generate_not_valid_answer(user_prompt:

  str)

  Optional[str]:"""当用户输入不在服务范围时，生成友好的回答（保留原有接口）"""log.info("[GeneralChat]

  获取提示词配置prompt_config

  get_prompt_setting(model_name=MODEL_NAME_CHAT,function_name=FUNCTION_NAME_USER_INPUT_NOT_VALID_ANSWER,)if

  not

  prompt_config:log.error("[GeneralChat]

  未找到提示词配置:

  "f"model_name={MODEL_NAME_CHAT},

  function_name={FUNCTION_NAME_USER_INPUT_NOT_VALID_ANSWER}")return

  Nonesystem_role_msg

  prompt_config.get("systemRoleMsg",

  "")user_role_msg

  prompt_config.get("userRoleMsg",

  "")#

  build_prompt_template_for_not_valid_answer(system_role_msg=system_role_msg,user_role_msg=user_role_msg,user_prompt=user_prompt,)#

  await

  get_line_settings(streaming=False)if

  not

  line_settings:log.error("[GeneralChat]

  未找到任何

  line_settings.get("master")slaver_line

  not

  slaver_line:log.error("[GeneralChat]

  主线路或备线路配置不完整")return

  call_llm_with_fallback(prompt_template=prompt_template,master_line=master_line,slaver_line=slaver_line,chain_name="UserInputNotValidAnswer",parse_json=True,)data

  result.get("data",

  ChatState(TypedDict):"""聊天工作流状态"""#

  输入user_prompt:

  fetch_knowledge_repos_node(state:

  ChatState)

  获取知识库列表"""log.info("[LangGraph]

  节点1:

  获取知识库列表...")try:knowledge_repos

  await

  get_all_knowledge_repos()log.info(f"[LangGraph]

  获取到

  state["user_prompt"]knowledge_repos

  [])#

  knowledge_repos:log.warning("[LangGraph]

  []}try:#

  get_prompt_setting(model_name=MODEL_NAME_CHAT,function_name=FUNCTION_NAME_USER_QUERY_IDENTIFY,)if

  not

  prompt_config:log.error("[LangGraph]

  未找到提示词配置")return

  prompt_config.get("systemRoleMsg",

  "")user_role_msg

  prompt_config.get("userRoleMsg",

  "")#

  格式化知识库列表knowledge_repo_list_json

  format_knowledge_repo_list_for_prompt(knowledge_repos)log.info(f"\n[LangGraph]

  内置知识库列表（共

  个）：\n"f"{knowledge_repo_list_json}\n")#

  build_prompt_template_for_query_identify(system_role_msg=system_role_msg,user_role_msg=user_role_msg,knowledge_repo_list_json=knowledge_repo_list_json,user_prompt=user_prompt,)log.info(f"[LangGraph]

  用户当前提示:

  get_line_settings(streaming=False)if

  not

  line_settings:log.error("[LangGraph]

  未找到

  line_settings.get("master")slaver_line

  not

  slaver_line:log.error("[LangGraph]

  主线路或备线路配置不完整")return

  call_llm_with_fallback(prompt_template=prompt_template,master_line=master_line,slaver_line=slaver_line,chain_name="UserQueryIdentify",parse_json=True,)data

  result.get("data",

  knowledge_list:log.info(f"\n[LangGraph]

  LLM

  个知识库：\n"f"{json.dumps(knowledge_list,

  ensure_ascii=False,

  indent=2)}\n")else:log.info("[LangGraph]

  LLM

  处理匹配到知识库的情况"""log.info("[LangGraph]

  节点3A:

  "knowledge_matched","result_data":

  matched_knowledge,}

  处理未匹配知识库的情况，生成友好回答"""log.info("[LangGraph]

  节点3B:

  用户输入不在服务范围，生成友好回答...")user_prompt

  获取提示词配置prompt_config

  get_prompt_setting(model_name=MODEL_NAME_CHAT,function_name=FUNCTION_NAME_USER_INPUT_NOT_VALID_ANSWER,)if

  not

  prompt_config:log.error("[LangGraph]

  未找到提示词配置")return

  prompt_config.get("systemRoleMsg",

  "")user_role_msg

  prompt_config.get("userRoleMsg",

  "")#

  build_prompt_template_for_not_valid_answer(system_role_msg=system_role_msg,user_role_msg=user_role_msg,user_prompt=user_prompt,)#

  await

  get_line_settings(streaming=False)if

  not

  line_settings:log.error("[LangGraph]

  未找到

  line_settings.get("master")slaver_line

  not

  slaver_line:log.error("[LangGraph]

  主线路或备线路配置不完整")return

  call_llm_with_fallback(prompt_template=prompt_template,master_line=master_line,slaver_line=slaver_line,chain_name="UserInputNotValidAnswer",parse_json=True,)data

  result.get("data",

  "friendly_answer","result_data":

  answer,}except

  "handle_not_matched"]:"""路由函数：根据意图识别结果决定走哪个分支"""matched_knowledge

  state.get("matched_knowledge")if

  matched_knowledge

  "handle_not_matched"log.info("[LangGraph]

  路由决策:

  添加节点workflow.add_node("fetch_repos",

  fetch_knowledge_repos_node)workflow.add_node("identify",

  identify_intent_node)workflow.add_node("handle_matched",

  handle_matched_node)workflow.add_node("handle_not_matched",

  设置入口点workflow.set_entry_point("fetch_repos")#

  添加边（流程连接）workflow.add_edge("fetch_repos",

  "identify")#

  添加条件边（分支路由）workflow.add_conditional_edges("identify",

  identify

  "handle_matched","handle_not_matched":

  添加结束边workflow.add_edge("handle_matched",

  END)workflow.add_edge("handle_not_matched",

  END)#

  workflow.compile()log.info("[LangGraph]

  工作流构建完成")return

  execute_general_chat_chain(user_prompt:

  str,

  app.ainvoke(initial_state)log.info(f"[LangGraph]

  工作流执行完成,

  "f"result_type={final_state.get('result_type')}")#

  {"result_type":

  final_state.get("result_type"),"result_data":

  final_state.get("result_data"),"error":

  ====================

  app.get_graph().draw_mermaid()#

  清理并简化为标准

  line.strip().startswith('%%{init:'):continueif

  line.strip().startswith('classDef'):continueif

  ':::first'

  '\n'.join(lines)log.info("[LangGraph]

  Mermaid

  流程图:")log.info(f"\n{cleaned_mermaid}\n")return

  0)">通过手工编程、RunnableBranch、LangGraph得到结论

  LangGraph

  ❌

  app.get_graph().draw_mermaid()#

  清理并简化为标准

  line.strip().startswith('%%{init:'):continueif

  line.strip().startswith('classDef'):continueif

  ':::first'

  '\n'.join(lines)log.info("[LangGraph]

  Mermaid

  流程图:")log.info(f"\n{cleaned_mermaid}\n")return

  这块代码会生成：Mermaid

  代码，把Mermaid代码放入：Mermaid在线免费生成图工具里后你应该会得到这样的一个图

  src="https://i-blog.csdnimg.cn/direct/52ae493a96bb480c845f35896b4ad9a2.png"

  width="651">

  你也可以把这个代码粘贴到支持

  Mermaid

  class="language-python">router.pyknowledge_repos

  await

  identify_user_query_intent(user_prompt,

  knowledge_repos)

  generate_not_valid_answer(user_prompt)yield

  answer

  我们可以看到问题：流程隐藏在代码逻辑里，无法可视化。

  要理解流程必须读代码。

  ✅

  class="language-python">workflow

  StateGraph(...)

  print(app.get_graph().draw_mermaid())

  0)">LangGraph在实际工作中怎么使用的流程

  1. 开发阶段：你用

     Python

  2. 调试阶段：在

     Jupyter

  3. 文档阶段：导出流程图放到文档/Wiki

  4. 维护阶段：新同事看流程图就能理解系统

  编写方式

  if-else

  函数调用

  流程可视化

  ❌

  无法自动生成

  理解成本

  需要读代码

  看图就懂

  适合场景

  简单流程（≤3步）

  rgba(0,

  0)">演示用LangGraph实现的工作流调用效果

  class="language-python">{"session_id":

  }

  但我很乐意帮助您解答问题或提供帮助。

  有什么我可以帮您的吗？

  "}

  data:

  class="language-python">{"session_id":

  }

  我们把上述的visualize_workflow通过本系列里的：langchain_start.py暴露成api

  general_chat_router.py

  APIRouter(prefix="/general_chat",

  tags=["General

  GeneralChatRequest(BaseModel):"""通用聊天请求模型"""session_id:

  Optional[str]

  GeneralChatRequest):"""通用聊天接口（SSE

  流式响应）接口地址:

  http://localhost:8001/v1/chat/general_chatArgs:request:

  请求参数-

  generate_stream():try:log.info("[GeneralChatRouter]

  收到通用聊天请求")log.info(f"[GeneralChatRouter]

  session_id={request.session_id}")log.info(f"[GeneralChatRouter]

  user_prompt={request.user_prompt}")#

  参数验证if

  request.user_prompt.strip():log.warning("[GeneralChatRouter]

  user_prompt

  ensure_ascii=False)}\n\n"returnuser_prompt

  LangGraph

  execute_general_chat_chainlog.info("[GeneralChatRouter]

  LangGraph

  execute_general_chat_chain(user_prompt,

  result.get("result_type")result_data

  error:log.error(f"[GeneralChatRouter]

  决策链执行失败:

  ensure_ascii=False)}\n\n"returnif

  result_type

  "knowledge_matched":log.info(f"[GeneralChatRouter]

  用户输入匹配到

  ensure_ascii=False)log.info(f"[GeneralChatRouter]

  返回知识库列表:\n{knowledge_list_json}")yield

  f"data:

  "friendly_answer":log.info(f"[GeneralChatRouter]

  返回友好回答:

  ensure_ascii=False)}\n\n"else:log.error(f"[GeneralChatRouter]

  {result_type}")yield

  ensure_ascii=False)}\n\n"returnexcept

  Exception

  e:log.error("[GeneralChatRouter]

  处理请求时发生异常")log.exception(e)yield

  f"data:

  ensure_ascii=False)}\n\n"return

  StreamingResponse(generate_stream(),media_type="text/event-stream",headers={"Cache-Control":

  "keep-alive","X-Accel-Buffering":

  "no"})

  http://localhost:8001/v1/chat/general_chat/visualizeReturns:Mermaid

  fastapi.responses

  PlainTextResponsetry:mermaid_code

  ("#

  PlainTextResponse(response_text)except

  Exception

  e:log.error(f"[GeneralChatRouter]

  生成流程图失败:

  langchain_start.py

  src.knowledgerepo.knowledge_repo_router

  import

  FastAPI):"""应用生命周期管理：启动和关闭时的资源初始化与清理"""#

  Initializing

  mongo_client.admin.command("ping")log.info(f"[Startup]

  MongoDB

  redis_client.ping()log.info(f"[Startup]

  Redis

  redis_client.connection_poollog.info(f"[Startup]

  Redis

  "f"max_connections={pool_info.max_connections},

  "f"connection_kwargs={{host={pool_info.connection_kwargs.get('host')},

  "f"port={pool_info.connection_kwargs.get('port')},

  "f"db={pool_info.connection_kwargs.get('db')}}}")log.info("[Startup]

  Redis

  shutdown...")log.info("[Shutdown]

  Closing

  close_mongo_connection()log.info("[Shutdown]

  MongoDB

  connection...")close_redis_connection()log.info("[Shutdown]

  Redis

  app.include_router(smartpen_router,

  prefix="/v1/chat")

  /v1/chat/demo/intension-identify

  app.include_router(demo_router,

  prefix="/v1/chat")

  /v1/chat/knowledgebase/label-identify

  app.include_router(knowledgebase_router,

  prefix="/v1/chat")

  app.include_router(general_chat_router,

  prefix="/v1/chat")

  'http://localhost:8001/v1/chat/general_chat/visualize'

  --header

  TD;__start__[__start__]fetch_repos(fetch_repos)identify(identify)handle_matched(handle_matched)handle_not_matched(handle_not_matched)__end__[__end__]__start__

  -->

  handle_not_matched;handle_matched

  -->

  TD开始复制到最后放到在线Mermaid网址里去即可就可以得到一幅当前的流程图了。

  src="https://i-blog.csdnimg.cn/direct/3bff3a4687b444b085a2a0eae64af104.png"

  width="1684">

  写在最后

  传统编码、RunnableBranch、LangGraph三者对比

  综上所述，面对“动态知识库路由

  多轮检索-推理”这类复杂流程，三种实现方式各有利弊：手工

  if-else

  编程逻辑直观但代码冗长、难以维护，扩展性差；RunnableBranch

  RunnableLambda 利用

  LCEL

  提升了声明式表达能力，适合简单分支，却无法自然表达循环与状态更新；而 LangGraph 通过图结构显式建模状态流转，天然支持循环、条件跳转与中间反馈，在复杂、动态、多步骤的

  Agent

  工作流中展现出强大优势——不仅代码清晰，还易于调试、监控和扩展。

  方式	可读性	循环支持	状态管理	扩展性	适用场景
  手工 if-else	低	弱	手动	差	超简单线性逻辑
  RunnableBranch	中	❌	无	一般	静态分支、无状态流程
  LangGraph	高	✅	内置	强	动态决策、多轮 Agent

  LangGraph与Dify、n8n等低代码开发工具的比较

  相比之下，DIFY

  应用的入门门槛，但其流程编排多为静态、线性、黑盒化设计，难以支持“根据

  LLM

  输出动态决定检索哪些知识库”或“在循环中实时生成中间反馈”等高级交互逻辑。

  DIFY

  的节点跳转依赖预设条件，无法在运行时动态构建执行路径，更缺乏对状态的细粒度控制。

  而

  LangGraph

  以代码为载体，赋予开发者完全的控制权——你可以自由定义节点行为、状态更新规则和跳转逻辑，轻松实现

  DIFY

  能力维度	LangGraph	DIFY（当前版本）
  编排方式	代码驱动，图结构显式定义（节点 边）	可视化低代码拖拽，隐式流程
  动态决策支持	✅ 支持运行时根据 仅支持预设条件分支，无法动态生成新路径
  嵌套循环/迭代	✅ DAG
  中间状态反馈	✅ 可在每一步输出中间结果（如“正在检索...”）	⚠️ 有限，通常仅支持最终输出
  多知识库动态路由	✅ LLM 需预先绑定固定知识库，无法运行时动态扩展

  当你的

  应用开始“思考”，LangGraph

  的利器。

  附：项目中要用LangGraph的额外安装

  它是完全开源免费的，LangChain官方专门用来应对复杂工作流的。

  • LangGraph 是独立的包，不包含在 langchain 核心包中

  • 需要 langchain-core

    >=

    0.2.0

  • Python

    >=

  你可以直接在你的requirement.txt里这样声明

  class="post-meta-container">