Flow 与工作流

AgenticX 中的 Flow 与工作流引擎。

Flow 与工作流引擎

进程内 Python 方法要成图时用 Flow（@start / @listen / @router）。要配置驱动的图用 WorkflowEngine。两者都不替代 Near 对话轮次——那是 AgentRuntime。

示意图：kickoff() 留在你的进程里，不会写 messages.json。

实践案例：三个节点的类

场景。 进程内抓取 → 分析 → 按路由发布或复核。

继承 Flow，入口 @start，下一步 @listen，分支 @router。
调用 kickoff() / kickoff_async()。这不会启动 Near，也不会写 messages.json。
需要暂停 / 续跑时用 ExecutionPlan（to_mermaid()、pause()、resume()）。

你会看到。 脚本里是返回的状态。文档里是一张能指着讲的流水线。Near 里不会自动出现，除非你自己把 Flow 包进去。

装饰器流水线 fetch → analyze → router — 界面示意：`@start` / `@listen` / `@router` —— 进程内的图，不是聊天窗格。

概述

Layer	Module	Best for
Flow	`agenticx.flow`	Python classes with `@start` / `@listen` / `@router`
Execution plan	`agenticx.flow.execution_plan`	Staged goals, pause/resume, persistence
Graph engine	`agenticx.core.graph`	Async nodes returning next node or `End`
Workflow engine	`agenticx.core.workflow_engine`	`Workflow` / `WorkflowGraph`, concurrent branches

Flow 系统

`Flow` 基类

泛型状态类型 `T`（dict 或 Pydantic model）。
入口：`kickoff()`（同步）与 `kickoff_async()`。

装饰器

Decorator	Role
`@start()`	Unconditional entry point
`@listen(...)`	Runs when trigger condition is satisfied
`@router(...)`	Returns a string label for routing

示例：简单数据流水线

python

1from agenticx.flow import Flow, start, listen, router
2
3class DataPipeline(Flow[dict]):
4    @start()
5    def fetch_data(self):
6        return {"data": [1, 2, 3]}
7
8    @listen("fetch_data")
9    def process_data(self, result):
10        return {"processed": [x * 2 for x in result["data"]]}
11
12    @router("process_data")
13    def branch(self, result):
14        values = result.get("processed", [])
15        return "NONEMPTY" if values else "EMPTY"
16
17    @listen("NONEMPTY")
18    def on_success(self):
19        self.state["status"] = "ok"
20
21    @listen("EMPTY")
22    def on_empty(self):
23        self.state["status"] = "empty"
24
25flow = DataPipeline()
26flow.kickoff()

执行计划

`ExecutionPlan`

Field	Description
`stages`	List of `ExecutionStage`
`current_stage_index`	Index of the active stage
`intervention_state`	`InterventionState` for external control

`ExecutionStage` 与 `Subtask`

`ExecutionStage`：name、subtasks、status
`Subtask`：id、name、query、status、result/error

核心能力

`ExecutionPlan.to_mermaid()`：返回 Mermaid 图字符串
`pause()` / `resume()` / `reset_node(subtask_id)`
`overall_progress`：已完成子任务占比

图引擎

Module: agenticx.core.graph

python

1from agenticx.core.graph import Graph, BaseNode, End, GraphRunContext
2
3class StartNode(BaseNode):
4    async def run(self, ctx: GraphRunContext) -> ProcessNode | End[str]:
5        # Do work
6        return ProcessNode()
7
8class ProcessNode(BaseNode):
9    async def run(self, ctx: GraphRunContext) -> End[str]:
10        return End(result="done")
11
12graph = Graph(nodes=[StartNode, ProcessNode])
13result = await graph.run(StartNode())

WorkflowEngine

Module: agenticx.core.workflow_engine

`WorkflowEngine` 运行 `Workflow` 模型或 `WorkflowGraph`
并发：入口节点与下游扇出通过 asyncio.gather 执行
条件路由：边可编码 JSON condition_config
可观测性：ExecutionContext 持有 event_log

快速参考 import

python

1# Flow
2from agenticx.flow import (
3    Flow,
4    FlowState,
5    start,
6    listen,
7    router,
8    or_,
9    and_,
10)
11
12# Execution plan
13from agenticx.flow import (
14    ExecutionPlan,
15    ExecutionStage,
16    Subtask,
17    ExecutionPlanManager,
18)
19
20# Graph
21from agenticx.core.graph import Graph, BaseNode, End, GraphRunContext
22
23# Workflow
24from agenticx.core.workflow_engine import WorkflowEngine, WorkflowGraph