One AI Is No Longer Enough — Build a Multi-Agent System with LangGraph [2026 Tutorial]

[Single AI]
User → AI (does everything alone) → Result
         └ Research, writing, verification — all solo
[Multi-Agent]
User → Supervisor (team lead)
             ├→ Researcher Agent (research specialist)
             ├→ Writer Agent     (writing specialist)
             └→ Fact-checker Agent (verification specialist)
                       ↓
                   Final output

[User Input: "Python 2026 Trends"]
        ↓
[Supervisor Agent]   ← Team lead. Decides the overall flow
        ↓
[Researcher Agent]   ← Web research, information gathering
        ↓
[Writer Agent]       ← Drafts a blog post from the research
        ↓
[Fact-checker Agent] ← Verifies claims; approves or requests revisions
        ↓
    [Final Output]   ← Verified blog draft

pip install langgraph langchain langchain-anthropic \
            langchain-community python-dotenv

ANTHROPIC_API_KEY=your-api-key-here

multi-agent/
├── .env
├── main.py          # Entry point
├── state.py         # Shared state definition
├── agents.py        # The 3 agents
├── supervisor.py    # Supervisor logic
└── graph.py         # Graph assembly

# Team shared notepad
state = {
    "topic": "Python 2026 Trends",     # Input
    "research": "",                     # Researcher fills this
    "draft": "",                        # Writer fills this
    "fact_check": "",                   # Fact-checker fills this
    "approved": False,                  # Final approval status
    "iteration": 0                      # Revision count
}

def researcher_node(state):
    # Read topic from state
    # Perform research
    # Update state's research field
    return {"research": "...research results..."}

# Branch based on fact-check result
def should_continue(state):
    if state["approved"]:
        return "END"          # Approved → stop
    elif state["iteration"] >= 3:
        return "END"          # 3+ revisions → stop anyway
    else:
        return "writer"       # Not approved → send back to writer

# state.py
from typing import TypedDict
class ResearchState(TypedDict):
    """State shared across the entire team"""
    topic: str              # Research topic
    research_notes: str     # Researcher's findings
    draft: str              # Writer's draft
    fact_check_result: str  # Fact-checker's review
    approved: bool          # Final approval
    revision_notes: str     # Revision requests
    iteration: int          # Current revision count (prevents infinite loops)
    final_output: str       # Final deliverable

# agents.py
import os
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from langchain.prompts import ChatPromptTemplate
from state import ResearchState
load_dotenv()
# Shared LLM — all agents use the same model; roles are defined by prompts
llm = ChatAnthropic(
    model="claude-sonnet-4-20250514",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    temperature=0.3,
    max_tokens=2000
)
# ────────────────────────────────────────
# Agent 1: Researcher
# ────────────────────────────────────────
def researcher_agent(state: ResearchState) -> dict:
    """
    Takes a topic and gathers key information.
    In production, wire in a real search tool (Tavily, SerpAPI, etc.).
    Here we rely on the LLM's training knowledge.
    """
    print("🔍 [Researcher] Starting research...")
    prompt = ChatPromptTemplate.from_template("""
You are a professional researcher. Gather key information about the given topic.
Topic: {topic}
Provide your findings in the following format:
## Key Facts (5–7 items)
- Fact 1: ...
- Fact 2: ...
## Notable Statistics & Data
- ...
## Latest Trends
- ...
## Relevant Examples
- ...
Include only accurate and specific information. Mark uncertain information as "(estimated)".
""")
    chain = prompt | llm
    result = chain.invoke({"topic": state["topic"]})
    print(f"✅ [Researcher] Complete — {len(result.content)} chars")
    return {"research_notes": result.content}
# ────────────────────────────────────────
# Agent 2: Writer
# ────────────────────────────────────────
def writer_agent(state: ResearchState) -> dict:
    """
    Writes a blog draft based on the researcher's notes.
    Incorporates any revision requests from the fact-checker.
    """
    print("✍️ [Writer] Starting draft...")
    revision_context = ""
    if state.get("revision_notes"):
        revision_context = f"""
⚠️ Fact-checker revision requests:
{state['revision_notes']}
Please address all of the above in your revision.
"""
    prompt = ChatPromptTemplate.from_template("""
You are a technical blog writer. Write a clear, engaging blog post using the researcher's notes.
Topic: {topic}
Research notes:
{research_notes}
{revision_context}
Writing guidelines:
- Include a title
- Structure: intro → body (3–4 sections) → conclusion
- Explain technical terms clearly
- Use specific numbers and examples
- Target length: ~800–1,000 words
""")
    chain = prompt | llm
    result = chain.invoke({
        "topic": state["topic"],
        "research_notes": state["research_notes"],
        "revision_context": revision_context
    })
    print(f"✅ [Writer] Draft complete — {len(result.content)} chars")
    return {
        "draft": result.content,
        "revision_notes": ""  # Clear revision notes after incorporating
    }
# ────────────────────────────────────────
# Agent 3: Fact-checker
# ────────────────────────────────────────
def fact_checker_agent(state: ResearchState) -> dict:
    """
    Reviews the writer's draft against the research notes.
    Identifies factual errors or exaggerations.
    Returns either APPROVED or a revision request.
    """
    print("🔎 [Fact-checker] Starting review...")
    prompt = ChatPromptTemplate.from_template("""
You are a rigorous fact-checker. Review the blog draft.
Original research notes:
{research_notes}
Draft to review:
{draft}
Check for:
1. Claims not supported by the research notes
2. Inaccurate statistics or figures
3. Exaggerated or misleading language
4. Logical inconsistencies
Respond ONLY in the following format:
Verdict: [APPROVED or NEEDS_REVISION]
Review notes:
(If approved: 1–2 sentences on why it passes)
(If revision needed: specific items and reasons)
""")
    chain = prompt | llm
    result = chain.invoke({
        "research_notes": state["research_notes"],
        "draft": state["draft"]
    })
    content = result.content
    approved = "APPROVED" in content.upper()
    if approved:
        print("✅ [Fact-checker] Approved!")
        return {
            "fact_check_result": content,
            "approved": True,
            "final_output": state["draft"],
            "iteration": state.get("iteration", 0) + 1
        }
    else:
        print(f"⚠️ [Fact-checker] Revision requested (attempt {state.get('iteration', 0) + 1})")
        return {
            "fact_check_result": content,
            "approved": False,
            "revision_notes": content,
            "iteration": state.get("iteration", 0) + 1
        }

# supervisor.py
from state import ResearchState
def should_continue(state: ResearchState) -> str:
    """
    Reads the fact-check result and decides the next step.
    Returns:
    - "writer"  → Send back to writer for revision
    - "end"     → Work complete
    """
    MAX_ITERATIONS = 3  # Prevent infinite loops
    current_iter = state.get("iteration", 0)
    if state.get("approved", False):
        print(f"\n🎉 [Supervisor] Final approval! (Total attempts: {current_iter})")
        return "end"
    if current_iter >= MAX_ITERATIONS:
        print(f"\n⏱️ [Supervisor] Max revisions reached. Finalizing last version.")
        return "end"
    print(f"\n🔄 [Supervisor] Sending revision back to writer (attempt {current_iter + 1}/{MAX_ITERATIONS})")
    return "writer"

# graph.py
from langgraph.graph import StateGraph, END
from state import ResearchState
from agents import researcher_agent, writer_agent, fact_checker_agent
from supervisor import should_continue
def build_research_graph():
    """
    Assembles the 3-agent team as a graph.
    Flow:
    researcher → writer → fact_checker
                   ↑            ↓
                   └── (revision loop) ──┘
                                ↓ (approved or max reached)
                               END
    """
    # 1. Initialize graph
    workflow = StateGraph(ResearchState)
    # 2. Register nodes
    workflow.add_node("researcher", researcher_agent)
    workflow.add_node("writer", writer_agent)
    workflow.add_node("fact_checker", fact_checker_agent)
    # 3. Set entry point
    workflow.set_entry_point("researcher")
    # 4. Regular edges (sequential flow)
    workflow.add_edge("researcher", "writer")
    workflow.add_edge("writer", "fact_checker")
    # 5. Conditional edge (branches based on fact-checker output)
    workflow.add_conditional_edges(
        "fact_checker",
        should_continue,
        {
            "writer": "writer",
            "end": END
        }
    )
    # 6. Compile into an executable app
    return workflow.compile()

# main.py
from dotenv import load_dotenv
from graph import build_research_graph
load_dotenv()
def run_research_team(topic: str) -> dict:
    print(f"\n{'='*60}")
    print(f"🚀 Research Team Activated — Topic: '{topic}'")
    print(f"{'='*60}\n")
    app = build_research_graph()
    initial_state = {
        "topic": topic,
        "research_notes": "",
        "draft": "",
        "fact_check_result": "",
        "approved": False,
        "revision_notes": "",
        "iteration": 0,
        "final_output": ""
    }
    result = app.invoke(initial_state)
    print(f"\n{'='*60}")
    print("📄 Final Output")
    print(f"{'='*60}")
    final = result.get("final_output") or result.get("draft", "No output")
    print(final)
    print(f"\n{'='*60}")
    print(f"📊 Run Summary")
    print(f"  - Total revisions: {result.get('iteration', 0)}")
    print(f"  - Final status: {'✅ Approved' if result.get('approved') else '⏱️ Timed out'}")
    print(f"{'='*60}\n")
    return result
if __name__ == "__main__":
    result = run_research_team("Python and AI Development Trends in 2026")

============================================================
🚀 Research Team Activated — Topic: 'Python and AI Development Trends in 2026'
============================================================
🔍 [Researcher] Starting research...
✅ [Researcher] Complete — 1243 chars
✍️ [Writer] Starting draft...
✅ [Writer] Draft complete — 987 chars
🔎 [Fact-checker] Starting review...
⚠️ [Fact-checker] Revision requested (attempt 1)
🔄 [Supervisor] Sending revision back to writer (attempt 1/3)
✍️ [Writer] Starting draft...
✅ [Writer] Draft complete — 1054 chars
🔎 [Fact-checker] Starting review...
✅ [Fact-checker] Approved!
🎉 [Supervisor] Final approval! (Total attempts: 2)
============================================================

# Multiple researchers running simultaneously
workflow.add_node("researcher_news", news_researcher)
workflow.add_node("researcher_data", data_researcher)
workflow.add_node("aggregator", combine_results)
# Both researchers run in parallel, results merged in aggregator
workflow.add_edge("researcher_news", "aggregator")
workflow.add_edge("researcher_data", "aggregator")

# Pause for human approval before sensitive actions
from langgraph.types import interrupt
def sensitive_action_node(state):
    approval = interrupt({
        "message": "Ready to publish this content. Proceed?",
        "draft": state["draft"]
    })
    if approval.get("approved"):
        return {"final_output": state["draft"]}
    else:
        return {"revision_notes": approval.get("feedback", "")}