StateGraph

A StateGraph models an agent as a directed graph: nodes (functions, agents or tool dispatchers) connected by edges, all reading and writing one shared state. You define the structure, call compile() to validate it, then run it with invoke() for a final result or stream() for incremental output. A recursion limit stops runaway loops.

flowchart LR
    S([START]) --> A[intake]
    A -->|urgent| B[urgent]
    A -->|backlog| C[backlog]
    B --> E([END])
    C --> E

What are the parts of a graph?

START and END live in tenxgraph.utils.constants. Nodes you register may return a Message, a list of messages, a plain string (stored as an assistant message), an AgentState, a dict or a Command.

How do I build a graph?

This example builds a support ticket triage graph. Each node is a function that receives the shared state and returns an update (either a string, a dict, or a list of messages). The conditional edge routes based on the ticket content.

from tenxgraph.core import StateGraph
from tenxgraph.core.state import AgentState, Message
from tenxgraph.storage.checkpointer import InMemoryCheckpointer
from tenxgraph.utils.constants import START, END


def intake(state: AgentState, config: dict) -> str:
    """First node: log the intake and prepare for routing."""
    return "Ticket received and queued for triage."


def handle_urgent(state: AgentState, config: dict) -> str:
    """Route path for urgent tickets: page on-call."""
    return "Escalating to on-call engineer. Incident ticket #12345."


def route_to_backlog(state: AgentState, config: dict) -> str:
    """Route path for non-urgent: file in backlog."""
    return "Ticket filed in product backlog for Q4 planning."


def decide_priority(state: AgentState) -> str:
    """Conditional function: returns the node name to visit next."""
    # Extract user message and check urgency indicator
    user_messages = [m for m in state.context if m.role == "user"]
    if not user_messages:
        return "backlog"  # default
    
    text = user_messages[-1].text().lower()
    return "urgent" if "outage" in text or "down" in text else "backlog"


# Build the graph structure
graph = StateGraph()
graph.add_node("intake", intake)
graph.add_node("urgent", handle_urgent)
graph.add_node("backlog", route_to_backlog)

# Set entry point (adds edge from START automatically)
graph.set_entry_point("intake")  

# Conditional edge: the function returns a key, path_map maps it to node name
graph.add_conditional_edges(
    "intake",
    decide_priority,
    {"urgent": "urgent", "backlog": "backlog"},  
)

# Terminal edges: both outcome paths lead to END
graph.add_edge("urgent", END)
graph.add_edge("backlog", END)

# Compile the graph: validates structure, binds runtime services
app = graph.compile(checkpointer=InMemoryCheckpointer())

# Run the graph
result = app.invoke(
    {"messages": [Message.text_message("AWS region down - checkout failing")]},
    config={"thread_id": "ticket-prod-001"},
)

# Print the final message (last state update)
print("Outcome:", result["messages"][-1].text())

Key points:

• set_entry_point("intake") adds an implicit edge from START to “intake”, so the graph always begins there.
• The conditional edge’s routing function decide_priority inspects the state and returns a string key (“urgent” or “backlog”), not a node name. The path_map dict maps that key to the actual node name.
• If you omit the path_map, the routing function must return the node name directly (e.g., return "urgent" instead of returning a key that the map looks up).
• Both “urgent” and “backlog” end with add_edge(..., END), so both paths terminate the graph.

How does state work?

Every run starts from an AgentState. Its main field is context, the list of messages, and it uses a reducer (add_messages) so new messages are appended rather than replacing the list. AgentState also carries context_summary and internal execution_meta that the runtime uses for steps, interrupts and resume.

Add your own fields by subclassing, then pass an instance to the graph:

from tenxgraph.core.state import AgentState


class TicketState(AgentState):
    customer_tier: str = "free"

graph = StateGraph(TicketState())

StateGraph(state=None, ...) creates a plain AgentState when you pass nothing. It also accepts context_manager, publisher, id_generator and container (an InjectQ container) for trimming context, emitting events, generating ids and dependency injection.

What does compile() do?

compile() validates the graph structure and returns a CompiledGraph object ready for execution. Compilation is where 10xGraph verifies that your graph is sound before you run it, catching structural errors early rather than during execution.

Compilation performs several checks:

• Entry point validation: Ensures you have called set_entry_point() or added an edge from START to a node. If not, raises GraphError with code GRAPH_002.
• Node reachability: Detects orphaned nodes (nodes with no incoming or outgoing edges), which are typically a sign of incomplete routing logic. Raises GraphError with code GRAPH_003.
• Edge target validation: Confirms that every edge points to a node you have actually registered with add_node(), otherwise GRAPH_004.
• Interrupt node validation: If you specified nodes to interrupt before or after, checks that they exist in the graph. Unknown names raise GraphError with code GRAPH_004, the same code as an edge that targets an unregistered node.
• Dependency resolution: Binds checkpointers, stores, publishers, and other runtime services into a dependency container so they are available to nodes during execution.

Once compiled, you can invoke the CompiledGraph many times with different inputs and thread ids without recompiling. Compile once at startup.

app = graph.compile(
    checkpointer=checkpointer,       # persist state per thread
    store=store,                     # long-term memory store
    media_store=media_store,         # multimodal content storage
    interrupt_before=["urgent"],     # pause before executing a node
    interrupt_after=["review"],      # pause after executing a node
    callback_manager=callback_manager,  # hooks into execution lifecycle
    shutdown_timeout=30.0,           # grace period for cleanup (seconds)
)

If you omit checkpointer, 10xGraph uses an InMemoryCheckpointer so the graph can always run. It keeps state only in process memory, so nothing survives a restart. See Checkpointing and threads for durable options and Interrupts for interrupt_before and interrupt_after.

Understanding recursion limits

Every graph execution has a recursion limit: a maximum number of steps before 10xGraph forces the run to stop. The limit prevents infinite loops caused by faulty routing logic. The default limit is 25 steps per run.

A step is one node execution. An agent node followed by a tool node counts as two steps, and each trip around an agent-tool loop adds two more. Conditional edges themselves do not count as steps; they only determine which node to visit next.

When you invoke a graph, you can override the default limit in the config:

result = app.invoke(
    {"messages": [Message.text_message("Your question")]},
    config={"thread_id": "session-1", "recursion_limit": 100},
)

If a run reaches the limit while the graph is still processing (the current node is not END), execution stops and raises GraphRecursionError with code RECURSION_001. The check runs after each node completes, when the step counter has reached the limit, so a run that needs exactly 25 steps also fails at the default. Its context holds max_steps, current_step and current_node. When streaming, an error chunk is yielded before the exception is raised. See Errors and limits.

When to raise the limit:

• Your graph has legitimate multi-step workflows that need more than 25 steps (e.g., a research agent that iterates over multiple search queries and summaries).
• You have nested graphs or multi-agent orchestration where one agent calls another, multiplying the step count.

When the limit is a signal of a bug:

• A conditional edge always routes back to itself, creating a loop.
• A tool node’s routing logic does not properly transition to the next stage.
• A node function returns an update that does not change the state, so the routing function sends it back to the same node.

Fix the routing logic first. Only raise the limit if you have verified that the loop is intentional and necessary.

from tenxgraph.core.state import AgentState


# A routing function that loops back to "agent" unconditionally will hit the
# recursion limit. Give it an exit condition based on state.
def route(state: AgentState) -> str:
    last = state.context[-1] if state.context else None
    if last is not None and last.role == "assistant" and not last.tools_calls:
        return "final"
    return "agent"

Should I use invoke or stream?

Both methods take the same input_data and config, but differ in how results are returned. Each also has an async variant: ainvoke and astream.

invoke calls asyncio.run internally, so from within an event loop, call ainvoke instead to avoid a “running event loop” error.

Example with stream: this reuses app from the triage example. At the default granularity only message and error chunks are delivered, so ask for PARTIAL to also receive state chunks.

from tenxgraph.core.state import Message
from tenxgraph.utils.constants import ResponseGranularity

for chunk in app.stream(
    {"messages": [Message.text_message("Checkout is down for all users")]},
    config={"thread_id": "session-42"},
    response_granularity=ResponseGranularity.PARTIAL,
):
    if chunk.event == "message":
        # Output from a node as it arrives
        print(chunk.message.text())
    elif chunk.event == "state":
        # State after a node has finished
        print("Step:", chunk.state.execution_meta.step)

Each StreamChunk has:

• event: "message", "state" (after a node completes), "updates" (run status data) or "error".
• message: The message if event == "message".
• state: The state if event == "state".

LOW yields message and error chunks, PARTIAL adds state, and FULL adds updates. See Streaming for the full model.

What goes in the run config?

The config dict passed to invoke() and stream() controls how the graph execution behaves for that specific run. It is separate from the response_granularity argument, which controls what data is returned.

response_granularity determines what the result or each stream chunk contains:

Pass response_granularity directly to invoke() or stream(), not in the config dict:

from tenxgraph.core.state import Message
from tenxgraph.utils.constants import ResponseGranularity

input_data = {"messages": [Message.text_message("Checkout is down")]}
result = app.invoke(
    input_data,
    config={"thread_id": "session-1", "recursion_limit": 50},
    response_granularity=ResponseGranularity.FULL,
)

Related pages