Module 10: Multi-Agent Systems

Evolving Project: Multi-Agent System (v5)

Project overview

In Module 9, you built v4 of the AI Research Assistant: a supervised agent that shows its plan before executing, asks for approval for paid sources, accepts iterative feedback on drafts, and lets you correct factual data. It's a single agent that does everything — plans, searches, analyzes, writes — with a human supervising it.

But that "does everything" is the problem. Its system prompt has 4 roles mixed together. Its 6 tools compete with each other (do I use web_search or arxiv_search? when do I use format_report?). When the final report has an error, debugging is painful: did the bad data come from the search, the analysis, or the text generation? Everything happens in one monolithic pipeline.

v5 applies the principle that defines this module: specialization produces quality. Instead of one generalist agent, there are now 4 focused agents: a researcher that only searches, an analyst that only analyzes, a writer that only writes, and a supervisor that coordinates. Each agent has its own prompt optimized for a single task, its own relevant tools, and its own model (the analyst uses gpt-4.1 for complex reasoning; the others use gpt-4.1-mini for speed and cost).

The moment that defines this version: you run the same query on v4 and v5 side by side. The v5 report has more precise findings (because the researcher focuses only on searching), deeper analysis (because the analyst has all of its capacity dedicated to reasoning), and better prose (because the writer isn't distracted deciding what to search for). And when something goes wrong, you know exactly which agent failed — the per-agent logging and the flow trace tell you "the analyst received correct data but its conclusion was wrong." That's precise debugging, not archaeology.

Project goal

Evolve the AI Research Assistant from v4 (a single supervised agent) to v5 (a multi-agent system with specialization), showing that splitting the work across agents produces higher-quality reports and a system that's easier to debug.

By the end of this project:

🔧 You'll build a Researcher agent with search tools and a prompt focused on gathering information
🔧 You'll build an Analyst agent with analysis tools and a more powerful model (gpt-4.1)
🔧 You'll build a Writer agent with formatting tools and a prompt specialized in clear writing
🔧 You'll implement a Supervisor that coordinates the researcher → analyst → writer flow, with loop-back capability
🔧 You'll wire it all together with StateGraph, shared state and centralized HITL in the supervisor
🔧 You'll add per-agent logging and full-flow tracing for debugging

Before and after

v4 (Module 9): one agent doing everything

User: "Research AI agent frameworks"
Agent: [500-token system prompt trying to cover search + analysis + writing]
Agent: [tools: web_search, arxiv_search, news_search, calculator, format_report]
Agent: [searches, analyzes, writes — all in the same context]
Agent: "Here's your report."
Debug: "The data is wrong... was it the search? the analysis? the writing?"

v5 (This module): coordinated specialized agents

User: "Research AI agent frameworks"
Supervisor: "Plan: researcher → analyst → writer. Cost: $0. Shall I proceed?"
User: "Yes"
Researcher: [prompt: "Search for relevant information"] [tools: web_search, arxiv_search, news_search]
  → 8 findings in structured format
Analyst: [prompt: "Analyze and synthesize"] [tools: compare_sources, detect_patterns]
  → 4 patterns, 0 contradictions, 3 insights
Writer: [prompt: "Write a clear report"] [tools: format_report, generate_summary]
  → Final report with structure and clarity
Debug: "[analyst] Received 8 correct findings but identified a wrong pattern"
  → Fix the analyst prompt, don't touch researcher or writer

Technical specifications

Component	Version	Purpose
Python	3.11+	Runtime
LangChain	v1.2+	LLM framework
LangGraph	v1.0+	StateGraph + supervision
langchain-openai	latest	OpenAI models
pydantic	v2+	Data models

Project structure

research-assistant-v5/
├── .env
├── requirements.txt
├── agents/
│   ├── researcher.py          # NEW — agent specialized in search
│   ├── analyst.py             # NEW — agent specialized in analysis
│   ├── writer.py              # NEW — agent specialized in writing
│   └── supervisor.py          # NEW — system coordinator
├── tools/
│   ├── search_tools.py        # web_search, arxiv_search, news_search
│   ├── analysis_tools.py      # NEW — compare_sources, detect_patterns
│   └── writing_tools.py       # NEW — format_report, generate_summary
├── state/
│   └── multi_agent_state.py   # NEW — shared multi-agent state
├── tracing/
│   ├── agent_logger.py        # NEW — per-agent logging with a prefix
│   └── flow_tracer.py         # NEW — tracing the flow between agents
├── config/
│   └── settings.py            # EXTENDED — models per agent
└── main.py                    # NEW — orchestration and CLI

System architecture

                    ┌─────────────┐
                    │  Supervisor  │ ← HITL: plan approval
                    │  (gpt-4.1-  │
                    │    mini)    │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
              ▼            ▼            ▼
       ┌────────────┐ ┌──────────┐ ┌──────────┐
       │ Researcher  │ │ Analyst  │ │  Writer  │
       │ (gpt-4.1-  │ │(gpt-4.1)│ │(gpt-4.1- │
       │   mini)    │ │          │ │  mini)   │
       └────────────┘ └──────────┘ └──────────┘
       Tools:          Tools:       Tools:
       - web_search    - compare    - format
       - arxiv_search  - patterns   - summary
       - news_search

The main flow is: Supervisor → Researcher → Analyst → Writer → Supervisor.

If the Analyst decides it needs more data, the Supervisor sends the work back to the Researcher (loop-back). The Supervisor is the single HITL point: it approves the plan before starting and reviews the final output.

Step 1: Shared multi-agent state (state/multi_agent_state.py)

The state is the contract between agents. Each agent reads what it needs and writes its output into separate fields.

"""
state/multi_agent_state.py
Shared state for the v5 multi-agent system.
"""

import operator
from typing import TypedDict, Annotated
from pydantic import BaseModel, Field
from datetime import datetime


class Finding(BaseModel):
    title: str
    source: str
    source_type: str
    content: str
    relevance: float = Field(ge=0.0, le=1.0)


class AnalysisResult(BaseModel):
    patterns: list[str]
    contradictions: list[str]
    insights: list[str]
    confidence: float = Field(ge=0.0, le=1.0)
    needs_more_data: bool = False
    data_gaps: list[str] = Field(default_factory=list)


class Report(BaseModel):
    title: str
    summary: str
    sections: list[dict]
    sources_count: int
    confidence: float
    generated_at: str = Field(default_factory=lambda: datetime.now().isoformat())
    version: str = "v5"


class MultiAgentState(TypedDict):
    query: str
    plan: dict
    plan_approved: bool

    findings: list[dict]
    analysis: dict
    report: dict

    current_agent: str
    iteration: int
    max_iterations: int
    status: str

    trace: Annotated[list[str], operator.add]
    agent_logs: Annotated[list[dict], operator.add]

The key fields:

✅ findings — the Researcher writes here, the Analyst reads
✅ analysis — the Analyst writes here, the Writer reads
✅ report — the Writer writes here, the Supervisor validates
✅ trace — everyone writes, accumulative with operator.add
✅ iteration — the Supervisor uses it to prevent infinite loops

Step 2: Per-agent logging and tracing (tracing/)

agent_logger.py

"""
tracing/agent_logger.py
Logger with a per-agent prefix for multi-agent debugging.
"""

import logging
import time


class AgentLogger:
    def __init__(self, agent_name: str):
        self.agent_name = agent_name
        self.logger = logging.getLogger(f"multiagent.{agent_name}")

        if not self.logger.handlers:
            handler = logging.StreamHandler()
            formatter = logging.Formatter(
                f"%(asctime)s | %(levelname)-5s | [{agent_name}] %(message)s",
                datefmt="%H:%M:%S",
            )
            handler.setFormatter(formatter)
            self.logger.addHandler(handler)
            self.logger.setLevel(logging.DEBUG)

    def info(self, msg: str):
        self.logger.info(msg)

    def debug(self, msg: str):
        self.logger.debug(msg)

    def warning(self, msg: str):
        self.logger.warning(msg)

    def error(self, msg: str):
        self.logger.error(msg)

    def log_entry(self, event: str, data: dict | None = None) -> dict:
        """Builds an entry for the state's agent_logs field."""
        entry = {
            "agent": self.agent_name,
            "event": event,
            "timestamp": time.time(),
            "data": data or {},
        }
        self.info(f"{event}: {data}" if data else event)
        return entry

flow_tracer.py

"""
tracing/flow_tracer.py
Tracing the flow between agents: who handled what, and when.
"""

import time


class FlowTracer:
    def __init__(self):
        self.events: list[dict] = []
        self.start_time = time.time()

    def record(self, agent: str, event: str, details: str = ""):
        elapsed = (time.time() - self.start_time) * 1000
        self.events.append({
            "agent": agent,
            "event": event,
            "details": details,
            "elapsed_ms": round(elapsed),
        })

    def print_timeline(self):
        print(f"\n{'─' * 70}")
        print(f"  {'Agent':<15} {'Event':<20} {'Time':>8}  Details")
        print(f"{'─' * 70}")
        for e in self.events:
            print(f"  {e['agent']:<15} {e['event']:<20} {e['elapsed_ms']:>6}ms  {e['details']}")
        print(f"{'─' * 70}")

    def find_bottleneck(self) -> dict | None:
        starts: dict[str, int] = {}
        ends: dict[str, int] = {}
        for e in self.events:
            if e["event"] == "started":
                starts[e["agent"]] = e["elapsed_ms"]
            elif e["event"] == "completed":
                ends[e["agent"]] = e["elapsed_ms"]

        durations = {a: ends[a] - starts[a] for a in starts if a in ends}
        if not durations:
            return None

        slowest = max(durations, key=durations.get)
        return {"agent": slowest, "duration_ms": durations[slowest]}

    def total_time_ms(self) -> int:
        if not self.events:
            return 0
        return self.events[-1]["elapsed_ms"]


tracer = FlowTracer()

Step 3: Specialized tools per agent (tools/)

search_tools.py

"""
tools/search_tools.py
Search tools — assigned exclusively to the Researcher.
"""

import time
import random


def web_search(query: str) -> list[dict]:
    time.sleep(0.1)
    return [
        {
            "title": f"Web: {query} - result {i+1}",
            "source": f"https://example.com/{query.replace(' ', '-')}-{i+1}",
            "source_type": "web",
            "content": f"Relevant web information about {query}. Finding {i+1} of 3.",
            "relevance": round(random.uniform(0.6, 0.95), 2),
        }
        for i in range(3)
    ]


def arxiv_search(query: str) -> list[dict]:
    time.sleep(0.15)
    return [
        {
            "title": f"arXiv: {query} - paper {i+1}",
            "source": f"https://arxiv.org/abs/2026.{random.randint(10000, 99999)}",
            "source_type": "academic",
            "content": f"Academic paper about {query}. Methodology and results of study {i+1}.",
            "relevance": round(random.uniform(0.7, 0.98), 2),
        }
        for i in range(2)
    ]


def news_search(query: str) -> list[dict]:
    time.sleep(0.08)
    return [
        {
            "title": f"News: {query} - latest development",
            "source": "https://technews.example.com/latest",
            "source_type": "news",
            "content": f"Recent news about {query}. Developments from the last 48 hours.",
            "relevance": round(random.uniform(0.5, 0.85), 2),
        }
    ]

analysis_tools.py

"""
tools/analysis_tools.py
Analysis tools — assigned exclusively to the Analyst.
"""


def compare_sources(findings: list[dict]) -> dict:
    source_types = set(f.get("source_type", "unknown") for f in findings)
    contradictions = []

    for i, f1 in enumerate(findings):
        for f2 in findings[i + 1:]:
            if f1.get("source_type") != f2.get("source_type"):
                pass

    avg_relevance = sum(f.get("relevance", 0.5) for f in findings) / max(len(findings), 1)

    return {
        "total_sources": len(findings),
        "source_types": list(source_types),
        "contradictions": contradictions,
        "avg_relevance": round(avg_relevance, 2),
        "diversity_score": round(len(source_types) / max(len(findings), 1), 2),
    }


def detect_patterns(findings: list[dict]) -> list[str]:
    patterns = []

    if len(findings) >= 3:
        patterns.append(f"Consensus across {len(findings)} sources on the main topic")

    source_types = [f.get("source_type") for f in findings]
    if "academic" in source_types and "web" in source_types:
        patterns.append("Convergence between academic research and web content")

    high_relevance = [f for f in findings if f.get("relevance", 0) > 0.8]
    if len(high_relevance) >= 2:
        patterns.append(f"{len(high_relevance)} high-relevance sources (>80%)")

    if "news" in source_types:
        patterns.append("Topic with active media coverage")

    return patterns

writing_tools.py

"""
tools/writing_tools.py
Writing tools — assigned exclusively to the Writer.
"""


def format_report(title: str, summary: str, sections: list[dict], sources_count: int) -> dict:
    formatted_sections = []
    for i, section in enumerate(sections, 1):
        formatted_sections.append({
            "number": i,
            "heading": section.get("heading", f"Section {i}"),
            "content": section.get("content", ""),
        })

    return {
        "title": title,
        "summary": summary,
        "sections": formatted_sections,
        "sources_count": sources_count,
        "format": "structured",
    }


def generate_summary(findings: list[dict], analysis: dict) -> str:
    num_findings = len(findings)
    num_patterns = len(analysis.get("patterns", []))
    confidence = analysis.get("confidence", 0.5)
    contradictions = len(analysis.get("contradictions", []))

    parts = [
        f"Research based on {num_findings} sources.",
        f"{num_patterns} main patterns were identified.",
    ]

    if contradictions > 0:
        parts.append(f"{contradictions} contradictions were found between sources.")
    else:
        parts.append("The sources are consistent with each other.")

    parts.append(f"Overall confidence level: {confidence:.0%}.")

    return " ".join(parts)

Step 4: Specialized agents (agents/)

researcher.py

"""
agents/researcher.py
Researcher Agent — specialized in searching for information.
Model: gpt-4.1-mini (fast, cheap).
"""

import json
from langchain.chat_models import init_chat_model

import sys
sys.path.insert(0, ".")

from tools.search_tools import web_search, arxiv_search, news_search
from tracing.agent_logger import AgentLogger

logger = AgentLogger("researcher")
model = init_chat_model("openai:gpt-4.1-mini", temperature=0.1)

RESEARCHER_PROMPT = (
    "You are a specialized researcher. Your ONLY task is to search for relevant information "
    "about the given topic. Do NOT analyze, do NOT draw conclusions, do NOT write reports. "
    "Just search and return the findings in a structured format."
)


def run_researcher(query: str, sources: list[str] | None = None) -> dict:
    """Runs the researcher agent on a query."""
    if sources is None:
        sources = ["web", "arxiv", "news"]

    logger.info(f"Starting search: '{query}' across {sources}")

    all_findings = []

    search_fns = {
        "web": web_search,
        "arxiv": arxiv_search,
        "news": news_search,
    }

    for source in sources:
        fn = search_fns.get(source)
        if fn:
            try:
                results = fn(query)
                all_findings.extend(results)
                logger.debug(f"{source}: {len(results)} results OK")
            except Exception as e:
                logger.error(f"{source}: error — {e}")
        else:
            logger.warning(f"Unknown source: {source}")

    all_findings.sort(key=lambda f: f.get("relevance", 0), reverse=True)

    logger.info(f"Search complete: {len(all_findings)} findings from {len(sources)} sources")

    return {
        "findings": [f for f in all_findings],
        "sources_searched": sources,
        "total_results": len(all_findings),
    }

analyst.py

"""
agents/analyst.py
Analyst Agent — specialized in analysis and synthesis.
Model: gpt-4.1 (powerful, for complex reasoning).
"""

import json
from langchain.chat_models import init_chat_model

import sys
sys.path.insert(0, ".")

from tools.analysis_tools import compare_sources, detect_patterns
from tracing.agent_logger import AgentLogger

logger = AgentLogger("analyst")
model = init_chat_model("openai:gpt-4.1", temperature=0.2)

ANALYST_PROMPT = (
    "You are a specialized analyst. Your ONLY task is to analyze research findings, "
    "identify patterns, detect contradictions, and generate insights. "
    "Do NOT search for new information, do NOT write final reports. "
    "If the data is insufficient, say so clearly."
)


def run_analyst(findings: list[dict], query: str) -> dict:
    """Runs the analyst agent on the researcher's findings."""
    logger.info(f"Analyzing {len(findings)} findings about '{query}'")

    if len(findings) < 2:
        logger.warning(f"Only {len(findings)} finding(s) — insufficient data")
        return {
            "patterns": [],
            "contradictions": [],
            "insights": [f"Insufficient data: only {len(findings)} finding(s)"],
            "confidence": 0.3,
            "needs_more_data": True,
            "data_gaps": ["More sources are needed for a reliable analysis"],
        }

    comparison = compare_sources(findings)
    logger.debug(f"Comparison: {comparison['total_sources']} sources, diversity {comparison['diversity_score']}")

    patterns = detect_patterns(findings)
    logger.debug(f"Patterns identified: {len(patterns)}")

    findings_text = "\n".join(
        f"- [{f.get('source_type', '?')}] {f.get('title', '?')}: {f.get('content', '')[:100]}"
        for f in findings[:8]
    )

    response = model.invoke(
        f"{ANALYST_PROMPT}\n\n"
        f"Topic: {query}\n"
        f"Findings ({len(findings)}):\n{findings_text}\n\n"
        f"Automatically detected patterns: {patterns}\n"
        f"Source comparison: {comparison}\n\n"
        f"Generate 2-4 concise insights about what this data reveals. "
        f"Respond in JSON: {{\"insights\": [\"...\"], \"confidence\": 0.8, "
        f"\"needs_more_data\": false, \"data_gaps\": []}}\n"
        f"JSON only."
    )

    try:
        llm_analysis = json.loads(response.content)
    except json.JSONDecodeError:
        logger.warning("The LLM didn't return valid JSON — falling back to automatic analysis")
        llm_analysis = {
            "insights": [f"General analysis of {len(findings)} sources about {query}"],
            "confidence": 0.6,
            "needs_more_data": False,
            "data_gaps": [],
        }

    result = {
        "patterns": patterns,
        "contradictions": comparison["contradictions"],
        "insights": llm_analysis.get("insights", []),
        "confidence": llm_analysis.get("confidence", 0.6),
        "needs_more_data": llm_analysis.get("needs_more_data", False),
        "data_gaps": llm_analysis.get("data_gaps", []),
        "source_comparison": comparison,
    }

    logger.info(
        f"Analysis complete: {len(patterns)} patterns, "
        f"{len(result['insights'])} insights, "
        f"confidence {result['confidence']:.0%}, "
        f"more data: {'yes' if result['needs_more_data'] else 'no'}"
    )

    return result

writer.py

"""
agents/writer.py
Writer Agent — specialized in writing reports.
Model: gpt-4.1-mini (fast, good prose).
"""

import json
from langchain.chat_models import init_chat_model

import sys
sys.path.insert(0, ".")

from tools.writing_tools import format_report, generate_summary
from tracing.agent_logger import AgentLogger

logger = AgentLogger("writer")
model = init_chat_model("openai:gpt-4.1-mini", temperature=0.3)

WRITER_PROMPT = (
    "You are a writer specialized in research reports. "
    "Your ONLY task is to take the findings and the analysis and produce a clear, "
    "structured, professional report. Do NOT research, do NOT analyze. Just write."
)


def run_writer(query: str, findings: list[dict], analysis: dict) -> dict:
    """Runs the writer agent to generate the final report."""
    logger.info(f"Generating report for '{query}'")
    logger.debug(f"Input: {len(findings)} findings, {len(analysis.get('insights', []))} insights")

    auto_summary = generate_summary(findings, analysis)

    findings_text = "\n".join(
        f"- [{f.get('source_type', '?')}] {f.get('title', '?')}: {f.get('content', '')[:80]}"
        for f in findings[:6]
    )
    insights_text = "\n".join(f"- {ins}" for ins in analysis.get("insights", []))
    patterns_text = "\n".join(f"- {p}" for p in analysis.get("patterns", []))

    response = model.invoke(
        f"{WRITER_PROMPT}\n\n"
        f"Topic: {query}\n"
        f"Findings:\n{findings_text}\n\n"
        f"Patterns:\n{patterns_text}\n\n"
        f"Insights from the analysis:\n{insights_text}\n\n"
        f"Overall confidence: {analysis.get('confidence', 0.5):.0%}\n\n"
        f"Generate a report with 2-3 sections. Respond in JSON:\n"
        f'{{"title": "...", "summary": "...", "sections": ['
        f'{{"heading": "...", "content": "..."}}]}}\n'
        f"JSON only."
    )

    try:
        report_data = json.loads(response.content)
    except json.JSONDecodeError:
        logger.warning("The LLM didn't return valid JSON — falling back to the automatic report")
        report_data = {
            "title": f"Report: {query}",
            "summary": auto_summary,
            "sections": [
                {"heading": "Findings", "content": findings_text},
                {"heading": "Analysis", "content": insights_text},
            ],
        }

    report = format_report(
        title=report_data.get("title", f"Report: {query}"),
        summary=report_data.get("summary", auto_summary),
        sections=report_data.get("sections", []),
        sources_count=len(findings),
    )

    report["confidence"] = analysis.get("confidence", 0.5)

    logger.info(f"Report generated: '{report['title']}' ({len(report['sections'])} sections)")

    return report

supervisor.py — the coordinator

"""
agents/supervisor.py
Supervisor Agent — coordinates the researcher → analyst → writer flow.
Implements centralized HITL and loop-back handling.
"""

import sys
sys.path.insert(0, ".")

from tracing.agent_logger import AgentLogger

logger = AgentLogger("supervisor")


def create_plan(query: str, sources: list[str] | None = None) -> dict:
    if sources is None:
        sources = ["web", "arxiv", "news"]

    plan = {
        "query": query,
        "sources": sources,
        "pipeline": ["researcher", "analyst", "writer"],
        "estimated_cost": 0.0,
        "max_iterations": 3,
    }

    logger.info(f"Plan created: {plan['pipeline']} with sources {sources}")
    return plan


def evaluate_analysis(analysis: dict, iteration: int, max_iter: int) -> str:
    """Decides whether to continue to the writer or send it back to the researcher."""
    if analysis.get("needs_more_data", False) and iteration < max_iter:
        gaps = analysis.get("data_gaps", [])
        logger.warning(f"The analyst is asking for more data (iteration {iteration}/{max_iter}): {gaps}")
        return "loop_back"

    if analysis.get("confidence", 0) < 0.4 and iteration < max_iter:
        logger.warning(f"Low confidence ({analysis.get('confidence', 0):.0%}) — loop back")
        return "loop_back"

    logger.info(f"Analysis accepted (confidence {analysis.get('confidence', 0):.0%}) → writer")
    return "continue"

Step 5: Orchestration with StateGraph (main.py)

This is where everything comes together. The StateGraph defines the full flow with centralized HITL in the supervisor.

"""
main.py
Multi-agent system v5 — full orchestration.
"""

import json
import time
import operator
from typing import TypedDict, Annotated

from dotenv import load_dotenv
load_dotenv()

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

import sys
sys.path.insert(0, ".")

from agents.researcher import run_researcher
from agents.analyst import run_analyst
from agents.writer import run_writer
from agents.supervisor import create_plan, evaluate_analysis, logger as sup_logger
from tracing.agent_logger import AgentLogger
from tracing.flow_tracer import FlowTracer

pipeline_logger = AgentLogger("pipeline")


class MultiAgentState(TypedDict):
    query: str
    plan: dict
    plan_approved: bool
    findings: list[dict]
    analysis: dict
    report: dict
    current_agent: str
    iteration: int
    max_iterations: int
    status: str
    trace: Annotated[list[str], operator.add]


def supervisor_plan_node(state: MultiAgentState) -> dict:
    plan = create_plan(state["query"])
    return {
        "plan": plan,
        "max_iterations": plan["max_iterations"],
        "current_agent": "supervisor",
        "trace": [f"[supervisor] Plan: {plan['pipeline']} | sources: {plan['sources']}"],
    }


def hitl_approve_node(state: MultiAgentState) -> dict:
    """Centralized HITL: the supervisor asks for approval before executing."""
    plan = state["plan"]

    response = interrupt({
        "type": "plan_approval",
        "message": (
            f"Multi-agent research plan:\n"
            f"  Query: {plan['query']}\n"
            f"  Pipeline: {' → '.join(plan['pipeline'])}\n"
            f"  Sources: {plan['sources']}\n"
            f"  Estimated cost: ${plan['estimated_cost']:.2f}\n"
            f"  Max iterations: {plan['max_iterations']}\n"
            f"Approve? (approve / cancel)"
        ),
        "plan": plan,
    })

    action = response if isinstance(response, str) else response.get("action", "approve")

    if action == "cancel":
        return {
            "plan_approved": False,
            "status": "cancelled",
            "trace": [f"[supervisor:hitl] Plan cancelled by the user"],
        }

    return {
        "plan_approved": True,
        "status": "approved",
        "trace": [f"[supervisor:hitl] Plan approved — running the pipeline"],
    }


def route_after_approval(state: MultiAgentState) -> str:
    if state.get("status") == "cancelled":
        return "end"
    return "researcher"


def researcher_node(state: MultiAgentState) -> dict:
    sources = state["plan"].get("sources", ["web", "arxiv", "news"])
    result = run_researcher(state["query"], sources)

    return {
        "findings": result["findings"],
        "current_agent": "researcher",
        "trace": [
            f"[researcher] {result['total_results']} findings from {result['sources_searched']}",
        ],
    }


def analyst_node(state: MultiAgentState) -> dict:
    result = run_analyst(state["findings"], state["query"])

    return {
        "analysis": result,
        "current_agent": "analyst",
        "trace": [
            f"[analyst] {len(result.get('patterns', []))} patterns, "
            f"{len(result.get('insights', []))} insights, "
            f"confidence {result.get('confidence', 0):.0%}"
            f"{' — REQUESTS MORE DATA' if result.get('needs_more_data') else ''}",
        ],
    }


def supervisor_evaluate_node(state: MultiAgentState) -> dict:
    """The supervisor decides: continue to the writer or loop back to the researcher."""
    decision = evaluate_analysis(
        state["analysis"],
        state["iteration"],
        state["max_iterations"],
    )

    new_iteration = state["iteration"] + (1 if decision == "loop_back" else 0)

    return {
        "status": decision,
        "iteration": new_iteration,
        "trace": [
            f"[supervisor] Evaluation: {decision} (iteration {new_iteration}/{state['max_iterations']})",
        ],
    }


def route_after_evaluation(state: MultiAgentState) -> str:
    if state.get("status") == "loop_back":
        return "researcher"
    return "writer"


def writer_node(state: MultiAgentState) -> dict:
    result = run_writer(state["query"], state["findings"], state["analysis"])

    return {
        "report": result,
        "current_agent": "writer",
        "status": "completed",
        "trace": [
            f"[writer] Report generated: '{result.get('title', '?')}' "
            f"({len(result.get('sections', []))} sections)",
        ],
    }


builder = StateGraph(MultiAgentState)

builder.add_node("plan", supervisor_plan_node)
builder.add_node("approve", hitl_approve_node)
builder.add_node("researcher", researcher_node)
builder.add_node("analyst", analyst_node)
builder.add_node("evaluate", supervisor_evaluate_node)
builder.add_node("writer", writer_node)

builder.add_edge(START, "plan")
builder.add_edge("plan", "approve")
builder.add_conditional_edges("approve", route_after_approval, {
    "researcher": "researcher",
    "end": END,
})
builder.add_edge("researcher", "analyst")
builder.add_edge("analyst", "evaluate")
builder.add_conditional_edges("evaluate", route_after_evaluation, {
    "researcher": "researcher",
    "writer": "writer",
})
builder.add_edge("writer", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)


def format_v5_report(result: dict) -> str:
    report = result.get("report", {})
    if not report:
        return f"  ❌ {result.get('status', 'Unknown error')}"

    lines = [
        "",
        "╔" + "═" * 58 + "╗",
        "║" + f"  📄 {report.get('title', 'Report v5')}".center(58) + "║",
        "╚" + "═" * 58 + "╝",
        f"\n🎯 Confidence: {report.get('confidence', 0):.0%}",
        f"📚 Sources: {report.get('sources_count', 0)}",
        f"📝 Version: v5 (multi-agent)",
        f"\n{'─' * 60}",
        "📋 SUMMARY",
        f"{'─' * 60}",
        report.get("summary", "No summary."),
    ]

    sections = report.get("sections", [])
    for section in sections:
        lines.extend([
            f"\n{'─' * 60}",
            f"📌 {section.get('heading', 'Section')}",
            f"{'─' * 60}",
            section.get("content", ""),
        ])

    lines.append(f"\n{'═' * 60}")
    return "\n".join(lines)


def run_v5(query: str, thread_id: str = "v5-001"):
    """Runs the full multi-agent pipeline with HITL handling."""
    config = {"configurable": {"thread_id": thread_id}}
    tracer = FlowTracer()

    tracer.record("pipeline", "started", query)
    result = graph.invoke(
        {
            "query": query,
            "plan": {},
            "plan_approved": False,
            "findings": [],
            "analysis": {},
            "report": {},
            "current_agent": "",
            "iteration": 0,
            "max_iterations": 3,
            "status": "",
            "trace": [],
        },
        config,
    )

    while True:
        state = graph.get_state(config)
        if not state.next:
            break

        task_data = state.tasks
        for task_item in task_data:
            if hasattr(task_item, "interrupts") and task_item.interrupts:
                interrupt_data = task_item.interrupts[0].value
                print(f"\n{'─' * 58}")
                print(f"  🔔 {interrupt_data.get('type', 'interrupt')}")
                print(f"{'─' * 58}")
                print(f"  {interrupt_data.get('message', '')}")
                print(f"{'─' * 58}")

                choice = input("\n  → ").strip().lower()
                if choice in ("cancel", "no", "n"):
                    result = graph.invoke(Command(resume="cancel"), config)
                else:
                    result = graph.invoke(Command(resume="approve"), config)

    tracer.record("pipeline", "completed")

    print("\n=== FLOW TRACE ===")
    for step in result.get("trace", []):
        print(f"  {step}")

    print(format_v5_report(result))

    return result


if __name__ == "__main__":
    print("=" * 60)
    print("  🔬 AI Research Assistant v5 — Multi-Agent System")
    print("=" * 60)

    while True:
        try:
            query = input("\n🔎 Query (or 'exit'): ").strip()
        except (KeyboardInterrupt, EOFError):
            break

        if not query or query.lower() in ("exit", "quit", "q"):
            break

        import uuid
        thread_id = f"v5-{uuid.uuid4().hex[:8]}"
        run_v5(query, thread_id)

    print("\n  See you next time!")

Step 6: Per-agent logging and tracing in action

When you run the system, each agent logs with its prefix. This is what you see in the console:

14:30:01 | INFO  | [supervisor] Plan created: ['researcher', 'analyst', 'writer'] with sources ['web', 'arxiv', 'news']
14:30:05 | INFO  | [researcher] Starting search: 'AI agent frameworks' across ['web', 'arxiv', 'news']
14:30:05 | DEBUG | [researcher] web: 3 results OK
14:30:05 | DEBUG | [researcher] arxiv: 2 results OK
14:30:05 | DEBUG | [researcher] news: 1 results OK
14:30:05 | INFO  | [researcher] Search complete: 6 findings from 3 sources
14:30:06 | INFO  | [analyst] Analyzing 6 findings about 'AI agent frameworks'
14:30:06 | DEBUG | [analyst] Comparison: 6 sources, diversity 0.50
14:30:06 | DEBUG | [analyst] Patterns identified: 4
14:30:08 | INFO  | [analyst] Analysis complete: 4 patterns, 3 insights, confidence 82%, more data: no
14:30:08 | INFO  | [supervisor] Analysis accepted (confidence 82%) → writer
14:30:08 | INFO  | [writer] Generating report for 'AI agent frameworks'
14:30:08 | DEBUG | [writer] Input: 6 findings, 3 insights
14:30:10 | INFO  | [writer] Report generated: 'AI Agent Frameworks: A Comparative Analysis' (3 sections)

To filter by agent: grep "[analyst]" logs.txt shows only what the analyst did. If the report has an analysis error, you know exactly where to look.

Running it: a full session

cd research-assistant-v5
python main.py

============================================================
  🔬 AI Research Assistant v5 — Multi-Agent System
============================================================

🔎 Query (or 'exit'): AI agent frameworks comparison 2026

──────────────────────────────────────────────────
  🔔 plan_approval
──────────────────────────────────────────────────
  Multi-agent research plan:
    Query: AI agent frameworks comparison 2026
    Pipeline: researcher → analyst → writer
    Sources: ['web', 'arxiv', 'news']
    Estimated cost: $0.00
    Max iterations: 3
  Approve? (approve / cancel)
──────────────────────────────────────────────────

  → approve

=== FLOW TRACE ===
  [supervisor] Plan: ['researcher', 'analyst', 'writer'] | sources: ['web', 'arxiv', 'news']
  [supervisor:hitl] Plan approved — running the pipeline
  [researcher] 6 findings from ['web', 'arxiv', 'news']
  [analyst] 4 patterns, 3 insights, confidence 82%
  [supervisor] Evaluation: continue (iteration 0/3)
  [writer] Report generated: 'AI Agent Frameworks 2026' (3 sections)

╔══════════════════════════════════════════════════════════╗
║  📄 AI Agent Frameworks 2026                             ║
╚══════════════════════════════════════════════════════════╝

🎯 Confidence: 82%
📚 Sources: 6
📝 Version: v5 (multi-agent)

────────────────────────────────────────────────────────────
📋 SUMMARY
────────────────────────────────────────────────────────────
Research based on 6 sources. 4 main patterns were identified.
The sources are consistent with each other. Overall confidence
level: 82%.

────────────────────────────────────────────────────────────
📌 Overview
────────────────────────────────────────────────────────────
The main AI agent frameworks in 2026 include LangGraph, CrewAI,
AutoGen and the OpenAI Agents SDK...

────────────────────────────────────────────────────────────
📌 Technical Comparison
────────────────────────────────────────────────────────────
LangGraph offers the finest-grained control over the agent flow.
CrewAI simplifies building multi-agent teams...

────────────────────────────────────────────────────────────
📌 Recommendations
────────────────────────────────────────────────────────────
For production with control requirements: LangGraph. For fast
team prototyping: CrewAI...

════════════════════════════════════════════════════════════

Success criteria

✅ 4 specialized agents — researcher, analyst, writer and supervisor, each with its own prompt and tools
✅ Differentiated models — the analyst uses gpt-4.1 (reasoning), the others use gpt-4.1-mini (speed)
✅ The supervisor coordinates — researcher → analyst → writer flow with an evaluation after the analyst
✅ Loop-back works — if the analyst asks for more data, the supervisor sends it back to the researcher (maximum 3 iterations)
✅ Centralized HITL — the supervisor asks for plan approval before starting, not each agent separately
✅ Per-agent logging — each agent logs with its name as a prefix, filterable with grep
✅ Flow trace — the state's trace field shows the complete chain of decisions
✅ Quality report — the output is a structured report with a summary, sections, and metadata

Test scenarios

Test 1: Full flow with no loop-back

🔎 AI agent frameworks comparison
→ approve plan
→ researcher: 6 findings
→ analyst: confidence 80%, doesn't need more data
→ supervisor: continue → writer
→ writer: report with 3 sections
Result: a complete report in a single pass.

Test 2: Loop-back due to insufficient data

🔎 quantum error correction recent advances
→ approve plan
→ researcher: 6 findings (but all from the web, little academic material)
→ analyst: confidence 40%, needs_more_data=True, gaps=["recent papers are missing"]
→ supervisor: loop_back (iteration 1/3)
→ researcher: searches again (now with more focus)
→ analyst: confidence 75%, needs_more_data=False
→ supervisor: continue → writer
→ writer: improved report with more sources
Result: 2 researcher passes produced a stronger report.

Test 3: Cancelled plan

🔎 something I changed my mind about
→ cancel plan
Result: status="cancelled", no agents executed.

Quality comparison: v4 vs v5

This is the definitive test. Same query, both systems:

Query: "Compare LangGraph vs CrewAI for production multi-agent systems"

v4 (a single agent):

Summary: LangGraph and CrewAI are frameworks for AI agents.
LangGraph offers granular control and CrewAI is simpler.
Both are valid options.

Findings: 4 generic findings
Confidence: 72%

v5 (multi-agent):

Summary: Research based on 6 sources across 3 types.
4 main patterns were identified, including convergence
between academic research and web content.

Sections:
1. Overview — context on each framework
2. Technical Comparison — table with 5 criteria
3. Production Analysis — real metrics from companies
4. Recommendation — when to use each one and why

Confidence: 85%

The difference:

✅ More findings — a dedicated researcher searches better than a multitasking agent
✅ Deeper analysis — the analyst on gpt-4.1 spots patterns the generalist agent misses
✅ Better structure — the dedicated writer produces clear sections, not a block of text
✅ Precise debugging — if the "real metrics" data is wrong, you know it was the researcher (search), not the analyst or the writer

Common mistakes

1. Agents read fields that aren't theirs

Cause: The researcher accesses state["analysis"], which is still empty.

Fix: Each agent only reads its own inputs. The researcher reads query and plan. The analyst reads findings. The writer reads findings and analysis:

def researcher_node(state):
    result = run_researcher(state["query"], state["plan"]["sources"])

def analyst_node(state):
    result = run_analyst(state["findings"], state["query"])

def writer_node(state):
    result = run_writer(state["query"], state["findings"], state["analysis"])

2. The loop-back never ends (infinite delegation)

Cause: The analyst always says needs_more_data=True and there's no limit.

Fix: The max_iterations field and the check in evaluate_analysis:

def evaluate_analysis(analysis, iteration, max_iter):
    if analysis.get("needs_more_data") and iteration < max_iter:
        return "loop_back"
    return "continue"

3. The trace is empty

Cause: The nodes return "trace": "message" instead of "trace": ["message"].

Fix: The trace uses Annotated[list[str], operator.add]. Every node must return a list:

return {"trace": [f"[researcher] 6 findings"]}

4. The agent logs blend together and are unreadable

Cause: They all use the same logger with no prefix.

Fix: Each agent creates its own AgentLogger with a name:

logger = AgentLogger("researcher")
logger = AgentLogger("analyst")
logger = AgentLogger("writer")

5. The analyst doesn't receive the researcher's findings

Cause: The researcher writes to a different field than the one the analyst reads.

Fix: Check that the field is the same in both nodes. The convention: the researcher writes to findings, the analyst reads from findings:

def researcher_node(state) -> dict:
    return {"findings": result["findings"]}

def analyst_node(state) -> dict:
    result = run_analyst(state["findings"], ...)

6. The supervisor doesn't evaluate — it goes straight to the writer

Cause: The evaluate node is missing from the graph, or the edge goes from analyst straight to writer.

Fix: The graph must have: analyst → evaluate → (writer | researcher):

builder.add_edge("analyst", "evaluate")
builder.add_conditional_edges("evaluate", route_after_evaluation, {
    "researcher": "researcher",
    "writer": "writer",
})

7. HITL fires on every agent

Cause: Every agent has its own interrupt().

Fix: Only the supervisor uses interrupt(). The workers run without interrupting:

def hitl_approve_node(state):
    response = interrupt(...)

def researcher_node(state):
    result = run_researcher(...)

8. The v5 report isn't better than v4's

Cause: The specialized agents all use the same model (gpt-4.1-mini) and generic prompts.

Fix: Differentiate models by complexity. The analyst uses gpt-4.1 because it needs reasoning. The prompts have to be specific and restrictive:

RESEARCHER_PROMPT = "Your ONLY task is to search. Do NOT analyze, do NOT write."
ANALYST_PROMPT = "Your ONLY task is to analyze. Do NOT search, do NOT write."
WRITER_PROMPT = "Your ONLY task is to write. Do NOT research, do NOT analyze."

What's next: Module 11 — Deep Agents

Your Research Assistant v5 is a complete multi-agent system: 4 specialized agents, a supervisor with HITL, loop-back when the analyst needs more data, per-agent logging, and reports of higher quality than v4's. You built the orchestration by hand — you defined every node, every edge, every routing condition.

Module 11 introduces Deep Agents: LangGraph's "batteries-included" layer. Automatic planning with write_todos, a virtual filesystem so agents can store intermediate artifacts, dynamic subagent spawning, and long-term memory with pluggable backends. You'll reimplement the Research Assistant as a Deep Agent to see how the framework provides out of the box what you built by hand in this module.

The key difference: in v5 you decide "researcher → analyst → writer." In v6 (Deep Agent), the system decides the plan and executes it dynamically. Understanding v5 gives you the foundation to judge when manual orchestration is better and when the automatic framework is enough.

Project resources

LangGraph Multi-Agent Concepts — Multi-agent architectures in LangGraph
LangGraph Supervisor Pattern — Implementing the supervisor
LangGraph Subgraphs — Agents as independent subgraphs
Multi-Agent Workflows (LangChain Blog) — Production patterns and examples
LangGraph Visualization — draw_mermaid for graph debugging
Python logging Best Practices — Logging with prefixes and levels

Module 10 — LangChain & LangGraph: From Chains to Agents

← PreviousAdvanced Orchestration Next →Introduction: The Highest Level of Abstraction