Shared State

Single source of truth for all agents

SharedState provides a single source of truth for all agents in a workflow. No more passing data around or wondering who has the latest version.

Basic Usage

from splinter.coordination import SharedState

state = SharedState(initial_data={"topic": "AI trends"})

# Set values (dot notation)
state.set("research.findings", ["trend1", "trend2"])
state.set("research.sources", ["source1", "source2"])

# Get values
findings = state.get("research.findings")
topic = state.get("topic")

Nested Access

Use dot notation for nested paths:

state.set("research.findings.primary", ["a", "b", "c"])
state.set("research.findings.secondary", ["d", "e"])

primary = state.get("research.findings.primary")  # ["a", "b", "c"]
all_findings = state.get("research.findings")  # {"primary": [...], "secondary": [...]}

Versioning

Every change increments the version:

state = SharedState()
print(state.version)  # 0

state.set("a", 1)
print(state.version)  # 1

state.set("b", 2)
print(state.version)  # 2

Snapshots

Save and restore state:

# Take snapshot
snapshot = state.snapshot()

# Make changes
state.set("oops", "mistake")
state.set("another", "error")

# Restore to snapshot
state.restore(snapshot)
# "oops" and "another" are gone

State History

Track what changed:

state.set("a", 1)
state.set("a", 2)
state.set("a", 3)

history = state.get_history("a")
# [
#   {"value": 1, "version": 1, "time": "..."},
#   {"value": 2, "version": 2, "time": "..."},
#   {"value": 3, "version": 3, "time": "..."},
# ]

Watching for Changes

def on_change(path, old_value, new_value):
    print(f"{path}: {old_value} → {new_value}")

state.watch("research.*", on_change)

state.set("research.findings", [1, 2, 3])
# Prints: research.findings: None → [1, 2, 3]

Atomic Updates

Multiple changes as one version:

with state.transaction():
    state.set("a", 1)
    state.set("b", 2)
    state.set("c", 3)
# All three are version N, not N, N+1, N+2

Getting All State

# Full state dict
all_data = state.to_dict()

# Just keys
keys = state.keys()

# Check existence
if state.has("research.findings"):
    # ...

Best Practices

Use descriptive paths

research.findings.primary is better than data.a.b.

Snapshot before risky operations

Easy rollback if something goes wrong.

Use transactions for related changes

Keeps versions meaningful.

Watch for changes in long workflows

Helps debug what happened.