Module 4: State Machines for Agents with LangGraph

8. Project: Research Agent with a State Machine

Next →

Project Overview

This project is different from the previous three. In modules 1-3 you built independent mini-projects — a ReAct agent, an agent with 5 tools, an extraction + routing system. Each lived in its own universe. From here on, you build one single evolving project that grows module by module until it becomes a production-ready multi-agent system in Module 10. And it all starts here, with the design of the base state machine.

You're going to build an AI Research Agent v1: an agent that takes a research question, breaks it into sub-questions, searches for information, evaluates the quality of what it found, and synthesizes a final answer. Four nodes — planning, research, analysis, synthesis — connected by a StateGraph with conditional routing, iteration limits, and typed state. It's the complete architecture of an agent controlled by a state machine.

Why does this foundation's design matter? Because every decision you make here — which fields the state has, how the nodes connect, where you put the stop conditions — directly affects the next 6 modules. In M5 you'll add deep planning and reflection. In M6, persistent memory. In M7, tools via MCP. In M8, multi-agent. In M9, testing. In M10, production. If your state machine is designed with clear interfaces and extensible state, every module will be a natural extension. If the design is coupled, every module will be a painful rewrite.

This isn't just a coding project — it's an architecture project. The quality of the agent you have in Module 10 depends on the decisions you make in the next 60-90 minutes.

Estimated time: 60-90 minutes.

Project Goal

Build a working Research Agent modeled as a StateGraph with 4 nodes, conditional routing, iteration limits, and extensible typed state.

By the end you'll be able to:

  • Model an agent as a StateGraph with functional nodes and edges as state transitions
  • Design typed state with TypedDict + Annotated that supports extensibility for future modules
  • Implement conditional routing based on quality_score and iteration_count
  • Build a research ↔ analysis cycle that iterates until it reaches a quality threshold
  • Visualize the agent's graph with draw_mermaid_png
  • Articulate why each state field exists and how it'll be used in later modules

Technical Specifications

Stack

TechnologyVersionUse
Python3.11+Runtime
langchainv1.2+init_chat_model, @tool, prompts
langchain-openailatestOpenAI provider
langgraphv1.0+StateGraph, conditional edges
tavily-pythonlatestWeb search (research node)
python-dotenvanyEnvironment variables

Setup

pip install langchain langchain-openai langgraph tavily-python python-dotenv

Create a .env file in the project directory:

OPENAI_API_KEY=sk-proj-your-api-key-here
TAVILY_API_KEY=tvly-your-api-key-here

File structure

research-agent-v1/
├── .env                    # API keys
├── state.py                # AgentState definition
├── nodes.py                # 4 nodes: planning, research, analysis, synthesis
├── graph.py                # StateGraph assembly + conditional edges
├── main.py                 # Entry point + test queries
└── visualize.py            # draw_mermaid_png of the graph

You can implement everything in a single file if you prefer. Splitting it into modules is recommended — in M8 you'll need to import individual nodes as subgraphs.

The Research Agent's Architecture

  [START] → [PLANNING] → [RESEARCH] ⇄ [ANALYSIS] → [SYNTHESIS] → [END]
                              ▲              │
                              │   score<0.7  │
                              │   & iter<max │
                              └──────────────┘

Execution flow

  1. START → planning: Takes the query, creates a plan with sub-questions.
  2. planning → research: Searches for information on the first pending sub-question.
  3. research → analysis: Evaluates the results: is this enough to answer?
  4. analysis → research (conditional): If quality_score < 0.7 and iteration_count < max_iterations, go back to research.
  5. analysis → synthesis (conditional): If quality_score >= 0.7 or iteration_count >= max_iterations, move to synthesis.
  6. synthesis → END: Combines all the research into a final answer.

Two stop conditions

  1. Quality threshold (0.7): If the research is sufficient, it stops iterating.
  2. Max iterations (3): A safety net against infinite loops.

The combination matters. A quality threshold alone with no max_iterations allows infinite loops. Max_iterations alone with no quality threshold always forces the same number of iterations.

Step 1: Define the State

The state is the contract between every node. Each field has to justify its existence — if no node reads or writes it, it doesn't belong here.

# state.py
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages


class ResearchPlan(TypedDict):
    """Structure of the research plan."""
    main_query: str
    sub_questions: list[str]
    completed_questions: list[str]


class AgentState(TypedDict):
    """State of the Research Agent v1.

    Every field has a specific purpose and will be extended
    in future modules:
    - messages: M5 adds reflection messages
    - plan: M5 adds dynamic re-planning
    - research_data: M6 persists it with checkpointing
    - quality_score: M9 evaluates it with golden datasets
    - metadata: M7 adds MCP tool info, M8 adds agent_id
    """
    messages: Annotated[list[BaseMessage], add_messages]
    plan: ResearchPlan
    iteration_count: int
    max_iterations: int
    research_data: list[str]
    quality_score: float
    final_answer: str
    metadata: dict

Why each field exists

FieldWho writes → readsWhat for
messagesEveryone → EveryoneConversation history with the LLM
planplanning → research, analysisSub-questions and progress
iteration_countanalysis → routingCycle control
max_iterationsinitial state → routingConfigurable limit
research_dataresearch → analysis, synthesisAccumulated data
quality_scoreanalysis → routingQuality assessment (0.0-1.0)
final_answersynthesis → callerThe final answer
metadataanyone → anyoneExtension without breaking changes

Design decisions for extensibility

plan as a TypedDict, not a string. A string like "1. Search X\n2. Analyze Y" requires parsing in every node. A TypedDict with sub_questions and completed_questions allows direct reading and updating. In M5, you'll add priority_order and decomposition_depth.

research_data as a list[str]. A list allows clean accumulation: each cycle appends elements. In M6, each element will have a timestamp and source tracking.

metadata as a dict. An escape hatch for extensions without breaking changes. M7 adds MCP tool info, M8 adds agent_id — without modifying the TypedDict.

max_iterations in the state. Configurable per query: simple = 2, complex = 5.

Step 2: Implement the Planning Node

The planning node takes the user's query and creates a research plan. The idea is that the agent doesn't rush straight into searching — it first thinks about what it needs to know.

# nodes.py
from dotenv import load_dotenv
load_dotenv()

import json
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

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


def planning_node(state: dict) -> dict:
    """Break the query into researchable sub-questions.

    Input: messages with the user's query.
    Output: a plan with sub-questions, and an assistant message with the plan.
    """
    query = ""
    for msg in reversed(state["messages"]):
        if isinstance(msg, HumanMessage):
            query = msg.content
            break

    planning_prompt = SystemMessage(content="""You are an expert researcher. Your task is to break a research question into specific, searchable sub-questions.

Rules:
- Generate between 2 and 4 sub-questions
- Each sub-question must be specific and searchable on the internet
- The sub-questions must cover different aspects of the original question
- Don't repeat the original question

Answer with ONLY valid JSON:
{
    "sub_questions": ["question 1", "question 2", "question 3"]
}""")

    response = model.invoke([planning_prompt, HumanMessage(content=query)])

    try:
        parsed = json.loads(response.content)
        sub_questions = parsed.get("sub_questions", [query])
    except (json.JSONDecodeError, AttributeError):
        sub_questions = [query]

    plan = {
        "main_query": query,
        "sub_questions": sub_questions,
        "completed_questions": [],
    }

    plan_summary = f"Research plan for: {query}\n"
    plan_summary += f"Sub-questions ({len(sub_questions)}):\n"
    for i, q in enumerate(sub_questions, 1):
        plan_summary += f"  {i}. {q}\n"

    return {
        "plan": plan,
        "messages": [AIMessage(content=plan_summary)],
    }

Key decisions

JSON parsing with a fallback. If the model doesn't return valid JSON (~5% of the time), the fallback uses the original query as the only sub-question. The agent doesn't break — it just researches without decomposing.

A partial dict as the return value. The node only returns the fields it modifies. LangGraph merges automatically: add_messages accumulates into messages, plan gets replaced entirely.

Step 3: Implement the Research Node

The research node takes the pending sub-questions from the plan and searches for information using Tavily. It's the node that interacts with the outside world.

from langchain_community.tools.tavily_search import TavilySearchResults

search_tool = TavilySearchResults(max_results=3)


def research_node(state: dict) -> dict:
    """Search for information on the pending sub-questions."""
    plan = state["plan"]
    completed = set(plan.get("completed_questions", []))
    pending = [q for q in plan["sub_questions"] if q not in completed]

    if not pending:
        return {"messages": [AIMessage(content="All sub-questions researched.")]}

    current_question = pending[0]

    try:
        search_results = search_tool.invoke(current_question)
    except Exception as e:
        search_results = [{"content": f"Search error: {str(e)}"}]

    gathered_data = [
        r["content"] if isinstance(r, dict) and "content" in r else str(r)
        for r in search_results
    ]

    existing_data = state.get("research_data", [])

    return {
        "plan": {**plan, "completed_questions": list(completed) + [current_question]},
        "research_data": existing_data + gathered_data,
        "messages": [AIMessage(content=f"Researching: {current_question}\nSources: {len(gathered_data)}")],
    }

Key decisions

One sub-question per iteration. The node searches only the first pending sub-question on each pass. That lets you evaluate quality after each search — more granular, and it allows early stopping.

research_data accumulates, it doesn't replace. Each pass appends new data to the existing list. If the cycle runs 3 times, research_data has the data from all 3 searches.

Error handling in the search. If Tavily fails, the node doesn't crash — it appends an error message. Analysis will see insufficient data and assign a low quality_score, triggering another cycle.

Updating the plan. The node marks the sub-question as completed so the next iteration moves on to the next pending one without repeating searches.

Step 4: Implement the Analysis Node

The analysis node evaluates the quality of the accumulated research. It's the "judge" that decides whether there's enough information to synthesize or whether it needs more research.

def analysis_node(state: dict) -> dict:
    """Evaluate the quality of the research and assign a quality_score."""
    plan = state["plan"]
    research_data = state.get("research_data", [])
    iteration = state.get("iteration_count", 0)

    if not research_data:
        return {"quality_score": 0.0, "iteration_count": iteration + 1,
                "messages": [AIMessage(content="No data. Quality score: 0.0")]}

    analysis_prompt = SystemMessage(content="""Evaluate whether the collected data is enough to answer the question.

Evaluate: coverage, depth, relevance, consistency.

Answer with ONLY valid JSON:
{"quality_score": 0.0 to 1.0, "reasoning": "explanation", "missing_aspects": ["aspect 1"]}""")

    data_summary = "\n\n".join(research_data[:10])
    completed = plan.get("completed_questions", [])
    total = len(plan.get("sub_questions", []))

    user_content = f"""Question: {plan['main_query']}
Sub-questions completed: {len(completed)}/{total}
Data ({len(research_data)} fragments):
{data_summary[:3000]}"""

    response = model.invoke([analysis_prompt, HumanMessage(content=user_content)])

    try:
        parsed = json.loads(response.content)
        quality_score = max(0.0, min(1.0, float(parsed.get("quality_score", 0.5))))
        reasoning = parsed.get("reasoning", "No reasoning")
    except (json.JSONDecodeError, AttributeError, ValueError):
        quality_score, reasoning = 0.5, "Parsing failed"

    return {
        "quality_score": quality_score,
        "iteration_count": iteration + 1,
        "messages": [AIMessage(content=f"Analysis (iter {iteration+1}): score={quality_score:.2f} — {reasoning}")],
    }

Key decisions

The LLM evaluates the quality. Not simple heuristics — the LLM evaluates coverage, depth, relevance and consistency. Criteria a len(research_data) > N can't capture.

quality_score clamped to [0.0, 1.0]. max(0.0, min(1.0, ...)) normalizes out-of-range values. It prevents bugs in the conditional routing.

Fallback to 0.5. If JSON parsing fails, 0.5 is a neutral value — it neither jumps to synthesis nor forces an unnecessary iteration.

iteration_count increments here, not in research. The analysis marks the end of a complete cycle. If you incremented in research, an error in analysis wouldn't count as an iteration.

Truncated to 3000 characters. The data can be lengthy. It's a balance between enough context to evaluate and not blowing up the context window.

Step 5: Implement the Synthesis Node

The synthesis node combines all the accumulated research into a structured final answer.

def synthesis_node(state: dict) -> dict:
    """Combine all the research into a structured final answer."""
    plan = state["plan"]
    research_data = state.get("research_data", [])
    quality_score = state.get("quality_score", 0.0)
    iteration_count = state.get("iteration_count", 0)

    synthesis_prompt = SystemMessage(content="""Synthesize the findings into a clear answer in English.

Rules: cite the data (don't invent), state what's missing if applicable.

Format:
## Summary
(2-3 sentences)

## Detailed Findings
(Points with evidence)

## Limitations
(What couldn't be determined)""")

    data_combined = "\n---\n".join(research_data[:15])
    completed = plan.get("completed_questions", [])

    user_content = f"""Question: {plan['main_query']}
Researched: {json.dumps(completed, ensure_ascii=False)}
Score: {quality_score:.2f}, Iterations: {iteration_count}
Data ({len(research_data)} fragments):
{data_combined[:5000]}"""

    response = model.invoke([synthesis_prompt, HumanMessage(content=user_content)])

    return {
        "final_answer": response.content,
        "messages": [AIMessage(content=f"Synthesis complete. Iter: {iteration_count}, Score: {quality_score:.2f}")],
        "metadata": {
            **state.get("metadata", {}),
            "total_iterations": iteration_count,
            "final_quality_score": quality_score,
            "data_fragments": len(research_data),
        },
    }

Key decisions

A prompt with an output format. A fixed format (Summary, Findings, Limitations) for consistent answers. Without a format, the LLM produces variable structures.

quality_score in the prompt. If the quality is low, the LLM knows it and can state what it couldn't research under "Limitations". Better than an answer that looks complete but isn't.

Metadata is extended, not replaced. The spread {**state.get("metadata", {}), ...} preserves existing metadata. In M8, another agent might have written metadata earlier.

Step 6: Build the Graph

Now we connect the 4 nodes with regular edges and conditional edges. This is the heart of the state machine.

# graph.py
from langgraph.graph import StateGraph, START, END

from state import AgentState
from nodes import planning_node, research_node, analysis_node, synthesis_node


def should_continue_research(state: dict) -> str:
    """Decide whether to keep researching or move to synthesis.

    Two stop conditions:
    1. quality_score >= 0.7 → there's enough information
    2. iteration_count >= max_iterations → safety net

    If neither holds → more research.
    """
    quality_score = state.get("quality_score", 0.0)
    iteration_count = state.get("iteration_count", 0)
    max_iterations = state.get("max_iterations", 3)

    if quality_score >= 0.7:
        return "synthesis"

    if iteration_count >= max_iterations:
        return "synthesis"

    return "research"


def build_research_agent() -> StateGraph:
    """Build the Research Agent v1's StateGraph."""

    graph = StateGraph(AgentState)

    # --- Nodes ---
    graph.add_node("planning", planning_node)
    graph.add_node("research", research_node)
    graph.add_node("analysis", analysis_node)
    graph.add_node("synthesis", synthesis_node)

    # --- Regular edges ---
    graph.add_edge(START, "planning")
    graph.add_edge("planning", "research")
    graph.add_edge("research", "analysis")
    graph.add_edge("synthesis", END)

    # --- Conditional edge: analysis → research or synthesis ---
    graph.add_conditional_edges(
        "analysis",
        should_continue_research,
        {
            "research": "research",
            "synthesis": "synthesis",
        }
    )

    return graph


agent = build_research_agent().compile()

Anatomy of the graph

Three types of connections:

  1. Regular edges (→). START → planning → research → analysis and synthesis → END. They always run.
  2. Conditional edge (◇). analysis ◇→ research|synthesis. The should_continue_research function decides.
  3. The cycle. research → analysis → research. The two stop conditions guarantee termination.

The routing function

should_continue_research is deliberately simple — two if statements. In M5, when you add reflection, you'll need to modify it. If it's already complex now, extending it will be a problem.

Notice the order: quality first, iterations second. If the quality is sufficient, the agent stops — no matter how many iterations are left. Max iterations is the safety net.

Visualizing the Graph

Visualizing the graph isn't optional — it's debugging and communication. Before testing the agent, verify visually that the connections are right.

# visualize.py
from graph import agent

png_data = agent.get_graph().draw_mermaid_png()
with open("research_agent_graph.png", "wb") as f:
    f.write(png_data)
print("Graph saved to research_agent_graph.png")

If the visualization doesn't show 4 nodes with a cycle between research and analysis, there's a bug in the connections.

The Complete Research Agent

This is the agent's full code in one runnable file. It's the consolidated version of steps 1-6.

The code below combines every step into a runnable file. The nodes are the same ones you implemented above — here they're consolidated with the graph and the execution function.

# main.py — Research Agent v1, complete
from dotenv import load_dotenv
load_dotenv()

import json
from typing import Annotated, TypedDict
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# --- STATE ---

class ResearchPlan(TypedDict):
    main_query: str
    sub_questions: list[str]
    completed_questions: list[str]

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    plan: ResearchPlan
    iteration_count: int
    max_iterations: int
    research_data: list[str]
    quality_score: float
    final_answer: str
    metadata: dict

# --- TOOLS & MODEL ---

model = init_chat_model("openai:gpt-4.1-mini")
search_tool = TavilySearchResults(max_results=3)

# --- NODES (see Steps 2-5 for the detailed explanation) ---

def planning_node(state: dict) -> dict:
    query = ""
    for msg in reversed(state["messages"]):
        if isinstance(msg, HumanMessage):
            query = msg.content
            break

    response = model.invoke([
        SystemMessage(content="You are an expert researcher. Break the question into 2-4 specific, searchable sub-questions.\n\nAnswer with ONLY valid JSON:\n{\"sub_questions\": [\"question 1\", \"question 2\"]}"),
        HumanMessage(content=query)
    ])

    try:
        sub_questions = json.loads(response.content).get("sub_questions", [query])
    except (json.JSONDecodeError, AttributeError):
        sub_questions = [query]

    plan = {"main_query": query, "sub_questions": sub_questions, "completed_questions": []}
    plan_text = f"Plan for: {query}\n" + "".join(f"  {i}. {q}\n" for i, q in enumerate(sub_questions, 1))
    return {"plan": plan, "messages": [AIMessage(content=plan_text)]}


def research_node(state: dict) -> dict:
    plan = state["plan"]
    completed = set(plan.get("completed_questions", []))
    pending = [q for q in plan["sub_questions"] if q not in completed]
    if not pending:
        return {"messages": [AIMessage(content="All sub-questions researched.")]}

    current_question = pending[0]
    try:
        search_results = search_tool.invoke(current_question)
    except Exception as e:
        search_results = [{"content": f"Error: {str(e)}"}]

    gathered = [r["content"] if isinstance(r, dict) and "content" in r else str(r) for r in search_results]
    summary = f"Researching: {current_question}\nSources: {len(gathered)}\n"

    return {
        "plan": {**plan, "completed_questions": list(completed) + [current_question]},
        "research_data": state.get("research_data", []) + gathered,
        "messages": [AIMessage(content=summary)],
    }


def analysis_node(state: dict) -> dict:
    plan, research_data = state["plan"], state.get("research_data", [])
    iteration = state.get("iteration_count", 0)
    if not research_data:
        return {"quality_score": 0.0, "iteration_count": iteration + 1,
                "messages": [AIMessage(content="No data. Score: 0.0")]}

    completed = plan.get("completed_questions", [])
    total = len(plan.get("sub_questions", []))
    data_summary = "\n\n".join(research_data[:10])[:3000]

    response = model.invoke([
        SystemMessage(content="Evaluate the research quality: coverage, depth, relevance.\n\nAnswer with ONLY JSON:\n{\"quality_score\": 0.0-1.0, \"reasoning\": \"...\", \"missing_aspects\": []}"),
        HumanMessage(content=f"Question: {plan['main_query']}\nCompleted: {len(completed)}/{total}\nData:\n{data_summary}")
    ])

    try:
        parsed = json.loads(response.content)
        score = max(0.0, min(1.0, float(parsed.get("quality_score", 0.5))))
        reasoning = parsed.get("reasoning", "")
    except (json.JSONDecodeError, AttributeError, ValueError):
        score, reasoning = 0.5, "Parsing failed"

    return {
        "quality_score": score, "iteration_count": iteration + 1,
        "messages": [AIMessage(content=f"Analysis (iter {iteration+1}): score={score:.2f} — {reasoning}")],
    }


def synthesis_node(state: dict) -> dict:
    plan, research_data = state["plan"], state.get("research_data", [])
    quality_score = state.get("quality_score", 0.0)
    iteration_count = state.get("iteration_count", 0)
    completed = plan.get("completed_questions", [])
    data_combined = "\n---\n".join(research_data[:15])[:5000]

    response = model.invoke([
        SystemMessage(content="Synthesize the findings in English.\n\nFormat:\n## Summary\n## Detailed Findings\n## Limitations"),
        HumanMessage(content=f"Question: {plan['main_query']}\nResearched: {json.dumps(completed, ensure_ascii=False)}\nScore: {quality_score:.2f}\nData:\n{data_combined}")
    ])

    return {
        "final_answer": response.content,
        "messages": [AIMessage(content=f"Synthesis complete. Iter: {iteration_count}, Score: {quality_score:.2f}")],
        "metadata": {**state.get("metadata", {}), "total_iterations": iteration_count,
                      "final_quality_score": quality_score, "data_fragments": len(research_data)},
    }


# --- GRAPH ---

def should_continue_research(state: dict) -> str:
    if state.get("quality_score", 0.0) >= 0.7:
        return "synthesis"
    if state.get("iteration_count", 0) >= state.get("max_iterations", 3):
        return "synthesis"
    return "research"

def build_research_agent():
    graph = StateGraph(AgentState)
    graph.add_node("planning", planning_node)
    graph.add_node("research", research_node)
    graph.add_node("analysis", analysis_node)
    graph.add_node("synthesis", synthesis_node)
    graph.add_edge(START, "planning")
    graph.add_edge("planning", "research")
    graph.add_edge("research", "analysis")
    graph.add_edge("synthesis", END)
    graph.add_conditional_edges("analysis", should_continue_research,
                                {"research": "research", "synthesis": "synthesis"})
    return graph

agent = build_research_agent().compile()


# --- EXECUTION ---

def run_research(query: str, max_iterations: int = 3, verbose: bool = True) -> str:
    initial_state = {
        "messages": [HumanMessage(content=query)],
        "plan": {"main_query": "", "sub_questions": [], "completed_questions": []},
        "iteration_count": 0, "max_iterations": max_iterations,
        "research_data": [], "quality_score": 0.0, "final_answer": "", "metadata": {},
    }

    if verbose:
        print(f"\n{'='*60}\n  Research Agent v1\n  Query: {query}\n  Max iterations: {max_iterations}\n{'='*60}")

    result = agent.invoke(initial_state)

    if verbose:
        print(f"\n{'='*60}\n  Complete — Iter: {result.get('iteration_count')}, "
              f"Score: {result.get('quality_score')}, "
              f"Data: {len(result.get('research_data', []))} fragments\n{'='*60}")

    return result.get("final_answer", "No answer.")


if __name__ == "__main__":
    answer = run_research("What are the current trends in AI agents and how do LangGraph and CrewAI compare?")
    print(f"\n{answer}")

Running it and the expected output

============================================================
  Research Agent v1
  Query: What are the current trends in AI agents...?
  Max iterations: 3
============================================================

Research plan for: What are the current trends...
  1. What are the main trends in AI agents in 2026?
  2. What is LangGraph and what are its main features?
  3. What is CrewAI and what are its main features?

Researching: What are the main trends in AI agents in 2026?
Sources found: 3
  [1] AI agents are increasingly being deployed in enterprise...

Analysis (iteration 1): score=0.35
  Only one of three sub-questions has been researched...

Researching: What is LangGraph and what are its main features?
Sources found: 3

Analysis (iteration 2): score=0.65
  Good coverage of trends and LangGraph. CrewAI is missing...

Researching: What is CrewAI and what are its main features?
Sources found: 3

Analysis (iteration 3): score=0.82
  Good coverage of all three aspects. Enough information.

============================================================
  Execution complete
  Iterations: 3
  Quality score: 0.82
  Data collected: 9 fragments
============================================================

## Summary
The trends in AI agents in 2026 center on multi-agent systems,
standardized tool use (MCP), and production deployment...

## Detailed Findings
...

## Limitations
...

Recommended Tests

Run these queries to validate different aspects of the agent:

Query 1: Broad topic (multiple iterations expected)

answer = run_research(
    "What's the current state of quantum computing and when will it be practical?",
    max_iterations=3
)

What it validates: The agent should decompose it into sub-questions (current state, recent advances, timeline to practicality), search progressively, and need at least 2 iterations.

Query 2: Specific topic (few iterations expected)

answer = run_research(
    "What is FastAPI and what are its advantages over Flask?",
    max_iterations=3
)

What it validates: With a well-defined topic, the quality_score should pass 0.7 in 1-2 iterations. The agent shouldn't need all 3 iterations.

Query 3: Niche topic (stress test)

answer = run_research(
    "What is the latest research on the effect of coffee on developer productivity?",
    max_iterations=3
)

What it validates: The agent can exhaust max_iterations without passing the threshold. The answer should acknowledge limitations.

Query 4: Low max iterations (stop condition)

answer = run_research("Explain the transformer architecture", max_iterations=1)

What it validates: With max_iterations=1, only one research → analysis → synthesis cycle. Check that the stop condition works.

Success Criteria

  1. The graph compiles and visualizes. build_research_agent().compile() with no errors. draw_mermaid_png() shows 4 nodes with the research ↔ analysis cycle.

  2. The 4 nodes work independently. Each node takes state, transforms it, and returns a partial dict with no errors.

  3. The cycle iterates correctly. The agent runs multiple research → analysis cycles. iteration_count increments in each one.

  4. Both stop conditions work. The quality threshold stops the cycle before exhausting iterations when quality is sufficient. Max iterations stops the cycle when quality_score doesn't reach the threshold.

  5. The final state is complete. plan with completed sub-questions, accumulated research_data, quality_score, final_answer, and metadata with metrics.

Checklist

  • AgentState with 8 typed fields, messages with the add_messages reducer
  • planning_node generates 2-4 sub-questions
  • research_node searches one sub-question per iteration, accumulates data
  • analysis_node assigns a quality_score [0.0-1.0], increments iteration_count
  • synthesis_node generates a structured answer
  • Graph: 4 nodes + a conditional edge on analysis
  • Stop condition: quality threshold (0.7) tested
  • Stop condition: max_iterations tested
  • run_research() works end-to-end with verbose mode
  • Tested with at least 3 different queries
  • draw_mermaid_png() generates the correct diagram

Common Errors

Error 1: research_data gets replaced instead of accumulating

Symptom: After 3 research iterations, research_data only has the data from the last search.

Cause: The research node returns {"research_data": gathered_data} instead of combining it with the existing data.

Solution:

# BAD: replaces the previous data
return {"research_data": gathered_data}

# GOOD: accumulates with the existing data
existing_data = state.get("research_data", [])
return {"research_data": existing_data + gathered_data}

Unlike messages (which has the add_messages reducer), research_data has no reducer. Without explicit accumulation, every node overwrites the previous value.

Error 2: The quality_score is always high and the agent never iterates

Symptom: The agent always goes from analysis to synthesis on the first iteration, regardless of the query.

Cause: The analysis node's prompt isn't demanding enough, or the LLM tends to give high scores.

Solution: Tighten the prompt:

# Add to the analysis node's prompt:
"Be strict in your evaluation. A score of 0.7+ requires that ALL the
sub-questions are covered with concrete data. If information is missing
about any sub-question, the score must not exceed 0.5."

You can also lower the threshold from 0.7 to 0.6, or raise max_iterations to give it more chances to iterate.

Error 3: KeyError when accessing state fields

Symptom: KeyError: 'plan' or KeyError: 'research_data' in some node.

Cause: The initial state doesn't include every field, and the node tries to access it without a default.

Solution: Always use .get() with a default value:

# BAD: crashes if the field doesn't exist
plan = state["plan"]
data = state["research_data"]

# GOOD: safe default
plan = state.get("plan", {"main_query": "", "sub_questions": [], "completed_questions": []})
data = state.get("research_data", [])

And make sure the initial state in run_research() includes every field of AgentState.

Error 4: The LLM doesn't return valid JSON

Symptom: json.JSONDecodeError in planning_node or analysis_node.

Cause: The model adds text before/after the JSON, or uses markdown code blocks.

Solution: A more robust parser that extracts JSON from mixed text:

def parse_json_response(text: str) -> dict:
    text = text.strip()
    if text.startswith("```"):
        text = "\n".join(text.split("\n")[1:-1])
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        import re
        match = re.search(r'\{[^{}]*\}', text, re.DOTALL)
        if match:
            return json.loads(match.group())
    return {}

Error 5: Tavily returns empty results or an API key error

Symptom: research_data always has errors. The agent iterates to max_iterations with no data.

Cause: TAVILY_API_KEY isn't configured or is invalid.

Solution: Check your API key. If you don't have Tavily, use a mock:

def mock_search(query: str) -> list[dict]:
    return [{"content": f"Simulated result for: {query}. [example data]"}]

try:
    search_tool = TavilySearchResults(max_results=3)
    search_tool.invoke("test")
except Exception:
    print("Tavily unavailable, using a mock")
    search_tool = None

Error 6: The conditional edge doesn't work — it always goes to synthesis

Symptom: The agent never goes back to research, it always goes straight to synthesis.

Cause: The strings the routing function returns don't match the mapping's keys.

Solution: Check that the return values match the keys exactly:

def should_continue_research(state):
    ...
    return "research"   # must match a key in the mapping
    return "synthesis"  # must match a key in the mapping

graph.add_conditional_edges("analysis", should_continue_research, {
    "research": "research",
    "synthesis": "synthesis",
})

Error 7: The agent is slow (30-60 seconds per query)

Cause: That's expected. With 3 iterations: 1 (planning) + 3 (research) + 3 (analysis) + 1 (synthesis) = 8 LLM calls + 3 to Tavily.

Mitigation: Lower max_iterations for simple queries. In M10, you'll implement caching and parallelization.

How M5-M10 Extend This Agent

This Research Agent v1 is the extensible foundation of the evolving project. Each module adds a layer without rewriting the state machine:

ModuleExtensionWhat changes in the agent
M5: Planning & ReflectionDeep planning + self-correctionplanning_node becomes a re-planner. A reflection node is added after synthesis that can reopen the cycle. AgentState gains reflection_notes and plan_revisions
M6: Memory SystemsCheckpointing + persistent memoryMemorySaver is added at compile: graph.compile(checkpointer=MemorySaver()). The agent can pause and resume. AgentState gains thread_id for persistent conversations
M7: MCP IntegrationDynamic tools via MCPresearch_node swaps TavilySearchResults for an MCP client that discovers tools dynamically. MCP servers get added for web search, file access, databases
M8: Multi-AgentMultiple specialized agentsEach node becomes a sub-agent with its own StateGraph. A supervisor coordinates: research_agent, analysis_agent, synthesis_agent. AgentState gains agent_id and handoff_data
M9: TestingEvaluation suiteGolden datasets get created with queries + expected answers. Tests verify that quality_score > threshold, iteration_count < limit, and final_answer contains the key facts. LangSmith tracing integrated
M10: ProductionDeployment with FastAPIThe agent gets exposed as an API endpoint. Streaming of partial responses, rate limiting, cost tracking, and monitoring with LangSmith get added

What does NOT change between modules

The fundamental structure — 4 nodes + StateGraph + conditional routing — holds from M4 all the way to M10. The planning → research ↔ analysis → synthesis flow is the same. What changes is what each node does internally and what the state contains. If your AgentState has clear fields with safe defaults, adding reflection_notes in M5 is trivial. If your nodes only read the fields they need, swapping Tavily for MCP in M7 doesn't affect the other nodes.

Resources

  1. LangGraph StateGraph Reference — The complete StateGraph API: nodes, edges, conditional edges
  2. LangGraph Conditional Edges — Documentation on conditional routing and path maps
  3. Tavily Search API — Setup, API key, and search options
  4. LangGraph Visualization — draw_mermaid_png and other visualization options
  5. Python TypedDict — Typed dictionaries for the agent's state
  6. LangGraph Agent Tutorial — The official tutorial showing the agent loop pattern as a StateGraph

Connection to the Next Module

Your Research Agent v1 researches, evaluates quality, iterates, and synthesizes. But it has a limitation: it doesn't think before acting or reflect on its work. If the first search heads in the wrong direction, it keeps searching in that direction until it exhausts its iterations.

In Module 5 (Multi-Step Reasoning and Planning) you solve exactly that:

  • Plan-and-execute: The planning node becomes a re-planner that adjusts the plan based on what it found
  • Reflection and self-correction: A reflection node that can reopen the cycle if it detects gaps
  • Task decomposition: Sub-questions that decompose recursively when the topic is complex

The transition: "Your agent follows a controlled flow (M4) → now make it plan before acting and reflect on its work (M5)." The state machine you designed here doesn't change — it gets extended with new nodes and fields.

← Previous7. Subgraphs as Agent ModulesNext →1. Introduction: Agents That Think Before They Act