Pydantic Graph Agents + Persistent State
Free · Open source (MIT) · Works with LangChain, CrewAI, AutoGen · No signup
Pydantic Graph agents typically lose their state between runs. Your carefully orchestrated nodes can't remember what happened in previous executions, breaking multi-turn workflows and forcing you to rebuild context from scratch every time.
The State Management Problem
When building pydantic graph agents, you quickly hit a wall: nodes execute, produce results, then vanish. The next time your pydantic-graph agent runs, it has zero memory of previous decisions, user preferences, or intermediate computations.
This breaks real-world scenarios:
- Multi-step research agents that need to remember findings across sessions
- Recommendation systems that should learn from user feedback
- Workflow orchestrators that pause and resume complex processes
- Agent conversations that span days or weeks
Most developers resort to hacky solutions: pickle files, JSON dumps, or passing massive state objects between function calls. These approaches are brittle, don't survive process restarts, and create tight coupling between your graph agent state logic and storage concerns.
You need persistent memory that survives restarts, works across different processes, and doesn't require complex infrastructure setup.
The Fix: Persistent Memory for Graph Agents
BotWire Memory gives your pydantic ai graph agents persistent key-value storage that works immediately:
pip install botwire
from botwire import Memory
from pydantic import BaseModel
from typing import Dict, Any
class AgentState(BaseModel):
user_preferences: Dict[str, Any] = {}
conversation_context: list = []
workflow_step: int = 0
def my_graph_node(state: AgentState, node_id: str):
# Initialize persistent memory for this agent
memory = Memory(f"agent-{node_id}")
# Load previous state
saved_state = memory.get("agent_state")
if saved_state:
state = AgentState.model_validate(saved_state)
# Your node logic here
state.workflow_step += 1
state.conversation_context.append("processed step")
# Persist updated state
memory.set("agent_state", state.model_dump())
return state
How It Works
The code above creates a persistent memory namespace for each agent instance. When your pydantic graph memory needs to survive restarts, the Memory class handles HTTP requests to a free backend service at botwire.dev.
Here's a more complete example showing cross-node state sharing:
from botwire import Memory
from datetime import datetime
class MultiNodeAgent:
def __init__(self, agent_id: str):
self.memory = Memory(f"graph-agent-{agent_id}")
def research_node(self, query: str):
# Store research findings
findings = {"query": query, "timestamp": datetime.now().isoformat()}
# Get existing research history
history = self.memory.get("research_history") or []
history.append(findings)
self.memory.set("research_history", history)
self.memory.set("last_activity", "research")
return findings
def analysis_node(self):
# Access research from previous node
history = self.memory.get("research_history") or []
last_activity = self.memory.get("last_activity")
if not history:
return {"error": "No research data found"}
analysis = {"findings_count": len(history), "last_research": history[-1]}
self.memory.set("analysis_result", analysis)
return analysis
# Works across different Python processes
agent = MultiNodeAgent("user-123")
agent.research_node("market trends")
# Process restarts, different machine, etc.
agent = MultiNodeAgent("user-123") # Same ID = same memory
result = agent.analysis_node() # Has access to research data
The memory persists automatically. You can also list all keys with memory.list_keys(), delete specific entries with memory.delete(key), or clear everything with memory.clear().
For time-sensitive data, set TTL: memory.set("temp_data", value, ttl_seconds=3600).
Framework Integration
BotWire includes helpers for popular agent frameworks. If you're using CrewAI alongside Pydantic graphs:
from botwire.memory import memory_tools
from crewai import Agent, Task, Crew
# Get memory tools for your crew
tools = memory_tools("my-crew-memory")
# Your crew agents can now remember/recall/list_memory
researcher = Agent(
role="Research Specialist",
tools=tools,
# Agent will have access to remember(), recall(), list_memory() functions
)
# The tools automatically namespace memory by crew ID
# and integrate with your existing pydantic graph workflows
When NOT to Use BotWire
BotWire Memory isn't suitable for:
- Vector similarity search — it's key-value storage, not a vector database
- High-frequency writes — free tier caps at 1000 writes/day per namespace
- Sub-millisecond latency — HTTP overhead makes it slower than in-memory solutions
FAQ
Why not just use Redis? Redis requires setup, hosting, and connection management. BotWire works immediately with zero configuration and has a generous free tier.
Is this actually free? Yes. 1000 writes/day per namespace, 50MB storage, unlimited reads. No signup required, no API keys, no credit card.
What about data privacy? Data is stored on botwire.dev servers. For sensitive use cases, you can self-host the open-source version (single FastAPI + SQLite service).
Get Started
Add persistent memory to your pydantic graph agents in under 60 seconds. Your nodes will finally remember what matters between executions.
pip install botwire
Full documentation and self-hosting guide at botwire.dev.