8.7 Multi-Agent Systems & Orchestration¶

WORKFLOW (control flow in code)                AGENT (control flow in model)

code: if intent == "math":                     model output: {"action": "calculator",
          call(math_model, query)                            "input": "sqrt(2)"}
      elif intent == "search":
          call(search_model, query)

"""
orchestrator_worker.py
A minimal orchestrator-worker multi-agent system using the OpenAI chat API.
Runs end-to-end: the orchestrator decomposes a task, dispatches workers,
and synthesises their results.

Requirements:
    pip install openai
    OPENAI_API_KEY must be set.
"""

import json
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Any

from openai import OpenAI

client = OpenAI()

# ---------------------------------------------------------------------------
# Data types
# ---------------------------------------------------------------------------

@dataclass
class WorkItem:
    """A sub-task produced by the orchestrator."""
    worker_id: str          # human-readable label, e.g. "researcher_1"
    system_prompt: str      # role specialisation
    task: str               # the concrete instruction
    context: str = ""       # any shared context the worker needs


@dataclass
class WorkResult:
    worker_id: str
    output: str
    success: bool = True
    error: str = ""


# ---------------------------------------------------------------------------
# Low-level model call
# ---------------------------------------------------------------------------

def call_model(
    system: str,
    user: str,
    model: str = "gpt-4o-mini",
    temperature: float = 0.3,
    max_tokens: int = 1024,
) -> str:
    """Thin wrapper around chat completions — easy to swap for any provider."""
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user",   "content": user},
        ],
        temperature=temperature,
        max_tokens=max_tokens,
    )
    return response.choices[0].message.content.strip()


# ---------------------------------------------------------------------------
# Orchestrator: produces a plan as structured JSON
# ---------------------------------------------------------------------------

ORCHESTRATOR_SYSTEM = """\
You are a task orchestrator. Given a high-level goal, decompose it into
independent sub-tasks that can run in parallel. Each sub-task should be
self-contained. Return a JSON array of objects, each with keys:
  worker_id   – short snake_case label
  system_prompt – the specialist role for this worker
  task        – the full instruction for this worker

Return ONLY the JSON array. No prose.
"""

def plan(goal: str, shared_context: str = "") -> list[WorkItem]:
    """Ask the orchestrator to produce a parallel work plan."""
    user_msg = f"Goal: {goal}\n\nShared context:\n{shared_context}" if shared_context else f"Goal: {goal}"
    raw = call_model(ORCHESTRATOR_SYSTEM, user_msg, temperature=0.0)

    # Strip markdown fences if model wrapped the JSON
    raw = raw.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip()

    items_data = json.loads(raw)
    return [
        WorkItem(
            worker_id=d["worker_id"],
            system_prompt=d["system_prompt"],
            task=d["task"],
            context=shared_context,
        )
        for d in items_data
    ]


# ---------------------------------------------------------------------------
# Worker: executes a single sub-task
# ---------------------------------------------------------------------------

def run_worker(item: WorkItem) -> WorkResult:
    """Execute one work item; capture any model or network error."""
    try:
        user_msg = item.task
        if item.context:
            user_msg = f"Context:\n{item.context}\n\nTask:\n{item.task}"
        output = call_model(item.system_prompt, user_msg)
        return WorkResult(worker_id=item.worker_id, output=output)
    except Exception as exc:
        return WorkResult(worker_id=item.worker_id, output="", success=False, error=str(exc))


# ---------------------------------------------------------------------------
# Synthesiser: combines parallel results
# ---------------------------------------------------------------------------

SYNTHESISER_SYSTEM = """\
You are a synthesis agent. You receive the outputs of multiple specialist
workers who each tackled a sub-task. Your job is to integrate their outputs
into a single coherent, well-structured response. Be concise.
"""

def synthesise(goal: str, results: list[WorkResult]) -> str:
    """Combine worker outputs into a final answer."""
    parts = [f"=== {r.worker_id} ===\n{r.output}" for r in results if r.success]
    failed = [r.worker_id for r in results if not r.success]
    user_msg = f"Original goal: {goal}\n\nWorker outputs:\n\n" + "\n\n".join(parts)
    if failed:
        user_msg += f"\n\nNote: workers {failed} failed and produced no output."
    return call_model(SYNTHESISER_SYSTEM, user_msg)


# ---------------------------------------------------------------------------
# Top-level orchestrate function
# ---------------------------------------------------------------------------

def orchestrate(goal: str, max_workers: int = 4) -> str:
    """Full orchestrator-worker pipeline for a given goal."""
    print(f"[Orchestrator] Planning: {goal}")
    work_items = plan(goal)
    print(f"[Orchestrator] {len(work_items)} sub-tasks: {[w.worker_id for w in work_items]}")

    results: list[WorkResult] = []
    # Run workers in parallel using a thread pool (each call is IO-bound)
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {pool.submit(run_worker, item): item for item in work_items}
        for future in as_completed(futures):
            result = future.result()
            status = "OK" if result.success else f"FAILED: {result.error}"
            print(f"  [Worker {result.worker_id}] {status}")
            results.append(result)

    print("[Orchestrator] Synthesising results …")
    answer = synthesise(goal, results)
    return answer


# ---------------------------------------------------------------------------
# Demo
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    goal = (
        "Write a brief market analysis report for a new AI-powered note-taking app: "
        "competitor landscape, target user segments, and key risks."
    )
    answer = orchestrate(goal)
    print("\n=== FINAL ANSWER ===")
    print(answer)

# One server hosts every "agent": the roles are system prompts, not separate models.
vllm serve ./stack-100m-sft --port 8000

client = OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY")
# call_model(..., model="./stack-100m-sft")   # the rest of the file is unchanged

"""
blackboard.py
A minimal blackboard for multi-agent coordination.
Thread-safe for concurrent workers using a threading.Lock.
"""

import threading
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Any


@dataclass
class BlackboardEntry:
    author: str           # which agent wrote this
    content: Any          # arbitrary Python object (str, dict, etc.)
    timestamp: float = 0.0


class Blackboard:
    """
    Shared key-value store for agent communication.
    Keys are namespaced: "agent_id:key_name".
    """

    def __init__(self):
        self._store: dict[str, list[BlackboardEntry]] = defaultdict(list)
        self._lock = threading.Lock()

    def write(self, key: str, content: Any, author: str) -> None:
        """Append a versioned entry. Never overwrites — full audit trail."""
        import time
        entry = BlackboardEntry(author=author, content=content, timestamp=time.time())
        with self._lock:
            self._store[key].append(entry)

    def read_latest(self, key: str) -> BlackboardEntry | None:
        """Return the most recently written entry for a key."""
        with self._lock:
            entries = self._store.get(key, [])
            return entries[-1] if entries else None

    def read_all(self, key: str) -> list[BlackboardEntry]:
        """Return full history for a key (useful for debate/critique)."""
        with self._lock:
            return list(self._store.get(key, []))

    def keys(self) -> list[str]:
        with self._lock:
            return list(self._store.keys())

    def snapshot(self) -> dict[str, Any]:
        """Return a plain dict of latest values — useful for debugging."""
        with self._lock:
            return {k: v[-1].content for k, v in self._store.items() if v}

# Example: two agents using the blackboard

def research_agent(board: Blackboard, topic: str) -> None:
    """Writes findings to the blackboard."""
    findings = call_model(
        "You are a research specialist. Be factual and concise.",
        f"Research: {topic}",
    )
    board.write(key="research:findings", content=findings, author="researcher")


def critic_agent(board: Blackboard) -> None:
    """Reads the latest research and writes a critique."""
    entry = board.read_latest("research:findings")
    if entry is None:
        board.write("critic:status", "No findings to critique yet.", author="critic")
        return
    critique = call_model(
        "You are a rigorous critic. Identify gaps and unsupported claims.",
        f"Critique this research:\n\n{entry.content}",
    )
    board.write(key="critic:critique", content=critique, author="critic")

"""
langgraph_example.py
A minimal LangGraph orchestrator-worker for a research task.
Requires: pip install langgraph langchain-openai
"""

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

# ---------------------------------------------------------------------------
# 1. Define state — the "blackboard" is strongly typed
# ---------------------------------------------------------------------------

class ResearchState(TypedDict):
    question: str
    research_notes: str     # filled by researcher node
    critique: str           # filled by critic node
    final_answer: str       # filled by synthesiser node
    iteration: int          # how many research-critique loops so far

# ---------------------------------------------------------------------------
# 2. Define nodes (each is a pure function: state → state patch)
# ---------------------------------------------------------------------------

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)


def researcher(state: ResearchState) -> dict:
    """Node: produce research notes on the question."""
    prompt = f"Research the following and produce concise notes:\n{state['question']}"
    if state.get("critique"):
        prompt += f"\n\nPrevious critique to address:\n{state['critique']}"
    result = llm.invoke(prompt)
    return {"research_notes": result.content}


def critic(state: ResearchState) -> dict:
    """Node: critique the research notes, or approve them."""
    prompt = (
        f"Question: {state['question']}\n\n"
        f"Research notes:\n{state['research_notes']}\n\n"
        "Identify missing information or unsupported claims. "
        "If the notes are sufficient, reply with exactly: APPROVED"
    )
    result = llm.invoke(prompt)
    return {
        "critique": result.content,
        "iteration": state.get("iteration", 0) + 1,
    }


def synthesiser(state: ResearchState) -> dict:
    """Node: produce the final answer from approved notes."""
    prompt = (
        f"Question: {state['question']}\n\n"
        f"Approved research notes:\n{state['research_notes']}\n\n"
        "Write a clear, concise final answer."
    )
    result = llm.invoke(prompt)
    return {"final_answer": result.content}


# ---------------------------------------------------------------------------
# 3. Routing: decide whether to loop or to synthesise
# ---------------------------------------------------------------------------

def route_after_critic(state: ResearchState) -> str:
    """Conditional edge: approve ends the loop; critique sends us back."""
    if "APPROVED" in state["critique"] or state.get("iteration", 0) >= 3:
        return "synthesise"   # edge label → goes to synthesiser node
    return "revise"           # edge label → goes back to researcher


# ---------------------------------------------------------------------------
# 4. Build the graph
# ---------------------------------------------------------------------------

builder = StateGraph(ResearchState)

builder.add_node("researcher",   researcher)
builder.add_node("critic",       critic)
builder.add_node("synthesiser",  synthesiser)

builder.set_entry_point("researcher")
builder.add_edge("researcher", "critic")
builder.add_conditional_edges(
    "critic",
    route_after_critic,
    {
        "revise":     "researcher",   # loop back
        "synthesise": "synthesiser",  # exit
    },
)
builder.add_edge("synthesiser", END)

graph = builder.compile()

# ---------------------------------------------------------------------------
# 5. Run
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    result = graph.invoke({
        "question": "What are the main differences between RLHF and DPO?",
        "research_notes": "",
        "critique": "",
        "final_answer": "",
        "iteration": 0,
    })
    print(result["final_answer"])

"""
autogen_debate.py
A two-agent debate using AutoGen's ConversableAgent API.
Requires: pip install pyautogen
"""

import autogen

# Shared config for both agents
llm_config = {"config_list": [{"model": "gpt-4o-mini", "api_key": "..."}]}

# Agent A: takes the PRO position
pro_agent = autogen.ConversableAgent(
    name="ProAgent",
    system_message=(
        "You argue FOR the proposition that large language models will "
        "replace most knowledge workers within 10 years. Be specific and "
        "evidence-based. Keep responses under 200 words."
    ),
    llm_config=llm_config,
    human_input_mode="NEVER",  # fully automated
)

# Agent B: takes the CON position
con_agent = autogen.ConversableAgent(
    name="ConAgent",
    system_message=(
        "You argue AGAINST the proposition that large language models will "
        "replace most knowledge workers within 10 years. Be specific and "
        "evidence-based. Keep responses under 200 words."
    ),
    llm_config=llm_config,
    human_input_mode="NEVER",
)

# Judge: produces a verdict after the debate
judge = autogen.ConversableAgent(
    name="Judge",
    system_message=(
        "You are a neutral judge. After reading the debate, you produce a "
        "balanced verdict on who made the stronger arguments and why."
    ),
    llm_config=llm_config,
    human_input_mode="NEVER",
    is_termination_msg=lambda msg: "VERDICT:" in msg.get("content", ""),
)

if __name__ == "__main__":
    # Initiate a 3-round debate between pro and con agents
    pro_agent.initiate_chat(
        recipient=con_agent,
        message="LLMs will replace most knowledge workers within 10 years.",
        max_turns=3,
    )
    # Then ask the judge to weigh in on the full transcript
    # (In practice you'd collect and pass the conversation history)

"""
crewai_example.py
A minimal CrewAI pipeline for writing a technical blog post.
Requires: pip install crewai
"""

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.4)

# ---------------------------------------------------------------------------
# Define specialist agents
# ---------------------------------------------------------------------------

researcher = Agent(
    role="Technical Researcher",
    goal="Find accurate, up-to-date information on the given topic.",
    backstory="A meticulous research scientist who double-checks every claim.",
    llm=llm,
    verbose=True,
)

writer = Agent(
    role="Technical Writer",
    goal="Write a clear, engaging technical blog post from research notes.",
    backstory="An experienced writer who makes complex ideas accessible.",
    llm=llm,
    verbose=True,
)

editor = Agent(
    role="Editor",
    goal="Polish the draft for clarity, accuracy, and style.",
    backstory="A demanding editor who improves every sentence.",
    llm=llm,
    verbose=True,
)

# ---------------------------------------------------------------------------
# Define tasks (sequential pipeline)
# ---------------------------------------------------------------------------

research_task = Task(
    description="Research the key concepts behind mixture-of-experts LLM architectures.",
    agent=researcher,
    expected_output="A structured set of notes covering architecture, training, and key papers.",
)

writing_task = Task(
    description="Write a 500-word technical blog post based on the research notes.",
    agent=writer,
    expected_output="A polished 500-word blog post draft.",
    context=[research_task],  # this task receives research_task's output
)

editing_task = Task(
    description="Edit the blog post draft for accuracy, flow, and style.",
    agent=editor,
    expected_output="A final, publication-ready blog post.",
    context=[writing_task],
)

# ---------------------------------------------------------------------------
# Assemble and run the crew
# ---------------------------------------------------------------------------

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,  # or Process.hierarchical for a manager LLM
    verbose=True,
)

if __name__ == "__main__":
    result = crew.kickoff()
    print(result)

"""
swarm_handoff.py
A triage-and-specialist handoff pattern using OpenAI Swarm.
Requires: pip install openai-swarm
"""

from swarm import Swarm, Agent

client = Swarm()

# ---------------------------------------------------------------------------
# Specialist agents
# ---------------------------------------------------------------------------

billing_agent = Agent(
    name="BillingAgent",
    instructions=(
        "You handle billing inquiries: invoices, payment methods, refunds. "
        "If the question is not billing-related, transfer back to triage."
    ),
)

technical_agent = Agent(
    name="TechnicalAgent",
    instructions=(
        "You handle technical support: bugs, API errors, configuration issues. "
        "If the question is not technical, transfer back to triage."
    ),
)

# ---------------------------------------------------------------------------
# Triage agent: routes to the right specialist
# ---------------------------------------------------------------------------

def transfer_to_billing() -> Agent:
    """Handoff function — returning an Agent transfers control."""
    return billing_agent

def transfer_to_technical() -> Agent:
    return technical_agent

triage_agent = Agent(
    name="TriageAgent",
    instructions=(
        "You are a customer service triage agent. "
        "For billing questions call transfer_to_billing(). "
        "For technical questions call transfer_to_technical(). "
        "Otherwise answer directly."
    ),
    functions=[transfer_to_billing, transfer_to_technical],
)

# ---------------------------------------------------------------------------
# Run a conversation
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    messages = [{"role": "user", "content": "My invoice shows the wrong amount."}]
    response = client.run(agent=triage_agent, messages=messages)
    print(response.messages[-1]["content"])
    print(f"Final agent: {response.agent.name}")

from pydantic import BaseModel

class ResearchHandoff(BaseModel):
    """Validated payload from researcher → writer."""
    topic: str
    key_facts: list[str]          # at most 10 bullet points
    source_urls: list[str]        # cited sources (may be empty)
    confidence: float             # researcher's self-assessed confidence 0–1

class WritingHandoff(BaseModel):
    """Validated payload from writer → editor."""
    draft_text: str
    word_count: int
    flagged_sections: list[str]   # writer's own uncertainty flags

def self_critique_loop(
    orchestrator_prompt: str,
    worker_fn,
    scoring_fn,
    task: str,
    threshold: float = 0.8,
    max_retries: int = 3,
) -> str:
    """
    Run a worker, score its output, and retry with critique if below threshold.
    scoring_fn: (task, output) -> float in [0, 1]
    """
    for attempt in range(max_retries):
        output = worker_fn(task)
        score = scoring_fn(task, output)
        if score >= threshold:
            return output
        # Generate a critique and append it to the task for the next attempt
        critique = call_model(
            "You are a quality assessor. Identify what is wrong or missing.",
            f"Task: {task}\n\nOutput:\n{output}\n\nScore: {score:.2f}",
        )
        task = f"{task}\n\n[Previous attempt scored {score:.2f}. Critique: {critique}]"
    return output  # return best-effort after max_retries