Graph Workflows

Graphs are the core abstraction in FlowgentraAI. You define nodes (functions), connect them with edges, compile the graph, and invoke it with state.

Building a Graph

from flowgentra_ai import StateGraphBuilder, SharedState, END

def step_a(state):
    state["a"] = True
    return state

def step_b(state):
    state["b"] = True
    return state

builder = StateGraphBuilder()
builder.add_node("step_a", step_a)
builder.add_node("step_b", step_b)
builder.set_entry_point("step_a")
builder.add_edge("step_a", "step_b")
builder.add_edge("step_b", END)
graph = builder.compile()

Node functions

Every node function must:

1. Accept a single SharedState argument
2. Return a SharedState (typically the same one, modified)

def my_node(state: SharedState) -> SharedState:
    state["result"] = do_something(state["input"])
    return state

Edge types

Method	Description
add_edge(from, to)	Fixed edge -- always goes from one node to another
add_conditional_edge(from, router_fn)	Dynamic routing based on state
add_edge(from, END)	Terminates the graph

Conditional edges

The router function receives state and returns the name of the next node (or "__end__"):

def router(state):
    score = state["score"]
    if score > 0.8:
        return "accept"
    elif score > 0.5:
        return "review"
    return "reject"

builder.add_conditional_edge("evaluate", router)

Invoking a Graph

# Basic invoke
result = graph.invoke(SharedState({"input": "data"}))

# With thread ID for checkpointing
result = graph.invoke_with_thread("thread-123", SharedState({"input": "data"}))

Inspecting a Graph

graph.node_names()     # ["step_a", "step_b"]
graph.entry_point()    # "step_a"
graph.to_mermaid()     # Mermaid diagram string
graph.to_dot()         # Graphviz DOT string
graph.to_json()        # JSON structure

Max Steps

Prevent infinite loops by setting a maximum number of steps:

builder.set_max_steps(50)  # default is 1000

Subgraphs

Compose graphs by embedding one graph inside another:

# Build inner graph
inner_builder = StateGraphBuilder()
inner_builder.add_node("process", process_fn)
inner_builder.set_entry_point("process")
inner_builder.add_edge("process", END)
inner_graph = inner_builder.compile()

# Embed as a node in outer graph
outer_builder = StateGraphBuilder()
outer_builder.add_node("prepare", prepare_fn)
outer_builder.add_subgraph("inner", inner_graph)
outer_builder.set_entry_point("prepare")
outer_builder.add_edge("prepare", "inner")
outer_builder.add_edge("inner", END)
outer_graph = outer_builder.compile()

result = outer_graph.invoke(SharedState({"data": "..."}))

Checkpointing

Persist graph state to disk for recovery and human-in-the-loop:

builder.set_checkpointer("./checkpoints")
graph = builder.compile()

# Invoke with a thread ID
result = graph.invoke_with_thread("thread-1", SharedState({"input": "data"}))

See Human-in-the-Loop for interrupt/resume patterns.

Message Graph

For chat-focused workflows, MessageGraphBuilder pre-configures message accumulation:

from flowgentra_ai import MessageGraphBuilder, Message

def echo(messages):
    """Receives list of Messages, returns list of Messages."""
    last = messages[-1]
    return messages + [Message.assistant(f"Echo: {last.content}")]

builder = MessageGraphBuilder()
builder.add_node("echo", echo)
builder.set_entry_point("echo")
builder.add_edge("echo", "__end__")
graph = builder.compile()

result = graph.invoke([Message.user("Hello")])
for msg in result:
    print(f"{msg.role}: {msg.content}")