Use custom state | 10xGraph

Every 10xGraph agent runs within an AgentState instance that persists data across all nodes in the graph. By default, AgentState holds the message conversation history and execution metadata. You extend it by subclassing to add typed, persistent application fields, like user IDs, ticket numbers, sentiment scores, or any domain-specific data your agent needs. The graph threads the state through every node and saves it with the checkpointer.

Why custom state

Custom state fields let you:

Track data that threads through the entire agent execution
Use placeholders in system prompts ("Help user {user_id}") interpolated at runtime
Persist application context across multiple invocations via checkpointing
Structure decision-making (conditional edges read custom fields to route the graph)
Pass initial values into a run without rebuilding the graph

Without custom state, you would store this data outside the graph, losing the automatic persistence and threading benefits of the framework.

Built-in fields

AgentState provides three fields you can read and build on:

Field	Type	Description
`context`	`list[Message]`	The conversation history (new messages are appended with the `add_messages` reducer and deduplicated by `message_id`)
`context_summary`	`str \| None`	Optional compressed summary of older messages
`execution_meta`	`ExecutionState`	Internal metadata tracking node progress, interrupts, and step counts. Read-only for most use cases.

Step 1: Define a custom state class

Create a Pydantic model that extends AgentState:

Python

from pydantic import Field
from tenxgraph.core.state import AgentState


class SupportTicketState(AgentState):
    """Custom state for a support agent."""
    
    user_id: str = ""
    ticket_id: str | None = None
    sentiment: str = "neutral"  # "positive" | "neutral" | "negative"
    escalation_count: int = 0
    resolved: bool = False
    tags: list[str] = Field(default_factory=list)

All Pydantic features work: type hints, validators, default_factory, optional fields, computed fields. The built-in fields (context, context_summary, execution_meta) are inherited automatically.

Step 2: Pass your state class to StateGraph

Instantiate the graph with your custom state class or instance:

Python

from tenxgraph.core.graph import StateGraph

# Option A: Pass the class (StateGraph instantiates it)
graph = StateGraph(SupportTicketState)

# Option B: Pass a pre-populated instance with defaults
initial_state = SupportTicketState(user_id="user-456")
graph = StateGraph(initial_state)

Both approaches work. The instance is a prototype: a new thread starts from a deep copy of it, and a thread that already has a saved state loads that state instead. Option B lets you bake in different defaults.

Step 3: Read and update custom fields in nodes

Node functions receive the current state (declare a state parameter). Change the fields you need on the state object and return it. A node can also return a str, a Message, a list of messages, or a Command; a plain dict is not a valid return value. See use custom nodes.

Python

from tenxgraph.core.state import Message


def classify_sentiment(state: SupportTicketState, config: dict) -> SupportTicketState:
    """Analyze the last user message and update sentiment."""
    last_user_msg = next(
        (m for m in reversed(state.context) if m.role == "user"), None
    )
    if not last_user_msg:
        return state

    text = last_user_msg.text().lower()
    if any(word in text for word in ["angry", "terrible", "worst", "unacceptable"]):
        state.sentiment = "negative"
        state.escalation_count += 1
    elif any(word in text for word in ["great", "thanks", "excellent", "happy"]):
        state.sentiment = "positive"
    else:
        state.sentiment = "neutral"
    return state


def resolve_ticket(state: SupportTicketState, config: dict) -> SupportTicketState:
    """Mark the ticket resolved and append a closing message."""
    state.resolved = True
    state.tags = ["handled", "closed"] if state.ticket_id else []
    state.context.append(
        Message.text_message(
            f"Ticket {state.ticket_id} has been resolved. Thank you for using our support.",
            role="assistant",
        )
    )
    return state

When a node returns the state, the runtime keeps the new field values and records any new entries in state.context as new messages of the run.

Step 4: Use custom fields in system prompts

The Agent class interpolates state field values into system prompts at runtime:

Python

from tenxgraph.core.graph import Agent

agent = Agent(
    model="gpt-4o",
    system_prompt=[
        {
            "role": "system",
            "content": (
                "You are a customer support agent assisting user {user_id}. "
                "Ticket: {ticket_id}. Current sentiment: {sentiment}. "
                "This is escalation #{escalation_count}."
            ),
        }
    ],
)

Placeholders in curly braces are filled from the current state with str.format. If any placeholder names a field the state does not have, a warning is logged and the whole prompt is sent without interpolation. A field set to None renders as None (so ticket_id above reads Ticket: None until it is set). Escape literal braces as {{ and }}.

Step 5: Pass initial values at invocation

Messages go under "messages". Custom field values go under a "state" key:

Python

from tenxgraph.utils import ResponseGranularity

result = app.invoke(
    {
        "messages": [Message.text_message("My order hasn't arrived.")],
        "state": {
            "user_id": "cust-789",
            "ticket_id": "TKT-2024-001",
            "sentiment": "negative",
        },
    },
    config={"thread_id": "support-session-1"},
    response_granularity=ResponseGranularity.FULL,
)
print(result["state"].ticket_id)

Only keys that already exist as fields on your state class are applied; context, context_summary and execution_meta are skipped. By default invoke returns only messages and token_usage. Pass response_granularity=ResponseGranularity.FULL (from tenxgraph.utils) to also get the final state object.

Reducers: combining concurrent updates

A reducer is a function (left, right) -> merged, attached to a field with Annotated. The runtime uses a field’s reducer when parallel tool calls in one ToolNode step both change the same field: their changes are combined with the reducer instead of one overwriting the other. Without a reducer the last write wins and a warning is logged. Reducers are not applied to values returned from ordinary function nodes; there you assign the field yourself.

10xGraph provides these in tenxgraph.core.state:

`add_messages` (append with deduplication)

Appends new messages and deduplicates by message_id. This is the default reducer for the context field. Use it for any message list field:

Python

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


class PipelineState(AgentState):
    # A separate log of intermediate messages, also deduplicated by ID
    processing_log: Annotated[list[Message], add_messages] = Field(default_factory=list)

When two parallel tool calls each add messages to processing_log, both sets are appended. A message whose ID already exists is skipped. The same function is what keeps context free of duplicate messages.

`replace_messages` (replace entire list)

Replaces the entire message list with the new one. Use this when the latest write should win outright:

Python

from typing import Annotated
from tenxgraph.core.state.reducers import replace_messages


class SummarizedState(AgentState):
    # This field is always replaced, never appended
    summary_messages: Annotated[list[Message], replace_messages] = Field(default_factory=list)

When the reducer runs, the old list is discarded and replaced by the new one.

`append_items` (append objects with id deduplication)

Appends items to a list and deduplicates by each item’s .id attribute. Use this for domain objects:

Python

from typing import Annotated
from tenxgraph.core.state.reducers import append_items
from pydantic import Field, BaseModel


class ToolResult(BaseModel):
    id: str
    name: str
    value: str


class ToolState(AgentState):
    # Append tool results, deduplicated by their .id
    tool_results: Annotated[list[ToolResult], append_items] = Field(default_factory=list)

When the reducer runs, new items are appended. An item with an .id that already exists is skipped.

`replace_value` (replace a value)

Returns the new value and ignores the old one. Use it to make last-write-wins explicit:

Python

from tenxgraph.core.state.reducers import replace_value


class MyState(AgentState):
    status: Annotated[str, replace_value] = "pending"
    counter: int = 0  # No reducer: last write wins, with a warning on a parallel conflict

`remove_tool_messages` (prune completed tool sequences)

Unlike the others, this takes a single list and returns the pruned list, so it is a plain helper, not a field reducer. It removes completed tool interaction sequences to keep context lean. A sequence is only removed if it is complete:

An assistant message with tool calls
One or more tool result messages
A final assistant message without tool calls (using the tool results)

If a sequence is incomplete (tool call made but no final response yet), all messages are kept:

Python

from tenxgraph.core.state.reducers import remove_tool_messages


def prune(state: AgentState, config: dict) -> AgentState:
    state.context = remove_tool_messages(state.context)
    return state

This is useful in long-running agents where you want to drop tool calls and results after the model has used them, reducing token usage on later calls.

Complete example

Here is a working example that ties it together. Install a provider extra first, for example pip install "10xgraph[openai]", and set OPENAI_API_KEY.

Python

from pydantic import Field
from tenxgraph.core.graph import StateGraph, Agent
from tenxgraph.core.state import AgentState, Message
from tenxgraph.utils import END, ResponseGranularity


# Define custom state
class ResearchState(AgentState):
    research_topic: str = ""
    findings: list[str] = Field(default_factory=list)
    sources: list[str] = Field(default_factory=list)
    confidence: float = 0.0


# Define a node that processes custom fields
def extract_findings(state: ResearchState, config: dict) -> ResearchState:
    """Extract key findings from the last response."""
    if not state.context:
        return state

    text = state.context[-1].text()

    # Simplified extraction: split by "Finding:" markers
    findings = [
        f.strip()
        for f in text.split("Finding:")[1:]
        if f.strip()
    ]

    state.findings = findings
    state.confidence = 0.85 if len(findings) > 2 else 0.5
    return state


# Build the graph
graph = StateGraph(ResearchState)

agent = Agent(
    model="gpt-4o",
    system_prompt=[
        {
            "role": "system",
            "content": (
                "You are a research assistant. "
                "Research topic: {research_topic}. "
                "Your confidence level: {confidence}. "
                "List your findings prefixed with 'Finding:'."
            ),
        }
    ],
)

graph.add_node("research", agent)
graph.add_node("extract", extract_findings)
graph.set_entry_point("research")
graph.add_edge("research", "extract")
graph.add_edge("extract", END)

app = graph.compile()

# Invoke with initial state values
result = app.invoke(
    {
        "messages": [Message.text_message("Research the benefits of remote work.")],
        "state": {"research_topic": "Remote work productivity"},
    },
    config={"thread_id": "research-session-1"},
    response_granularity=ResponseGranularity.FULL,
)

final = result["state"]
print("Findings:", final.findings)
print("Confidence:", final.confidence)

How to verify it worked

Invoke with response_granularity=ResponseGranularity.FULL so the result includes the final state, then check:

State is threaded: custom fields hold the values you expect.

Python

state = result["state"]
print(state.user_id)      # Should match what you passed
print(state.sentiment)    # Should reflect updates from nodes
print(len(state.context)) # Messages should be accumulated

Checkpointing preserves state: compile() uses an in-memory checkpointer by default, so invoking again with the same thread_id in the same process continues from the saved state.

Python

result2 = app.invoke(
    {"messages": [Message.text_message("Follow-up question.")]},
    config={"thread_id": "support-session-1"},
    response_granularity=ResponseGranularity.FULL,
)
assert result2["state"].user_id == state.user_id

Common errors and fixes

Error: AttributeError or a Pydantic error mentioning a missing field

You accessed or assigned a field that is not defined in your state class. Add it:

Python

class SupportTicketState(AgentState):
    my_field: str = ""  # Add the missing field

My field in "state" was ignored.

Only keys that already exist on the state class are applied, and context, context_summary and execution_meta are always skipped. Check the spelling, and pass messages under "messages".

The system prompt shows raw {placeholders}.

A placeholder names a field the state does not have, so interpolation was skipped for the whole prompt and a warning was logged. Add the field to your state class with a default:

Python

class MyState(AgentState):
    user_id: str = "unknown"

My custom state updates do not persist across restarts.

The default checkpointer is in memory. Pass a durable one to compile(), for example SqliteCheckpointer (pip install "10xgraph[sqlite_checkpoint]"):

Python

from tenxgraph.storage.checkpointer import SqliteCheckpointer

app = graph.compile(checkpointer=SqliteCheckpointer(db_path="./state.db"))

Variations and best practices

Typed reducers for custom types: If your field is a list of domain objects, use append_items:

Python

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


class Item(BaseModel):
    id: str
    name: str

class MyState(AgentState):
    items: Annotated[list[Item], append_items] = Field(default_factory=list)

Combining multiple reducers: You can have different fields with different reducers in the same state:

Python

from typing import Annotated
from pydantic import BaseModel, Field
from tenxgraph.core.state import AgentState, Message
from tenxgraph.core.state.reducers import add_messages, replace_value, append_items


class MultiReducerState(AgentState):
    messages_log: Annotated[list[Message], add_messages] = Field(default_factory=list)
    latest_status: Annotated[str, replace_value] = "pending"
    collected_items: Annotated[list[Item], append_items] = Field(default_factory=list)

Initializing with factory defaults: For complex fields, use Field(default_factory=...):

Python

class MyState(AgentState):
    metadata: dict = Field(default_factory=dict)
    tags: list[str] = Field(default_factory=list)

This ensures each graph instance gets its own dict/list, not a shared one (a common Python pitfall).

Next steps

Build a graph for the full workflow assembly guide
Set up checkpointing to persist custom state across requests
Use dependency injection to pass runtime config into nodes
State and messages (concept) for a deeper dive into message content blocks

Why custom state

Built-in fields

Step 1: Define a custom state class

Step 2: Pass your state class to StateGraph

Step 3: Read and update custom fields in nodes

Step 4: Use custom fields in system prompts

Step 5: Pass initial values at invocation

Reducers: combining concurrent updates

add_messages (append with deduplication)

replace_messages (replace entire list)

append_items (append objects with id deduplication)

replace_value (replace a value)

remove_tool_messages (prune completed tool sequences)

Complete example

How to verify it worked

Common errors and fixes

Variations and best practices

Next steps

`add_messages` (append with deduplication)

`replace_messages` (replace entire list)

`append_items` (append objects with id deduplication)

`replace_value` (replace a value)

`remove_tool_messages` (prune completed tool sequences)