8. Project: Research Agent v5 — Multi-Agent System

Project Overview

In Module 7 you built Research Agent v4: an individual agent that discovers tools dynamically from 3 MCP servers (filesystem, web search, papers DB), plans with priorities, researches with reflection and quality gates, remembers across sessions with checkpointing and long-term memory, and saves reports to disk. It's an individually complete agent — it has everything. But it's still a single agent doing all the work inside a single StateGraph, sharing a single context window, with a single set of instructions that mix search, analysis, and writing.

In this project you transform that individual agent into a system of 4 specialized agents that coordinate to solve complex research queries:

1. Supervisor: Receives the user's query, breaks it into sub-tasks, assigns each sub-task to the right specialized agent, monitors progress, validates partial results, and delivers the final report. It doesn't execute tools — it coordinates the ones who do.

2. Researcher: An agent specialized in information search. It has access to web search and the paper database (via M7's MCP). Its only task is to find relevant, reliable sources. It doesn't analyze data or write reports.

3. Analyst: An agent specialized in data evaluation. It has calculation and processing tools. It receives raw data from the Researcher, identifies patterns, compares sources, extracts key findings. It doesn't search for information or write.

4. Writer: An agent specialized in writing. It has filesystem access (via M7's MCP). It receives the processed findings from the Analyst and composes a coherent, structured research report with cited sources. It doesn't search or analyze — it synthesizes.

The Supervisor uses the central orchestration pattern (capsule 02): it decomposes, assigns, reviews, reassigns if necessary, and finishes. The workers are subagents (capsule 04) with isolated context: each one receives only what it needs for its task, executes in its own StateGraph, and returns a clean result. The system combines Supervisor + Subagents — the most robust composition for research tasks.

Estimated duration: 90-120 minutes.

---

Project Goal

Transform Research Agent v4 (an individual agent with MCP tools) into a system of 4 specialized agents coordinated by a Supervisor, where each agent has its own StateGraph, its own tools, and its own optimized context.

By the end you'll be able to:

• Design a coordination state that the Supervisor uses to track progress without sharing the entire context with each worker
• Implement a Researcher Agent with its own StateGraph and search tools (web search + papers via MCP)
• Implement an Analyst Agent with calculation and data-processing tools
• Implement a Writer Agent with filesystem tools (via MCP) to generate and save reports
• Build a Supervisor that decomposes queries, assigns agents, validates results, and handles reassignments
• Connect the 4 agents into a complete system with conditional edges and stop conditions
• Verify that the multi-agent system produces higher-quality results than the individual v4 agent

---

What Changes vs v4 (M7)

The Research Agent's internal architecture transforms completely. The nodes that used to live inside a single graph are now independent agents with their own graphs.

Before (v4): one agent, one graph, all the tools

graph = StateGraph(AgentState)
graph.add_node("planning", planning_node)
graph.add_node("research", research_node)       # uses search_web, search_papers
graph.add_node("analysis", analysis_node)        # uses the same model
graph.add_node("synthesis", synthesis_node)       # uses write_file
graph.add_node("reflection", reflection_node)
# One graph, one context window, 6+ tools competing for attention
agent = graph.compile(checkpointer=checkpointer)

After (v5): four agents, four graphs, distributed tools

researcher = build_researcher_agent()   # tools: search_web, search_papers, get_paper
analyst = build_analyst_agent()         # tools: calculate, extract_findings, compare
writer = build_writer_agent()           # tools: write_file, read_file

supervisor = StateGraph(SupervisorState)
supervisor.add_node("decompose", decompose_task)
supervisor.add_node("researcher", researcher_node)
supervisor.add_node("analyst", analyst_node)
supervisor.add_node("writer", writer_node)
supervisor.add_node("validate", validate_results)
system = supervisor.compile()

What gets distributed

Aspect	v4 (one agent)	v5 (4 agents)
Tools	6+ tools in one agent	3-4 tools per specialized agent
System prompt	500+ words mixing instructions	50-100 focused words per agent
Context window	Everything together: sources + analysis + drafts	Each agent sees only what it needs
Tool selection accuracy	~85% (6 tools compete)	~95% (3 clear tools)
Debugging	One point of failure, but an enormous context	Failure isolated per agent

What stays the same

• M7's MCP servers (filesystem, web search, papers DB) get reused — now distributed between the Researcher and the Writer
• Checkpointing works at the level of the complete system
• The user's query is still the entry point

---

Technical Specifications

Stack

Technology	Version	Use
Python	3.11+	Runtime
langchain, langchain-openai	v1.2+	LLM, prompts, structured output
langgraph	v1.0+	StateGraph, subgraphs, conditional edges
mcp	latest	SDK for MCP servers (from M7)
langchain-mcp-adapters	latest	Bridge MCP → LangChain tools
duckduckgo-search	latest	Backend of the web search server

pip install langchain langchain-openai langgraph mcp langchain-mcp-adapters duckduckgo-search python-dotenv

No new dependencies compared to M7. Multi-agent is built with the same tools — LangGraph natively handles multiple graphs and subgraphs.

Environment variables

# .env
OPENAI_API_KEY=sk-proj-your-api-key-here
RESEARCH_DIR=./research_output

Models per agent

Agent	Model	Reason
Supervisor	gpt-4.1	Routing and validation decisions are critical
Researcher	gpt-4.1-mini	Search with tools — the tool does the heavy lifting
Analyst	gpt-4.1	Analysis requires deep reasoning
Writer	gpt-4.1-mini	Good prose without complex reasoning

Optimizing models per agent is a direct advantage of multi-agent: in v4, a single model did everything.

---

v5 Architecture

                          ┌─────────────────────────────────┐
                          │          SUPERVISOR              │
                          │                                  │
        Query ──────────► │  1. Decomposes into sub-tasks    │
                          │  2. Assigns to the right worker  │
                          │  3. Reviews the partial result   │
                          │  4. Decides: another worker/end? │
                          │  5. Validates the final result   │
                          └──┬──────────┬──────────┬─────────┘
                             │          │          │
                    ┌────────┘          │          └────────┐
                    ▼                   ▼                   ▼
          ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐
          │   RESEARCHER     │ │    ANALYST        │ │     WRITER       │
          │                  │ │                   │ │                  │
          │ StateGraph:      │ │ StateGraph:       │ │ StateGraph:      │
          │  reason → tools  │ │  analyze          │ │  draft → save    │
          │  → reason → done │ │  → extract        │ │  → done          │
          │                  │ │  → done           │ │                  │
          │ Tools (MCP):     │ │ Tools:            │ │ Tools (MCP):     │
          │ • search_web     │ │ • calculate       │ │ • write_file     │
          │ • search_papers  │ │ • extract_findings│ │ • read_file      │
          │ • get_paper      │ │ • compare_sources │ │                  │
          │                  │ │                   │ │ Prompt:          │
          │ Prompt:          │ │ Prompt:           │ │ "Write a coherent│
          │ "Find relevant   │ │ "Evaluate data,   │ │  report with     │
          │  sources"        │ │  identify         │ │  cited sources"  │
          │                  │ │  patterns"        │ │                  │
          └──────────────────┘ └──────────────────┘ └──────────────────┘
               │ stdio              (local)              │ stdio
               ▼                                         ▼
     ┌──────────────────┐                      ┌──────────────────┐
     │  MCP SERVERS:    │                      │  MCP SERVER:     │
     │  web_search +    │                      │  filesystem      │
     │  papers_db       │                      │                  │
     └──────────────────┘                      └──────────────────┘
       (from M7, unchanged)                     (from M7, unchanged)

A typical flow for a research query:

User → Supervisor [decompose] → Researcher → Supervisor [review]
    → Analyst → Supervisor [review] → Writer → Supervisor [validate] → User

The Supervisor can reassign: if the Analyst reports missing data, the Supervisor sends it back to the Researcher. If the Writer produces a weak draft, the Supervisor can ask for a revision. The routing is dynamic.

---

Step 1: Design the Shared State

The state design is the most important architectural decision. The Supervisor needs global visibility to coordinate. The workers need minimal context to work cleanly. The solution: a coordination SupervisorState and isolated states per worker.

Supervisor state

import operator
from typing import Annotated, Literal, Optional
from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage


class SubTask(TypedDict):
    id: str
    description: str
    assigned_to: str
    status: str  # "pending" | "in_progress" | "completed" | "failed"
    result: str


class SupervisorState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    original_query: str
    sub_tasks: list[SubTask]
    current_agent: str
    agent_results: dict       # {agent_name: result_string}
    iteration_count: int
    max_iterations: int
    workers_called: list[str]
    final_report: Optional[str]
    status: str               # "decomposing" | "executing" | "validating" | "complete"

Field	Purpose
original_query	The original query — it isn't lost even after 10 iterations
sub_tasks	The list of sub-tasks with status and assignment — the Supervisor tracks them
current_agent	Who's working now — the conditional edges read this field
agent_results	Partial results per agent — they accumulate until synthesis
workers_called	The history of assignments — it prevents loops and gives the Supervisor context
status	The system's global status — it enables debugging and monitoring

Worker states

Each worker has its own typed state, decoupled from the Supervisor:

from langgraph.graph.message import add_messages


class ResearcherState(TypedDict):
    messages: Annotated[list, add_messages]
    query: str
    sources_found: list[dict]
    search_iterations: int


class AnalystState(TypedDict):
    messages: Annotated[list, add_messages]
    raw_data: str
    findings: list[str]
    confidence: float


class WriterState(TypedDict):
    messages: Annotated[list, add_messages]
    findings_summary: str
    sources: list[str]
    draft: str

The Researcher doesn't know a final_report exists. The Writer doesn't know there are sub_tasks. Each worker sees only what it needs. The Supervisor is the one that translates between the global state and each worker's inputs/outputs.

---

Step 2: Implement the Researcher Agent

The Researcher has one goal: find relevant, reliable sources. It has its own StateGraph with a reason-tools-reason loop, and it connects to M7's web search and papers DB MCP servers.

The Researcher's tools

from langchain_core.tools import tool
from langchain_mcp_adapters.client import MultiServerMCPClient

RESEARCHER_SERVER_CONFIG = {
    "web_search": {
        "command": "python",
        "args": ["servers/web_search_server.py"],
        "transport": "stdio",
    },
    "papers_db": {
        "command": "python",
        "args": ["servers/papers_server.py"],
        "transport": "stdio",
    },
}

The Researcher's StateGraph

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage
from langgraph.graph import StateGraph, START, END

researcher_model = init_chat_model("openai:gpt-4.1-mini")

RESEARCHER_PROMPT = """You are a specialized research agent.
Your ONLY task is to find relevant, reliable sources.

Rules:
- Use search_web for current information and trends
- Use search_papers for academic papers and theoretical foundations
- Use get_paper to get the details of a specific paper
- ALWAYS report: title, source/URL, and the key finding
- Don't analyze in depth — just find and report
- When you have at least 3 relevant sources, answer directly with a summary

DO NOT write reports. DO NOT do statistical analysis. Just search."""


def build_researcher_agent(mcp_tools: list) -> object:
    """Builds the Researcher as a compiled StateGraph."""
    tool_map = {t.name: t for t in mcp_tools}
    model_with_tools = researcher_model.bind_tools(mcp_tools)

    def reason(state: ResearcherState) -> dict:
        response = model_with_tools.invoke(
            [SystemMessage(content=RESEARCHER_PROMPT)]
            + state["messages"]
        )
        return {
            "messages": [response],
            "search_iterations": state.get("search_iterations", 0) + 1,
        }

    def execute_tools(state: ResearcherState) -> dict:
        last = state["messages"][-1]
        tool_messages = []
        sources = list(state.get("sources_found", []))

        for tc in last.tool_calls:
            if tc["name"] in tool_map:
                try:
                    result = tool_map[tc["name"]].invoke(tc["args"])
                    tool_messages.append(
                        ToolMessage(content=str(result), tool_call_id=tc["id"])
                    )
                    sources.append({
                        "tool": tc["name"],
                        "args": tc["args"],
                        "result_preview": str(result)[:200],
                    })
                except Exception as e:
                    tool_messages.append(
                        ToolMessage(content=f"Error: {e}", tool_call_id=tc["id"])
                    )

        return {"messages": tool_messages, "sources_found": sources}

    def route(state: ResearcherState) -> str:
        if state.get("search_iterations", 0) >= 4:
            return "done"
        last = state["messages"][-1]
        if hasattr(last, "tool_calls") and last.tool_calls:
            return "tools"
        return "done"

    graph = StateGraph(ResearcherState)
    graph.add_node("reason", reason)
    graph.add_node("tools", execute_tools)
    graph.add_edge(START, "reason")
    graph.add_conditional_edges("reason", route, {"tools": "tools", "done": END})
    graph.add_edge("tools", "reason")

    return graph.compile()

The Researcher has an internal loop: reason → tools → reason → ... → done. A maximum of 4 search iterations. Each iteration can invoke one or more tools. When it has enough sources or hits the limit, it finishes and returns its result.

---

Step 3: Implement the Analyst Agent

The Analyst receives raw data from the Researcher and processes it: it evaluates source quality, identifies patterns, extracts comparative findings. It has local tools (not MCP) because its work is computational, not about connectivity.

The Analyst's tools

@tool
def calculate(expression: str) -> str:
    """Evaluates a simple mathematical expression.

    expression: the expression to evaluate (e.g. '(120000 + 95000) / 2').
    Supports basic arithmetic operations.
    """
    try:
        allowed = set("0123456789+-*/().%, ")
        if not all(c in allowed for c in expression):
            return f"Error: characters not allowed in '{expression}'"
        result = eval(expression)
        return f"{expression} = {result}"
    except Exception as e:
        return f"Error evaluating '{expression}': {e}"


@tool
def extract_findings(text: str, focus: str = "") -> str:
    """Extracts the main findings from a research text.

    text: text with research data.
    focus: optional focus area (e.g. 'comparison', 'trends').
    Returns the 3-5 most relevant findings in a structured format.
    """
    analysis_model = init_chat_model("openai:gpt-4.1-mini")
    prompt = f"Extract the 3-5 most important findings"
    if focus:
        prompt += f" focusing on: {focus}"
    prompt += f".\n\nFormat: numbered, with evidence.\n\nText:\n{text}"
    response = analysis_model.invoke(prompt)
    return response.content


@tool
def compare_sources(source_a: str, source_b: str) -> str:
    """Compares two research sources and identifies agreements and contradictions.

    source_a: summary or content of the first source.
    source_b: summary or content of the second source.
    Returns: agreements, contradictions, and a reliability assessment.
    """
    comparison_model = init_chat_model("openai:gpt-4.1-mini")
    response = comparison_model.invoke(
        f"Compare these two sources:\n\n"
        f"SOURCE A:\n{source_a}\n\n"
        f"SOURCE B:\n{source_b}\n\n"
        f"Identify: 1) Agreements, 2) Contradictions, "
        f"3) Which is more reliable and why."
    )
    return response.content

The Analyst's StateGraph

analyst_model = init_chat_model("openai:gpt-4.1")

ANALYST_PROMPT = """You are a specialized analysis agent.
Your ONLY task is to evaluate data, identify patterns, and extract findings.

Rules:
- Use extract_findings to pull key points out of the data
- Use compare_sources when you have multiple sources to contrast
- Use calculate for numeric operations (averages, percentages)
- Prioritize findings with concrete evidence
- State the confidence level of each finding

DO NOT search for new information. DO NOT write reports. Just analyze."""

analyst_tools = [calculate, extract_findings, compare_sources]


def build_analyst_agent() -> object:
    """Builds the Analyst as a compiled StateGraph."""
    tool_map = {t.name: t for t in analyst_tools}
    model_with_tools = analyst_model.bind_tools(analyst_tools)

    def analyze(state: AnalystState) -> dict:
        response = model_with_tools.invoke(
            [SystemMessage(content=ANALYST_PROMPT)]
            + state["messages"]
        )
        return {"messages": [response]}

    def execute_tools(state: AnalystState) -> dict:
        last = state["messages"][-1]
        tool_messages = []
        findings = list(state.get("findings", []))

        for tc in last.tool_calls:
            if tc["name"] in tool_map:
                try:
                    result = tool_map[tc["name"]].invoke(tc["args"])
                    tool_messages.append(
                        ToolMessage(content=str(result), tool_call_id=tc["id"])
                    )
                    if tc["name"] == "extract_findings":
                        findings.append(str(result))
                except Exception as e:
                    tool_messages.append(
                        ToolMessage(content=f"Error: {e}", tool_call_id=tc["id"])
                    )

        return {"messages": tool_messages, "findings": findings}

    def route(state: AnalystState) -> str:
        last = state["messages"][-1]
        if hasattr(last, "tool_calls") and last.tool_calls:
            return "tools"
        return "done"

    graph = StateGraph(AnalystState)
    graph.add_node("analyze", analyze)
    graph.add_node("tools", execute_tools)
    graph.add_edge(START, "analyze")
    graph.add_conditional_edges("analyze", route, {"tools": "tools", "done": END})
    graph.add_edge("tools", "analyze")

    return graph.compile()

The Analyst has a shorter loop than the Researcher. It typically analyzes in 1-2 iterations: it invokes extract_findings or compare_sources, receives the results, and synthesizes its analysis.

---

Step 4: Implement the Writer Agent

The Writer receives processed findings from the Analyst and turns them into a coherent research report. It connects to M7's filesystem MCP server to save the final report.

The Writer's tools

WRITER_SERVER_CONFIG = {
    "filesystem": {
        "command": "python",
        "args": ["servers/filesystem_server.py"],
        "transport": "stdio",
        "env": {"RESEARCH_DIR": "./research_output"},
    },
}

The Writer's StateGraph

writer_model = init_chat_model("openai:gpt-4.1-mini")

WRITER_PROMPT = """You are a specialized technical writing agent.
Your ONLY task is to synthesize findings into a coherent, well-structured report.

Rules:
- Structure: Title, Executive Summary, Findings, Analysis, Conclusions, Sources
- Cite sources whenever possible (author, year, URL)
- Use clear, professional language
- The report must be self-contained (readable with no extra context)
- Use write_file to save the final report as markdown
- Use read_file if you need to consult previous research

DO NOT search for new information. DO NOT do analysis. Just write."""


def build_writer_agent(mcp_tools: list) -> object:
    """Builds the Writer as a compiled StateGraph."""
    tool_map = {t.name: t for t in mcp_tools}
    model_with_tools = writer_model.bind_tools(mcp_tools)

    def draft(state: WriterState) -> dict:
        response = model_with_tools.invoke(
            [SystemMessage(content=WRITER_PROMPT)]
            + state["messages"]
        )
        draft_content = response.content if not response.tool_calls else ""
        return {
            "messages": [response],
            "draft": draft_content if draft_content else state.get("draft", ""),
        }

    def execute_tools(state: WriterState) -> dict:
        last = state["messages"][-1]
        tool_messages = []

        for tc in last.tool_calls:
            if tc["name"] in tool_map:
                try:
                    result = tool_map[tc["name"]].invoke(tc["args"])
                    tool_messages.append(
                        ToolMessage(content=str(result), tool_call_id=tc["id"])
                    )
                except Exception as e:
                    tool_messages.append(
                        ToolMessage(content=f"Error: {e}", tool_call_id=tc["id"])
                    )

        return {"messages": tool_messages}

    def route(state: WriterState) -> str:
        last = state["messages"][-1]
        if hasattr(last, "tool_calls") and last.tool_calls:
            return "tools"
        return "done"

    graph = StateGraph(WriterState)
    graph.add_node("draft", draft)
    graph.add_node("tools", execute_tools)
    graph.add_edge(START, "draft")
    graph.add_conditional_edges("draft", route, {"tools": "tools", "done": END})
    graph.add_edge("tools", "draft")

    return graph.compile()

The Writer typically does 2-3 iterations: it writes the draft, invokes write_file to save it, and confirms it was saved successfully.

---

Step 5: Implement the Supervisor

The Supervisor is the system's brain. It has no execution tools — its power is decision-making. It uses structured output to guarantee valid routing and explainable reasoning.

Structured output for decisions

from pydantic import BaseModel, Field


class SupervisorDecision(BaseModel):
    next_agent: Literal["researcher", "analyst", "writer", "FINISH"] = Field(
        description="The next agent to assign, or FINISH if everything is complete"
    )
    task_for_agent: str = Field(
        description="A specific instruction for the assigned agent"
    )
    reasoning: str = Field(
        description="Justification for the decision"
    )

The decomposition node

from langchain_openai import ChatOpenAI

supervisor_model = ChatOpenAI(model="gpt-4.1", temperature=0)


def decompose_task(state: SupervisorState) -> dict:
    """Breaks the query into sub-tasks and plans the execution."""
    query = state["original_query"]

    response = supervisor_model.invoke([
        SystemMessage(content=(
            "You are the supervisor of a research team. "
            "Break this query into 2-4 concrete sub-tasks.\n\n"
            "For each sub-task specify:\n"
            "- id: a short identifier\n"
            "- description: what must be done\n"
            "- assigned_to: researcher, analyst, or writer\n\n"
            "The typical flow is: researcher → analyst → writer.\n"
            "Answer in JSON format: a list of objects with id, description, assigned_to."
        )),
        HumanMessage(content=f"Query to research: {query}"),
    ])

    import json
    try:
        tasks_raw = json.loads(
            response.content.strip().removeprefix("```json").removesuffix("```").strip()
        )
    except json.JSONDecodeError:
        tasks_raw = [
            {"id": "research", "description": f"Research: {query}", "assigned_to": "researcher"},
            {"id": "analyze", "description": "Analyze the findings", "assigned_to": "analyst"},
            {"id": "write", "description": "Write the final report", "assigned_to": "writer"},
        ]

    sub_tasks = [
        SubTask(
            id=t["id"],
            description=t["description"],
            assigned_to=t["assigned_to"],
            status="pending",
            result="",
        )
        for t in tasks_raw
    ]

    first_pending = next((t for t in sub_tasks if t["status"] == "pending"), None)
    next_agent = first_pending["assigned_to"] if first_pending else "FINISH"

    return {
        "sub_tasks": sub_tasks,
        "current_agent": next_agent,
        "status": "executing",
        "messages": [HumanMessage(
            content=f"[supervisor] Decomposition: {len(sub_tasks)} sub-tasks. "
                    f"First: {next_agent}",
            name="supervisor",
        )],
    }

The routing node (after each worker)

structured_supervisor = supervisor_model.with_structured_output(SupervisorDecision)


def supervisor_route(state: SupervisorState) -> dict:
    """Reviews results and decides the next step."""
    iteration = state.get("iteration_count", 0) + 1
    workers_called = state.get("workers_called", [])
    agent_results = state.get("agent_results", {})

    if iteration > state.get("max_iterations", 10):
        return {
            "current_agent": "FINISH",
            "status": "complete",
            "iteration_count": iteration,
            "messages": [HumanMessage(
                content="[supervisor] Iteration limit reached. Finishing.",
                name="supervisor",
            )],
        }

    results_summary = "\n".join([
        f"- {agent}: {result[:200]}..."
        for agent, result in agent_results.items()
    ]) or "No results yet."

    sub_tasks_summary = "\n".join([
        f"- [{t['status']}] {t['id']}: {t['description']} (→ {t['assigned_to']})"
        for t in state.get("sub_tasks", [])
    ])

    decision: SupervisorDecision = structured_supervisor.invoke([
        SystemMessage(content=(
            "You are a research supervisor. Review the progress and decide what comes next.\n\n"
            f"ORIGINAL QUERY: {state['original_query']}\n\n"
            f"SUB-TASKS:\n{sub_tasks_summary}\n\n"
            f"RESULTS OBTAINED:\n{results_summary}\n\n"
            f"WORKERS EXECUTED: {', '.join(workers_called) or 'none'}\n"
            f"ITERATION: {iteration}/{state.get('max_iterations', 10)}\n\n"
            "Rules:\n"
            "- If the researcher hasn't worked yet and data is needed, assign researcher\n"
            "- If there's data but no analysis, assign analyst\n"
            "- If there's analysis but no report, assign writer\n"
            "- If a result is insufficient, reassign the same agent with more specific instructions\n"
            "- If everything is complete and the report is satisfactory, answer FINISH\n"
        )),
        *state.get("messages", [])[-10:],
    ])

    is_done = decision.next_agent == "FINISH"

    return {
        "current_agent": decision.next_agent.lower() if not is_done else "FINISH",
        "status": "complete" if is_done else "executing",
        "iteration_count": iteration,
        "workers_called": workers_called + ([decision.next_agent] if not is_done else []),
        "messages": [HumanMessage(
            content=(
                f"[supervisor] Iter {iteration}: → {decision.next_agent} "
                f"| Task: {decision.task_for_agent[:100]} "
                f"| Reason: {decision.reasoning}"
            ),
            name="supervisor",
        )],
    }

The final validation node

def validate_and_deliver(state: SupervisorState) -> dict:
    """Synthesizes all the results into the final delivery."""
    agent_results = state.get("agent_results", {})

    all_results = "\n\n---\n\n".join([
        f"### {agent.upper()}\n{result}"
        for agent, result in agent_results.items()
    ])

    response = supervisor_model.invoke([
        SystemMessage(content=(
            "Review the results from all the agents and generate a coherent final "
            "answer for the user. If the Writer already produced a report, use it as the base. "
            "If not, synthesize the findings from the Researcher and the Analyst."
        )),
        HumanMessage(content=(
            f"QUERY: {state['original_query']}\n\n"
            f"RESULTS:\n{all_results}"
        )),
    ])

    return {
        "final_report": response.content,
        "status": "complete",
        "messages": [response],
    }

---

Step 6: Build the System

Now you connect everything: the 4 agents, the conditional edges, and the stop conditions.

Wrapper nodes for the workers

Each worker node invokes its sub-agent with isolated context and returns the result to the Supervisor:

def make_researcher_node(researcher_agent):
    """A wrapper that invokes the Researcher with isolated context."""
    def node(state: SupervisorState) -> dict:
        query = state["original_query"]
        agent_results = state.get("agent_results", {})

        task_context = query
        if agent_results:
            task_context += f"\n\nAdditional context: {list(agent_results.values())[-1][:300]}"

        result = researcher_agent.invoke({
            "messages": [HumanMessage(content=task_context)],
            "query": query,
            "sources_found": [],
            "search_iterations": 0,
        })

        output = result["messages"][-1].content
        updated_results = dict(agent_results)
        updated_results["researcher"] = output

        updated_tasks = []
        for t in state.get("sub_tasks", []):
            if t["assigned_to"] == "researcher" and t["status"] == "pending":
                updated_tasks.append({**t, "status": "completed", "result": output[:500]})
            else:
                updated_tasks.append(t)

        return {
            "agent_results": updated_results,
            "sub_tasks": updated_tasks,
            "messages": [HumanMessage(
                content=f"[researcher] {output[:300]}...",
                name="researcher",
            )],
        }
    return node


def make_analyst_node(analyst_agent):
    """A wrapper that invokes the Analyst with the Researcher's data."""
    def node(state: SupervisorState) -> dict:
        agent_results = state.get("agent_results", {})
        research_data = agent_results.get("researcher", "No research data yet.")

        result = analyst_agent.invoke({
            "messages": [HumanMessage(content=(
                f"Analyze the following research data:\n\n{research_data}"
            ))],
            "raw_data": research_data,
            "findings": [],
            "confidence": 0.0,
        })

        output = result["messages"][-1].content
        updated_results = dict(agent_results)
        updated_results["analyst"] = output

        updated_tasks = []
        for t in state.get("sub_tasks", []):
            if t["assigned_to"] == "analyst" and t["status"] == "pending":
                updated_tasks.append({**t, "status": "completed", "result": output[:500]})
            else:
                updated_tasks.append(t)

        return {
            "agent_results": updated_results,
            "sub_tasks": updated_tasks,
            "messages": [HumanMessage(
                content=f"[analyst] {output[:300]}...",
                name="analyst",
            )],
        }
    return node


def make_writer_node(writer_agent):
    """A wrapper that invokes the Writer with the Analyst's findings."""
    def node(state: SupervisorState) -> dict:
        agent_results = state.get("agent_results", {})
        research = agent_results.get("researcher", "")
        analysis = agent_results.get("analyst", "")
        query = state["original_query"]

        from datetime import datetime
        filename = f"research-{datetime.now().strftime('%Y%m%d-%H%M%S')}.md"

        result = writer_agent.invoke({
            "messages": [HumanMessage(content=(
                f"Write a complete research report.\n\n"
                f"TOPIC: {query}\n\n"
                f"RESEARCH FINDINGS:\n{research}\n\n"
                f"ANALYSIS:\n{analysis}\n\n"
                f"Save the report as '{filename}' using write_file."
            ))],
            "findings_summary": analysis,
            "sources": [],
            "draft": "",
        })

        output = result["messages"][-1].content
        draft = result.get("draft", output)
        updated_results = dict(agent_results)
        updated_results["writer"] = draft if draft else output

        updated_tasks = []
        for t in state.get("sub_tasks", []):
            if t["assigned_to"] == "writer" and t["status"] == "pending":
                updated_tasks.append({**t, "status": "completed", "result": output[:500]})
            else:
                updated_tasks.append(t)

        return {
            "agent_results": updated_results,
            "sub_tasks": updated_tasks,
            "messages": [HumanMessage(
                content=f"[writer] {output[:300]}...",
                name="writer",
            )],
        }
    return node

Each wrapper follows the subagents pattern (capsule 04): it extracts the minimum context from the SupervisorState, invokes the worker with an isolated state, and returns only the result to the Supervisor. The Researcher doesn't see the analysis. The Writer doesn't see the raw search queries. Context isolation in action.

The complete system graph

def build_multi_agent_system(researcher_agent, analyst_agent, writer_agent):
    """Builds the complete multi-agent system."""

    graph = StateGraph(SupervisorState)

    graph.add_node("decompose", decompose_task)
    graph.add_node("supervisor", supervisor_route)
    graph.add_node("researcher", make_researcher_node(researcher_agent))
    graph.add_node("analyst", make_analyst_node(analyst_agent))
    graph.add_node("writer", make_writer_node(writer_agent))
    graph.add_node("validate", validate_and_deliver)

    graph.add_edge(START, "decompose")

    def route_after_decompose(state: SupervisorState) -> str:
        agent = state.get("current_agent", "FINISH")
        if agent in ("researcher", "analyst", "writer"):
            return agent
        return "validate"

    graph.add_conditional_edges("decompose", route_after_decompose, {
        "researcher": "researcher",
        "analyst": "analyst",
        "writer": "writer",
        "validate": "validate",
    })

    graph.add_edge("researcher", "supervisor")
    graph.add_edge("analyst", "supervisor")
    graph.add_edge("writer", "supervisor")

    def route_from_supervisor(state: SupervisorState) -> str:
        if state.get("status") == "complete" or state.get("current_agent") == "FINISH":
            return "validate"
        agent = state.get("current_agent", "FINISH")
        if agent in ("researcher", "analyst", "writer"):
            return agent
        return "validate"

    graph.add_conditional_edges("supervisor", route_from_supervisor, {
        "researcher": "researcher",
        "analyst": "analyst",
        "writer": "writer",
        "validate": "validate",
    })

    graph.add_edge("validate", END)

    return graph.compile()

The flow: START → decompose → [worker] → supervisor → [worker] → supervisor → ... → validate → END. Every pass through the Supervisor is a routing decision. The cycle ends when the Supervisor answers FINISH or max_iterations is reached.

---

The Complete Multi-Agent System

To run the system end-to-end, you need to initialize the MCP clients, build the agents, and assemble everything:

import asyncio
from datetime import datetime
from dotenv import load_dotenv

load_dotenv()


async def run_research_v5(
    query: str,
    max_iterations: int = 10,
    verbose: bool = True,
) -> dict:
    """Runs the multi-agent Research Agent v5."""

    async with MultiServerMCPClient(RESEARCHER_SERVER_CONFIG) as research_mcp:
        research_tools = research_mcp.get_tools()

        async with MultiServerMCPClient(WRITER_SERVER_CONFIG) as writer_mcp:
            writer_tools = writer_mcp.get_tools()

            researcher_agent = build_researcher_agent(research_tools)
            analyst_agent = build_analyst_agent()
            writer_agent = build_writer_agent(writer_tools)

            system = build_multi_agent_system(
                researcher_agent, analyst_agent, writer_agent
            )

            if verbose:
                r_tools = [t.name for t in research_tools]
                w_tools = [t.name for t in writer_tools]
                a_tools = [t.name for t in analyst_tools]
                print(f"\n{'='*60}")
                print(f"  Research Agent v5 — Multi-Agent System")
                print(f"  Query: {query}")
                print(f"  Agents: Supervisor, Researcher, Analyst, Writer")
                print(f"  Researcher tools: {r_tools}")
                print(f"  Analyst tools: {a_tools}")
                print(f"  Writer tools: {w_tools}")
                print(f"{'='*60}")

            initial_state = {
                "messages": [HumanMessage(content=query)],
                "original_query": query,
                "sub_tasks": [],
                "current_agent": "",
                "agent_results": {},
                "iteration_count": 0,
                "max_iterations": max_iterations,
                "workers_called": [],
                "final_report": None,
                "status": "decomposing",
            }

            result = system.invoke(initial_state)

            if verbose:
                print(f"\n{'─'*60}")
                print(f"  Status: {result.get('status')}")
                print(f"  Iterations: {result.get('iteration_count')}")
                print(f"  Workers used: {result.get('workers_called')}")
                tasks = result.get('sub_tasks', [])
                completed = sum(1 for t in tasks if t['status'] == 'completed')
                print(f"  Sub-tasks: {completed}/{len(tasks)} completed")
                print(f"{'─'*60}")

            return result


if __name__ == "__main__":
    result = asyncio.run(run_research_v5(
        query="What are the most effective advanced RAG techniques "
              "and how do they compare to fine-tuning for improving LLMs?",
    ))

    report = result.get("final_report", "No report")
    print(f"\n{report[:500]}")

Expected output (schematic)

============================================================
  Research Agent v5 — Multi-Agent System
  Query: What are the most effective advanced RAG techniques...?
  Agents: Supervisor, Researcher, Analyst, Writer
  Researcher tools: ['search_web', 'search_papers', 'get_paper']
  Analyst tools: ['calculate', 'extract_findings', 'compare_sources']
  Writer tools: ['write_file', 'read_file', 'list_files']
============================================================

[supervisor] Decomposition: 3 sub-tasks. First: researcher
[researcher] Search: RAG techniques → 5 web sources, 2 papers
[supervisor] Iter 1: → analyst | Task: Analyze the RAG vs fine-tuning sources
[analyst] Analysis: 4 key findings, confidence 0.82
[supervisor] Iter 2: → writer | Task: Write the report with the findings
[writer] Report written → write_file("research-20260313-151022.md")
[supervisor] Iter 3: → FINISH | Everything completed satisfactorily

──────────────────────────────────────────────────────────────
  Status: complete
  Iterations: 3
  Workers used: ['researcher', 'analyst', 'writer']
  Sub-tasks: 3/3 completed
──────────────────────────────────────────────────────────────

---

Recommended Tests

Test 1: Each agent works in isolation

async def test_agents_isolated():
    async with MultiServerMCPClient(RESEARCHER_SERVER_CONFIG) as mcp:
        tools = mcp.get_tools()
        researcher = build_researcher_agent(tools)
        result = researcher.invoke({
            "messages": [HumanMessage(content="Search for papers about RAG")],
            "query": "RAG", "sources_found": [], "search_iterations": 0,
        })
        assert result["messages"][-1].content, "The Researcher produced no result"
        print("✓ The Researcher works in isolation")

    analyst = build_analyst_agent()
    result = analyst.invoke({
        "messages": [HumanMessage(content="Analyze: RAG improves accuracy by 30%")],
        "raw_data": "RAG improves accuracy by 30%", "findings": [], "confidence": 0.0,
    })
    assert result["messages"][-1].content, "The Analyst produced no result"
    print("✓ The Analyst works in isolation")

asyncio.run(test_agents_isolated())

What it validates: Each agent produces results independently of the system. If an agent fails in isolation, the bug is in the agent — not in the coordination.

Test 2: The complete system end-to-end

result = asyncio.run(run_research_v5(
    query="What is the Model Context Protocol and why does it matter for AI agents?",
    max_iterations=8,
))

assert result.get("final_report"), "No final report"
assert result.get("status") == "complete", "It didn't complete"
assert len(result.get("workers_called", [])) >= 2, "Too few workers executed"
print("✓ The complete system works end-to-end")

What it validates: The Supervisor → Researcher → Analyst → Writer flow produces a coherent final result.

Test 3: The Supervisor reassigns when data is missing

result = asyncio.run(run_research_v5(
    query="Compare the architectures of 5 AI agent frameworks "
          "with specific performance metrics",
    max_iterations=12,
))

workers = result.get("workers_called", [])
researcher_count = workers.count("researcher")
assert researcher_count >= 1, "The Researcher should run at least once"
print(f"✓ The Researcher ran {researcher_count} times")
print(f"  Full sequence: {workers}")

What it validates: For complex queries, the Supervisor can send work to the Researcher more than once if the analysis reveals that data is missing.

Test 4: The report saved to the filesystem

from pathlib import Path

result = asyncio.run(run_research_v5(
    query="A summary of advanced prompting techniques",
))

research_dir = Path("./research_output")
md_files = list(research_dir.glob("research-*.md"))
assert len(md_files) > 0, "No report was saved"
latest = max(md_files, key=lambda f: f.stat().st_mtime)
content = latest.read_text()
assert len(content) > 200, "The report is empty or too short"
print(f"✓ Report saved: {latest.name} ({len(content)} chars)")

What it validates: The Writer uses the filesystem MCP server to persist the report — the result doesn't just live in the state.

Test 5: Max iterations prevents loops

result = asyncio.run(run_research_v5(
    query="Research absolutely everything about artificial intelligence",
    max_iterations=5,
))

assert result.get("iteration_count", 0) <= 6, "It exceeded the limit"
assert result.get("status") == "complete", "It didn't finish"
print(f"✓ Finished in {result['iteration_count']} iterations (max: 5)")

What it validates: Ambiguous or overly broad queries don't cause infinite loops.

---

Success Criteria

1. The 3 workers are built as independent StateGraphs. Each one has its own typed state, its own set of tools, and compiles without depending on the others.

2. The Supervisor decomposes queries into sub-tasks. One query produces 2-4 sub-tasks with a clear assignment to researcher/analyst/writer.

3. The routing is dynamic. The Supervisor decides at runtime who works next — it isn't a fixed pipeline. For simple queries it can skip agents. For complex ones it can reassign.

4. Each worker receives isolated context. The Researcher doesn't see the analysis. The Writer doesn't see the raw search queries. The wrappers control what goes in and what comes out.

5. The system produces a final report. The complete research run generates a coherent final_report and saves it to the filesystem via MCP.

6. Anti-loop works. max_iterations prevents infinite Supervisor loops. workers_called gives the Supervisor visibility into what has already run.

7. Higher quality than the individual agent. For complex, multi-faceted queries, the multi-agent system produces more structured reports with better coverage than v4.

---

Checklist

• ResearcherState, AnalystState, WriterState, SupervisorState defined with types
• build_researcher_agent() compiles a StateGraph with MCP tools (search_web, search_papers)
• build_analyst_agent() compiles a StateGraph with local tools (calculate, extract_findings, compare_sources)
• build_writer_agent() compiles a StateGraph with MCP tools (write_file, read_file)
• Specialized system prompts: each agent has instructions focused on its expertise
• SupervisorDecision with structured output: next_agent, task_for_agent, reasoning
• decompose_task generates sub-tasks from the query
• supervisor_route makes routing decisions with progress and results as context
• The wrapper nodes isolate context: each worker receives only what it needs
• build_multi_agent_system() connects everything with conditional edges
• validate_and_deliver synthesizes the results from all the agents
• max_iterations implemented as a stop condition
• A complete end-to-end research run produces a final_report
• The report is saved as .md in ./research_output/
• Each agent works in isolation (independent test)

---

Common Errors

Error 1: The Supervisor loops between two agents

Symptom: Researcher → Supervisor → Researcher → Supervisor → ... The Supervisor thinks the results are always insufficient.

Cause: The Supervisor's prompt is too demanding about completeness criteria, or the Researcher's results aren't formatted in a way the Supervisor recognizes as sufficient.

Solution: Verify that agent_results is updated correctly in the wrapper node. Add workers_called to the Supervisor's prompt so it sees what has already run. If the Researcher has already run twice, the Supervisor should move on to the Analyst with what it has.

Error 2: The Analyst receives "No research data yet"

Symptom: The Supervisor assigns the Analyst before the Researcher completes.

Cause: decompose_task generated sub-tasks in an order that doesn't respect dependencies, or the Supervisor's routing doesn't check that there's data before sending work to the Analyst.

Solution: In decompose_task's prompt, emphasize that the typical sequence is researcher → analyst → writer. In supervisor_route, add a check: if "researcher" not in agent_results and current_agent == "analyst", redirect to the researcher first.

Error 3: Context bloat in the wrapper nodes

Symptom: The agents get slower and the answers lose quality after several iterations.

Cause: The wrappers pass too much context to the sub-agent. If you include all of agent_results as text in the HumanMessage, each iteration adds more context.

Solution: Limit the context each wrapper passes. The Researcher receives only the query. The Analyst receives only the Researcher's result (truncated to 2000 chars). The Writer receives the processed analysis, not the raw data. Apply the subagents principle: the minimum necessary context.

Error 4: The MCP clients time out while building the agents

Symptom: MultiServerMCPClient hangs when starting the MCP servers in the complete system.

Cause: Two MCP clients try to launch the same server simultaneously, or M7's servers have initialization errors that aren't reported.

Solution: Test each server individually with mcp dev server.py before integrating. Verify that the paths in SERVER_CONFIG are correct relative to the execution directory. If you use the same server (e.g. filesystem) in multiple agents, use a single instance of MultiServerMCPClient and share the tools.

Error 5: The Writer doesn't invoke write_file

Symptom: The report is generated in the state but isn't saved as a file.

Cause: The Writer doesn't have the filesystem MCP tools bound, or the prompt doesn't explicitly tell it to save the file.

Solution: Verify that build_writer_agent(mcp_tools) receives the tools from the filesystem client. Verify that the Writer's prompt includes the explicit instruction: "Use write_file to save the final report as markdown." Check that tool_map in the Writer includes write_file.

Error 6: The Supervisor's structured output fails with an invalid response

Symptom: A Pydantic ValidationError when parsing the Supervisor's decision.

Cause: The LLM generates a next_agent that isn't in the defined Literal, or the response doesn't match the SupervisorDecision schema.

Solution: LangChain's with_structured_output handles this automatically in most cases. If it fails, add a fallback: parse the raw response and map it to a valid value. Consider adding a retry with a lower temperature if the first invocation fails.

Error 7: The sub-tasks never get marked as "completed"

Symptom: sub_tasks shows every task as "pending" even after the workers ran.

Cause: The wrapper nodes don't update the status of the sub-tasks in the state, or the match between assigned_to and the agent that ran doesn't line up.

Solution: Verify that the loop in each wrapper (for t in state.get("sub_tasks", [])) correctly matches t["assigned_to"] == "researcher" (or the corresponding agent). Print sub_tasks before and after each wrapper to identify where the update gets lost.

---

Connection to M9

Your Research Agent v5 is a functional system of 4 coordinated agents. But how do you know it works well? With an individual agent, you could read the output and evaluate it manually. With 4 agents, the points of failure multiply:

• Does the Supervisor decompose correctly? Does it assign to the right agent?
• Does the Researcher find relevant sources or generic ones?
• Does the Analyst identify real patterns or invent findings?
• Does the Writer produce a coherent report or just concatenate text with no structure?
• Does the coordination between agents preserve the necessary information?

Module 9 (Testing and Evaluating Agents) tackles exactly this:

• Golden datasets: Questions with expected answers to evaluate each individual agent and the complete system
• Trajectory evaluation: You don't just evaluate the final result — you evaluate the sequence of the Supervisor's decisions. Was the order right? Were the reassignments justified?
• Per-agent evaluation: Each agent gets evaluated independently against metrics for its specialty. The Researcher against source precision/recall. The Writer against coherence and structure
• Regression testing: When you change the Supervisor's prompt or add a tool to the Analyst, does the system still produce quality results? Automated tests that catch regressions

The transition is direct: M8 builds the system → M9 verifies it works → M10 deploys it to production.

---

Resources

1. LangGraph Multi-Agent Supervisor Tutorial — The official tutorial for implementing the Supervisor pattern with workers
2. LangGraph Multi-Agent Concepts — Conceptual documentation: subgraphs, state management, communication between agents
3. LangGraph Subgraphs How-to — A guide to creating subgraphs with state mapping — the foundation of the worker nodes
4. Building Effective Agents — Anthropic — Patterns for coordinating, delegating, and orchestrating agents
5. OpenAI: Orchestrating Agents — OpenAI's guide to multi-step agents and coordination
6. langchain-mcp-adapters — The MCP → LangChain/LangGraph bridge for integrating tools into workers

---