10xGraphv0.9.2Search

    State

    In shortAgentState, custom state, reducers, and the add_messages pattern for 10xGraph.

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

    When to use this

    Read this page when you need to understand how data flows through graph nodes, how to add your own fields to the state, or how message deduplication works.

    Import paths

    Python
    from tenxgraph.core.state import AgentState
from tenxgraph.core.state.reducers import add_messages, replace_messages, append_items, replace_value

    AgentState

    The default state class for all 10xGraph graphs. Pydantic BaseModel subclass — all fields are validated and serialised automatically.

    Python
    from tenxgraph.core.state import AgentState

state = AgentState()

    Fields

    Field Type Default Description
    context Annotated[list[Message], add_messages] [] Conversation message history. Uses add_messages reducer — new messages are appended; duplicates by message_id are dropped.
    context_summary str | None None Optional rolling summary of the conversation. Written by summary nodes.
    execution_meta ExecutionState auto Internal execution tracking. Managed by the framework — read-only for application code.

    Status methods

    These methods delegate to execution_meta and are safe to call in any node:

    Method Signature Description
    is_running() () -> bool True while the graph is executing.
    is_interrupted() () -> bool True when execution paused at an interrupt point.
    is_stopped_requested() () -> bool True when compiled.stop() was called for this thread.

    Lifecycle methods (framework use only)

    Method Description
    set_interrupt(node, reason, status, data=None) Record an interrupt at the given node.
    clear_interrupt() Clear the interrupt, allowing execution to resume.
    advance_step() Increment the step counter in execution_meta.
    set_current_node(node) Update the current_node pointer.
    complete() Mark execution as completed.
    error(error_msg) Record an error message in execution_meta.

    Subclassing AgentState

    Add your own fields by inheriting from AgentState. All internal framework fields are preserved.

    Python
    from typing import Annotated
from pydantic import BaseModel, Field
from tenxgraph.core.state import AgentState
from tenxgraph.core.state.reducers import add_messages

class OrderState(AgentState):
    """State for an order-processing agent."""
    order_id: str | None = None
    cart: list[dict] = Field(default_factory=list)
    confirmed: bool = False
    discount_code: str | None = None

    Use the custom state when building the graph:

    Python
    from tenxgraph.core.graph import StateGraph

graph = StateGraph(OrderState())

    Inside nodes, the state parameter is typed as your custom class:

    Python
    def process_order(state: OrderState, config: dict) -> list:
    if state.confirmed:
        order_id = state.order_id
        # ...
    return [Message.text_message("Order confirmed!", role="assistant")]

    Adding custom reducers to fields

    Use Annotated and a reducer function to control how a field merges across node updates:

    Python
    from typing import Annotated
from tenxgraph.core.state import AgentState
from tenxgraph.core.state.reducers import append_items

class AnalysisState(AgentState):
    # Append new items; deduplicate by .id
    findings: Annotated[list[Finding], append_items] = Field(default_factory=list)
    # Full replacement — always overwritten
    current_step: str = "init"

    Reducers

    Reducers are functions that control how field values merge when a node returns a partial update. They are applied automatically by the framework when a node returns a dict or list.

    add_messages

    Python
    from tenxgraph.core.state.reducers import add_messages

    The default reducer for AgentState.context. Merges two lists of Message objects:

    • Appends right to left.
    • Skips any message in right whose message_id already exists in left.
    • Skips delta messages (message.delta == True) — these are streaming intermediates, not stored.

    replace_messages

    Python
    from tenxgraph.core.state.reducers import replace_messages

    Replaces the entire message list. Use when you want a node to overwrite all history.

    Python
    class MyState(AgentState):
    context: Annotated[list[Message], replace_messages] = Field(default_factory=list)

    append_items

    Python
    from tenxgraph.core.state.reducers import append_items

    Appends new items by .id deduplication. Use for any list of Pydantic models that have an id field.

    replace_value

    Python
    from tenxgraph.core.state.reducers import replace_value

    Always replaces with the new value. Equivalent to the default Pydantic field update behavior.

    How nodes update state

    A node function may return one of:

    Return type Effect
    list[Message] Appended to state.context via add_messages.
    dict Each key-value pair updates the corresponding field using its reducer.
    None No state update — the state is passed through unchanged.
    Python
    # Return a list of messages
def answer_node(state: AgentState, config: dict) -> list:
    return [Message.text_message("Here is the answer.", role="assistant")]

# Return a partial state dict
def summarise_node(state: AgentState, config: dict) -> dict:
    summary = generate_summary(state.context)
    return {"context_summary": summary}

# Update both messages and a custom field
def checkout_node(state: OrderState, config: dict) -> dict:
    return {
        "context": [Message.text_message("Order placed!", role="assistant")],
        "confirmed": True,
    }

    ExecutionState (internal)

    The execution_meta field of AgentState. Holds framework-managed execution tracking. Application code should not mutate this directly.

    Field Type Description
    current_node str The node currently executing.
    step int Step counter (number of node transitions).
    status ExecutionStatus RUNNING, INTERRUPTED, COMPLETED, ERROR.
    interrupt_node str | None Node where the last interrupt occurred.
    interrupt_reason str | None Reason string for the interrupt.
    stop_current_execution StopRequestStatus NONE or STOP_REQUESTED.

    Common patterns

    Seed initial data into state

    Pass a state instance with pre-filled fields as input_data:

    Python
    initial = OrderState(order_id="ORD-001", cart=[{"sku": "ABC", "qty": 1}])
result = app.invoke(
    {"messages": [Message.text_message("Process my order")]},
    config={"thread_id": "order-session-1"},
)

    Read state after invoke (FULL granularity)

    Python
    from tenxgraph.utils import ResponseGranularity

result = await app.ainvoke(
    {"messages": [Message.text_message("Hello")]},
    config={"thread_id": "t1"},
    response_granularity=ResponseGranularity.FULL,
)
state: OrderState = result["state"]
print(state.confirmed, state.order_id)
    Last updated for v0.9.2Edit this page on GitHubReport an issue