orchestrator

Multi-model AI agent orchestration via LangGraph + Composio. Routes tasks to the best-fit model and connects to external services (Slack, Notion, GitHub, Google Drive). Memory-aware — each run is enriched with the top-3 most relevant notes from Adam's project/feedback/reference memory store.

Architecture

Claude Code ──► run.py ──► LangGraph StateGraph
                              │
                              ▼
                          retriever  ──► top-3 memories from
                              │           ~/.claude/projects/.../memory/
                              ▼
                          router (Sonnet, structured output)
                              │
                ┌─────────────┼─────────────────────┐
                ▼             ▼                     ▼
        ┌──────────────┐  ┌───────────┐      ┌──────────┐
        │ implementer  │  │ connector │      │ unknown  │
        │ reviewer     │  │ (Composio)│      │ (no fit) │
        │ researcher   │  └───────────┘      └──────────┘
        │ cross_reviewer│       │
        │ scanner       │       ▼
        │ fast_coder    │  tool_executor ──► summarizer
        └──────────────┘

The retriever embeds Adam's memory files once and caches vectors to .cache/embeddings.json (mtime-keyed; only changed files re-embed). Each run picks the top-3 most relevant memories and surfaces them in the CLI output before the route line.

The router uses Pydantic structured output (RouteDecision) and returns an explicit "unknown" route when no agent fits — no silent fallback. All LLM invocations are wrapped with retry-on-transient-error.

Files

File	Purpose
run.py	CLI entry point — python3 run.py "<task>"
graph.py	LangGraph graph: retriever, router, connector, summarizer, unknown nodes
agents.py	AGENTS registry (label → model_fn, prompt, description) + make_agent_node factory
models.py	LLM factories, model-ID constants, with_retries() helper
state.py	OrchestratorState TypedDict
retriever.py	Memory loader, embedder, cache, top-k retrieval
tools.py	Composio tool loading (Slack, Notion, GitHub, Google Drive)
tests/test_routing_golden.py	20-case golden-set regression test for the router

Usage

# Full execution — retrieves memory, routes, and runs the task
python3 run.py "What is the LangGraph checkpoint API?"

# Route-only — retrieves memory and prints the agent that would handle the task
python3 run.py --route-only "Review this code for security issues"

Output shape:

[retrieved: project_seahaven_slack_bot, feedback_secrets_manager, reference_sea_haven_aws]
[reviewer]

<agent output>

From Claude Code (via CLAUDE.md hybrid delegation):

python3 ~/Documents/repositories/orchestrator/run.py "<task description>"
python3 ~/Documents/repositories/orchestrator/run.py --route-only "<task description>"

When Claude Code delegates vs. handles natively

Claude Code uses a hybrid model — it delegates to the orchestrator when a different model has a genuine advantage, and handles everything else natively:

Delegate to orchestrator	Handle natively in Claude Code
Cross-family code review (GPT-4.1)	File editing, refactoring, bug fixes
Large codebase scanning (Gemini)	Git operations, PRs, merges
Quick bounded coding (DeepSeek)	AWS/SAM/CDK deployments
External service actions (Composio)	Shell commands, system admin
	Interactive planning and conversation

Agents

Agent	Model	Use Case
implementer	Claude Sonnet	Write code with a clear spec
reviewer	Claude Sonnet	Code review (BLOCK/FIX/NIT/QUESTION)
researcher	Claude Haiku	Doc lookups, API research
cross_reviewer	GPT-4.1	Independent second-opinion review
scanner	Gemini 2.5 Pro	Large codebase analysis
fast_coder	DeepSeek Coder	Quick, bounded coding tasks
connector	Sonnet + Composio	Slack, Notion, GitHub, Google Drive

The router can also return done (no agent needed) or unknown (no clear fit). Model IDs are centralized as constants in models.py.

Memory retrieval

The retriever reads ~/.claude/projects/-Users-adammoussa-Documents-repositories/memory/*.md (skipping the MEMORY.md index), embeds each file once with text-embedding-3-small, and caches the vectors to .cache/embeddings.json. On subsequent runs:

• Only files whose mtime changed are re-embedded.
• Top-3 memories by cosine similarity are injected as system context into both the router and the agent.
• Retrieved names are printed as the first line of every run so bad retrieval is visible.
• Retrieval is read-only. The orchestrator never writes back to the memory store.

If retrieval fails (network, missing key), the run continues with no memory context and logs the failure into the message trail.

Connectors (via Composio)

All connections authenticated under Composio user amoussa:

• Slack: send messages, read channels/threads, find users, add reactions
• Notion: search/read/create/update pages, add content
• GitHub: create issues, list issues, get repo info
• Google Drive: find files, get metadata

The connector node is restricted to one tool call per run — a load-bearing rule learned from a 1.9M-token incident with meta-tool routing.

Setup

1. Install dependencies: pip install -r requirements.txt
2. Copy .env.example to .env and fill in API keys
3. Authenticate Composio integrations at app.composio.dev

Configuration

All API keys are stored in .env (gitignored):

• ANTHROPIC_API_KEY — Claude models + router
• OPENAI_API_KEY — GPT-4.1 cross-reviewer + text-embedding-3-small
• GOOGLE_API_KEY — Gemini scanner
• DEEPSEEK_API_KEY — DeepSeek fast-coder
• COMPOSIO_API_KEY — Composio connectors
• LANGSMITH_API_KEY — LangSmith tracing

Tracing is enabled via LangSmith (project: orchestration).

Testing

pytest tests/test_routing_golden.py -v

20 labelled tasks → expected agent. Skipped cleanly if ANTHROPIC_API_KEY or COMPOSIO_API_KEY are unset.