🤖 Multi-Agent System

Cooperative multi-agent orchestration with pipeline workflows, shared memory, and task execution

Overview

A multi-agent orchestration framework where specialized agents (Researcher, Writer, Coder, Reviewer) collaborate through pipeline workflows with shared memory, message passing, and cooperative task completion. Built with production patterns including retry logic, middleware, serialization, and factory-based instantiation.

Features

• 🤖 4 Specialized Agents — Researcher, Writer, Coder, Reviewer with distinct roles
• 🔗 Pipeline Orchestration — Sequential, conditional, broadcast, and round-robin workflows
• 🧠 Shared Memory — Conversation history, key-value context, task queues, event bus
• 💬 Message Passing — Structured inter-agent communication protocol with metadata
• 📊 Performance Metrics — Agent stats, latency, quality scoring, run comparison
• 🔄 Retry & Resilience — Exponential backoff with jitter for transient failure recovery
• 🧅 Middleware System — Composable pre/post-processing hooks (logging, validation, token budget)
• 🏭 Agent Factory — Configuration-driven agent creation with extensible registry
• 💾 Run Persistence — Save/load pipeline runs to JSON for auditing and replay
• 🖥️ CLI Interface — Run pipelines from the terminal with verbose and JSON output modes
• 🌐 REST API — FastAPI endpoints for remote pipeline execution
• 🧪 90+ Tests — Comprehensive test coverage with parametrized tests and edge cases

Architecture

┌─────────────────────────────────────────────────────────┐
│                    Entry Points                          │
│  CLI (cli.py) │ REST API (api/) │ Streamlit (streamlit/) │
├─────────────────────────────────────────────────────────┤
│                  Middleware Stack                         │
│  LoggingMW → ValidationMW → TokenBudgetMW → Agent       │
├─────────────────────────────────────────────────────────┤
│               Orchestration Layer                        │
│  Pipeline │ Workflow │ Coordinator │ ExecutionTracer      │
├─────────────────────────────────────────────────────────┤
│                  Agent Layer                              │
│  Researcher │ Writer │ Coder │ Reviewer │ (custom...)     │
├─────────────────────────────────────────────────────────┤
│               Infrastructure                             │
│  Memory │ EventBus │ TaskQueue │ SharedContext │ Retry    │
│  Factory │ Serialization │ Evaluation │ Exceptions       │
└─────────────────────────────────────────────────────────┘

Project Structure

multi-agent-system/
├── src/
│   ├── __init__.py          # Package exports and version
│   ├── agent_base.py        # AgentMessage, BaseAgent ABC, Tool
│   ├── agents.py            # Researcher, Writer, Coder, Reviewer
│   ├── orchestration.py     # Pipeline, Workflow, Coordinator, Tracer
│   ├── memory.py            # ConversationMemory, SharedContext, TaskQueue, EventBus
│   ├── evaluation.py        # Metrics, latency, quality scoring, reports
│   ├── middleware.py         # Middleware ABC and built-in middleware
│   ├── retry.py             # @retry decorator and RetryPolicy
│   ├── factory.py           # AgentFactory and create_agent/create_team
│   ├── serialization.py     # PipelineRun save/load to JSON
│   ├── cli.py               # Command-line interface
│   ├── config.py            # Centralized configuration
│   ├── exceptions.py        # Custom exception hierarchy
│   └── api/
│       └── main.py          # FastAPI REST endpoints
├── tests/
│   ├── conftest.py          # Shared fixtures
│   ├── test_agent_base.py   # Base class and message tests
│   ├── test_agents.py       # Agent implementation tests
│   ├── test_orchestration.py # Pipeline, Workflow, Coordinator tests
│   ├── test_memory.py       # Memory subsystem tests
│   ├── test_evaluation.py   # Metrics and report tests
│   ├── test_middleware.py   # Middleware stack tests
│   ├── test_retry.py        # Retry logic tests
│   ├── test_factory.py      # Factory pattern tests
│   ├── test_serialization.py # Persistence tests
│   └── test_api.py          # API endpoint tests
├── streamlit_app/           # Interactive dashboard
├── requirements.txt
└── README.md

Quick Start

git clone https://github.com/mohamed-elkholy95/multi-agent-system.git
cd multi-agent-system
pip install -r requirements.txt
python -m pytest tests/ -v

Run from CLI

# Default pipeline (all agents)
python -m src.cli "Analyze the impact of transformers on NLP"

# Custom agent selection
python -m src.cli "Write a sorting algorithm" --agents coder reviewer

# Verbose output with execution trace
python -m src.cli "Research AI trends" --verbose --trace --report

# Save results to file
python -m src.cli "Analyze data" --output outputs/run_001.json --json

Run as API Server

uvicorn src.api.main:app --reload --port 8004
# Then visit http://localhost:8004/docs for interactive API docs

Run Streamlit Dashboard

streamlit run streamlit_app/app.py

Python API

from src import Pipeline, create_team, MiddlewareStack, LoggingMiddleware

# Create agents using the factory
team = create_team(["researcher", "writer", "reviewer"])

# Build a pipeline with middleware
pipeline = Pipeline("research_pipeline")
for agent in team:
    pipeline.add_agent(agent)

# Run the pipeline
messages = pipeline.run("Analyze current AI safety research")

# Print results
for msg in messages:
    print(f"[{msg.agent_name or 'user'}] {msg.content[:80]}")

Using Middleware

from src import MiddlewareStack, LoggingMiddleware, TokenBudgetMiddleware, create_agent

stack = MiddlewareStack()
stack.add(LoggingMiddleware(log_content=True))
stack.add(TokenBudgetMiddleware(max_tokens=5000))

agent = create_agent("researcher")
messages = [AgentMessage("user", "What are the latest ML trends?")]

# Execute with middleware wrapping
response = stack.execute(agent, messages)
print(response.metadata["token_budget"])  # Token usage stats

Saving and Loading Runs

from src import PipelineRun, save_run, load_run

# After running a pipeline...
run = PipelineRun("experiment_001", messages, metadata={"model": "gpt-4"})
run.finalize()
save_run(run, "outputs/experiment_001.json")

# Later, load and analyze
loaded = load_run("outputs/experiment_001.json")
print(loaded.summary)
print(f"Agents: {loaded.agent_names}")

Design Patterns

This project demonstrates several GoF and production design patterns:

Pattern	Where	Purpose
Strategy	BaseAgent.process()	Agents are interchangeable strategies
Chain of Responsibility	Pipeline	Sequential agent processing
Mediator	Coordinator	Centralized agent communication
Observer	EventBus, ExecutionTracer	Decoupled event notifications
Factory	AgentFactory, create_agent()	Config-driven agent creation
Decorator	@retry, Middleware	Transparent behavior extension
Template Method	BaseAgent	Common structure, custom processing

API Endpoints

Method	Endpoint	Description
GET	/health	System health check
GET	/agents	List registered agents
POST	/pipeline/run	Execute the default pipeline
POST	/broadcast	Send message to all agents
POST	/round-robin	Multi-turn agent discussion
GET	/metrics/summary	Last run metrics
GET	/events/log	Event bus history

Key Concepts

Multi-Agent Orchestration

Agents are specialized workers that each handle one aspect of a task. The orchestration layer decides WHO does WHAT and in WHAT ORDER, turning individual capabilities into collaborative workflows.

Middleware Stack

Composable processing layers wrap agent execution (like Express.js middleware). Before hooks can validate inputs or check budgets; after hooks can log results or track tokens. The onion model ensures proper ordering.

Exponential Backoff

When agents call external services that fail transiently, the retry decorator waits progressively longer between attempts (1s → 2s → 4s) with random jitter to prevent thundering herd problems.

Pipeline Persistence

Complete pipeline runs can be serialized to JSON for audit trails, debugging, caching, and A/B comparison between runs.

Author

Mohamed Elkholy — GitHub · melkholy@techmatrix.com