Multi-Agent Base Framework

A reusable Python foundation for building multi-agent AI systems with comprehensive observability, cost tracking, and A2A protocol support.

Features

• 🤖 Multi-Provider LLM Support: Ollama, OpenAI, Anthropic via Microsoft Agent Framework
• 📊 Full Observability: Agent conversations, tool usage, inputs/outputs via Arize Phoenix
• 💰 Cost Tracking: Token usage and cost calculation per model/provider
• 🏗️ Parametric Architectures: SingleAgent, Supervisor, Swarm patterns
• 🎴 A2A Agent Cards: Agent metadata and capability declaration
• 🔍 Skill Auto-Discovery: Automatic skill extraction from tool functions
• 📝 Structured Logging: OpenTelemetry-based tracing
• 🧠 Conversation Memory: Multiple backends including buffer, sliding window, vector, and Redis
• 🔄 Resilience Patterns: Retry strategies, circuit breaker, timeout handling, fallbacks
• ⏱️ Rate Limiting: Token bucket, sliding window, and composite limiters
• 🗄️ Response Caching: LRU, TTL, and semantic caching strategies
• 📡 Event System: Pub/sub event bus for inter-agent communication
• 🔒 Security: Input validation, injection detection, permission management

Installation

# Basic installation
pip install multi-agent-base

# With specific provider support
pip install multi-agent-base[ollama]
pip install multi-agent-base[openai]
pip install multi-agent-base[anthropic]

# All providers
pip install multi-agent-base[all]

# Development
pip install multi-agent-base[dev]

Quick Start

Single Agent

from multi_agent_base import SystemConfig, SingleAgentPattern
from multi_agent_base.providers import ModelClientFactory

# Configure system
config = SystemConfig(
    provider="openai",
    model="gpt-4o-mini",
    observability_enabled=True,
)

# Create agent
pattern = SingleAgentPattern(config)
agent = pattern.create_agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
)

# Run
response = await agent.run("Hello, how are you?")
print(response)

Supervisor Team

from multi_agent_base import SystemConfig, SupervisorPattern

config = SystemConfig(
    provider="ollama",
    model="llama3.2",
    observability_enabled=True,
)

pattern = SupervisorPattern(config)
team = pattern.create_team(
    supervisor_name="manager",
    worker_configs=[
        {"name": "researcher", "system_prompt": "You research topics."},
        {"name": "writer", "system_prompt": "You write content."},
    ]
)

response = await team.run("Write a blog post about AI agents.")

Agent Cards

from multi_agent_base.a2a import AgentCard, SkillDiscoverer

# Auto-discover skills from tools
discoverer = SkillDiscoverer()
skills = discoverer.discover_from_tools([my_tool_function])

# Create agent card
card = AgentCard(
    name="research-agent",
    description="An agent that researches topics",
    skills=skills,
    capabilities=["text-generation", "web-search"],
)

# Export as JSON
card.to_json("agent_card.json")

Memory System

from multi_agent_base.memory import BufferMemory, MemoryConfig

# Create memory with configuration
memory = BufferMemory(MemoryConfig(max_entries=100))

# Store conversation
await memory.store(role="user", content="Hello!")
await memory.store(role="assistant", content="Hi there!")

# Retrieve history
history = await memory.retrieve(limit=10)

Rate Limiting

from multi_agent_base.ratelimit import RateLimiter, RateLimitConfig

# Create rate limiter
limiter = RateLimiter(RateLimitConfig(
    requests_per_minute=60,
    tokens_per_minute=10000,
))

# Check before making API calls
if await limiter.can_acquire():
    await limiter.acquire(tokens=100)
    # Make API call

Resilience

from multi_agent_base.resilience import retry_with_backoff, CircuitBreaker

# Retry with exponential backoff
@retry_with_backoff(max_attempts=3, delay=1.0)
async def call_api():
    return await risky_operation()

# Circuit breaker pattern
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30)
async with breaker:
    result = await external_service()

Security

from multi_agent_base.security import (
    validate_input,
    check_injection,
    SecretManager,
)

# Input validation
validation = validate_input(user_input, max_length=1000)
if not validation.is_valid:
    raise ValueError(validation.errors)

# Injection detection
result = check_injection(user_input)
if not result.is_safe:
    log_security_event(result.threats)

# Secure secrets management
secrets = SecretManager()
api_key = secrets.get("OPENAI_API_KEY")

Architecture

┌─────────────────────────────────────────────────────────────────┐
│                     Multi-Agent Base                            │
├─────────────────────────────────────────────────────────────────┤
│  Patterns                                                       │
│  ┌──────────────┐ ┌──────────────┐ ┌──────────────┐            │
│  │ SingleAgent  │ │  Supervisor  │ │    Swarm     │            │
│  └──────────────┘ └──────────────┘ └──────────────┘            │
├─────────────────────────────────────────────────────────────────┤
│  Cross-Cutting Concerns                                         │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐  │
│  │ Memory  │ │Resilience│ │  Rate   │ │  Cache  │ │Security │  │
│  │         │ │         │ │ Limiting│ │         │ │         │  │
│  └─────────┘ └─────────┘ └─────────┘ └─────────┘ └─────────┘  │
├─────────────────────────────────────────────────────────────────┤
│  Core Services                                                  │
│  ┌──────────────┐ ┌──────────────┐ ┌──────────────┐            │
│  │  A2A Cards   │ │    Cost      │ │ Observability│            │
│  │ & Discovery  │ │   Tracker    │ │   (Phoenix)  │            │
│  └──────────────┘ └──────────────┘ └──────────────┘            │
│  ┌──────────────┐                                               │
│  │ Event System │                                               │
│  └──────────────┘                                               │
├─────────────────────────────────────────────────────────────────┤
│  LLM Providers (via Microsoft Agent Framework)                  │
│  ┌──────────────┐ ┌──────────────┐ ┌──────────────┐            │
│  │   Ollama     │ │   OpenAI     │ │  Anthropic   │            │
│  └──────────────┘ └──────────────┘ └──────────────┘            │
└─────────────────────────────────────────────────────────────────┘

Documentation

• Getting Started
• Configuration Guide
• Architecture Patterns
• A2A Agent Cards
• Observability
• Cost Tracking
• Memory System
• Resilience Patterns
• Rate Limiting
• Caching
• Event System
• Security
• API Reference

Development

# Clone repository
git clone https://github.com/yourusername/multi-agent-base.git
cd multi-agent-base

# Create virtual environment
python -m venv .venv
source .venv/bin/activate

# Install with dev dependencies
pip install -e ".[dev,all]"

# Run tests
pytest

# Run linting
ruff check src tests
mypy src

License

MIT License - See LICENSE for details.