10xGraphv0.9.2Search

    How to build a graph

    In shortStep-by-step guide to constructing, compiling, and executing a StateGraph with Agent and ToolNode nodes.

    • 4 min read
    • 15 sections
    • Updated
    • v0.9.2
    • Markdown

    StateGraph is the core orchestration primitive in 10xGraph. You construct a workflow by adding nodes (functions, Agent instances, or ToolNode instances), connecting them with edges, and compiling to get a runnable CompiledGraph.

    Prerequisites

    Terminal
    pip install 10xgraph

    Set your provider API key:

    Terminal
    export OPENAI_API_KEY=sk-...      # for OpenAI
export GOOGLE_API_KEY=...         # for Google

    Step 1: Import the essentials

    Python
    from tenxgraph.core.graph import StateGraph, Agent, ToolNode
from tenxgraph.core.state import AgentState, Message
from tenxgraph.utils import START, END

    Step 2: Define tools

    Tools are plain Python functions. Type-annotate parameters so the LLM receives an accurate schema.

    Python
    def get_weather(city: str) -> str:
    """Return current weather for a city."""
    return f"Weather in {city}: 22°C, partly cloudy."

def calculate(expression: str) -> str:
    """Evaluate a safe math expression."""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Error: {e}"

    Step 3: Create a ToolNode

    Group tools in a ToolNode. The node registers every function by its __name__.

    Python
    tool_node = ToolNode([get_weather, calculate])

    Step 4: Create an Agent

    Agent wraps the LLM call as a graph node.

    Python
    agent = Agent(
    model="gpt-4o",
    system_prompt=[{"role": "system", "content": "You are a helpful assistant."}],
    tool_node=tool_node,
)

    Step 5: Build and wire the graph

    Python
    graph = StateGraph()

graph.add_node("MAIN", agent)
graph.add_node("TOOL", tool_node)

# Route: if agent produced tool calls, go to TOOL, else END
def should_use_tools(state: AgentState) -> str:
    last = state.context[-1] if state.context else None
    if last and last.role == "assistant" and getattr(last, "tools_calls", None):
        return "TOOL"
    return END

graph.add_conditional_edges("MAIN", should_use_tools, {"TOOL": "TOOL", END: END})
graph.add_edge("TOOL", "MAIN")   # loop back after tool execution
graph.set_entry_point("MAIN")    # also adds START → MAIN edge

    Graph methods

    Method Signature Purpose
    add_node (name_or_func, func=None) Register a node. Pass a function (name inferred) or an explicit name + callable/Agent/ToolNode.
    add_edge (from_node, to_node) Static route between nodes. add_edge(START, "X") sets the entry point.
    add_conditional_edges (from_node, condition, path_map=None) Dynamic routing. condition(state) returns a key; path_map maps keys to node names. Without path_map, condition must return the node name directly.
    set_entry_point (node_name) Shorthand for add_edge(START, node_name).
    override_node (name, func) Replace an existing node (useful in tests).
    compile (checkpointer, store, interrupt_before, interrupt_after, ...) Returns a CompiledGraph.

    Step 6: Compile

    Python
    app = graph.compile()

    Without a checkpointer argument, compilation defaults to InMemoryCheckpointer. For persistent state see how-to/python/set-up-checkpointing.

    Step 7: Invoke

    Synchronous

    Python
    result = app.invoke(
    {"messages": [Message.text_message("What is the weather in Paris?")]},
    config={"thread_id": "session-1", "user_id": "user-42"},
)

for msg in result["messages"]:
    print(msg.role, msg.content)

    Asynchronous

    Python
    import asyncio

async def main():
    result = await app.ainvoke(
        {"messages": [Message.text_message("Calculate 123 * 456")]},
        config={"thread_id": "session-2"},
    )
    for msg in result["messages"]:
        print(msg.role, msg.content)

asyncio.run(main())

    Config keys

    Use config for runtime metadata. For most cases, you only need these keys:

    Key Default Notes
    thread_id auto (UUID) Conversation/thread identifier used by the checkpointer.
    user_id "anonymous" Passed to tools and publisher events. If not provided, it is set automatically.
    recursion_limit 25 Maximum node-execution steps before GraphRecursionError.

    Other reserved keys are auto-populated by the runtime and should not be set manually:

    • run_id
    • is_stream
    • timestamp

    When API authentication is enabled, two values are injected into config automatically:

    • user_id
    • user (whatever object/value your auth layer returns)

    These are reserved keys; beyond them, you can add any custom keys you want in config.

    For partial state updates, return a dictionary with only the fields you want to change. State is available inside input_data, which is the first dictionary passed into node functions.

    Python
    result = app.invoke(
    {"messages": [Message.text_message("What is the weather in Paris?")], "state": {"location": "Paris"}},
    config={"thread_id": "session-1", "user_id": "user-42"},
)

    Step 8: Stream responses

    Python
    from tenxgraph.core.state import StreamEvent
from tenxgraph.utils import ResponseGranularity

async def stream_example():
    async for chunk in app.astream(
        {"messages": [Message.text_message("Tell me a short story.")]},
        config={"thread_id": "stream-1"},
        response_granularity=ResponseGranularity.LOW,
    ):
        if chunk.event == StreamEvent.MESSAGE and chunk.message:
            print(chunk.message.text(), end="", flush=True)

asyncio.run(stream_example())

    See how-to/python/stream-graph for the full streaming reference.

    Response granularity

    Pass response_granularity to invoke(), ainvoke(), or astream():

    Value invoke() returns
    ResponseGranularity.LOW (default) {"messages": [...]} - only the final messages
    ResponseGranularity.PARTIAL {"messages": [...], "context": [...], "context_summary": ...}
    ResponseGranularity.FULL Complete state dict including execution_meta

    Stop a running graph

    Python
    # From another coroutine or task
await app.astop({"thread_id": "session-1"})

    stop() is the sync wrapper. The graph checks the stop flag between nodes and halts before the next node.

    Override a node for testing

    Python
    # Production graph
app = graph.compile()

# In tests: swap the real agent with a stub
def stub_agent(state, config, **deps):
    return {"messages": [Message.text_message("stub response", role="assistant")]}

app.override_node("MAIN", stub_agent)

    override_node is available on both StateGraph (before compile) and CompiledGraph (after compile).

    Complete example

    Python
    import asyncio
from tenxgraph.core.graph import StateGraph, Agent, ToolNode
from tenxgraph.core.state import AgentState, Message
from tenxgraph.utils import START, END

def get_weather(city: str) -> str:
    """Return current weather for a city."""
    return f"22°C, partly cloudy in {city}."

tool_node = ToolNode([get_weather])

agent = Agent(
    model="gpt-4o",
    system_prompt=[{"role": "system", "content": "You are a helpful assistant."}],
    tool_node=tool_node,
)

graph = StateGraph()
graph.add_node("MAIN", agent)
graph.add_node("TOOL", tool_node)

def should_use_tools(state: AgentState) -> str:
    last = state.context[-1] if state.context else None
    if last and last.role == "assistant" and getattr(last, "tools_calls", None):
        return "TOOL"
    return END

graph.add_conditional_edges("MAIN", should_use_tools, {"TOOL": "TOOL", END: END})
graph.add_edge("TOOL", "MAIN")
graph.set_entry_point("MAIN")

app = graph.compile()

result = app.invoke(
    {"messages": [Message.text_message("What is the weather in Tokyo?")]},
    config={"thread_id": "demo-1"},
)

for msg in result["messages"]:
    print(f"[{msg.role}]", msg.content)

    What you learned

    • Use StateGraph to build the workflow, then call compile() to get a runnable CompiledGraph.
    • add_node, add_edge, add_conditional_edges, set_entry_point wire the graph topology.
    • invoke() / ainvoke() run the graph; astream() streams chunks token by token.
    • config dict controls thread_id, user_id, recursion_limit, and more.

    Next steps

    Last updated for v0.9.2Edit this page on GitHubReport an issue