Framework Integrations

ZeroLock works with any AI framework. This guide shows how to add coordination to agents built with LangGraph, CrewAI, AutoGen, and LlamaIndex.

How Integration Works

ZeroLock doesn't replace your AI framework—it adds coordination. Your agents still use LangGraph, CrewAI, or whatever you prefer. ZeroLock ensures they don't step on each other.

plaintext

┌─────────────────────────────────────────────────────────────┐
│                    Your AI Framework                         │
│      (LangGraph / CrewAI / AutoGen / LlamaIndex)            │
│                                                              │
│   Agent 1        Agent 2        Agent 3                      │
│   ┌─────┐        ┌─────┐        ┌─────┐                      │
│   │ LLM │        │ LLM │        │ LLM │                      │
│   └──┬──┘        └──┬──┘        └──┬──┘                      │
└──────┼──────────────┼──────────────┼────────────────────────┘
       │              │              │
       └──────────────┼──────────────┘
                      │ coordinates through
                      ▼
┌─────────────────────────────────────────────────────────────┐
│                   ZeroLock Gateway                           │
│         (shared state, task claiming, locks)                 │
└─────────────────────────────────────────────────────────────┘

Prerequisites

bash

# Start the gateway
cd packages/gateway && pnpm dev
 
# For LLM-powered examples (uses Google Gemini)
export GOOGLE_API_KEY=your-gemini-api-key

LangGraph: Research Team

Location: examples/frameworks/langgraph/research-team/

What It Does

Three AI researchers claim topics from a queue, research them with Gemini, and submit findings. A synthesizer agent combines everything into a final report.

Why You Need ZeroLock

Without coordination, all three researchers might grab the same topic:

plaintext

❌ Without ZeroLock:
[researcher-1] Researching: AI Agent Architectures
[researcher-2] Researching: AI Agent Architectures  ← Same topic!
[researcher-3] Researching: AI Agent Architectures  ← Same topic!
 
Result: 3 LLM calls for 1 topic, wasted time and money

plaintext

✓ With ZeroLock:
[researcher-1] Claimed: AI Agent Architectures
[researcher-2] Lost race, trying next → Claimed: Vector Databases
[researcher-3] Claimed: LLM Frameworks
 
Result: Each topic researched exactly once

Architecture

plaintext

┌──────────────────────────────────────────────────────────┐
│                     Research Team                         │
│                                                           │
│  ┌───────────┐  ┌───────────┐  ┌───────────┐             │
│  │Researcher1│  │Researcher2│  │Researcher3│             │
│  │ (Gemini)  │  │ (Gemini)  │  │ (Gemini)  │             │
│  └─────┬─────┘  └─────┬─────┘  └─────┬─────┘             │
│        │              │              │                    │
│        └──────────────┼──────────────┘                    │
│                       ▼                                   │
│        ┌─────────────────────────────┐                   │
│        │      ZeroLock Gateway       │                   │
│        │  ┌─────────┬────────────┐   │                   │
│        │  │ Topics  │  Findings  │   │                   │
│        │  │ (claim) │   (CRDT)   │   │                   │
│        │  └─────────┴────────────┘   │                   │
│        └──────────────┬──────────────┘                   │
│                       ▼                                   │
│              ┌───────────────┐                           │
│              │  Synthesizer  │                           │
│              │   (Gemini)    │                           │
│              └───────────────┘                           │
└──────────────────────────────────────────────────────────┘

Run It

bash

cd examples/frameworks/langgraph/research-team
pip install -r requirements.txt
GOOGLE_API_KEY=xxx python main.py

Key Code: Claiming Topics in a LangGraph Node

python

from langgraph.graph import StateGraph, START, END
from zerolock_sdk import ZeroLockClient
from zerolock_sdk.task_claim import claim_task, get_pending_tasks
 
async def claim_topic_node(state: ResearcherState) -> ResearcherState:
    """LangGraph node that claims a research topic."""
 
    async with ZeroLockClient(GATEWAY, agent_id=state["agent_id"]) as client:
        # Step 1: See what topics are available
        pending = await get_pending_tasks(client, "topics")
 
        if not pending:
            return {**state, "done": True}  # All topics taken
 
        # Step 2: Try to claim the first available topic
        topic = pending[0]
        result = await claim_task(client, "topics", topic.id, state["agent_id"])
 
        if not result.claimed:
            # Another researcher got it first — try again
            return {**state, "current_topic": None}
 
        # We got the topic!
        return {**state, "current_topic": topic.payload}

Key Code: Submitting Findings

python

async def submit_findings_node(state: ResearcherState) -> ResearcherState:
    """Submit findings using CRDT append — never conflicts with other researchers."""
 
    async with ZeroLockClient(GATEWAY, agent_id=state["agent_id"]) as client:
        await client.append_item("findings", {
            "topic": state["current_topic"]["title"],
            "researcher": state["agent_id"],
            "content": state["research_output"],
        })
 
    return state

LangGraph Flow

plaintext

START → claim_topic → [route] → research_topic → submit_findings → claim_topic
              │                                                         │
              └─────────────── (if no more topics) ───────────────────►END

CrewAI: Coordinated Crew

Location: examples/frameworks/crewai/coordinated-crew/

What It Does

A CrewAI crew where agents use ZeroLock tools to claim tasks and submit results. The framework handles agent orchestration; ZeroLock ensures they coordinate properly.

How It Works

ZeroLock provides a create_zerolock_tools() adapter that creates standard CrewAI tools:

python

from crewai import Agent, Task, Crew
from zerolock_crewai import create_zerolock_tools
 
# Create ZeroLock tools for your agents
tools = create_zerolock_tools(
    gateway_url="http://localhost:8080",
    queue_shard="tasks",
    findings_shard="results"
)
 
# Give tools to your agent
researcher = Agent(
    role="Research Analyst",
    goal="Research assigned topics thoroughly",
    tools=tools,  # ← ZeroLock tools
    backstory="You claim topics before researching them."
)
 
# Create crew as normal
crew = Crew(
    agents=[researcher],
    tasks=[Task(
        description="Claim a topic and research it",
        agent=researcher
    )]
)
 
crew.kickoff()

Available Tools

Tool	What It Does
`claim_task_tool`	Claim a task from the queue
`complete_task_tool`	Mark task as done
`submit_finding_tool`	Add a finding (CRDT append)
`check_pending_tool`	List available tasks

Run It

bash

cd examples/frameworks/crewai/coordinated-crew
pip install -r requirements.txt
GOOGLE_API_KEY=xxx python main.py

AutoGen: Group Chat

Location: examples/frameworks/autogen/group-chat/

What It Does

AutoGen agents in a group chat use ZeroLock to coordinate turn-taking. Only one agent speaks at a time—no interruptions.

The Problem

Without coordination, agents might speak simultaneously:

plaintext

❌ Without ZeroLock:
[Agent-A] "I think we should..."
[Agent-B] "Let me add that..."   ← Interrupting!
[Agent-C] "Actually, I believe..." ← Talking over!
 
Result: Chaotic, unreadable conversation

plaintext

✓ With ZeroLock:
[Agent-A] Claims turn → speaks
[Agent-B] Waits for turn → speaks
[Agent-C] Waits for turn → speaks
 
Result: Orderly discussion

Key Code: Turn-Taking

python

async def claim_turn(client: ZeroLockClient, agent_id: str) -> bool:
    """Claim the right to speak."""
 
    shard = await client.get_shard("turn-order")
    current_turn = shard.value
 
    if current_turn["status"] == "open":
        try:
            await client.commit_shard("turn-order", {
                "status": "claimed",
                "speaker": agent_id,
                "turn_number": current_turn["turn_number"],
            }, shard.version)
            return True  # We got the turn!
 
        except StaleCommitError:
            return False  # Another agent claimed it
 
    return False

Run It

bash

cd examples/frameworks/autogen/group-chat
pip install -r requirements.txt
GOOGLE_API_KEY=xxx python main.py

LlamaIndex: Distributed RAG

Location: examples/frameworks/llamaindex/distributed-rag/

What It Does

Multiple workers index documents in parallel. ZeroLock ensures each document is indexed exactly once, and all embeddings merge into a shared index.

The Problem

Without coordination, workers duplicate effort:

plaintext

❌ Without ZeroLock:
[indexer-1] Indexing: document-1.pdf
[indexer-2] Indexing: document-1.pdf  ← Same document!
[indexer-3] Indexing: document-1.pdf  ← Same document!
 
Result: 3x compute, wasted embeddings

plaintext

✓ With ZeroLock:
[indexer-1] Claimed: document-1.pdf → indexing
[indexer-2] Lost race → Claimed: document-2.pdf → indexing
[indexer-3] Claimed: document-3.pdf → indexing
 
Result: Each document indexed once, 3x throughput

Architecture

plaintext

┌───────────────────────────────────────────────────────────┐
│                   Document Corpus                          │
│  [doc1.pdf] [doc2.pdf] [doc3.pdf] ... [docN.pdf]          │
└─────────────────────────┬─────────────────────────────────┘
                          ▼
┌───────────────────────────────────────────────────────────┐
│                   ZeroLock Queue                           │
│  (Task Claiming - each doc claimed by one worker)         │
└─────────────────────────┬─────────────────────────────────┘
          ┌───────────────┼───────────────┐
          ▼               ▼               ▼
    ┌──────────┐    ┌──────────┐    ┌──────────┐
    │ Indexer 1│    │ Indexer 2│    │ Indexer 3│
    └────┬─────┘    └────┬─────┘    └────┬─────┘
         └────────────────┼────────────────┘
                          ▼
┌───────────────────────────────────────────────────────────┐
│                   Shared Index                             │
│  (CRDT Merge - embeddings from all workers)               │
└───────────────────────────────────────────────────────────┘

Key Code

python

from llama_index.core import Document
from zerolock_sdk import ZeroLockClient
from zerolock_sdk.task_claim import claim_task, complete_task, get_pending_tasks
 
async def indexer_worker(worker_id: str):
    async with ZeroLockClient(GATEWAY, agent_id=worker_id) as client:
        while True:
            # Step 1: Get documents waiting to be indexed
            pending = await get_pending_tasks(client, "docs-queue")
            if not pending:
                break
 
            # Step 2: Claim one
            doc = pending[0]
            result = await claim_task(client, "docs-queue", doc.id, worker_id)
 
            if not result.claimed:
                continue  # Another worker got it
 
            # Step 3: Index the document (LlamaIndex)
            embedding = embed_model.get_text_embedding(doc.payload["text"])
 
            # Step 4: Add to shared index (CRDT - never conflicts)
            await client.append_item("shared-index", {
                "doc_id": doc.id,
                "embedding": embedding,
                "metadata": doc.payload["metadata"]
            })
 
            # Step 5: Mark complete
            await complete_task(client, "docs-queue", doc.id, worker_id)

Run It

bash

cd examples/frameworks/llamaindex/distributed-rag
pip install -r requirements.txt
python main.py

Summary: Integration Patterns

Framework	Integration Method	Primary Use Case
LangGraph	SDK calls in graph nodes	Research teams, workflows
CrewAI	Tool adapter	Agent crews with shared tasks
AutoGen	Decision shards	Group chat coordination
LlamaIndex	Task claiming + CRDT	Distributed indexing

Common Pattern: Async Context Manager

All integrations use the same basic pattern:

python

async with ZeroLockClient(GATEWAY, agent_id="my-agent") as client:
    # ZeroLock operations here
    await client.get_shard("my-data")
    await client.append_item("my-list", item)

Agent ID Best Practices

Use descriptive agent IDs:

python

# Good
agent_id = "researcher-1"
agent_id = "pr-reviewer-security"
agent_id = "indexer-worker-3"
 
# Bad
agent_id = "a1"
agent_id = "x"

Next Steps

Quickstart Examples — Core patterns without frameworks
Use Case Examples — PR review, codebase migration, deep research
Advanced Patterns — Deep dives into task claiming, locking, approval gates