Async Safety

How knowledge search, memory writes, and the in-memory adapter behave under concurrent async tasks

Under async execution, PraisonAI dispatches blocking I/O to worker threads so parallel tasks actually run in parallel, and the built-in in-memory store stays safe under concurrent writes.

import asyncio
from praisonaiagents import Agent

# Two unnamed agents in the same process — skill grants stay scoped per instance.
reviewer = Agent(skills=["./skills/file-reviewer"])
worker   = Agent(instructions="Do a job.")

async def main():
    await reviewer.astart("Review report.pdf")   # knowledge.search runs on a worker thread
    await worker.astart("Summarise the review")  # store_in_memory in the callback also off-loop

asyncio.run(main())

Both knowledge.search(...) and task-callback memory writes are offloaded automatically — you don’t call anything new. What you see is parallel async tasks that share a knowledge base or memory now actually running in parallel.

Quick Start

1. Parallel agents sharing knowledge

   An asyncio.gather(...) of agents that query the same knowledge base runs in parallel — no lookup blocks the loop.

   import asyncio
   from praisonaiagents import Agent
   
   agent = Agent(
       name="Research Assistant",
       instructions="Answer questions using the knowledge base.",
       knowledge=["handbook.pdf"],
   )
   
   async def main():
       await asyncio.gather(
           agent.astart("Summarise the onboarding policy"),
           agent.astart("What is the refund window?"),
       )
   
   asyncio.run(main())

2. Parallel tasks writing memory

   Task-callback memory writes offload to worker threads, so a fan-out of tasks doesn’t stall on one slow write.

   import asyncio
   from praisonaiagents import Agent, Task, PraisonAIAgents
   
   worker = Agent(name="Worker", instructions="Do the work.", memory=True)
   
   agents = PraisonAIAgents(
       agents=[worker],
       tasks=[
           Task(description="Task A", agent=worker, async_execution=True),
           Task(description="Task B", agent=worker, async_execution=True),
           Task(description="Task C", agent=worker, async_execution=True),
       ],
       # process defaults to "sequential"; async_execution on tasks fans them out.
   )
   
   asyncio.run(agents.astart())

3. Parallel agents with guardrails

   An asyncio.gather(...) of agents where one carries a string guardrail runs in parallel — the guardrail’s validation LLM call offloads to a worker thread instead of stalling the fan-out.

   import asyncio
   from praisonaiagents import Agent
   
   agent1 = Agent(
       name="Writer",
       instructions="Write a short blurb.",
       guardrail="Ensure the blurb makes no unverifiable claims.",
   )
   agent2 = Agent(
       name="Summarizer",
       instructions="Summarise the release notes.",
   )
   
   async def main():
       await asyncio.gather(
           agent1.astart("Blurb for the new dashboard"),
           agent2.astart("Summarise v2.0 release notes"),
       )
   
   asyncio.run(main())

How It Works

sequenceDiagram
    participant Loop as Event Loop
    participant Agent
    participant Worker as Worker Thread
    participant Knowledge
    participant Memory

    Agent->>Loop: astart(...)
    Loop->>Worker: to_thread(knowledge.search)
    Worker->>Knowledge: embedding + vector search
    Knowledge-->>Worker: results
    Worker-->>Loop: results (event loop stayed responsive)
    Loop->>Worker: to_thread(store_in_memory)
    Worker->>Memory: embedding + DB write (RLock-guarded)
    Memory-->>Worker: id
    Worker-->>Loop: done
    Loop-->>Agent: continues next coroutine

What runs off the event loop

Both were synchronous embedding + vector-store / DB calls that previously blocked the event loop for the whole coroutine lifetime. An asyncio.gather(...) fan-out would serialise on the slowest lookup or write instead of running concurrently. Offloading them to threads keeps the loop free.

A string / LLMGuardrail output guardrail fires a blocking LLM call under the hood to validate the agent’s output. On the async path this call is now offloaded via loop.run_in_executor(...), so parallel tasks don’t serialise on one task’s guardrail round-trip (PR #4469). Contextvars are preserved across the offload with copy_context_to_callable, so trace emission and session context stay intact.

Session auto-save (Agent(memory=MemoryConfig(auto_save="…"))) does a FileLock + JSON read-modify-write on the process-wide DefaultSessionStore singleton. Before PR #5258 that synchronous write ran inline on the async call sites, so one async agent’s auto-save blocked every other async agent sharing the loop — a fast agent gathered alongside a slow-writing one would visibly stall. The write is now offloaded via asyncio.to_thread at every async call site, and an internal lock serialises the shared _auto_save_last_index bookkeeping so two concurrent async runs sharing one Agent can’t double-persist the same turns.

Thread-safe in-memory adapter

The built-in InMemoryAdapter guards its store, search, delete, and reset methods with a re-entrant lock (RLock). Once writes move off the event loop, parallel async writes from an asyncio.gather(...) fan-out cannot produce duplicate ids or lose entries.

If you write a custom adapter that will be shared across async agents, do the same — guard its mutating methods with a lock.

import threading

class MyAdapter:
    def __init__(self):
        self._lock = threading.RLock()
        self._store = {}

    def store_long_term(self, memory_id, content, **kwargs):
        with self._lock:
            self._store[memory_id] = content

Best Practices

Related