Distributed agent framework for production-grade AI systems

Quick Start • Features • Agent Example • Tool Example • Dashboard

What is Laddr?
Features
Quick Start
Agent Example
Tool Example
FastAPI Runtime
Dashboard & Observability
Architecture
License

What is Laddr?

Laddr is a python framework for building multi-agent systems where agents communicate, delegate tasks, and execute work in parallel. Think of it as a microservices architecture for AI agents — with built-in message queues, observability, and horizontal scalability.

Two Operating Modes

Coordinator-Orchestrator Mode
Dynamic, autonomous workflows where a coordinator agent analyzes tasks, delegates to specialist agents, and synthesizes results. The coordinator makes real-time decisions about routing and orchestration.

Sequential Deterministic Workflow Mode
Pre-defined, predictable pipelines where tasks flow through agents in a fixed sequence. Each step is explicitly defined with inputs, outputs, and dependencies.

Features

Scalability & Performance

Horizontal scaling — Scale each agent independently based on workload
Multi-worker support — Run multiple workers per agent for high throughput
Automatic load balancing — Redis Streams distribute tasks across workers
Fault tolerance — Worker failures don't stop the system

Observability & Monitoring

Complete tracing — Every agent action automatically logged (SQLite/PostgreSQL)
Langfuse integration — Optional external tracing with automatic span creation
Real-time metrics — Job status, duration, success rates, queue depth
Interactive dashboard — Beautiful UI to monitor agents, traces, batches, and logs
Playground — Test agents interactively with live streaming responses

Developer Experience

Clean CLI — Minimal, modern interface for project management
Hot reload — Code changes reflect instantly in development mode
Type safety — Full Python type hints throughout the codebase
Simple setup — One command (laddr init) to scaffold complete projects

AI-Optimized Architecture

Built-in delegation — Agents can delegate tasks to other agents seamlessly
Batch operations — Execute multiple tasks in parallel with full tracking and WebSocket monitoring
Parallel execution — Process multiple independent tasks simultaneously
Context management — Automatic artifact storage for large payloads (>1MB)
Smart retries — Configurable retry logic and timeout handling per agent

Production Ready

Docker native — Everything runs in containers with docker-compose
Queue-based messaging — Redis Streams for reliable, distributed communication
Artifact storage — MinIO/S3 for handling large files and documents
REST API — FastAPI server with auto-generated OpenAPI documentation
API security — Optional API key authentication for all endpoints

Extensibility

Custom tools — Add any Python function as an agent tool with @tool decorator
Override system tools — Extend delegation and storage with your own implementations
LLM agnostic — Works with Gemini, OpenAI, Anthropic, Groq, Ollama, and local models
Pluggable backends — Swap Redis, PostgreSQL, or storage providers easily

Quick Start

Installation

pip install laddr

Create Your First Project

# Initialize a new project
laddr init my-agent-system
cd my-agent-system

Configure API Keys

Edit .env file:

# Required for web search tools
SERPER_API_KEY=your_serper_api_key  # Get from https://serper.dev

# LLM API Keys (choose one or more)
GEMINI_API_KEY=your_gemini_key      # Get from https://aistudio.google.com
OPENAI_API_KEY=your_openai_key      # Get from https://platform.openai.com

# Or use Ollama for local models (free, private, offline)
# Install: curl -fsSL https://ollama.ai/install.sh | sh
# Then: ollama pull gemma2:2b
OLLAMA_BASE_URL=http://localhost:11434
LLM_BACKEND=ollama  # Use local models instead of cloud APIs

# Optional: API key authentication
LADDR_API_KEY=your_api_key_here  # Protect API endpoints

# Optional: Langfuse external tracing
LANGFUSE_PUBLIC_KEY=your_public_key
LANGFUSE_SECRET_KEY=your_secret_key
LANGFUSE_HOST=https://cloud.langfuse.com

💡 Using Ollama? See the Ollama Integration Guide for complete setup instructions including Docker configuration.

Start the System

# Start all services (API, workers, database, Redis, MinIO)
laddr run dev

This starts:

✅ API Server at http://localhost:8000
✅ Dashboard at http://localhost:5173
✅ SQLite for traces and job history (default, or PostgreSQL if configured)
✅ Redis for message queue
✅ MinIO for artifact storage
✅ 2 agents: coordinator (orchestrator) and researcher (specialist)

Test Your Agents

Open the dashboard at http://localhost:5173/playground or use the API:

curl -X POST http://localhost:8000/api/jobs \
  -H "Content-Type: application/json" \
  -d '{
    "agent_name": "coordinator",
    "inputs": {
      "query": "What are the best hotels in Shimla?"
    }
  }'

Agent Example

Agents are defined in simple Python files. Here's what laddr init generates:

"""
Coordinator agent - orchestrates research tasks
"""
from laddr import Agent
from laddr.llms import gemini

coordinator = Agent(
    name="coordinator",
    role="Research Task Coordinator",
    goal="Coordinate research tasks by delegating to specialist agents",
    backstory="""You are a research coordinator who manages information 
    gathering tasks. You delegate specific queries to researcher agents 
    and compile their findings into comprehensive summaries.""",
    
    llm=gemini("gemini-2.0-flash"),
    
    # Coordinator can delegate to other agents
    is_coordinator=True,
    available_agents=["researcher"],
    
    # Behavior settings
    max_iterations=15,
    max_tool_calls=50,
    timeout=600,
    
    # Built-in delegation tools provided automatically
    tools=[],
    
    instructions="""
    ## Your workflow:
    1. Receive user query
    2. Delegate research to 'researcher' agent using system_delegate_task
    3. Wait for results (use wait_for_response=true)
    4. Synthesize findings into comprehensive answer
    5. Return final result to user
    
    ## Delegation format:
    {
      "type": "tool",
      "tool": "system_delegate_task",
      "params": {
        "agent_name": "researcher",
        "task": "Search for best hotels in Shimla with reviews",
        "wait_for_response": true,
        "timeout_seconds": 90
      }
    }
    """
)

Key Features:

Coordinators delegate work to specialists
Specialists use tools to complete tasks
LLM-powered decision making
Configurable behavior and limits
Clear instructions guide agent behavior

Tool Example

Tools are Python functions that agents can call. Here's a web search tool from the template:

"""
Web search tool using Serper.dev API
"""
from typing import Dict
import os
import requests
from laddr import tool

@tool(
    name="web_search",
    description="Search the web using Serper.dev API. Returns title, URL, and snippet for each result.",
    parameters={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Search query (be specific for better results)"
            },
            "max_results": {
                "type": "integer",
                "description": "Maximum number of results (1-10, default 5)",
                "default": 5
            }
        },
        "required": ["query"]
    }
)
def web_search(query: str, max_results: int = 5) -> Dict:
    """
    Search the web and return structured results.
    
    Returns:
        {
            "query": str,
            "results": [
                {
                    "title": str,
                    "link": str,
                    "snippet": str,
                    "site": str
                },
                ...
            ],
            "count": int,
            "status": "success" | "error"
        }
    """
    api_key = os.getenv("SERPER_API_KEY")
    if not api_key:
        return {
            "status": "error",
            "error": "SERPER_API_KEY not set"
        }
    
    response = requests.post(
        "https://google.serper.dev/search",
        headers={
            "X-API-KEY": api_key,
            "Content-Type": "application/json",
        },
        json={"q": query, "num": max_results},
        timeout=15,
    )
    response.raise_for_status()
    data = response.json()
    
    results = []
    for item in data.get("organic", [])[:max_results]:
        results.app

Laddr

Install / Use

README

Table of Contents