intermediate⏱45 minutesLesson 10 of 10
Intermediate Project: Multi-Agent Research System
Build a production-ready multi-agent research system with a supervisor agent coordinating specialized research agents for comprehensive report generation.
Intermediate Project: Multi-Agent Research System
Build a multi-agent research system that coordinates specialized agents to research topics, analyze findings, and generate comprehensive reports. This project brings together advanced state, persistence, multi-agent patterns, and streaming.
System Architecture
User Query
↓
Supervisor Agent ──→ Planner Agent
↓ ↓
Researcher Agent ←── Research Plan
↓
Analyst Agent ──→ Synthesizes findings
↓
Writer Agent ──→ Generates final report
↓
Reviewer Agent ──→ Quality check
↓
Final Report
Step 1: Define State and Messages
python
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from typing_extensions import TypedDict, Annotated
from typing import List, Any, Optional
from operator import add
from datetime import datetime
import json
# Structured agent message
class AgentMessage(TypedDict):
sender: str
recipient: str
content: str
message_type: str # plan, research, analysis, draft, review, final
timestamp: str
class ResearchState(TypedDict):
query: str # Original user query
messages: Annotated[List[AgentMessage], add] # Agent conversation log
research_plan: str # Structured research plan
research_findings: List[str] # Raw research data
analysis: str # Synthesized analysis
draft_report: str # Written report
review_feedback: str # Quality review
final_report: str # Approved final report
next_agent: str # Supervisor's decision
errors: Annotated[List[str], add] # Error log
status: str # Overall statusStep 2: Initialize LLMs
python
# Different models for different agents
planner_llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
researcher_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
analyst_llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
writer_llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
reviewer_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
supervisor_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)Note
Different models optimize for cost. Simple tasks (research, review) use gpt-4o-mini. Complex reasoning (planning, analysis) uses gpt-4o.
Step 3: Define Agents
Supervisor Agent
python
def supervisor_agent(state: ResearchState) -> dict:
context = "\n".join(
f"{m['sender']} → {m['recipient']}: {m['content'][:200]}"
for m in state["messages"][-5:]
) if state["messages"] else "No activity yet."
prompt = f"""Research Query: {state['query']}
Current Status: {state.get('status', 'starting')}
Recent Activity:
{context}
Available agents and their order:
1. planner — Creates a research plan
2. researcher — Executes the research
3. analyst — Analyzes research findings
4. writer — Writes the report draft
5. reviewer — Reviews the draft for quality
6. complete — Task is finished
Which agent should work next? Respond with one word."""
response = supervisor_llm.invoke(prompt)
next_agent = response.content.strip().lower()
if next_agent not in ["planner", "researcher", "analyst", "writer", "reviewer"]:
next_agent = "complete"
return {"next_agent": next_agent,
"messages": [{"sender": "supervisor",
"recipient": next_agent,
"content": f"Proceed to {next_agent}",
"message_type": "instruction",
"timestamp": datetime.now().isoformat()}]}Planner Agent
python
def planner_agent(state: ResearchState) -> dict:
prompt = f"""Create a detailed research plan for: {state['query']}
The plan should include:
1. Key topics to investigate
2. Search queries to execute
3. Data sources to consult
4. Analysis methodology
Format as a structured plan with numbered sections."""
response = planner_llm.invoke(prompt)
plan = response.content
return {"research_plan": plan,
"status": "planned",
"messages": [{"sender": "planner",
"recipient": "supervisor",
"content": plan,
"message_type": "plan",
"timestamp": datetime.now().isoformat()}]}Researcher Agent
python
def researcher_agent(state: ResearchState) -> dict:
plan = state.get("research_plan", state["query"])
prompt = f"""Research Plan: {plan}
Conduct thorough research on each topic. For each area:
1. Find key facts and statistics
2. Note different perspectives
3. Identify authoritative sources
4. Highlight recent developments
Provide detailed findings for each topic."""
response = researcher_llm.invoke(prompt)
return {"research_findings": state.get("research_findings", []) + [response.content],
"status": "researched",
"messages": [{"sender": "researcher",
"recipient": "supervisor",
"content": response.content[:500],
"message_type": "research",
"timestamp": datetime.now().isoformat()}]}Analyst Agent
python
def analyst_agent(state: ResearchState) -> dict:
findings = "\n\n".join(state.get("research_findings", ["No findings"]))
prompt = f"""Research Findings:
{findings}
Analyze these findings and provide:
1. Key insights and patterns
2. Contradictions or debates
3. Gaps in the research
4. Implications and conclusions
5. Recommendations for the report
Be critical and thorough."""
response = analyst_llm.invoke(prompt)
return {"analysis": response.content,
"status": "analyzed",
"messages": [{"sender": "analyst",
"recipient": "supervisor",
"content": response.content[:500],
"message_type": "analysis",
"timestamp": datetime.now().isoformat()}]}Writer Agent
python
def writer_agent(state: ResearchState) -> dict:
report_prompt = f"""Write a comprehensive research report.
Query: {state['query']}
Analysis: {state['analysis']}
Research Findings: {json.dumps(state.get('research_findings', []))}
The report should include:
1. Executive Summary
2. Introduction
3. Methodology
4. Key Findings (with data and evidence)
5. Analysis and Discussion
6. Conclusions
7. Recommendations
8. References
Write in a professional, academic style. Use markdown formatting."""
response = writer_llm.invoke(report_prompt)
return {"draft_report": response.content,
"status": "drafted",
"messages": [{"sender": "writer",
"recipient": "supervisor",
"content": "Draft report completed",
"message_type": "draft",
"timestamp": datetime.now().isoformat()}]}Reviewer Agent
python
def reviewer_agent(state: ResearchState) -> dict:
review_prompt = f"""Review this research report for quality:
Report Draft:
{state['draft_report']}
Original Query:
{state['query']}
Check for:
1. Accuracy of claims and data
2. Completeness — does it answer the query?
3. Clarity and organization
4. Grammar and style
5. Missing sections or information
6. Bias or unsupported statements
Provide specific, actionable feedback."""
response = reviewer_llm.invoke(review_prompt)
return {"review_feedback": response.content,
"status": "reviewed",
"messages": [{"sender": "reviewer",
"recipient": "supervisor",
"content": response.content[:500],
"message_type": "review",
"timestamp": datetime.now().isoformat()}]}Finalizer Agent
python
def finalizer_agent(state: ResearchState) -> dict:
final_prompt = f"""Based on the draft report and review feedback, produce the final report.
Draft: {state['draft_report']}
Review Feedback: {state['review_feedback']}
Incorporate the feedback and polish the report. Ensure it is comprehensive,
well-structured, and directly addresses the query: {state['query']}"""
response = writer_llm.invoke(final_prompt)
return {"final_report": response.content,
"status": "completed",
"messages": [{"sender": "finalizer",
"recipient": "supervisor",
"content": "Final report ready",
"message_type": "final",
"timestamp": datetime.now().isoformat()}]}Step 4: Router and Graph Construction
python
def supervisor_router(state: ResearchState) -> str:
if state.get("status") == "completed":
return "finalize"
return state["next_agent"]
builder = StateGraph(ResearchState)
# Add nodes
builder.add_node("supervisor", supervisor_agent)
builder.add_node("planner", planner_agent)
builder.add_node("researcher", researcher_agent)
builder.add_node("analyst", analyst_agent)
builder.add_node("writer", writer_agent)
builder.add_node("reviewer", reviewer_agent)
builder.add_node("finalizer", finalizer_agent)
# Build edges
builder.add_edge(START, "supervisor")
# Supervisor routes to any agent
builder.add_conditional_edges(
"supervisor",
supervisor_router,
{
"planner": "planner",
"researcher": "researcher",
"analyst": "analyst",
"writer": "writer",
"reviewer": "reviewer",
"finalize": "finalizer"
}
)
# All agents return to supervisor
builder.add_edge("planner", "supervisor")
builder.add_edge("researcher", "supervisor")
builder.add_edge("analyst", "supervisor")
builder.add_edge("writer", "supervisor")
builder.add_edge("reviewer", "supervisor")
builder.add_edge("finalizer", END)
# Compile with persistence
app = builder.compile(checkpointer=MemorySaver())Success
The supervisor loop pattern: START → supervisor → agent → supervisor → agent → ... → finalizer → END. Each agent returns control to the supervisor after completing its work.
Step 5: Run the System
python
def research_topic(query: str, thread_id: str = "research-1") -> str:
config = {"configurable": {"thread_id": thread_id}}
# Stream the execution
for event in app.stream(
{
"query": query,
"messages": [],
"research_plan": "",
"research_findings": [],
"analysis": "",
"draft_report": "",
"review_feedback": "",
"final_report": "",
"next_agent": "",
"errors": [],
"status": "starting"
},
config,
stream_mode="updates"
):
for node, update in event.items():
if node == "__end__":
continue
if "status" in update:
print(f"[{update['status'].upper()}] {node} completed")
# Get the final result
final = app.get_state(config)
return final.values.get("final_report", "No report generated.")
# Run the research system
report = research_topic(
"What are the environmental impacts of quantum computing?",
"research-quantum-1"
)
print(report)Step 6: Adding Human Review
Add an optional human review step:
python
def human_review_node(state: ResearchState) -> dict:
response = interrupt({
"draft": state["draft_report"],
"prompt": "Review the draft. Approve or request changes."
})
if response.get("approved"):
return {"status": "approved"}
return {"status": "rejected",
"review_feedback": response.get("feedback", "Revise")}
def review_router(state: ResearchState) -> str:
if state.get("status") == "approved":
return "finalize"
return "revise"
# Add human review to graph
builder.add_node("human_review", human_review_node)
builder.add_edge("reviewer", "human_review")
builder.add_conditional_edges("human_review", review_router, {
"finalize": "finalizer",
"revise": "writer" # Send back to writer with feedback
})Tip
Adding human review at strategic points (before finalization) catches errors while keeping most of the workflow autonomous.
System Diagram
Practice Questions
Practice Question
What pattern does the multi-agent research system use?
Practice Question
How does the supervisor communicate its decision?
Practice Question
Why does the project use different LLMs for different agents?
Practice Question
What is the role of the planner agent?
Practice Question
Where does the analyst agent get its input from?
Practice Question
What terminates the supervisor loop?
Practice Question
What type of message structure enables agent communication tracking?
Practice Question
What advantage does streaming ('updates' mode) provide in this system?
Practice Question
How can you add a human review step to the research workflow?
Practice Question
What is the purpose of the errors field in the research state?
Success
Key Takeaways
- Multi-agent research system with supervisor loop pattern
- Each agent specializes in one task (planning, research, analysis, writing, review)
- Structured AgentMessage enables full communication traceability
- Different LLM models for different agents optimizes cost vs. quality
- Streaming provides real-time visibility into agent progress
- Human review can be added with interrupt() for quality control
- The supervisor decides when the work is complete, breaking the loop
- Checkpointing enables pause/resume for long-running research tasks
Progress100%