repos/Laddr/README.md · tensense/code_repo

code_repo_finetuning / repos /Laddr /README.md

tensense

Upload folder using huggingface_hub

4e909c7 verified about 1 month ago

preview code

raw

history blame contribute delete

17.9 kB

Your framework for building scalable multi-agent 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 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

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

💡 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
✅ 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:

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.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

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

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

# 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:

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 for complete documentation with advanced patterns like rate limiting, circuit breakers, retries, and metrics.

License

Apache License 2.0 - see 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!

tensense
/

code_repo_finetuning

Table of Contents