Flows Flows — Rust examples backed by real provider calls. rust examples examples/flows src/examples/rust/flows example Flows
ax
Rust axllm
TypeScript Python Java C++ Go Rust

These Rust examples are real runnable files. Edit the source file first; this page is rebuilt from the checked-in example and its metadata header.

Rust Sequential Flow

Runs a two-step Ax flow against OpenAI.

Rust 
use axllm::{ax, flow, AxResult, OpenAICompatibleClient};
use serde_json::json;
use std::env;


fn openai_client() -> AxResult<OpenAICompatibleClient> {
    let api_key = env::var("OPENAI_API_KEY").or_else(|_| env::var("OPENAI_APIKEY")).map_err(|_| axllm::AxError::runtime("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example."))?;
    let model = env::var("AX_OPENAI_MODEL").unwrap_or_else(|_| "gpt-5.4-mini".to_string());
    Ok(OpenAICompatibleClient::new(api_key, model).with_model_config(json!({"temperature": 0})))
}

fn main() -> AxResult<()> {
    let mut client = openai_client()?;
    let step = ax("documentText:string -> summaryText:string")?;
    let mut program = axllm::flow("examples.sequentialFlow").execute("step", step).returns(json!({"step": "step"}));
    let output = program.forward(&mut client, json!({"documentText": "Ax gives developers signatures, provider clients, agents, flows, tracing, and optimization."}))?;
    println!("{}", serde_json::to_string_pretty(&output)?);
    Ok(())
}

Rust Branching Flow

Routes a classification through follow-up flow logic backed by OpenAI.

Rust 
use axllm::{ax, flow, AxResult, OpenAICompatibleClient};
use serde_json::json;
use std::env;

fn openai_client() -> AxResult<OpenAICompatibleClient> {
    let api_key = env::var("OPENAI_API_KEY")
        .or_else(|_| env::var("OPENAI_APIKEY"))
        .map_err(|_| {
            axllm::AxError::runtime("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.")
        })?;
    let model = env::var("AX_OPENAI_MODEL").unwrap_or_else(|_| "gpt-5.4-mini".to_string());
    Ok(OpenAICompatibleClient::new(api_key, model).with_model_config(json!({"temperature": 0})))
}

fn main() -> AxResult<()> {
    let mut client = openai_client()?;
    let classifier = ax("request:string -> route:class \"support, sales, engineering\"")?;
    let responder = ax("request:string, route:string -> response:string")?;
    let mut program = flow("examples.branchFlow")
        .execute_with_options(
            "classifier",
            classifier,
            &json!({"reads": ["request"], "writes": ["classifierResult", "route"]}),
        )
        .execute_with_options(
            "responder",
            responder,
            &json!({
                "reads": ["request", "route"],
                "writes": ["responderResult", "response"]
            }),
        )
        .returns(json!({"route": "route", "response": "response"}));
    let output = program.forward(
        &mut client,
        json!({"request": "A customer says checkout is down for their enterprise account."}),
    )?;
    println!("{}", serde_json::to_string_pretty(&output)?);
    Ok(())
}

Rust Parallel Flow

Runs two independent OpenAI-backed steps in parallel before joining their results.

Rust 
use axllm::{ax, flow, AxResult, OpenAICompatibleClient};
use serde_json::json;
use std::env;

fn openai_client() -> AxResult<OpenAICompatibleClient> {
    let api_key = env::var("OPENAI_API_KEY")
        .or_else(|_| env::var("OPENAI_APIKEY"))
        .map_err(|_| {
            axllm::AxError::runtime("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.")
        })?;
    let model = env::var("AX_OPENAI_MODEL").unwrap_or_else(|_| "gpt-5.4-mini".to_string());
    Ok(OpenAICompatibleClient::new(api_key, model).with_model_config(json!({"temperature": 0})))
}

fn main() -> AxResult<()> {
    let mut client = openai_client()?;
    let research = ax("topicText:string -> factList:string[]")?;
    let audience = ax("topicText:string -> audienceAngle:string")?;
    let join = ax("factList:string[], audienceAngle:string -> briefText:string")?;
    let mut program = flow("examples.parallelFlow")
        .execute_with_options(
            "research",
            research,
            &json!({"reads": ["topicText"], "writes": ["researchResult", "factList"]}),
        )
        .execute_with_options(
            "audience",
            audience,
            &json!({"reads": ["topicText"], "writes": ["audienceResult", "audienceAngle"]}),
        )
        .execute_with_options(
            "join",
            join,
            &json!({
                "reads": ["factList", "audienceAngle"],
                "writes": ["joinResult", "briefText"]
            }),
        )
        .returns(json!({"briefText": "briefText"}));
    let output = program.forward(
        &mut client,
        json!({"topicText": "Why typed contracts make multi-step LLM systems easier to maintain"}),
    )?;
    println!("{}", serde_json::to_string_pretty(&output)?);
    Ok(())
}

Rust Controlled Background Flow

Uses ordinary generation with background tools, steering, and a reasoning update.

Rust 
use axllm::{ai, ax, flow, run_control, tool, AxForwardOptions, AxResult};
use serde_json::json;
use std::{
    env,
    sync::{
        atomic::{AtomicBool, AtomicUsize, Ordering},
        Arc,
    },
    time::Duration,
};

fn main() -> AxResult<()> {
    let key = env::var("OPENAI_API_KEY")
        .or_else(|_| env::var("OPENAI_APIKEY"))
        .expect("Set OPENAI_API_KEY or OPENAI_APIKEY.");
    let mut client = ai(
        "openai",
        json!({"api_key":key,"model":"gpt-6-astra","model_config":{"thinkingTokenBudget":"low","max_tokens":4096}}),
    )?;
    let pending = Arc::new(AtomicBool::new(false));
    let finished = Arc::new(AtomicBool::new(false));
    let overlap = Arc::new(AtomicBool::new(false));
    let applied = Arc::new(AtomicUsize::new(0));
    let control = run_control();
    let steering = control.clone();
    let updates = applied.clone();
    control.on_event(move |event| {
        if event["type"] == "applied" {
            updates.fetch_add(1, Ordering::SeqCst);
        }
    });
    let started = pending.clone();
    let done = finished.clone();
    let steered = AtomicBool::new(false);
    let slow = tool("slow_reference")
        .description("Look up a reference; takes a few seconds.")
        .execution("background")
        .handler(move |_| {
            started.store(true, Ordering::SeqCst);
            if !steered.swap(true, Ordering::SeqCst) {
                steering.steer("Include the word VERIFIED in the final answer.")?;
                steering.set_thinking_token_budget("medium")?;
            }
            std::thread::sleep(Duration::from_secs(6));
            done.store(true, Ordering::SeqCst);
            Ok(json!("REF-42"))
        });
    let observed = overlap.clone();
    let label = tool("local_label")
        .description("Read an independent local label immediately.")
        .handler(move |_| {
            for _ in 0..300 {
                if pending.load(Ordering::SeqCst) {
                    break;
                }
                std::thread::sleep(Duration::from_millis(10));
            }
            observed.fetch_or(
                pending.load(Ordering::SeqCst) && !finished.load(Ordering::SeqCst),
                Ordering::SeqCst,
            );
            Ok(json!("LAUNCH"))
        });
    let mut program = ax("question -> answer")?.with_tool(slow).with_tool(label);
    let mut workflow=flow("astra-flow").execute_with_options("lookup",program,&json!({"writes":["answer"]})).execute_with_options("verify",ax(r#"answer -> report "Repeat the exact reference, label, and verification word from the answer.""#)?,&json!({"reads":["answer"]})).returns(json!({"answer":"report"}));
    let result=workflow.forward_with_options(&mut client,json!({"question":"First call slow_reference. While it is pending, call local_label. Call each tool only once; do not call a tool again while its result is pending. If a required tool result is still pending, end this response with a brief progress message. The application will continue with the result when it arrives; do not spend reasoning tokens waiting for it. Once both results arrive, return them in one sentence."}),AxForwardOptions::from(json!({"serviceTier":"standard","maxSteps":6})).with_control(control))?;
    let answer = result.to_string();
    for word in ["REF-42", "LAUNCH", "VERIFIED"] {
        assert!(answer.contains(word), "Missing final result: {answer}");
    }
    assert!(
        overlap.load(Ordering::SeqCst),
        "No independent work while background tool was pending"
    );
    assert_eq!(applied.load(Ordering::SeqCst), 4);
    println!("{answer}\nBackground overlap verified; steering and reasoning applied at the next response.");
    Ok(())
}

Rust Concurrent Astra Flow

Independent conversations overlap, retain their tool results, and receive scoped controls.

Rust 
use axllm::{ai, ax, flow, run_control, tool, AxForwardOptions, AxResult};
use serde_json::{json, Value};
use std::{collections::BTreeSet, env, sync::{Arc, Mutex, Condvar}, time::Duration};

fn main()->AxResult<()> {
    let key=env::var("OPENAI_API_KEY").or_else(|_|env::var("OPENAI_APIKEY")).expect("Set OPENAI_API_KEY or OPENAI_APIKEY.");
    let mut client=ai("openai",json!({"api_key":key,"model":"gpt-6-astra","model_config":{"thinkingTokenBudget":"low","max_tokens":4096}}))?;
    let control=run_control();let paths=Arc::new(Mutex::new(BTreeSet::new()));let applied=Arc::new(Mutex::new(Vec::<Value>::new()));
    let observed_paths=paths.clone();let observed_updates=applied.clone();
    control.on_event(move |event|{if event["type"]=="tool.started"{observed_paths.lock().unwrap().insert(event["path"].as_str().unwrap().to_owned());}if event["type"]=="applied"{observed_updates.lock().unwrap().push(event);}});
    let gate=Arc::new((Mutex::new(0usize),Condvar::new()));let steering=control.clone();
    let lookup=tool("lookup").description("Look up the exact reference once.").execution("background").handler(move |_|{
        let (mutex,ready)=&*gate;let mut calls=mutex.lock().unwrap();*calls+=1;assert!(*calls<=2,"Lookup was called more than once per node");
        if *calls==2{steering.steer("Include VERIFIED with the exact reference in your final answer.")?;steering.set_thinking_token_budget_at("medium","root/left")?;ready.notify_all();}
        let (calls,timeout)=ready.wait_timeout_while(calls,Duration::from_secs(45),|calls|*calls<2).unwrap();assert!(!timeout.timed_out()&&*calls==2,"Independent nodes did not overlap");Ok(json!("REF-42"))
    });
    let program=ax("question -> answer")?.with_tool(lookup);
    let mut workflow=flow("parallel").execute("left",program.clone()).execute("right",program).returns(json!({"left":"leftResult","right":"rightResult"}));
    let result=workflow.forward_with_options(&mut client,json!({"question":"Call lookup exactly once. If its result is pending, return a brief progress message without calling it again. Return the exact reference when its result arrives."}),AxForwardOptions::from(json!({"serviceTier":"standard","maxSteps":6})).with_control(control))?;
    assert_eq!(*paths.lock().unwrap(),BTreeSet::from(["root/left".to_owned(),"root/right".to_owned()]));assert_eq!(applied.lock().unwrap().len(),3);
    for node in ["left","right"]{let answer=result[node].to_string();assert!(answer.contains("REF-42")&&answer.contains("VERIFIED"),"Missing final result: {answer}");}
    println!("{}",json!({"result":result,"parallel_overlap":true,"applied_controls":*applied.lock().unwrap()}));Ok(())
}

Rust Composed Flow

Composes multiple typed programs into one OpenAI-backed flow.

Rust 
use axllm::{ax, flow, AxResult, OpenAICompatibleClient};
use serde_json::json;
use std::env;


fn openai_client() -> AxResult<OpenAICompatibleClient> {
    let api_key = env::var("OPENAI_API_KEY").or_else(|_| env::var("OPENAI_APIKEY")).map_err(|_| axllm::AxError::runtime("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example."))?;
    let model = env::var("AX_OPENAI_MODEL").unwrap_or_else(|_| "gpt-5.4-mini".to_string());
    Ok(OpenAICompatibleClient::new(api_key, model).with_model_config(json!({"temperature": 0})))
}

fn main() -> AxResult<()> {
    let mut client = openai_client()?;
    let step = ax("topic:string -> outline:string[]")?;
    let mut program = axllm::flow("examples.composedFlow").execute("step", step).returns(json!({"step": "step"}));
    let output = program.forward(&mut client, json!({"topic": "How Ax moves from typed generation to agents, flows, and optimization"}))?;
    println!("{}", serde_json::to_string_pretty(&output)?);
    Ok(())
}

Rust Refinement Flow

Drafts, critiques, and revises an answer through three OpenAI-backed steps.

Rust 
use axllm::{ax, flow, AxResult, OpenAICompatibleClient};
use serde_json::json;
use std::env;

fn openai_client() -> AxResult<OpenAICompatibleClient> {
    let api_key = env::var("OPENAI_API_KEY")
        .or_else(|_| env::var("OPENAI_APIKEY"))
        .map_err(|_| {
            axllm::AxError::runtime("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.")
        })?;
    let model = env::var("AX_OPENAI_MODEL").unwrap_or_else(|_| "gpt-5.4-mini".to_string());
    Ok(OpenAICompatibleClient::new(api_key, model).with_model_config(json!({"temperature": 0})))
}

fn main() -> AxResult<()> {
    let mut client = openai_client()?;
    let draft = ax("topicText:string -> draftText:string")?;
    let critique = ax("draftText:string -> critiqueText:string")?;
    let revise = ax("draftText:string, critiqueText:string -> revisedText:string")?;
    let mut program = flow("examples.refineFlow")
        .execute_with_options(
            "draft",
            draft,
            &json!({"reads": ["topicText"], "writes": ["draftResult", "draftText"]}),
        )
        .execute_with_options(
            "critique",
            critique,
            &json!({"reads": ["draftText"], "writes": ["critiqueResult", "critiqueText"]}),
        )
        .execute_with_options(
            "revise",
            revise,
            &json!({
                "reads": ["draftText", "critiqueText"],
                "writes": ["reviseResult", "revisedText"]
            }),
        )
        .returns(json!({"revisedText": "revisedText"}));
    let output = program.forward(
        &mut client,
        json!({"topicText": "Explain automatic flow parallelism to a backend engineer."}),
    )?;
    println!("{}", serde_json::to_string_pretty(&output)?);
    Ok(())
}
Docs
Start
Quick Start How Ax Fits Together Advanced Start FAQ
Agents
Overview Micro Agents Standard Agents Long-Horizon Agents Performance Internals
Examples
Overview Generation Agents Long-Horizon Agents Flows Audio MCP Optimization
Concepts
DSPy Signatures Tools MCP MCP Subscriptions Event Runtime Optimization Playbook LLMs Telemetry
Subsystems
ai() LLM Models ax() Generation s() Signatures agent() Agents flow() Workflows optimize() GEPA
Skills
Overview Quick Reference AI Typesafe / Jev Audio Signatures Generation Flow Agent Agent RLM Memory & Skills Observability Agent Optimize Agent Context GEPA Refinement Playbook
API Docs
ai() API ax() API s() API agent() API flow() API optimize() API