For AI agents: the complete documentation index is available at https://a3s-lab.github.io/Flow/v0.12.0/llms.txt, the full documentation bundle is available at https://a3s-lab.github.io/Flow/v0.12.0/llms-full.txt, and this page is available as Markdown at https://a3s-lab.github.io/Flow/v0.12.0/guide/index.md.

在 0.12.0 跑通第一个工作流

Cargo.toml
[dependencies]
a3s-flow = "=0.12.0"
async-trait = "0.1"
serde_json = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

完整示例

src/main.rs
use a3s_flow::{
    FlowEngine, FlowError, FlowRuntime, RuntimeCommand,
    StepInvocation, WorkflowInvocation, WorkflowSpec,
};
use async_trait::async_trait;
use serde_json::json;
use std::sync::Arc;

struct GreetingRuntime;

#[async_trait]
impl FlowRuntime for GreetingRuntime {
    async fn run_workflow(
        &self,
        invocation: WorkflowInvocation,
    ) -> a3s_flow::Result<RuntimeCommand> {
        let ctx = invocation.context();
        if let Some(output) = ctx.step_output("greet") {
            return Ok(ctx.complete(output.clone()));
        }

        Ok(ctx.schedule_step(
            "greet",
            "greet_user",
            json!({ "name": ctx.input()["name"] }),
        ))
    }

    async fn run_step(
        &self,
        invocation: StepInvocation,
    ) -> a3s_flow::Result<serde_json::Value> {
        match invocation.step_name.as_str() {
            "greet_user" => Ok(json!({
                "message": format!(
                    "hello {}",
                    invocation.input["name"]
                        .as_str()
                        .unwrap_or("unknown"),
                ),
            })),
            name => Err(FlowError::Runtime(
                format!("unknown step: {name}"),
            )),
        }
    }
}

#[tokio::main]
async fn main() -> a3s_flow::Result<()> {
    let engine = FlowEngine::in_memory(Arc::new(GreetingRuntime));
    let spec = WorkflowSpec::rust_embedded(
        "demo.greeting",
        "0.1.0",
        "demo",
        "main",
    );

    let run_id = engine
        .start_with_id(
            "greeting-ada",
            spec,
            json!({ "name": "Ada" }),
        )
        .await?;
    let snapshot = engine.snapshot(&run_id).await?;

    println!("status={:?}", snapshot.status);
    println!("output={:?}", snapshot.output);
    Ok(())
}
cargo run

第一次重放安排 greet,步骤完成事件写入后,第二次重放读取输出并完成。工作流函数没有执行网络或文件操作。

幂等创建

start_with_id() 允许调用方重试创建请求。运行 ID、WorkflowSpec 和初始输入完全一致时返回原运行。任一项变化都会得到 RunConflict

业务系统可以把订单号或任务号变成运行 ID,但一个 ID 只能长期表达同一份运行权威。

检查原始历史

let history = engine.history(&run_id).await?;
for event in history {
    println!("{} {}", event.sequence, event.event.event_key());
}

步骤输出进入历史后才可见。外部动作成功而事件尚未提交时进程退出,步骤会再次执行。步骤输入应带稳定幂等键,例如 greeting-ada:greet

可选 SQL 功能

0.12.0 使用 ORM 0.3。

a3s-flow = { version = "=0.12.0", features = ["sqlite"] }

应用如果直接使用 ORM 或 Boot,要把直接依赖同步到 0.3.0 和 0.2.0,避免同时出现两套不相同的公共类型。

可选 TypeScript 适配器

NativeTsRuntime 要求宿主提供编译器路径。

let runtime = NativeTsRuntime::new(NativeTsRuntimeConfig::new(
    "/opt/a3s/bin/a3s-flow-native-compiler",
    ".a3s/flow/native-ts",
    ".",
));

编译器要支持 compile <entrypoint.ts> -o <artifact>。0.12.0 不提供依赖清单扫描,入口导入或配置变化时必须更新 WorkflowSpec.version

下一步