Custom State

Source example: examples/custom-state/custom_state.py

What you will build

An HR assistant that evaluates candidates by matching their CVs against job descriptions. Unlike a simple chatbot that only remembers conversation history, this agent carries domain-specific data: the candidate’s CV text, the job description, a computed match score, and detailed analysis results. You will learn how to subclass AgentState to add typed fields, wire them into the graph and checkpointer, seed them before running, and update them selectively without affecting other state.

Why custom state matters

The default AgentState holds context (the conversation history), context_summary, and internal execution_meta. These are sufficient for a chatbot, but real agents need to carry structured context. For an HR assistant, you need the candidate information, job details, and scores. For a data analyst, you might track data sources and computed metrics. For a support agent, you track the customer account, ticket ID, and resolution state.

By extending AgentState, you make that data:

• Typed: your IDE and type checker catch field name and type errors.
• Persistent: checkpointers save and restore it across turns automatically.
• Mergeable: you update individual fields at invoke time without losing the rest.
• Introspectable: the agent can read these fields in routing logic and tool implementations.

classDiagram
    class AgentState {
        +list context
        +str context_summary
        +ExecMeta execution_meta
    }
    class MyState {
        +str candidate_cv
        +str jd
        +float match_score
        +dict analysis_results
    }
    AgentState <|-- MyState : extends

How to run this example

Clone the repository and navigate to the example directory:

git clone https://github.com/10xGraph/10xGraph.git
cd 10xGraph/examples/custom-state

Install the core library with Google Gemini support:

pip install "10xgraph[google-genai]"

Set your API key:

export GEMINI_API_KEY="your-api-key-here"

Run the example:

python custom_state.py

The script runs three test functions that demonstrate basic invocation, pre-populating custom fields, and partial state updates.

Building the custom state class

Start by subclassing AgentState and adding typed fields with defaults:

from typing import Any
from tenxgraph.core.state import AgentState

class MyState(AgentState):
    """Custom state with additional fields for resume matching."""
    candidate_cv: str = ""
    jd: str = ""
    match_score: float = 0.0
    analysis_results: dict[str, Any] = {}

Give every field a default value: StateGraph uses the state instance you pass as the prototype for new threads, and checkpoints serialize it. You can use any JSON-serializable type: strings, numbers, booleans, lists, dicts, or nested Pydantic models.

Creating a typed checkpointer

The checkpointer preserves state across turns. Pass your custom state as a generic parameter so type checkers know which state class it stores:

from tenxgraph.storage.checkpointer import InMemoryCheckpointer

checkpointer = InMemoryCheckpointer[MyState]()

For production, use PgCheckpointer (it needs a redis_url, and calls need a user_id) to persist state across server restarts. The type parameter is for static typing only.

Building the graph with custom state

Pass an instance of your custom state class to StateGraph:

from tenxgraph.core import Agent, StateGraph

def create_app(initial_state: MyState | None = None):
    state = initial_state or MyState()

    agent = Agent(
        model="gemini-2.5-flash",
        provider="google",
        system_prompt=[
            {
                "role": "system",
                "content": "You are a helpful HR assistant. Analyse CVs against job descriptions.",
            }
        ],
        trim_context=True,
    )

    graph = StateGraph[MyState](state)
    graph.add_node("MAIN", agent)
    graph.set_entry_point("MAIN")

    return graph.compile(checkpointer=checkpointer)

The state instance you pass is the prototype for new threads; the StateGraph[MyState] generic is for static typing. compile(checkpointer=...) wires the checkpointer in, so state persists across invocations within the same thread.

Three ways to invoke the agent

1. Basic invocation

The simplest case: just send a message and let the default state fields be used:

from tenxgraph.core.state import Message

app = create_app()
res = app.invoke(
    {"messages": [Message.text_message("Hello, can you help me with CV analysis?")]},
    config={"thread_id": "basic_test", "recursion_limit": 10},
)

The state starts with empty CV, job description, and match score fields.

2. Pre-populate custom fields before creation

Create a state instance and populate fields, then pass it to create_app:

custom_state = MyState()
custom_state.candidate_cv = "John Doe, Senior Python Engineer, 5 years experience"
custom_state.jd = "Looking for Senior Python Developer with 3+ years experience"
custom_state.match_score = 0.85
custom_state.analysis_results = {"skills_match": True, "experience_match": True}

app = create_app(custom_state)
res = app.invoke(
    {"messages": [Message.text_message("What's the match score for this candidate?")]},
    config={"thread_id": "custom_test"},
)

The custom fields are in the graph state, so routing functions and tools can read them. The Agent node itself only sends the system prompt and conversation messages to the model, so to have it use these fields, reference them in a tool or build the prompt from them.

3. Partial state update at invoke time

Update only the fields you need without touching the rest. Pass a state dict in the input:

from tenxgraph.utils import ResponseGranularity

res = app.invoke(
    {
        "messages": [Message.text_message("Update the job description only.")],
        "state": {"jd": "Looking for Data Scientist with deep learning experience"},
    },
    config={"thread_id": "partial_update_test"},
    response_granularity=ResponseGranularity.FULL,
)

# The returned state reflects the partial update
updated_state = res["state"]
print(updated_state.jd)           # new value
print(updated_state.candidate_cv)  # unchanged from before

This is the most powerful pattern for multi-turn interactions. You can update the job description without rewriting the CV or resetting scores. Other fields remain exactly as they were in the checkpoint.

Understanding partial state merge

When you invoke with a state dict, 10xGraph merges it with the existing checkpoint state by updating only the keys you provide. This is crucial for multi-step workflows where different API calls handle different aspects of the agent’s context.

flowchart LR
    A([invoke input]) -->|messages| B[Graph runtime]
    A -->|state dict partial| B
    B -->|merge: only listed keys updated| C[MyState snapshot]
    C --> D[MAIN Agent Node]
    D --> E([Output state])

    style A fill:#4A90D9,color:#fff
    style B fill:#7B68EE,color:#fff
    style C fill:#50C878,color:#fff
    style D fill:#F5A623,color:#fff
    style E fill:#FF6B6B,color:#fff

Running the complete example

The example file includes three test functions that you can run sequentially:

if __name__ == "__main__":
    # Run tests
    try:
        test_basic_functionality()           # Invoke with default state
        test_custom_state_fields()           # Pre-populate fields
        test_partial_state_update()          # Update one field at invoke time
        print("\n=== All tests completed successfully! ===")
    except Exception as e:
        print(f"Error during testing: {e}")
        raise

Each test demonstrates a pattern:

1. Basic: no custom data, just chat
2. Custom fields: pre-seed state before creating the app
3. Partial update: merge new values at invoke time while preserving the rest

Key patterns to remember

What you learned

• How to extend AgentState with custom typed fields for domain context.
• How to use generics (StateGraph[MyState], InMemoryCheckpointer[MyState]) to type state through the runtime.
• Three ways to populate state: defaults, pre-seeding, and partial updates at invoke time.
• How partial state merge preserves untouched fields across turns.
• Why typing your state fields matters for IDE support and production reliability.

Next steps

• Read the custom state guide for complete details on reducers and serialization.
• See state and messages concepts to understand the full state model.
• Explore checkpointing and threads to choose the right persistence strategy for production.
• Try the tool decorator example to learn how agents access and use state inside tools.