**Your framework for building scalable multi-agent systems**
[](https://opensource.org/licenses/Apache-2.0)
[](https://www.python.org/downloads/)
[](https://www.docker.com/)
[](https://fastapi.tiangolo.com/)
[](https://redis.io/)
[](https://www.postgresql.org/)
[Quick Start](#quick-start) • [Features](#features) • [Agent Example](#agent-example) • [Tool Example](#tool-example) • [Dashboard](#dashboard--observability)
---
## Table of Contents
- [What is Laddr?](#what-is-laddr)
- [Features](#features)
- [Quick Start](#quick-start)
- [Agent Example](#agent-example)
- [Tool Example](#tool-example)
- [FastAPI Runtime](#fastapi-runtime--rest-api)
- [Dashboard & Observability](#dashboard--observability)
- [Architecture](#architecture)
- [License](#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 to PostgreSQL
- **Real-time metrics** — Job status, duration, success rates, queue depth
- **Interactive dashboard** — Beautiful UI to monitor agents, traces, 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
- **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
### 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
```bash
pip install laddr
```
### Create Your First Project
```bash
# Initialize a new project
laddr init my-agent-system
cd my-agent-system
```
### Configure API Keys
Edit `.env` file:
```bash
# 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
```
> 💡 **Using Ollama?** See the [Ollama Integration Guide](docs/guides/ollama-integration.md) for complete setup instructions including Docker configuration.
### Start the System
```bash
# Start all services (API, workers, database, Redis, MinIO)
laddr run dev
```
This starts:
- ✅ **API Server** at `http://localhost:8000`
- ✅ **Dashboard** at `http://localhost:5173`
- ✅ **PostgreSQL** for traces and job history
- ✅ **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:
```bash
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:
```python
"""
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:
```python
"""
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.append({
"title": item.get("title", ""),
"link": item.get("link", ""),
"snippet": item.get("snippet", "")[:500],
"site": item.get("domain", "")
})
return {
"query": query,
"results": results,
"count": len(results),
"status": "success"
}
```
**Features:**
- **@tool decorator** — Automatic registration
- **JSON Schema** — Parameters defined with validation
- **Structured output** — Consistent return format
- **Fast** — 15 second timeout
- **Error handling** — Graceful failures
---
## FastAPI Runtime & REST API
Laddr includes a **production-ready FastAPI server** with comprehensive REST endpoints:
### Core Endpoints
| Method | Endpoint | Description |
|--------|----------|-------------|
| `POST` | `/api/jobs` | Submit a job to an agent |
| `GET` | `/api/jobs/{job_id}` | Get job status and result |
| `GET` | `/api/jobs` | List all jobs with filters |
| `POST` | `/api/jobs/{job_id}/replay` | Replay a failed job |
| `GET` | `/api/agents` | List all available agents |
| `GET` | `/api/agents/{agent_name}/tools` | Get agent's tools |
| `GET` | `/api/agents/{agent_name}/chat` | Interactive chat with agent |
### Observability Endpoints
| Method | Endpoint | Description |
|--------|----------|-------------|
| `GET` | `/api/traces` | Get execution traces for jobs |
| `GET` | `/api/traces/grouped` | Get traces grouped by job |
| `GET` | `/api/traces/{trace_id}` | Get specific trace details |
| `GET` | `/api/metrics` | Get system metrics (jobs, latency, success rate) |
| `GET` | `/api/logs/containers` | List Docker containers |
| `GET` | `/api/logs/containers/{name}` | Stream container logs |
### Playground Endpoints
| Method | Endpoint | Description |
|--------|----------|-------------|
| `POST` | `/api/prompts` | Create interactive prompt session |
| `GET` | `/api/prompts/{prompt_id}` | Get prompt status and messages |
| `GET` | `/api/prompts` | List all prompt sessions |
| `POST` | `/api/prompts/{prompt_id}/cancel` | Cancel running prompt |
### Example: Submit a Job
```bash
curl -X POST http://localhost:8000/api/jobs \
-H "Content-Type: application/json" \
-d '{
"agent_name": "researcher",
"inputs": {
"query": "Latest AI trends 2025"
}
}'
# Response:
{
"job_id": "job_abc123",
"status": "queued",
"created_at": "2025-01-15T10:30:00Z"
}
```
### Example: Get Job Result
```bash
curl http://localhost:8000/api/jobs/job_abc123
# Response:
{
"job_id": "job_abc123",
"status": "completed",
"result": {
"answer": "Based on my research, the top AI trends in 2025 are...",
"sources": [...]
},
"agent_name": "researcher",
"created_at": "2025-01-15T10:30:00Z",
"completed_at": "2025-01-15T10:31:45Z",
"duration_seconds": 105
}
```
### Interactive API Documentation
Full OpenAPI documentation available at:
http://localhost:8000/docs — Swagger UI
---
## Dashboard & Observability
Laddr includes a **beautiful React dashboard** with real-time monitoring:
### Dashboard Features
**Playground**
- Test agents interactively
- Live streaming responses
- Token usage tracking
- Save and replay sessions
**Traces**
- Complete execution history
- Tool calls and results
- LLM interactions
- Delegation flows
- Filter by job, agent, or time range
**Metrics**
- System health overview
- Jobs per agent (success/failed)
- Average latency per agent
- Queue depth monitoring
- Worker status
**Logs**
- Real-time container logs
- Filter by agent or service
- Error highlighting
- Download logs
**Agents**
- List all agents and their tools
- Agent configuration viewer
- Worker scaling controls
- Health status
### Access the Dashboard
```bash
# Start the system
laddr run dev
# Open dashboard in browser
open http://localhost:5173
```
**Views:**
- `/` — Dashboard home with metrics
- `/playground` — Interactive agent testing
- `/traces` — Execution traces and history
- `/agents` — Agent management
- `/logs` — Container logs viewer
- `/settings` — Configuration
---
## Architecture
### Message Bus (Redis Streams)
Laddr uses **Redis Streams** for reliable, distributed messaging:
- **Agent Queues** — Each agent has a dedicated stream (`laddr:agent:{name}`)
- **Response Streams** — Temporary streams for delegation responses
- **Consumer Groups** — Multiple workers consume from the same stream
- **Automatic Load Balancing** — Redis distributes tasks across workers
- **Persistence** — Messages persisted until acknowledged
- **Backpressure** — Queue depth monitoring prevents overload
**Example flow:**
```
API → Redis Stream → Worker 1, Worker 2, Worker 3
↓
Process task
↓
Store result in Postgres
```
### Trace Storage (PostgreSQL)
All agent executions are **automatically traced** to PostgreSQL:
- **Complete history** — Every tool call, LLM interaction, delegation
- **Structured data** — JSON traces with metadata
- **Fast queries** — Indexed by job_id, agent_name, timestamp
- **No external dependencies** — Built-in, no Jaeger or DataDog needed
- **Retention policies** — Configurable trace retention
**Trace data includes:**
- Tool calls and results
- LLM prompts and responses
- Delegation events
- Error stack traces
- Token usage
- Latency breakdown
### Artifact Storage (MinIO/S3)
Large payloads are **automatically stored** in object storage:
- **Automatic threshold** — Messages >1MB stored as artifacts
- **S3-compatible** — MinIO (local) or AWS S3 (production)
- **Efficient messaging** — Only artifact reference sent via Redis
- **Retrieval on demand** — Workers fetch artifacts when needed
- **Configurable** — Set size threshold, retention, bucket names
**Benefits:**
- Reduces Redis memory usage
- Prevents message size limits
- Enables large document processing
- Supports binary data (images, PDFs, etc.)
---
## Custom System Tools
Laddr allows you to **override built-in system tools** (delegation, artifact storage) with your own custom implementations:
```python
from laddr import override_system_tool, TaskDelegationTool
@override_system_tool("system_delegate_task")
async def custom_delegation(
agent_name: str,
task_description: str,
task: str,
task_data: dict = None,
timeout_seconds: int = 300,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
"""Custom delegation with logging and metrics."""
# Add your custom logic (logging, rate limiting, etc.)
logger.info(f"Delegating to {agent_name}: {task_description}")
# Reuse base tool for actual delegation
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
result = await delegation_tool.delegate_task(
agent_name=agent_name,
task_description=task_description,
task=task,
task_data=task_data,
timeout_seconds=timeout_seconds
)
logger.info(f"Delegation completed: {result}")
return result
```
**Available base tools:**
- `TaskDelegationTool` — Single-task delegation
- `ParallelDelegationTool` — Parallel multi-task delegation
- `ArtifactStorageTool` — Artifact storage and retrieval
**System tools you can override:**
- `system_delegate_task` — Single task delegation
- `system_delegate_parallel` — Parallel task delegation
- `system_store_artifact` — Store data artifacts
- `system_retrieve_artifact` — Retrieve data artifacts
**Learn more:** See the [Custom System Tools Guide](docs/guides/custom-system-tools.md) for complete documentation with advanced patterns like rate limiting, circuit breakers, retries, and metrics.
---
## License
Apache License 2.0 - see [LICENSE](LICENSE) for details.
---
## 🔗 Links
- **Visit us at AgnetLabs**: https://agnetlabs.com
- **Documentation**: https://laddr.agnetlabs.com
- **GitHub**: https://github.com/AgnetLabs/Laddr
- **Issues**: https://github.com/AgnetLabs/Laddr/issues
---
**Built for production. Designed for scale. Made transparent.**
⭐ **Star us on GitHub** if Laddr helps you build better agent systems!