StateGraph

A StateGraph describes an agent as a set of nodes (Python functions, agents or tool nodes) joined by edges, all reading and writing one shared state object. You build it, call compile() to get a runnable CompiledGraph, then run it with invoke or stream.

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 graph routes a support ticket. It uses no model, so you can run it as is:

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


def intake(state: AgentState, config: dict) -> str:
    return "Ticket received."


def urgent(state: AgentState, config: dict) -> str:
    return "Paging the on-call engineer."


def backlog(state: AgentState, config: dict) -> str:
    return "Filed in the backlog."


def route(state: AgentState) -> str:
    user_messages = [m for m in state.context if m.role == "user"]
    return "urgent" if "outage" in user_messages[-1].text().lower() else "backlog"


graph = StateGraph()
graph.add_node("intake", intake)
graph.add_node("urgent", urgent)
graph.add_node("backlog", backlog)

graph.set_entry_point("intake")  
graph.add_conditional_edges("intake", route, {"urgent": "urgent", "backlog": "backlog"})  
graph.add_edge("urgent", END)
graph.add_edge("backlog", END)

app = graph.compile(checkpointer=InMemoryCheckpointer())

result = app.invoke(
    {"messages": [Message.text_message("Checkout outage in EU")]},
    config={"thread_id": "ticket-1"},
)
print(result["messages"][-1].text())

set_entry_point("intake") adds the edge from START. The routing function returns a key, and the path_map turns that key into a node name. If you omit path_map, the function must return the node name itself.

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() checks the graph and returns a CompiledGraph. It fails early on two mistakes: no entry point (GraphError, code GRAPH_002) and nodes that no edge touches (orphans). It also rejects an edge that targets a node you never added.

app = graph.compile(
    checkpointer=checkpointer,      # persist state per thread
    store=store,                    # long-term memory store
    interrupt_before=["urgent"],    # pause before a node
    interrupt_after=None,           # pause after a node
)

The remaining arguments are media_store, callback_manager and shutdown_timeout (30 seconds by default).

Should I use invoke or stream?

Both take the same input_data, config and response_granularity. Each has an async twin, ainvoke and astream.

invoke calls asyncio.run internally, so call ainvoke from inside an event loop.

for chunk in app.stream(
    {"messages": [Message.text_message("Checkout outage in EU")]},
    config={"thread_id": "ticket-2"},
):
    print(chunk.event, chunk.message.text() if chunk.message else "")

A chunk has an event (message, state, updates or error) and carries a message or state accordingly.

What goes in the run config?

response_granularity is a separate argument: LOW returns only messages, PARTIAL adds context and summary, FULL adds the complete state.

Related pages