🤖 Agent Orchestrator

A lightweight Python framework for building and orchestrating AI agent pipelines.

Chain multiple AI agents with tools, memory, and fallback chains — in under 50 lines of code.

---

🧭 What It Does

Agent Orchestrator lets you build autonomous AI agent systems that can:

• Think → Act → Observe — ReAct-style agent loop with tool calling
• Chain agents together — Sequential, parallel, and fan-out pipeline modes
• Remember context — Short-term (sliding window) + long-term (vector similarity) memory
• Fall back gracefully — Automatic provider failover (OpenAI → Anthropic → xAI → Mythos → local)
• Serve via API — FastAPI REST + WebSocket server out of the box

🏗 Architecture

┌─────────────────────────────────────────────────────┐
│                   Orchestrator                       │
│  ┌──────────┐  ┌──────────┐  ┌──────────────────┐  │
│  │Sequential │  │ Parallel │  │    Fan-Out       │  │
│  │ A → B → C │  │A,B,C @t │  │ in → [A,B] → out │  │
│  └──────────┘  └──────────┘  └──────────────────┘  │
├─────────────────────────────────────────────────────┤
│                     Agent                            │
│  ┌─────────┐   ┌────────┐   ┌───────────────────┐  │
│  │ Think   │ → │  Act   │ → │     Observe       │  │
│  │  LLM    │   │ Tools  │   │  Tool Results     │  │
│  └─────────┘   └────────┘   └───────────────────┘  │
├─────────────────────────────────────────────────────┤
│  Memory (STM + LTM)  │  Tools Registry  │  Fallback │
│  ┌────────┐┌───────┐  │  ┌────────────┐  │  Chain   │
│  │Sliding ││Vector │  │  │calc/search │  │  OpenAI  │
│  │Window  ││  Sim  │  │  │file/custom │  │  → Anth  │
│  └────────┘└───────┘  │  └────────────┘  │  → xAI   │
└─────────────────────────────────────────────────────┘

⚡ Quick Start

import asyncio
from agent_orchestrator.core import Agent, AgentConfig, Orchestrator
from agent_orchestrator.providers import OpenAIProvider
from agent_orchestrator.tools import CalculatorTool, ToolRegistry

async def main():
    # Set up tools
    tools = ToolRegistry()
    tools.register(CalculatorTool())

    # Create an agent
    agent = Agent(
        config=AgentConfig(
            name="math-bot",
            system_prompt="You are a math assistant. Use the calculator for arithmetic.",
            model="gpt-5.5",
        ),
        provider=OpenAIProvider(model="gpt-5.5", api_key="sk-..."),
        tools=tools,
    )

    # Run it
    response = await agent.run("What is 15 * 23 + 47?")
    print(response.message.content)

asyncio.run(main())

🚀 Multi-Agent Pipeline

from agent_orchestrator.core import Agent, AgentConfig, Orchestrator

orch = Orchestrator()
orch.register("researcher", research_agent)
orch.register("analyst", analyst_agent)
orch.register("writer", writer_agent)

# Sequential: researcher → analyst → writer
results = await orch.run_sequential(
    "Explain quantum computing",
    steps=["researcher", "analyst", "writer"],
)

# Parallel: all agents process simultaneously
results = await orch.run_parallel({
    "researcher": "What are renewable energy trends?",
    "analyst": "Analyze AI automation impact.",
    "writer": "Summarize quantum computing apps.",
})

# Fan-out: one input → all agents → aggregated
result = await orch.run_fan_out(
    input_text="Future of space exploration",
    agent_names=["researcher", "analyst", "writer"],
    aggregation="concat",
)

🛠 Custom Tools

from agent_orchestrator.tools import tool, ToolParameter, ToolRegistry

# Option 1: Decorator
@tool(name="weather", description="Get weather for a city")
def get_weather(city: str) -> str:
    return f"Weather in {city}: Sunny, 22°C"

# Option 2: Class-based
from agent_orchestrator.tools import Tool

class DatabaseTool(Tool):
    def __init__(self):
        super().__init__(
            name="db_query",
            description="Query the database",
            parameters=[
                ToolParameter(name="sql", type="string", description="SQL query"),
            ],
        )

    def execute(self, sql: str = "", **kwargs) -> list:
        # Your database logic here
        return [{"id": 1, "name": "result"}]

# Register
registry = ToolRegistry()
registry.register(get_weather)
registry.register(DatabaseTool())

🧠 Memory

from agent_orchestrator.core.memory import ShortTermMemory, LongTermMemory

# Short-term: sliding window conversation history
stm = ShortTermMemory(window_size=50)

# Long-term: vector similarity search (numpy-only, no external DB)
ltm = LongTermMemory(dimension=128)
ltm.add(Message(role=Role.USER, content="Python is great for AI"))

# Search by relevance
results = ltm.search("programming languages", top_k=5)
for msg, score in results:
    print(f"[{score:.2f}] {msg.content}")

🌐 REST API

# Start server
python -m agent_orchestrator.server.app

# Create agent
curl -X POST http://localhost:8000/agents/create \
  -H "Content-Type: application/json" \
  -d '{"name": "assistant", "system_prompt": "You are helpful."}'

# Chat
curl -X POST http://localhost:8000/agents/assistant/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!"}'

# List agents
curl http://localhost:8000/agents

# WebSocket streaming
wscat -c ws://localhost:8000/agents/assistant/stream

API Endpoints

Method	Endpoint	Description
POST	/agents/create	Create a new agent
POST	/agents/{name}/chat	Send a message
GET	/agents/{name}/history	Get conversation history
GET	/agents	List all agents
DELETE	/agents/{name}	Delete an agent
POST	/pipelines/run	Run a multi-agent pipeline
WS	/agents/{name}/stream	Stream responses

📦 Installation

# Clone
git clone https://github.com/your-org/agent-orchestrator.git
cd agent-orchestrator

# Install
pip install -e ".[dev]"

# Run tests
pytest

# Start API server
python -m agent_orchestrator.server.app

Docker

docker build -f docker/Dockerfile -t agent-orchestrator .
docker run -p 8000:8000 agent-orchestrator

📁 Project Structure

agent-orchestrator/
├── src/agent_orchestrator/
│   ├── core/                  # Core: agent, orchestrator, memory, types
│   ├── tools/                 # Tool system + built-in tools
│   ├── providers/             # LLM providers (OpenAI, Anthropic, fallback)
│   └── server/                # FastAPI REST + WebSocket API
├── examples/                  # Runnable examples
│   ├── basic_agent.py
│   ├── multi_agent.py
│   ├── research_pipeline.py
│   └── customer_support.py
├── tests/                     # Pytest test suite
├── docker/                    # Dockerfile
├── pyproject.toml
└── README.md

🔑 Key Design Decisions

• Minimal dependencies — numpy, pydantic, httpx, FastAPI. No heavy ML frameworks.
• Provider-agnostic — OpenAI, Anthropic, xAI, Mythos, or any provider. Swap without changing agent code.
• Memory built-in — Short-term sliding window + long-term vector search. No external DB required.
• Production-ready — Async/await throughout, proper error handling, Pydantic validation.
• Extensible — Add custom tools with one decorator. Add providers with one class.

📜 License

MIT License — see LICENSE for details.

---

Built with ❤️ for the AI engineering community