多 Agent 编排：子工作流与条件路由

学习目标

• 理解 Langflow 中多 Agent 的图表示：子图合并（Graph.__add__）
• 掌握条件路由机制（conditionally_excluded_vertices）
• 学会处理循环图（max_iterations + 循环检测）
• 了解预构建 Agent 模板（助手、翻译、系统消息）

前置知识

本章涉及多 Agent 编排的通用原理，建议先阅读：

• 多 Agent 路由与隔离

下文假设你已理解多 Agent 的基本概念，直接聚焦 Langflow 的具体实现。

---

项目实践

1. 子图合并：Graph.__add__

Langflow 的 Graph 支持加法运算，将多个 Agent 子图合并为一个协作图：

# 创建两个独立的 Agent 子图
graph_a = Graph(start=agent_a_input, end=agent_a_output)
graph_b = Graph(start=agent_b_input, end=agent_b_output)

# 合并为一个多 Agent 图
combined = graph_a + graph_b

合并语义：

• 所有顶点和边从两个子图复制到新图
• 使用 vertex_map 去重：同 ID 的顶点只保留一份
• 重新构建路由表（predecessor_map、successor_map、in_degree_map）
• 重新执行拓扑排序，确定执行层序

2. 条件路由

Graph 运行期间，根据上游输出的结果动态决定下游顶点是否执行：

# Graph 中的条件路由状态
self.conditionally_excluded_vertices: set[str] = set()  # 被条件排除的顶点
self.conditional_exclusion_sources: dict[str, set[str]] = {}  # 排除源映射

工作模式：

典型场景：

• 情感分析 LLM 判断用户意图为”咨询”或”投诉”，路由到不同的处理 Agent
• 代码审查 Agent 仅在检测到问题时触发”修复建议”子图

3. 循环图处理

当 Flow 中存在循环（如 Agent 反复尝试直到成功），Langflow 通过以下机制防止无限循环：

async def async_start(self, max_iterations: int | None = None, ...):
    yielded_counts = defaultdict(int)  # 记录每个顶点的执行次数

    while should_continue(yielded_counts, max_iterations):
        result = await self.astep(...)
        yield result
        yielded_counts[result.vertex.id] += 1

循环检测：

检测算法：find_all_cycle_edges() 使用 DFS 后序遍历检测回边，标记所有参与循环的顶点。

4. 预构建 Agent 模板

Langflow 的 agentic/flows/ 模块提供预构建的 Agent 模板：

模板	文件	用途
Langflow Assistant	LangflowAssistant.json	帮助用户搜索组件、构建 Flow 的助手
System Message Generator	SystemMessageGen.json	根据上下文生成系统提示
Translation Flow	TranslationFlow.json	多语言翻译 Agent

这些模板作为 Starter Projects 内置，用户加载后可直接使用或修改。

---

问题与规避

陷阱	现象	对策
循环图未设置 max_iterations	运行时报错 “You must specify a max_iterations if the graph is cyclic”	对包含循环的 Flow 必须设置最大迭代次数
条件路由遗漏	某个条件下的下游顶点既未被执行也未被排除	确保 If-Else 的所有分支都有对应的输出连线
子图合并后 ID 冲突	两个子图使用相同的组件 ID 导致覆盖	每个组件实例使用唯一 ID（nanoid 生成）
Agent 模板过时	组件名称或输入已变更，模板无法加载	组件变更时更新 starter_projects/ 中的 JSON

---

设计取舍

图合并 vs 子工作流引用：

维度	图合并（Graph.__add__）	子工作流引用
执行模型	合并为单一图，一次拓扑排序	子图独立执行，主图调用结果
状态共享	顶点间可直接传递数据	通过输入/输出接口通信
复杂度	大图的性能开销高	子图可独立缓存
适用场景	紧密协作的多 Agent，需要数据共享	松耦合的多 Agent，各司其职

Langflow 同时支持两种模式：画布内直接连线使用图合并，通过 SubFlow 组件引用其他 Flow 使用子工作流模式。

---

参考来源

• 源码：src/lfx/src/lfx/graph/graph/base.py
• 源码：src/backend/base/langflow/agentic/flows/
• 源码：src/lfx/src/lfx/components/flow_controls/