vstash-swarm

Multi-agent shared memory built on vstash.

Agents communicate through a shared SQLite database instead of passing state between them. Each agent publishes findings to a layer and subscribes to others — no message brokers, no servers, no configuration.

from vstash_swarm import SwarmMemory
from vstash_swarm.models import Layer

# Each agent gets shared + private memory
mem = SwarmMemory(agent_id="researcher_1", project="my_task")

mem.publish("Found that X correlates with Y...", layer=Layer.RESEARCH)
results = mem.subscribe("key findings", layer=Layer.SYNTHESIS)
mem.think("Private reasoning, not visible to other agents")

How it works

START → researchers (parallel) → synthesizer → executor → judge → END
           │                          │              │          │
           └──── Layer.RESEARCH ──────┘              │          │
                                    Layer.SYNTHESIS──┘          │
                                                   Layer.RESULT─┘

Agents don't share a state dict — they share a vstash database. The SQLite WAL mode handles concurrent writes safely. Memory persists across runs: a second run benefits from the first run's research.

Install

pip install vstash-swarm

Requires vstash ≥ 0.5.3:

pip install "vstash[cerebras]"   # or [ollama] / [openai]
export CEREBRAS_API_KEY=...

Quick start

from vstash_swarm import SwarmMemory, ResearcherAgent, SynthesizerAgent, ExecutorAgent, JudgeAgent
from vstash_swarm.models import Layer

shared_db = "~/.vstash/swarm.db"
project = "my_research"

# Phase 1: researchers ingest sources (run in parallel threads)
with SwarmMemory("researcher", project=project, shared_db=shared_db) as mem:
    agent = ResearcherAgent("researcher", mem, sources=["docs/paper.pdf"])
    agent.run("What are the key contributions?")

# Phase 2: synthesizer distills research with LLM
with SwarmMemory("synthesizer", project=project, shared_db=shared_db) as mem:
    SynthesizerAgent("synthesizer", mem).run("What are the key contributions?")

# Phase 3: executor produces final answer
with SwarmMemory("executor", project=project, shared_db=shared_db) as mem:
    result = ExecutorAgent("executor", mem).run("What are the key contributions?")

# Phase 4: judge evaluates quality
with SwarmMemory("judge", project=project, shared_db=shared_db) as mem:
    score = JudgeAgent("judge", mem).run("What are the key contributions?")

LangGraph integration

from vstash_swarm.langgraph_integration import build_swarm_graph

graph = build_swarm_graph(
    researchers={
        "researcher_1": ["docs/paper.pdf"],
        "researcher_2": ["https://arxiv.org/abs/2310.06825"],
    },
    project="my_research",
    shared_db="~/.vstash/swarm.db",
)

result = graph.invoke({"task": "What are the key contributions?", "final_answer": "", "evaluation": ""})
print(result["final_answer"])
print(result["evaluation"])

Memory layers

Layer	Who writes	Who reads
TASK	Orchestrator	All agents
RESEARCH	Researcher agents	Synthesizer
SYNTHESIS	Synthesizer	Executor
DECISION	Planner	Executor
RESULT	Executor	Judge, user

Custom agents

from vstash_swarm.agents import BaseAgent
from vstash_swarm.models import Layer

class ValidatorAgent(BaseAgent):
    def process(self, task: str) -> str:
        synthesis = self.memory.subscribe(task, layer=Layer.SYNTHESIS, top_k=5)
        # validate, cross-check, flag inconsistencies...
        verdict = "All claims verified." if synthesis else "Nothing to validate."
        self.memory.publish(verdict, layer=Layer.RESULT, tags="validation")
        return verdict

Scaling

vstash-swarm uses SQLite WAL for concurrent reads and serialized writes. Comfortable up to ~50k chunks per shared DB. For larger workloads, swap the vector backend to LanceDB (HNSW index) — the SwarmMemory API stays the same.

License

MIT