Flow 流程编排详解

本文是 Flow 子系统的完整开发参考。Flow 是 Familiars 的"多 agent 工作流引擎"：以主 agent 轨道为主干，若干 worker 轨道通过跨角色 jump 拉起/跳回，适合跨角色协作类任务。

1. 概念模型

1.1 什么是 Flow

Flow（流程） 是一条以「主 agent 轨道」为主干的多 agent 工作流：

[Starter] ─ Phase1 ─ Phase2 ─ … ─ PhaseN ─ [Summary]
     │ jump ▲ jump
     ▼      │
[子 agent 轨道 A：unit ─ unit ─ jump back to main]

主轨道（main）：首尾固定是 Starter 模块 和 Summary 模块，中间是若干 Phase（checkpoint）；
Worker 轨道：通过主轨道的 jump 拉起；每个 worker 跑完自己的 unit 序列后，通过 jump 跳回主轨道某个锚点；
Trigger：主轨道在两个 jump 之间可以等待某个条件（例如某 worker 完成）。

1.2 Template vs Instance

FlowTemplate（创作期）：作者写的静态图 + 单元/检查点/触发器定义。磁盘文件在 workspace/manager/flows/<template_id>/template.json。
FlowInstance（运行期）：一次具体的执行。Instance 在启动时 pin 模板的当前版本（pinned_template_json），后续即使作者改了模板也不影响这个 instance。
Mutation log：运行期对图的改写（加 unit、rewire、expand_repeat 等）以 mutation 形式记录在 SQLite flow_mutations 表；resume 时重放 mutation + hash 校验，确保 landing 状态一致。

1.3 三大不变量（engine.py）

只有 owner 写。Worker 只 post message，owner 把 message 折叠进 FlowState 并应用 mutation。唯一例外是 role status，由 Scheduler 在自己的锁下写（agent_state 谓词必须立刻观察到）。
图冻结不撕裂。rewrite.apply_spec 复制+校验+owner 原子替换引用。Worker 拿着的是完整一致的图，可能陈旧但不会半重写。
陈旧靠 catch 不靠 prevent。Worker 在每个 unit 前比较 generation，stale 就 reseat。Worker 发起的 mutation 同步 acknowledge，请求者不会走过刚要求改写的节点。

2. 目录与文件

2.1 源码结构

src/flow/
├── engine.py             # FlowEngine：owner thread + 每 role worker thread
├── resume.py             # rehydrate()：从 mutation log + pinned template 重建 engine
├── rewrite.py            # apply_and_validate：mutation → 新图（单源真相）
├── state.py              # FlowState：run state（dotted-path 写）
├── bus.py                # FlowEventBus
├── cancel.py             # CancelToken + Cancelled
├── compile.py            # 模板 → 运行时图编译
├── trace.py              # FlowTrace（事件流）
├── scheduler.py          # RoleCursor + Scheduler
├── template_store.py     # 模板磁盘 store
├── packaging_cli.py      # CLI 打包
├── adapters/             # legacy pipeline 适配
├── bus.py                # 事件总线
├── checkpoints/          # checkpoint 处理器
├── expr/                 # 表达式求值（AST 白名单）
├── models/               # ★ 数据模型
│   ├── template.py       # FlowTemplate / RendererDecl / FLOW_SCHEMA_VERSION
│   ├── graph.py          # FlowGraph + graph_hash（canonical）
│   ├── instance.py       # FlowInstance
│   ├── mutation.py       # MutationSpec
│   ├── unit.py           # WorkUnit
│   ├── workflow.py       # AgentWorkflow
│   ├── checkpoint.py     # Checkpoint (V1)
│   ├── checkpoint_v2.py  # CheckpointV2 + JumpTarget
│   ├── trigger.py        # Trigger
│   ├── event.py          # FlowEvent / FlowEventType
│   ├── field_spec.py     # FieldSpec（IO 字段）
│   └── workflow.py       # AgentWorkflow
├── render/               # .vue 渲染器编译
├── units/                # unit 类型
│   ├── agent_unit.py     # agent unit（LLM 调用）
│   └── tool_unit.py      # tool unit
└── _builtin/             # ★ 内置 flow 模板（只读）
    ├── hello_world/      # 最小跨 agent demo
    │   ├── template.json
    │   ├── layout.json
    │   └── README.md
    └── weekly_report/    # 多角色周报协作
        ├── template.json
        ├── layout.json
        └── README.md

2.2 运行时磁盘布局

每个自定义 flow 模板：

workspace/manager/flows/<template_id>/
├── template.json           # FlowTemplate（核心）
├── layout.json             # V2 track/module/variable 布局
├── docs/                   # Per-flow 文档（DESIGN.md / DEVELOPMENT.md / *.md）
│   ├── DESIGN.md
│   └── DEVELOPMENT.md
├── renderers/              # 自定义 .vue 渲染器
│   └── *.vue
├── contract.md             # 可选契约说明
└── README.md               # 可选

2.3 内置模板

install_built_ins() 启动时注册，前端"系统"分组，只读：

ID	路径	说明
`hello_world`	`src/flow/_builtin/hello_world/`	最小跨 agent 协作 demo：starter → dispatch jump 到 agent1 → agent1 跑完 jump 回 summary
`weekly_report`	`src/flow/_builtin/weekly_report/`	多角色协作：协调者 → 数据采集员 → 撰写员 → 审核员（条件分支回退）→ 汇总

3. Schema 详解

3.1 `FlowTemplate`（template.json 顶层）

python

class FlowTemplate(BaseModel):
    template_id: str                       # [a-z0-9_-]{1,64}
    name: str = ""
    schema_version: int = FLOW_SCHEMA_VERSION  # 当前 = 3
    template_version: int = 1              # 作者每次编辑 +1
    main_agent_description: str = ""       # 主 agent 职责描述
    main_recipe_id: str = ""               # 主 agent 绑定的 recipe
    main_agent_id: Optional[int] = None
    workflows: list[AgentWorkflow]         # ★ 每个 role 的轨道
    units: dict[str, WorkUnit]             # 所有 unit（id → unit）
    checkpoints: dict[str, Checkpoint | CheckpointV2]  # V1/V2 共存
    triggers: dict[str, Trigger]
    renderers: list[RendererDecl]          # 自定义 .vue 渲染器
    contract_md: str = ""
    created_at: float
    updated_at: float

3.2 `AgentWorkflow`（一个 role 的轨道）

python

class AgentWorkflow(BaseModel):
    role: str                              # 角色名（"main" / "agent1" / "collector" / ...）
    title: str = ""
    planning: {"strategy": "fixed", "needs_main_agent": bool}
    entry: str                             # 入口 node_id
    nodes: list[WorkflowNode]              # 节点序列

3.3 `WorkflowNode`（轨道里的节点）

python

class WorkflowNode(BaseModel):
    node_id: str
    node_kind: str                         # starter | summary | checkpoint | unit | trigger
    ref: str                               # 指向 checkpoint_id / unit_id / trigger_id
    next: list[str]                        # 后继 node_id

3.4 `WorkUnit`（执行单元）

python

class WorkUnit(BaseModel):
    unit_id: str
    kind: str                              # "agent" | "tool"
    owner_role: str
    title: str = ""
    instruction: str                       # ★ 给 agent 的指令（system prompt 的一部分）
    input: {"fields": {name: type_str}}
    output: {"fields": {name: type_str}}
    tools: list[str]                       # 工具白名单
    subflow_template_id: str = ""          # 子 flow 模板
    input_fields: list[FieldSpec] = []     # starter default [prompt, attachments]
    output_fields: list[FieldSpec] = []    # summary default [answer, attachments]

3.5 `CheckpointV2`（决策点，含分支/jump）

python

class CheckpointV2(BaseModel):
    checkpoint_id: str
    title: str = ""
    sub_modules: list[SubModule]           # ★ 分支 / 跳转 / 容器
    blocking: bool = True
    input_fields: list[FieldSpec] = []

SubModule 类型：

type	含义
`branch`	条件分支（case list + branch_bodies）
`jump`	跨 role 跳转（target: {track_role, module_id, anchor}）
`parallel_group`	并行组（容纳并排 chain）
`combination_group`	组合组（强制 head/tail 相邻）
`chain`	顺序链（一组顺序执行的节点）

分支体存储位置（V2 关键）：每个 case 的分支体放在 branch_bodies[case.id]，else 分支放在 branch_bodies["__else__"]。不是 case.modules 或 else_modules（已废弃）。迁移是幂等的。

3.6 `Trigger`（等待条件）

python

class Trigger(BaseModel):
    trigger_id: str
    title: str = ""
    kind: str                              # "agent_state" | "unit_state" | ...
    params: dict

agent_state 是引擎唯一自动注入的 trigger：当某 role 状态变成 done 时触发。

3.7 `FieldSpec`（IO 字段）

python

class FieldSpec(BaseModel):
    name: str
    type: str = "str"
    required: bool = True
    description: str = ""

Starter 默认 input_fields = [prompt, attachments]；Summary 默认 output_fields = [answer, attachments]。这些是保留名，自定义字段不能同名。

3.8 `RendererDecl`（自定义渲染器）

python

class RendererDecl(BaseModel):
    name: str
    source_file: str                       # 相对 renderers/ 的 .vue 路径
    description: str = ""

保存时后端 src.flow.render.compiler 预编译为 ESM JS，前端按名挂载。

4. REST API

4.1 Flow Templates CRUD（`/api/flow-templates`）

Method	Path	说明
GET	`/api/flow-templates`	列所有模板（summary rows）
POST	`/api/flow-templates`	新建
GET	`/api/flow-templates/{id}`	取完整 body
PUT	`/api/flow-templates/{id}`	保存（validate + persist）
DELETE	`/api/flow-templates/{id}`	删除（有 instance 引用时拒）
POST	`/api/flow-templates/{id}/validate`	dry-run 校验
POST	`/api/flow-templates/{id}/duplicate`	复制

4.2 Per-flow 文档（`/api/flow-templates/{id}/docs`）

Method	Path	说明
GET	`/{id}/docs`	列 docs/ 下所有 markdown
GET	`/{id}/docs/{name}`	读一个文档（DESIGN / DEVELOPMENT / 自定义）
PUT	`/{id}/docs/{name}`	写一个文档（最大 256KB）
DELETE	`/{id}/docs/{name}`	删

DESIGN.md：该 flow 的设计文档（无则返回模板骨架）；
DEVELOPMENT：特殊 —— 永远指向根 FLOW_DEVELOPMENT.md（framework doc），is_global: true，前端编辑前会警告；
名字白名单 ^[A-Z0-9_-]{1,32}$，自动大写。

4.3 Bundle 协议（V2 track layout，schema_version=3）

Method	Path	说明
POST	`/api/flow-templates/preview_bundle`	dry-run 校验 bundle
POST	`/api/flow-templates/import_bundle`	上传 .flow.zip 或 folder 路径
GET	`/{id}/export_bundle`	下载 .flow.zip

Bundle = 包含 manifest.json + template.json + layout.json (+ contract.md / README.md / schemas/) 的目录。Layout 是 V2 track/module/variable 层；template 是 V1 pool 层。两者必须交叉校验。

4.4 渲染器

Method	Path	说明
GET	`/{id}/renderers`	列渲染器
GET	`/{id}/renderers/{name}`	读 .vue 源码
POST	`/{id}/renderers/{name}`	写 .vue 源码
DELETE	`/{id}/renderers/{name}`	删

4.5 Flow Instances（`/api/flows`）

Method	Path	说明
GET	`/api/flows`	列实例（支持 status/template_id 过滤）
GET	`/api/flows/{id}`	取一个实例 + 最近 trace_tail 事件
GET	`/api/flows/{id}/trace`	取完整 trace（事后分析）
POST	`/api/flows/{id}/cancel`	取消（幂等）

启动 flow instance 走 src.pipeline.Generation._run_flow_recipe_chain，不是这些 route。

5. FlowEngine 运行期

5.1 启动序列

python

engine = FlowEngine(
    graph=template.to_graph(),
    trace=trace,
    repo=repo,
    instance_id=instance_id,
    fields={"prompt": "...", "attachments": [...]},
    executor=default_executor,  # src.flow.units.dispatch
    ...
)
engine.run()

5.2 Thread 模型

Owner thread：调用 run() 的线程，处理 message queue、写 FlowState、应用 mutation。
Worker thread per role：每个 AgentWorkflow 一个 worker。Worker 从 Scheduler 拿下一个 node，调用 unit executor（agent_unit / tool_unit），把结果 post 回 owner。

5.3 Mutation 类型

kind	含义	是否重写图
`add_unit`	加 unit	是
`remove_unit`	删 unit	是
`rewire`	重连边	是
`replace_workflow`	替换整个 role 的 workflow	是
`expand_repeat`	展开 repeat 模块	是
`set_config`	改 config	是
`set_field`	写 run state（dotted path）	否

5.4 状态机

pending → running → done | failed
              ↓
         cancelled

pending：instance 创建但未启动
running：engine.run() 在执行
done：所有 role 都完成（cursor.ended）
failed：未捕获异常 / Cancelled
cancelled：CancelToken 触发

5.5 Resume（`src/flow/resume.py`）

rehydrate(instance_id, repo) 读 pinned_template_json 重建图，按 seq 顺序重放 flow_mutations，每步的 result_graph_hash 与 apply_and_validate 计算结果比对；不一致抛 FlowResumeError。

新 instance（无 mutation）→ 直接返回 base graph 的 engine（等价于新跑）
set_field 重放：写 FlowState，不影响图 hash

6. 表达式求值

src/flow/expr/ 提供 AST 白名单求值（与 Recipe 共享语法）：

支持：Python 比较/布尔运算符、变量引用、len()、字面量
禁止：函数调用（除白名单）、attribute access（除安全子集）、import、class 定义
边缘 condition= 都通过此求值

7. 常见坑

坑	症状	解决
并行组节点互相等待	死锁	保证依赖方向单向，避免循环等待
分支体位置错	前端看不到分支 / 后端拒加载	用 `branch_bodies[case.id]` / `branch_bodies["__else__"]`
模板被改后续跑	hash 不匹配	resume 拒绝续跑；必须先 pin 版本
VueFlow 节点组件不 markRaw	"Component made reactive" 警告 + 点击失效	`markRaw(nodeTypes)`
VueFlow 自定义 edge 两 root	每渲染都报 "Extraneous non-props attributes"	用 `<g>` 单根包 BaseEdge + EdgeLabelRenderer
`watch(() => store.graph, ..., {deep: true})`	渲染循环（全屏卡）	改用 fingerprint string watcher
Starter 字段名被占	校验报错	不用 `prompt` / `attachments` / `answer` 作为自定义字段名

8. 开发流程（端到端）

在流编辑器（CreativeWorkshop）新建模板 → 拖节点/连线；
添加 Starter 头（自动建）、Summary 尾（自动建）、中间加 workflow/checkpoint/trigger；
配置各 node 参数：
- unit：instruction、input/output fields、tools 白名单
- checkpoint：分支条件（AST 表达式）、jump target（track_role + module_id）
- trigger：等待条件（agent_state / unit_state）
写 docs/DESIGN.md（设计）+ docs/DEVELOPMENT.md（开发笔记）；
POST /api/flow-templates/{id}/validate 校验；
绑到 agent → 启动 instance（chain 自动跑）；
在 /api/flows/{id} 看 trace；
中断可 rehydrate 续跑（模板被改则拒续）。

9. 与 Recipe 的关系

Recipe 是"单 agent 单轮的状态图"；Flow 是"多 agent 跨轮的工作流"；
Flow 的 unit（agent unit）内部会调 Recipe（flow_template_id 绑到 agent 后，unit 跑的就是该 recipe）；
Flow 通过 filtered_schemas 把工具暴露给 agent，不走 tools.json；
Flow 编辑器的"编译/干跑"按钮走 POST /api/flow-templates/{id}/validate；
Flow 模板的 docs/ 会作为前端 DESIGN / DEVELOPMENT 分页展示。

参考源码

模板模型：src/flow/models/template.py
图模型：src/flow/models/graph.py
Engine：src/flow/engine.py
Resume：src/flow/resume.py
REST（templates）：src/bridge/routes/flow_templates.py
REST（instances）：src/bridge/routes/flows.py
内置：src/flow/_builtin/{hello_world,weekly_report}/

Flow 流程编排详解 ​

1. 概念模型 ​

1.1 什么是 Flow ​

1.2 Template vs Instance ​

1.3 三大不变量（engine.py） ​

2. 目录与文件 ​

2.1 源码结构 ​

2.2 运行时磁盘布局 ​

2.3 内置模板 ​

3. Schema 详解 ​

3.1 FlowTemplate（template.json 顶层） ​

3.2 AgentWorkflow（一个 role 的轨道） ​

3.3 WorkflowNode（轨道里的节点） ​

3.4 WorkUnit（执行单元） ​

3.5 CheckpointV2（决策点，含分支/jump） ​

3.6 Trigger（等待条件） ​

3.7 FieldSpec（IO 字段） ​

3.8 RendererDecl（自定义渲染器） ​

4. REST API ​

4.1 Flow Templates CRUD（/api/flow-templates） ​

4.2 Per-flow 文档（/api/flow-templates/{id}/docs） ​

4.3 Bundle 协议（V2 track layout，schema_version=3） ​

4.4 渲染器 ​

4.5 Flow Instances（/api/flows） ​

5. FlowEngine 运行期 ​

5.1 启动序列 ​

5.2 Thread 模型 ​

5.3 Mutation 类型 ​

5.4 状态机 ​

5.5 Resume（src/flow/resume.py） ​

6. 表达式求值 ​

7. 常见坑 ​

8. 开发流程（端到端） ​

9. 与 Recipe 的关系 ​

参考源码 ​

Flow 流程编排详解

1. 概念模型

1.1 什么是 Flow

1.2 Template vs Instance

1.3 三大不变量（engine.py）

2. 目录与文件

2.1 源码结构

2.2 运行时磁盘布局

2.3 内置模板

3. Schema 详解

3.1 `FlowTemplate`（template.json 顶层）

3.2 `AgentWorkflow`（一个 role 的轨道）

3.3 `WorkflowNode`（轨道里的节点）

3.4 `WorkUnit`（执行单元）

3.5 `CheckpointV2`（决策点，含分支/jump）

3.6 `Trigger`（等待条件）

3.7 `FieldSpec`（IO 字段）

3.8 `RendererDecl`（自定义渲染器）

4. REST API

4.1 Flow Templates CRUD（`/api/flow-templates`）

4.2 Per-flow 文档（`/api/flow-templates/{id}/docs`）

4.3 Bundle 协议（V2 track layout，schema_version=3）

4.4 渲染器

4.5 Flow Instances（`/api/flows`）

5. FlowEngine 运行期

5.1 启动序列

5.2 Thread 模型

5.3 Mutation 类型

5.4 状态机

5.5 Resume（`src/flow/resume.py`）

6. 表达式求值

7. 常见坑

8. 开发流程（端到端）

9. 与 Recipe 的关系

参考源码