Multi-Agent
Orchestration

touch multi_agent.py

import os, json
import anthropic

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
MODEL = "claude-haiku-4-5-20251001"  # cheap model for all three agents

# ── Shared knowledge base (stands in for real search) ─────────────
KNOWLEDGE = {
    "transformer architecture":
        "The Transformer uses self-attention to process all tokens in parallel. "
        "Introduced by Vaswani et al. (2017). Replaced recurrent networks for NLP.",
    "rlhf":
        "RLHF fine-tunes LLMs using human preference rankings. Three stages: SFT, "
        "reward model training, PPO. Introduced at scale by Ouyang et al. (2022).",
    "rag":
        "RAG retrieves relevant documents at query time and injects them as context. "
        "Solves the knowledge staleness problem without retraining. Lewis et al. (2020).",
    "react agents":
        "ReAct agents interleave Thought, Action, and Observation in a loop. "
        "Enables dynamic tool use with visible reasoning. Yao et al. (2022).",
    "mcp":
        "Model Context Protocol is Anthropic's open standard for connecting LLM agents "
        "to tools and data sources via a standard wire protocol. Released Nov 2024.",
}

SEARCH_TOOLS = [{
    "name": "search",
    "description": "Search the knowledge base for information on a topic.",
    "input_schema": {
        "type": "object",
        "properties": {"query": {"type": "string"}},
        "required": ["query"],
    },
}]

def search(query: str) -> str:
    q = query.lower()
    for key, val in KNOWLEDGE.items():
        if key in q:
            return val
    return f'No results found for "{query}".'

ORCHESTRATOR_SYSTEM = """You are a research orchestrator.
Given a research question, decompose it into 2–3 focused subtasks.
Each subtask should be answerable with a single targeted search.
Return ONLY a JSON array of subtask strings.
Example: ["What is X?", "How does Y work?", "What are the tradeoffs of Z?"]"""

SYNTHESIZER_SYSTEM = """You are a research synthesizer.
Given a research question and a list of (subtask, answer, quality_score) triples,
write a concise 3–5 sentence summary that answers the original question.
Only use information from the provided answers — do not add outside knowledge."""


def orchestrate(question: str) -> list[str]:
    """Decompose a research question into focused subtasks."""
    response = client.messages.create(
        model=MODEL, max_tokens=256,
        system=ORCHESTRATOR_SYSTEM,
        messages=[{"role": "user", "content": f"Research question: {question}"}],
    )
    raw = response.content[0].text.strip()
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        start, end = raw.find("["), raw.rfind("]") + 1
        return json.loads(raw[start:end])


def synthesize(question: str, results: list[dict]) -> str:
    """Synthesize all research results into a final answer."""
    results_text = "\n\n".join(
        f"Subtask: {r['subtask']}\nAnswer: {r['answer']}\nQuality: {r['score']}/5"
        for r in results
    )
    response = client.messages.create(
        model=MODEL, max_tokens=400,
        system=SYNTHESIZER_SYSTEM,
        messages=[{
            "role": "user",
            "content": f"Original question: {question}\n\nResearch results:\n{results_text}",
        }],
    )
    return response.content[0].text

RESEARCHER_SYSTEM = """You are a focused research assistant.
You will receive a single subtask. Use the search tool to find relevant information,
then write a concise 2–3 sentence answer grounded in what you found.
Do not add information beyond what the search tool returns."""


def research(subtask: str) -> str:
    """Run the researcher agent on a single subtask. Returns a text answer."""
    messages = [{"role": "user", "content": subtask}]

    for _ in range(4):  # small iteration cap per subagent
        response = client.messages.create(
            model=MODEL, max_tokens=256,
            system=RESEARCHER_SYSTEM,
            tools=SEARCH_TOOLS,
            messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            return "".join(b.text for b in response.content if hasattr(b, "text"))

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = search(block.input["query"])
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })
            messages.append({"role": "user", "content": tool_results})

    return "Researcher did not complete within iteration budget."

CRITIC_SYSTEM = """You are a research quality critic.
Given a subtask and a proposed answer, evaluate the answer on:
- Relevance: does it address the subtask?
- Groundedness: does it only use verifiable information?
- Conciseness: is it clear and appropriately brief?

Return ONLY a JSON object with this exact structure:
{"score": integer 1-5, "feedback": "one sentence of specific feedback"}
5 = excellent, 3 = acceptable but could improve, 1 = missing or wrong."""


def critique(subtask: str, answer: str) -> dict:
    """Run the critic agent. Returns {"score": int, "feedback": str}."""
    response = client.messages.create(
        model=MODEL, max_tokens=128,
        system=CRITIC_SYSTEM,
        messages=[{
            "role": "user",
            "content": f"Subtask: {subtask}\n\nProposed answer: {answer}",
        }],
    )
    raw = response.content[0].text.strip()
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        # Fallback: extract JSON from response if wrapped in prose
        start = raw.find("{")
        end = raw.rfind("}") + 1
        return json.loads(raw[start:end])

def run_pipeline(question: str) -> None:
    print(f"\n{'='*60}\nQUESTION: {question}\n{'='*60}")

    # 1. Orchestrator decomposes the question
    subtasks = orchestrate(question)
    print(f"\n[Orchestrator] Decomposed into {len(subtasks)} subtasks:")
    for i, s in enumerate(subtasks, 1):
        print(f"  {i}. {s}")

    results = []
    for subtask in subtasks:
        print(f"\n[Researcher] Working on: {subtask}")

        # 2. Researcher answers the subtask
        answer = research(subtask)
        print(f"  Answer: {answer[:100]}...")

        # 3. Critic scores the answer
        review = critique(subtask, answer)
        score = review.get("score", 0)
        feedback = review.get("feedback", "no feedback")
        print(f"  [Critic] Score: {score}/5 — {feedback}")

        results.append({"subtask": subtask, "answer": answer, "score": score})

    # 4. Orchestrator synthesizes the final report
    report = synthesize(question, results)
    avg_score = sum(r["score"] for r in results) / len(results)

    print(f"\n{'='*60}")
    print(f"FINAL REPORT (avg quality: {avg_score:.1f}/5):")
    print(f"{'='*60}")
    print(report)


if __name__ == "__main__":
    run_pipeline(
        "How do modern LLM agents learn to follow instructions "
        "and use tools effectively?"
    )

python multi_agent.py

============================================================
QUESTION: How do modern LLM agents learn to follow instructions and use tools?
============================================================

[Orchestrator] Decomposed into 3 subtasks:
  1. How does RLHF train LLMs to follow instructions?
  2. How do ReAct agents use tools at runtime?
  3. What role does the Transformer architecture play?

[Researcher] Working on: How does RLHF train LLMs to follow instructions?
  Answer: RLHF fine-tunes LLMs using human preference rankings...
  [Critic] Score: 4/5 — Clear and grounded but could mention the KL penalty.

[Researcher] Working on: How do ReAct agents use tools at runtime?
  Answer: ReAct agents interleave Thought, Action, and Observation...
  [Critic] Score: 5/5 — Concise, accurate, and well-sourced.

[Researcher] Working on: What role does the Transformer architecture play?
  Answer: The Transformer uses self-attention to process all tokens...
  [Critic] Score: 4/5 — Good but could connect more directly to tool use.

============================================================
FINAL REPORT (avg quality: 4.3/5):
============================================================
Modern LLM agents learn to follow instructions through RLHF, which fine-tunes
the model on human preference rankings across three stages... [continues]

from concurrent.futures import ThreadPoolExecutor, as_completed

def research_and_critique(subtask: str) -> dict:
    """Run researcher + critic for one subtask. Safe to call in parallel."""
    answer = research(subtask)
    review = critique(subtask, answer)
    return {
        "subtask": subtask,
        "answer": answer,
        "score": review.get("score", 0),
        "feedback": review.get("feedback", ""),
    }


# Replace the sequential loop with this:
with ThreadPoolExecutor(max_workers=len(subtasks)) as executor:
    futures = {executor.submit(research_and_critique, s): s for s in subtasks}
    results = [future.result() for future in as_completed(futures)]

print(f"\nAll {len(results)} subtasks completed in parallel.")
for r in results:
    print(f"  [{r['score']}/5] {r['subtask'][:60]}")