How to Use FileStateStore for Persistent State in aisuite

Use FileStateStore from aisuite/agents/state_store.py to persist agent run state across sessions by initializing it with a directory path, then passing it to your agent runner or calling save_state() and load_state() directly with thread IDs.

The FileStateStore class provides the default persistence layer in the andrewyng/aisuite framework, enabling agents to resume execution across process restarts without external databases. By implementing the StateStore protocol, it serializes RunState objects to JSON files while providing optimistic concurrency control through revision tracking. Understanding how to configure and interact with this store is essential for building reliable, long-running agent workflows.

What is FileStateStore?

FileStateStore is the concrete file-based implementation of aisuite's abstract StateStore protocol. Located in aisuite/agents/state_store.py, this class manages the lifecycle of agent execution state by mapping each unique thread ID to a dedicated JSON file on disk.

When instantiated, the store creates a root directory—defaulting to .aisuite/state—where it persists StoredRunState objects. These objects wrap the in-memory RunState (containing conversation history, tool outputs, and execution context) alongside metadata and revision numbers for conflict detection.

Initializing the FileStateStore

Create a FileStateStore by passing a pathlib.Path object pointing to a writable directory. If the directory does not exist, the constructor creates it automatically.

import aisuite as ai
from pathlib import Path

# Use default location (.aisuite/state)

store = ai.FileStateStore()

# Or specify a custom directory

store = ai.FileStateStore(Path("/var/lib/my_agent_states"))

Each thread's state is stored as an individual JSON file named after the URL-escaped thread ID, preventing filesystem conflicts while maintaining human-readable identifiers.

Saving and Loading Agent State

The store provides three primary operations defined in the StateStore protocol: save_state(), load_state(), and delete_state().

Persisting State with Version Control

When calling save_state(), you must provide a thread_id, the RunState object, and optionally a revision number for concurrency control. The method returns a StoredRunState dataclass containing the assigned revision.

# Obtain a RunState from your agent execution

run_state = ai.RunState(...)  

# Save with optional metadata

stored = store.save_state(
    thread_id="thread/user:1",
    state=run_state,
    metadata={"description": "customer onboarding flow"},
)
print(f"Persisted revision {stored.revision}")

As implemented in aisuite/agents/state_store.py at line 105, the store checks the supplied revision against the current stored revision. If they mismatch, the method raises StateConflictError, providing optimistic concurrency protection against simultaneous updates from multiple processes.

Handling Concurrent Updates with Optimistic Locking

To safely update state in concurrent environments, always read the current revision before writing:

# Load existing state to get current revision

loaded = store.load_state("thread/user:1")
if loaded:
    # Modify the state...

    new_state = modify_state(loaded.state)
    
    # Save only if no other process modified it

    try:
        store.save_state(
            thread_id="thread/user:1",
            state=new_state,
            revision=loaded.revision  # Must match current

        )
    except ai.StateConflictError:
        print("State was modified by another process, retry...")

Attaching Metadata to State Snapshots

The metadata parameter accepts arbitrary dictionaries that are deep-copied during storage, ensuring subsequent modifications to your local dictionary do not affect the persisted snapshot. This metadata is retrieved intact when calling load_state().

Retrieving and Deleting State

The load_state() method (line 124) reconstructs the StoredRunState from JSON, returning None if the thread ID has never been persisted. Use delete_state() (line 131) for cleanup:

# Load previously saved state

loaded = store.load_state("thread/user:1")
if loaded:
    print(f"Resuming from revision {loaded.revision}")
    resume_execution(loaded.state)

# Cleanup when thread completes

store.delete_state("thread/user:1")

Integrating FileStateStore with Agent Runners

While you can interact with FileStateStore directly, aisuite's Runner class automatically manages persistence when you pass a store instance via the state_store parameter. According to the implementation in aisuite/agents/runner.py, the runner loads state before execution and persists it upon completion.

from aisuite import Agent, FileStateStore

# Initialize store and agent

store = FileStateStore()  # Defaults to .aisuite/state

agent = Agent(name="support-agent", ...)

# Runner automatically handles save/load cycles

runner = agent.get_runner(state_store=store)

# First call creates state file

result1 = runner.run(input="Hello, I need help")

# Subsequent call loads previous state automatically

result2 = runner.run(input="Actually, never mind")

Complete Working Example

This example demonstrates the full lifecycle: initialization, manual state management, and cleanup.

import aisuite as ai
from pathlib import Path

def run_persistent_workflow():
    # 1. Initialize store

    store = ai.FileStateStore(Path("./my_states"))
    
    thread_id = "workflow/invoice:12345"
    
    # 2. Check for existing state (resumable workflow)

    loaded = store.load_state(thread_id)
    
    if loaded:
        print(f"Resuming from revision {loaded.revision}")
        current_state = loaded.state
    else:
        print("Starting fresh workflow")
        current_state = ai.RunState(...)
    
    # 3. Execute agent steps...

    # current_state = agent.step(current_state)

    
    # 4. Save progress with metadata

    stored = store.save_state(
        thread_id=thread_id,
        state=current_state,
        metadata={"last_step": "extract_invoice_data", "confidence": 0.95}
    )
    
    print(f"Checkpoint saved: revision {stored.revision}")
    
    # 5. Cleanup on completion

    # store.delete_state(thread_id)

if __name__ == "__main__":
    run_persistent_workflow()

Summary

• FileStateStore provides file-based persistence for aisuite agents via JSON serialization in aisuite/agents/state_store.py.
• Default location is .aisuite/state, but custom paths are supported through the constructor.
• Optimistic concurrency is enforced through monotonic revision numbers; mismatched revisions raise StateConflictError.
• Deep-copied metadata allows attaching arbitrary context to state snapshots without side effects.
• Runner integration enables automatic persistence by passing the store to agent.get_runner(state_store=store).

Frequently Asked Questions

How does FileStateStore handle concurrent writes from multiple processes?

FileStateStore implements optimistic concurrency control. Each save_state() call accepts an optional revision parameter. If provided, the store compares it against the current stored revision; if they differ, it raises StateConflictError (as implemented at line 105 of state_store.py). To handle conflicts, catch this exception and implement a retry logic that reloads the latest state before reapplying your changes.

What is the default storage directory for FileStateStore?

When initialized without arguments, FileStateStore() defaults to the .aisuite/state directory relative to the current working directory. You can override this by passing a pathlib.Path object to the constructor, such as FileStateStore(Path("/custom/path")).

Can I store arbitrary metadata alongside the agent state?

Yes. The save_state() method accepts a metadata dictionary parameter that accepts any JSON-serializable data. The implementation deep-copies this metadata to prevent reference mutations from affecting the stored snapshot. Retrieve this metadata later through the stored.metadata attribute of the StoredRunState object returned by load_state().

How do I migrate from FileStateStore to a database-backed store?

Since FileStateStore implements the StateStore protocol defined in aisuite/agents/state_store.py, you can swap it for any alternative implementation (such as PostgreSQLStateStore) without changing your application logic. Simply pass the alternative store instance to agent.get_runner(state_store=...) or use it directly in place of FileStateStore in your persistence code.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}