Async State Store

Non-blocking key-value state persistence with the AsyncStateStore protocol

Async state stores provide non-blocking key-value and hash persistence so native-async backends never block the event loop.

from praisonaiagents import Agent

agent = Agent(
    name="Assistant",
    instructions="Remember the user's preferences across turns.",
)
agent.start("What's my preferred language?")

The user asks a follow-up; an async state store fetches cached preferences without blocking other agents on the same loop.

Quick Start

1. Subclass AsyncStateStore

   Implement a native-async backend by inheriting from AsyncStateStore:

   from praisonai.persistence import AsyncStateStore
   from typing import Any, Dict, List, Optional
   
   class MyAsyncStateStore(AsyncStateStore):
       async def get(self, key: str) -> Optional[Any]:
           ...
   
       async def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
           ...
   
       async def delete(self, key: str) -> bool:
           ...
   
       async def exists(self, key: str) -> bool:
           ...
   
       async def keys(self, pattern: str = "*") -> List[str]:
           ...
   
       async def ttl(self, key: str) -> Optional[int]:
           ...
   
       async def expire(self, key: str, ttl: int) -> bool:
           ...
   
       async def hget(self, key: str, field: str) -> Optional[Any]:
           ...
   
       async def hset(self, key: str, field: str, value: Any) -> None:
           ...
   
       async def hgetall(self, key: str) -> Dict[str, Any]:
           ...
   
       async def hdel(self, key: str, *fields: str) -> int:
           ...
   
       async def close(self) -> None:
           ...

2. Wire It Into an Agent

   Pass the store through the persistence config so the agent dispatches async calls natively:

   from praisonaiagents import Agent
   
   store = MyAsyncStateStore()
   
   agent = Agent(
       name="Assistant",
       instructions="Remember the user's preferences across turns.",
       state_store=store,
   )
   agent.start("What's my preferred language?")

How It Works

A sync caller inside a running loop dispatches to the async store through PraisonAIDB._dispatch_async.

Abstract Methods

Concrete helpers — get_json(key), set_json(key, value, ttl=None), __aenter__, __aexit__ — ship on the ABC and need no override.

Configuration Options

Common Patterns

Custom Async Backend Inheritance

Subclass AsyncStateStore (not StateStore) when your backend is natively async:

from praisonai.persistence import AsyncStateStore

class RedisAsyncStateStore(AsyncStateStore):
    def __init__(self, client):
        self._client = client

    async def get(self, key):
        return await self._client.get(key)

    async def set(self, key, value, ttl=None):
        await self._client.set(key, value, ex=ttl)

    async def close(self):
        await self._client.aclose()
    # ... implement the remaining abstract methods

async with Context Manager

The ABC defines __aenter__ / __aexit__, so async with closes the store automatically:

async with MyAsyncStateStore() as store:
    await store.set("prefs:lang", "en", ttl=3600)
    lang = await store.get("prefs:lang")

get_json / set_json Helpers

The concrete helpers serialize to JSON on write. On read, a stored str is parsed with json.loads; any non-str value is returned as-is:

async with MyAsyncStateStore() as store:
    await store.set_json("prefs", {"lang": "en", "theme": "dark"})
    prefs = await store.get_json("prefs")  # {"lang": "en", "theme": "dark"}

Best Practices

# ✅ Good - native async dispatch
class MyStore(AsyncStateStore):
    async def get(self, key): ...

# ❌ Bad - sync ABC forces run_sync shims
class MyStore(StateStore):
    def get(self, key): ...

# ✅ Good - automatic cleanup
async with MyAsyncStateStore() as store:
    await store.get("key")

# ❌ Bad - leaked connections
store = MyAsyncStateStore()
await store.get("key")

# ✅ Good - pure async
async with store:
    await store.set("key", "value")

# ❌ Bad - calling an async method without await
store.set("key", "value")  # returns an un-awaited coroutine

Related