Graphs and state

​

What it is

• State — a struct of yours that travels between nodes. You implement GraphState, which says how an Update is applied to the state.
• Nodes — async functions that read the state and emit NodeOut { update, goto }.
• A start node — declared with .start_at("name"). The engine walks edges from there until a node returns Goto::end().

​

When to reach for a graph

• You need a tool-calling loop with custom termination logic.
• You want time travel — checkpoint after every step, rewind, branch.
• You need human-in-the-loop pauses at specific points.
• Branches need to run in parallel and rejoin.
• You want to subgraph and namespace state.

​

Quick example

use cognis::prelude::*;

#[derive(Default, Clone, Debug)]
struct State { count: u32 }

#[derive(Default, Clone)]
struct Update { count: u32 }

impl GraphState for State {
    type Update = Update;
    fn apply(&mut self, u: Update) { self.count += u.count; }
}

#[tokio::main]
async fn main() -> Result<()> {
    let tick = node_fn::<State, _, _>("tick", |s, _ctx| {
        let cur = s.count;
        async move {
            if cur >= 5 {
                Ok(NodeOut { update: Update { count: 0 }, goto: Goto::end() })
            } else {
                Ok(NodeOut { update: Update { count: 1 }, goto: Goto::node("tick") })
            }
        }
    });

    let graph = Graph::<State>::new()
        .node("tick", tick)
        .start_at("tick")
        .compile()?;

    let final_state = graph.invoke(State::default(), RunnableConfig::default()).await?;
    println!("count: {}", final_state.count);
    Ok(())
}

​

State and reducers

#[derive(Default, Clone, Debug)]
struct State {
    messages: Vec<Message>,
    counter: u32,
}
#[derive(Default, Clone)]
struct Update {
    messages: Vec<Message>,
    counter: u32,
}
impl GraphState for State {
    type Update = Update;
    fn apply(&mut self, u: Update) {
        self.messages.extend(u.messages);
        self.counter += u.counter;
    }
}

use cognis::prelude::*;

#[derive(Default, Clone, Debug, GraphStateV2)]
struct State {
    #[reducer(append)]   messages: Vec<Message>,
    #[reducer(add)]      counter: u32,
    #[reducer(merge)]    metadata: serde_json::Value,
    // unannotated fields default to last_value
    pending: Option<ToolCall>,
}

​

Control flow

let g = Graph::<S>::new()
    .node("plan", plan)
    .node("execute", execute)
    .node("review", review)
    .edge("plan", "execute")
    .edge("execute", "review")
    .start_at("plan")
    .compile()?;

​

How it works

• Each invoke is a sequence of supersteps. The engine runs all eligible nodes in a step, applies their updates atomically, then advances.
• Goto::Multiple triggers parallel fan-out. All targets run in the same superstep.
• State is applied, not replaced. Your apply(&mut self, Update) controls how. Reducers are just apply written for one field at a time.
• Compiled graphs are Runnable<S, S>. Wrappers (with_retry, with_timeout) and observability work the same as anywhere else.

​

Beyond the basics

• Persist state across runs? Add a checkpointer. See Graph workflows → Checkpointing.
• Pause for a human? Mark interrupts. See Building agents → Human-in-the-loop.
• Inspect intermediate state? graph.get_state, get_state_history, update_state. See Graph workflows → Checkpointing.
• Stream node deltas in real time? Graph workflows → Streaming.
• Visualize? graph.to_dot(), to_mermaid(), to_ascii(). See Graph workflows → Visualization.

​

See also