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:
- Accept a single
SharedStateargument - 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:
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}")