State-first Agent Runtime: Build AI agents driven by business state, not conversation history.
StateFlow is a lightweight runtime for building state-driven AI agents. Instead of managing ever-growing conversation histories, StateFlow agents operate on strongly-typed business states that represent only what matters for future decisions.
Traditional agent frameworks suffer from context bloat:
# Traditional approach
messages = [
{"role": "user", "content": "I want to return order 12345"},
{"role": "assistant", "content": "Let me check that order..."},
{"role": "function", "content": '{"status": "delivered", "amount": 599}'},
{"role": "assistant", "content": "I see the order was delivered..."},
# ... continues growing ...
]As the conversation grows:
- ❌ Token costs increase linearly
- ❌ Context gets cluttered with irrelevant history
- ❌ Debugging requires reading entire message logs
- ❌ State is implicit in natural language
# StateFlow approach
state = RefundState(
order_id="12345",
order_status="DELIVERED",
refund_eligible=True,
phase="AWAITING_APPROVAL",
completed_steps=["FETCH_ORDER", "VERIFY_USER"]
)With StateFlow:
- ✅ Constant token usage (state size doesn't grow)
- ✅ Explicit, validated business state
- ✅ Easy debugging via state diffs
- ✅ Built-in state history and replay
StateFlow is built on four simple concepts:
┌─────────────────────────────────────────┐
│ Skill │
│ "What rules should the agent follow?" │
└────────────────┬────────────────────────┘
↓
┌─────────────────────────────────────────┐
│ State │
│ "What does the agent currently know?" │
└────────────────┬────────────────────────┘
↓
┌─────────────────────────────────────────┐
│ Observation │
│ "What just happened in the world?" │
└────────────────┬────────────────────────┘
↓
┌─────────────────────────────────────────┐
│ Action │
│ "What should the agent do next?" │
└─────────────────────────────────────────┘
State(t) + Observation(t) + Skill
↓
LLM Decision
↓
State Patch + Action
↓
Validate & Apply Patch
↓
Execute Action
↓
Observation(t+1) → State(t+1)
# Using poetry (recommended)
poetry add stateflow
# Using pip
pip install stateflowimport asyncio
from stateflow import AgentRuntime, AgentState
from stateflow.llm.openai_adapter import OpenAIAdapter
from stateflow.tools.builtin import MockDatabaseTool
# Define your skill (what the agent should do)
SKILL_SPEC = """
You are a customer service agent processing refund requests.
Your goal:
1. Fetch the order information
2. Verify the order is eligible for refund
3. Mark the state as COMPLETED when done
Use the query_database tool to fetch order information.
"""
# Create tools
tools = [MockDatabaseTool()]
# Create runtime
runtime = AgentRuntime(
state_class=AgentState,
llm_adapter=OpenAIAdapter(model="gpt-4o-mini"),
skill_spec=SKILL_SPEC,
tools=tools,
max_iterations=10,
)
# Run the agent
async def main():
result = await runtime.run(
entity_id="ORDER_12345",
context={"user_id": "USER_001"}
)
if result.success:
print(f"✅ Task completed in {result.iterations} iterations")
print(f"Final state: {result.final_state}")
else:
print(f"❌ Task failed: {result.error}")
asyncio.run(main())Define domain-specific states using Pydantic:
from stateflow.core.state import StateModel, TaskPhase
from typing import Optional
class RefundState(StateModel):
"""State for refund processing agent."""
# Entity & Context
order_id: str
user_id: str
# Verified Facts
order_status: Optional[str] = None
refund_eligible: Optional[bool] = None
refund_amount: Optional[float] = None
# Progress
phase: TaskPhase = TaskPhase.INIT
completed_steps: list[str] = []
def is_finished(self) -> bool:
return self.phase in {TaskPhase.COMPLETED, TaskPhase.FAILED}
# Use your custom state
runtime = AgentRuntime(
state_class=RefundState, # Your custom state
llm_adapter=OpenAIAdapter(),
skill_spec=REFUND_SKILL_SPEC,
tools=[get_order_tool, create_refund_tool],
)from stateflow.tools.base import Tool, ToolResult
class GetOrderTool(Tool):
def __init__(self, db_client):
super().__init__(
name="get_order",
description="Retrieve order information by order ID"
)
self.add_parameter("order_id", "string", "The order ID", required=True)
self.db_client = db_client
async def execute(self, order_id: str) -> ToolResult:
try:
order = await self.db_client.fetch_order(order_id)
return ToolResult(success=True, data=order)
except Exception as e:
return ToolResult(success=False, error=str(e))StateFlow distinguishes between State and Memory:
| State | Memory |
|---|---|
| Current execution context | Long-term knowledge |
| Strongly typed, validated | Retrieved as needed |
| Fixed schema per domain | Unstructured or semi-structured |
Example: order_status="DELIVERED" |
Example: "User complained about shipping last year" |
State = What I need to know right now to make the next decision
Memory = Background knowledge that might be relevant
| Feature | StateFlow | LangChain/LangGraph |
|---|---|---|
| Context Model | Explicit state | Message history |
| Token Usage | Constant | Grows with history |
| Debugging | State diffs | Message logs |
| Type Safety | Full (Pydantic) | Partial |
| Complexity | Minimal (~3K LOC) | Large framework |
| Focus | Long-running business agents | General-purpose chains |
✅ Good fit:
- Customer service workflows
- Order processing
- IT troubleshooting
- Multi-step business processes
- Agents that need auditability
❌ Not ideal for:
- Simple one-shot queries
- Conversational chatbots focused on dialogue
- Agents that need extensive RAG
┌─────────────────────────────────────────────────────────────┐
│ FastAPI Gateway │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Create │ │ Query │ │ Continue │ │ Replay │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└────────────────────────┬────────────────────────────────────┘
│
┌────────────────────────┴────────────────────────────────────┐
│ StateFlow Runtime │
│ ┌────────────────────────────────────────────────────┐ │
│ │ Agent Execution Loop │ │
│ │ State → LLM → Patch → Validate → Action │ │
│ └────────────────────────────────────────────────────┘ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ State │ │ Store │ │ Tools │ │
│ │ Schema │ │ (Redis) │ │ Registry │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────┘
# Clone the repository
git clone https://github.com/yourusername/stateflow.git
cd stateflow
# Install dependencies
poetry install
# Start development services
docker-compose up -d
# Copy environment template
cp .env.example .env
# Edit .env with your API keys# Run all tests
poetry run pytest
# Run with coverage
poetry run pytest --cov=stateflow --cov-report=html
# Run only unit tests
poetry run pytest tests/unit -m unit
# Run integration tests
poetry run pytest tests/integration -m integration# Format code
poetry run ruff format .
# Lint
poetry run ruff check .
# Type checking
poetry run mypy src/stateflow- Core state management
- Basic runtime loop
- OpenAI adapter
- Tool system
- Unit tests (>80% coverage)
- FastAPI REST API
- Redis/PostgreSQL persistence
- State replay functionality
- Integration tests with Docker
- Prometheus metrics
- State diff visualization
- CLI debugging tool
- Multi-LLM support (Claude, Qwen)
- Plugin system
- Web dashboard
StateFlow is inspired by the SKILL.state paper, which proposes using structured execution states instead of conversation histories for long-horizon agent tasks.
Key insight from the paper:
"Agents should maintain what they currently know (state), not what they previously said (history)."
Contributions are welcome! Please read our Contributing Guide for details.
MIT License - see LICENSE for details.
If you use StateFlow in your research, please cite:
@software{stateflow2024,
title = {StateFlow: State-first Agent Runtime},
author = {StateFlow Contributors},
year = {2024},
url = {https://github.com/yourusername/stateflow}
}Built with ❤️ by the open source community