Build Your First AI Agent with LangGraph — Step-by-Step Python Tutorial (2026)

AI agents are not chatbots. A chatbot answers questions. An agent decides what to do, uses tools, evaluates results, and loops until the job is done.

If you have tried building agents with raw API calls and prompt chains, you know the pain: managing state across steps, handling tool failures, deciding when to loop and when to stop. It gets messy fast.

LangGraph solves this. It lets you model your agent as a graph — nodes are actions, edges are decisions — and it handles state, persistence, and control flow for you. It reached v1.0 in late 2025, and as of Q1 2026, it is the most widely adopted Python framework for building production AI agents. Companies like Klarna, Uber, Replit, and Elastic run LangGraph agents in production.

In this tutorial, you will build a working research agent from scratch. Not a toy demo — a real agent that takes a research question, searches the web, reads results, decides if it has enough information, and produces a structured summary. By the end, you will understand LangGraph's core concepts well enough to build your own agents for any task.

---

Why LangGraph Is the Leading AI Agent Framework in 2026

Before writing code, it helps to understand why LangGraph won the framework race.

The AI agent framework landscape in 2026 has consolidated. Microsoft shifted AutoGen to maintenance mode in favor of their broader Agent Framework. CrewAI remains popular for quick team-based prototypes. But for production workloads that need fine-grained control, durable execution, and human-in-the-loop capabilities, LangGraph is the default choice.

Here is what makes it different:

Graph-based architecture. Instead of a linear chain of prompts, you define your agent as a directed graph. Each node is a function. Each edge is a transition. Conditional edges let the agent decide its own path. This makes complex workflows — loops, branches, parallel execution — natural to express.

Durable execution. LangGraph agents can persist their state. If an agent fails midway through a 10-step research task, it resumes from where it left off. This matters in production where network failures and API timeouts are facts of life.

Human-in-the-loop. You can insert approval steps anywhere in the graph. The agent pauses, waits for human input, and continues. This is critical for agents that take real-world actions like sending emails or modifying databases.

First-class memory. Both short-term (within a conversation) and long-term (across sessions) memory are built in. Your agent can remember what it learned yesterday.

MIT licensed and free. The core library is open source. You only pay if you want LangGraph Cloud for hosted execution and monitoring.

LangGraph is built on top of LangChain, but you do not need to use LangChain's abstractions. You can use LangGraph with raw OpenAI or Anthropic API calls if you prefer. The graph orchestration layer is the value — not the LLM wrapper.

---

LangGraph Architecture: Nodes, Edges, and State — Explained Simply

LangGraph has three core concepts. Once you understand these, everything else clicks.

State

State is a Python dictionary (or TypedDict) that flows through your entire graph. Every node reads from it and writes to it. Think of it as the agent's working memory — it holds the current question, search results, draft answers, and any other data your agent needs.

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    search_results: list[str]
    final_answer: str

The Annotated[list, add_messages] pattern tells LangGraph to append new messages instead of replacing the list. This is how conversation history accumulates naturally.

Nodes

A node is a Python function that takes the current state and returns an update. That is it. No special base classes, no decorators required.

def search_web(state: AgentState) -> dict:
    query = state["messages"][-1].content
    results = tavily_client.search(query)
    return {"search_results": results}

Each node does one thing. Search the web. Analyze results. Write a summary. Small, focused functions that are easy to test individually.

Edges

Edges connect nodes and define the flow. There are three types:

• Normal edges: Always go from A to B. graph.add_edge("search", "analyze")
• Conditional edges: A routing function decides where to go next. This is where agents make decisions.
• Entry/exit edges: START and END mark where the graph begins and terminates.

def should_continue(state: AgentState) -> str:
    if len(state["search_results"]) >= 3:
        return "summarize"
    return "search"

graph.add_conditional_edges("analyze", should_continue)

The routing function returns a string — the name of the next node. This is how your agent decides whether to keep searching or move on to writing its answer.

---

Prerequisites

You need:

• Python 3.11 or higher (LangGraph 1.x requires 3.11+)
• An OpenAI API key (we use GPT-4o-mini for affordability; you can swap in any LLM)
• A Tavily API key (for web search — free tier gives 1,000 searches/month)
• Basic Python knowledge (functions, dictionaries, type hints)

If you prefer using a local LLM instead of OpenAI, check our Ollama + Open WebUI self-hosting guide for setting up a local model, or our Gemma 4 local setup guide for running Google's open model on your machine.

---

Step 1: Install LangGraph and Dependencies

Create a project directory and set up a virtual environment:

mkdir langgraph-research-agent
cd langgraph-research-agent
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

Install the dependencies with pinned versions:

pip install langgraph==0.3.34 \
    langchain-openai==0.3.12 \
    langchain-community==0.3.19 \
    tavily-python==0.5.0 \
    python-dotenv==1.1.0

Note on versions: LangGraph follows semantic versioning post-1.0. The langgraph PyPI package is at 0.3.x as the pip-installable runtime while the framework is branded as LangGraph 1.0+. Pin your versions in production to avoid breaking changes. Always check pip index versions langgraph for the latest release before starting a new project.

Create a .env file for your API keys:

OPENAI_API_KEY=sk-your-openai-key-here
TAVILY_API_KEY=tvly-your-tavily-key-here

Verify the installation:

import langgraph
print(langgraph.__version__)

If you see a version number without errors, you are ready to build.

---

Step 2: Define Your Agent's State Schema

Create a file called agent.py. This is where our entire agent will live.

The state schema is the most important design decision. It defines what your agent knows and what it can track. Get this right and everything downstream is easier.

"""Research agent built with LangGraph."""

import os
from typing import TypedDict, Annotated
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

load_dotenv()

# --- State Schema ---

class ResearchState(TypedDict):
    """The agent's working memory."""
    messages: Annotated[list, add_messages]  # Conversation history
    research_topic: str                       # What we are researching
    search_queries: list[str]                 # Queries we have run
    sources: list[dict]                       # Raw search results
    analysis: str                             # Our analysis of the sources
    final_report: str                         # The finished research report
    iteration: int                            # How many research loops we have done
    max_iterations: int                       # Safety limit on loops

A few things to notice:

• messages uses the add_messages annotation so conversation history accumulates automatically.
• iteration and max_iterations prevent infinite loops. This is not optional — any agent that can loop must have a safety limit.
• Each field has a clear purpose. When you debug your agent (and you will), having well-named state fields saves hours.

---

Step 3: Build Tool Nodes

Now we create the functions that our agent will use as tools. Each function is a node in the graph.

The LLM Setup

# --- LLM Setup ---

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.1,  # Low temperature for factual research
)

We use gpt-4o-mini because it is cheap, fast, and good enough for research tasks. You can swap this for gpt-4o, Claude, or even a local model served through Docker Model Runner or Ollama.

Node 1: Generate Search Queries

# --- Node Functions ---

def generate_queries(state: ResearchState) -> dict:
    """Turn the research topic into specific search queries."""
    topic = state["research_topic"]
    iteration = state.get("iteration", 0)

    # On later iterations, refine based on what we already found
    existing_info = ""
    if state.get("analysis"):
        existing_info = f"\n\nWe already know:\n{state['analysis']}\n\nGenerate queries to fill gaps in our knowledge."

    response = llm.invoke([
        SystemMessage(content=(
            "You are a research assistant. Generate 3 specific, diverse "
            "search queries to research the given topic. Return only the "
            "queries, one per line. No numbering, no extra text."
            f"{existing_info}"
        )),
        HumanMessage(content=f"Research topic: {topic}"),
    ])

    new_queries = [q.strip() for q in response.content.strip().split("\n") if q.strip()]

    return {
        "search_queries": state.get("search_queries", []) + new_queries,
        "messages": [response],
    }

This node demonstrates a key agent pattern: adaptive behavior across iterations. On the first pass, it generates broad queries. On later passes, it reads its own previous analysis and generates queries to fill knowledge gaps. The agent literally gets smarter each loop.

Node 2: Search the Web

from tavily import TavilyClient

tavily = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))

def search_web(state: ResearchState) -> dict:
    """Execute search queries and collect results."""
    queries = state.get("search_queries", [])
    # Only search the latest batch of queries (last 3)
    recent_queries = queries[-3:]
    all_results = state.get("sources", [])

    for query in recent_queries:
        try:
            response = tavily.search(
                query=query,
                max_results=3,
                include_raw_content=False,
            )
            for result in response.get("results", []):
                # Avoid duplicate URLs
                if not any(s["url"] == result["url"] for s in all_results):
                    all_results.append({
                        "title": result.get("title", ""),
                        "url": result.get("url", ""),
                        "content": result.get("content", ""),
                        "query": query,
                    })
        except Exception as e:
            # Log but do not crash — the agent can work with partial results
            print(f"Search failed for '{query}': {e}")

    return {"sources": all_results}

Notice the deduplication and error handling. A search API failing should not crash your entire agent. The agent works with whatever results it gets, even if some queries fail.

Node 3: Analyze Results

def analyze_results(state: ResearchState) -> dict:
    """Analyze search results and assess if we have enough information."""
    sources = state.get("sources", [])

    if not sources:
        return {
            "analysis": "No search results found. Need to try different queries.",
            "iteration": state.get("iteration", 0) + 1,
        }

    # Format sources for the LLM
    source_text = ""
    for i, source in enumerate(sources, 1):
        source_text += f"\n[{i}] {source['title']}\nURL: {source['url']}\n{source['content']}\n"

    response = llm.invoke([
        SystemMessage(content=(
            "You are a research analyst. Analyze the following search results "
            "about the given topic. Provide:\n"
            "1. Key findings (what we know)\n"
            "2. Gaps (what we still need to find out)\n"
            "3. Confidence level (low/medium/high) in our overall understanding\n\n"
            "Be specific and cite source numbers."
        )),
        HumanMessage(content=f"Topic: {state['research_topic']}\n\nSources:{source_text}"),
    ])

    return {
        "analysis": response.content,
        "iteration": state.get("iteration", 0) + 1,
        "messages": [response],
    }

Node 4: Write the Final Report

def write_report(state: ResearchState) -> dict:
    """Write a structured research report from our findings."""
    sources = state.get("sources", [])
    analysis = state.get("analysis", "")

    source_text = ""
    for i, source in enumerate(sources, 1):
        source_text += f"\n[{i}] {source['title']}\nURL: {source['url']}\n{source['content']}\n"

    response = llm.invoke([
        SystemMessage(content=(
            "You are a research writer. Write a clear, well-structured research "
            "report based on the analysis and sources provided. Include:\n"
            "- Executive summary (2-3 sentences)\n"
            "- Key findings with citations [1], [2], etc.\n"
            "- Conclusions\n"
            "- Sources list\n\n"
            "Write for a technical audience. Be factual and specific."
        )),
        HumanMessage(content=(
            f"Topic: {state['research_topic']}\n\n"
            f"Analysis:\n{analysis}\n\n"
            f"Sources:{source_text}"
        )),
    ])

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

---

Step 4: Create the Agent Graph with Conditional Routing

This is where LangGraph shines. We wire our nodes together and define the decision logic.

# --- Routing Logic ---

def should_continue_research(state: ResearchState) -> str:
    """Decide whether to keep researching or write the report."""
    iteration = state.get("iteration", 0)
    max_iterations = state.get("max_iterations", 3)
    analysis = state.get("analysis", "")

    # Hard stop: prevent infinite loops
    if iteration >= max_iterations:
        return "write_report"

    # If analysis mentions low confidence or significant gaps, keep going
    analysis_lower = analysis.lower()
    if "low" in analysis_lower and "confidence" in analysis_lower:
        return "generate_queries"
    if "significant gaps" in analysis_lower or "need more" in analysis_lower:
        return "generate_queries"

    # Otherwise, we have enough to write
    return "write_report"


# --- Build the Graph ---

workflow = StateGraph(ResearchState)

# Add nodes
workflow.add_node("generate_queries", generate_queries)
workflow.add_node("search_web", search_web)
workflow.add_node("analyze_results", analyze_results)
workflow.add_node("write_report", write_report)

# Add edges
workflow.add_edge(START, "generate_queries")
workflow.add_edge("generate_queries", "search_web")
workflow.add_edge("search_web", "analyze_results")

# Conditional edge: the agent decides whether to loop or finish
workflow.add_conditional_edges(
    "analyze_results",
    should_continue_research,
    {
        "generate_queries": "generate_queries",
        "write_report": "write_report",
    },
)

workflow.add_edge("write_report", END)

# Compile the graph
agent = workflow.compile()

Let me break down what this graph does:

1. START → generate_queries: The agent creates search queries from the topic.
2. generate_queries → search_web: It executes those queries.
3. search_web → analyze_results: It analyzes what it found.
4. analyze_results → ???: The conditional edge kicks in. If the analysis says we need more information, we loop back to generate new queries. If we have enough, we move to writing.
5. write_report → END: The agent outputs its final report.

This is the agent loop — the core pattern that separates agents from simple chains. The agent evaluates its own work and decides what to do next.

---

Step 5: Add Memory and Conversation History

LangGraph supports two types of memory: checkpointing (short-term, within a session) and persistent stores (long-term, across sessions).

For our research agent, checkpointing is the key feature. It lets you pause and resume research, and it gives you a full audit trail of every step the agent took.

from langgraph.checkpoint.memory import MemorySaver

# Add memory to the compiled graph
memory = MemorySaver()
agent_with_memory = workflow.compile(checkpointer=memory)

That is it — two lines. Now every state transition is saved. You can:

• Resume a failed research session from the last successful step
• Inspect exactly what the agent did at each node
• Replay the agent's decision-making for debugging

To use memory, pass a thread_id in the config:

config = {"configurable": {"thread_id": "research-session-1"}}

result = agent_with_memory.invoke(
    {
        "research_topic": "Impact of AI coding assistants on developer productivity in 2026",
        "messages": [],
        "search_queries": [],
        "sources": [],
        "analysis": "",
        "final_report": "",
        "iteration": 0,
        "max_iterations": 3,
    },
    config=config,
)

Each unique thread_id maintains its own conversation history and state. Use different thread IDs for different research sessions.

For long-term memory across sessions (e.g., remembering user preferences or past research topics), LangGraph provides the Store interface:

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
agent_with_long_memory = workflow.compile(
    checkpointer=memory,
    store=store,
)

The store gives your agent a key-value memory that persists across different threads. This is useful for agents that need to build up knowledge over time — but for our research agent, checkpointing alone is sufficient.

---

Step 6: Run and Test Your Agent

Add the main execution block to agent.py:

# --- Run the Agent ---

if __name__ == "__main__":
    print("=" * 60)
    print("LangGraph Research Agent")
    print("=" * 60)

    topic = input("\nEnter a research topic: ").strip()
    if not topic:
        topic = "How are companies using AI agents in production in 2026?"

    print(f"\nResearching: {topic}")
    print("-" * 60)

    initial_state = {
        "research_topic": topic,
        "messages": [],
        "search_queries": [],
        "sources": [],
        "analysis": "",
        "final_report": "",
        "iteration": 0,
        "max_iterations": 3,
    }

    config = {"configurable": {"thread_id": "session-001"}}

    # Stream to see the agent's progress in real time
    for event in agent_with_memory.stream(initial_state, config=config):
        for node_name, output in event.items():
            print(f"\n>> Node: {node_name}")
            if node_name == "generate_queries" and "search_queries" in output:
                print(f"   Queries: {output['search_queries'][-3:]}")
            elif node_name == "search_web" and "sources" in output:
                print(f"   Found {len(output['sources'])} total sources")
            elif node_name == "analyze_results" and "analysis" in output:
                print(f"   Iteration: {output.get('iteration', '?')}")
                print(f"   Analysis preview: {output['analysis'][:200]}...")
            elif node_name == "write_report" and "final_report" in output:
                print(f"\n{'=' * 60}")
                print("RESEARCH REPORT")
                print("=" * 60)
                print(output["final_report"])

    print(f"\n{'=' * 60}")
    print("Research complete.")

Run it:

python agent.py

You should see the agent work through its graph — generating queries, searching, analyzing, potentially looping back for more research, and finally writing its report.

Debugging Tips

If the agent loops too many times, check your should_continue_research function. The routing logic is where most agent bugs live.

To visualize the graph (useful for complex agents):

# Requires: pip install pygraphviz
print(agent.get_graph().draw_mermaid())

This outputs a Mermaid diagram you can paste into any Mermaid renderer to see your agent's flow visually.

---

Step 7: Deploy with LangServe

Once your agent works locally, you might want to expose it as an API. LangServe turns any LangGraph agent into a REST API with streaming support.

pip install langserve[all]==0.3.1 fastapi==0.115.0 uvicorn==0.34.0

Create server.py:

"""Serve the research agent as a REST API."""

from fastapi import FastAPI
from langserve import add_routes
from agent import agent_with_memory

app = FastAPI(
    title="Research Agent API",
    description="AI research agent powered by LangGraph",
)

add_routes(app, agent_with_memory, path="/research")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run the server:

python server.py

Your agent is now available at http://localhost:8000/research. LangServe automatically provides:

• A playground UI at /research/playground
• Streaming endpoints for real-time output
• Input/output schema documentation

For production deployment, consider LangGraph Cloud, which adds monitoring, horizontal scaling, and managed infrastructure. But for most use cases, a simple FastAPI server on a VPS works fine.

---

LangGraph vs Alternatives: CrewAI, AutoGen, Semantic Kernel

Choosing an agent framework matters. Here is an honest comparison based on the state of these tools in Q1 2026.

Feature	LangGraph	CrewAI	AutoGen	Semantic Kernel
Architecture	Graph (nodes + edges)	Role-based teams	Conversational agents	Plugin-based
Learning curve	Medium	Low	Medium	High
Control over flow	Full (conditional edges, cycles)	Limited (sequential/hierarchical)	Moderate (conversation patterns)	Moderate
Durable execution	Built-in checkpointing	No	No	No
Human-in-the-loop	First-class support	Basic	Basic	Moderate
Memory	Short-term + long-term	Short-term only	Conversation-based	Plugin-based
Production readiness	High (v1.0, used at Klarna/Uber)	Medium	Low (maintenance mode)	Medium
MCP/A2A support	Community integrations	A2A native	Community integrations	Plugin-based
License	MIT	MIT	MIT	MIT

When to Choose Each

Choose LangGraph when:

• You need fine-grained control over agent behavior
• Your workflow has loops, branches, or complex decision points
• Production reliability matters (retries, persistence, monitoring)
• You want human-in-the-loop approval steps

Choose CrewAI when:

• You are prototyping a multi-agent team quickly
• Your workflow is straightforward (researcher → writer → editor)
• You want the easiest possible getting-started experience
• You need native A2A protocol support

Choose AutoGen when:

• Your use case is primarily multi-party conversations
• You are already in the Microsoft ecosystem
• Note: Microsoft has shifted AutoGen to maintenance mode, so evaluate long-term viability

Choose Semantic Kernel when:

• You are building within the Microsoft/Azure stack
• You need tight integration with Azure AI services
• Your team primarily writes C# or Java

For most Python developers building production agents in 2026, LangGraph is the safest bet. It has the largest community, the most active development, and the strongest production track record.

---

Next Steps: Where to Go From Here

You have built a working research agent. Here are the paths forward:

Multi-Agent Systems

LangGraph supports multiple agents communicating within a single graph. You can build:

• A supervisor agent that delegates to specialist sub-agents
• A debate system where two agents argue for and against a position
• A pipeline where a researcher feeds a writer who feeds an editor

The pattern is the same — nodes and edges — but now some nodes are themselves compiled graphs.

Human-in-the-Loop

Add approval gates for high-stakes actions:

from langgraph.types import interrupt

def send_email(state):
    # Pause and wait for human approval
    approval = interrupt(
        {"question": f"Send report to {state['recipient']}?"}
    )
    if approval.get("approved"):
        # Actually send the email
        ...

Tool Expansion

Give your agent more capabilities:

• File I/O: Read and write local files for persistent research
• Code execution: Run Python code to analyze data
• API calls: Interact with external services
• Browser automation: Navigate and extract from web pages

If you are building AI-powered development tools, our guide on Claude Code advanced workflows covers how production agent systems handle subagents and multi-session orchestration.

Using Local Models

You do not need to pay for API calls during development. Swap ChatOpenAI for a local model:

from langchain_community.llms import Ollama

llm = Ollama(model="llama3.1:8b")

See our Docker Model Runner vs Ollama comparison for choosing the right local inference setup, or grab a completely free AI coding stack for development.

---

Complete Code Reference

Here is the full agent.py in one block for easy copy-paste:

"""Research agent built with LangGraph — complete code."""

import os
from typing import TypedDict, Annotated
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from tavily import TavilyClient

load_dotenv()

# --- State ---

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]
    research_topic: str
    search_queries: list[str]
    sources: list[dict]
    analysis: str
    final_report: str
    iteration: int
    max_iterations: int

# --- Setup ---

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
tavily = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))

# --- Nodes ---

def generate_queries(state: ResearchState) -> dict:
    topic = state["research_topic"]
    existing_info = ""
    if state.get("analysis"):
        existing_info = (
            f"\n\nWe already know:\n{state['analysis']}\n\n"
            "Generate queries to fill gaps in our knowledge."
        )
    response = llm.invoke([
        SystemMessage(content=(
            "You are a research assistant. Generate 3 specific, diverse "
            "search queries to research the given topic. Return only the "
            "queries, one per line. No numbering, no extra text."
            f"{existing_info}"
        )),
        HumanMessage(content=f"Research topic: {topic}"),
    ])
    new_queries = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
    return {
        "search_queries": state.get("search_queries", []) + new_queries,
        "messages": [response],
    }

def search_web(state: ResearchState) -> dict:
    queries = state.get("search_queries", [])
    recent_queries = queries[-3:]
    all_results = state.get("sources", [])
    for query in recent_queries:
        try:
            response = tavily.search(query=query, max_results=3, include_raw_content=False)
            for result in response.get("results", []):
                if not any(s["url"] == result["url"] for s in all_results):
                    all_results.append({
                        "title": result.get("title", ""),
                        "url": result.get("url", ""),
                        "content": result.get("content", ""),
                        "query": query,
                    })
        except Exception as e:
            print(f"Search failed for '{query}': {e}")
    return {"sources": all_results}

def analyze_results(state: ResearchState) -> dict:
    sources = state.get("sources", [])
    if not sources:
        return {
            "analysis": "No search results found. Need to try different queries.",
            "iteration": state.get("iteration", 0) + 1,
        }
    source_text = ""
    for i, source in enumerate(sources, 1):
        source_text += f"\n[{i}] {source['title']}\nURL: {source['url']}\n{source['content']}\n"
    response = llm.invoke([
        SystemMessage(content=(
            "You are a research analyst. Analyze the following search results "
            "about the given topic. Provide:\n"
            "1. Key findings (what we know)\n"
            "2. Gaps (what we still need to find out)\n"
            "3. Confidence level (low/medium/high) in our overall understanding\n\n"
            "Be specific and cite source numbers."
        )),
        HumanMessage(content=f"Topic: {state['research_topic']}\n\nSources:{source_text}"),
    ])
    return {
        "analysis": response.content,
        "iteration": state.get("iteration", 0) + 1,
        "messages": [response],
    }

def write_report(state: ResearchState) -> dict:
    sources = state.get("sources", [])
    analysis = state.get("analysis", "")
    source_text = ""
    for i, source in enumerate(sources, 1):
        source_text += f"\n[{i}] {source['title']}\nURL: {source['url']}\n{source['content']}\n"
    response = llm.invoke([
        SystemMessage(content=(
            "You are a research writer. Write a clear, well-structured research "
            "report based on the analysis and sources provided. Include:\n"
            "- Executive summary (2-3 sentences)\n"
            "- Key findings with citations [1], [2], etc.\n"
            "- Conclusions\n"
            "- Sources list\n\n"
            "Write for a technical audience. Be factual and specific."
        )),
        HumanMessage(content=(
            f"Topic: {state['research_topic']}\n\n"
            f"Analysis:\n{analysis}\n\n"
            f"Sources:{source_text}"
        )),
    ])
    return {
        "final_report": response.content,
        "messages": [response],
    }

# --- Routing ---

def should_continue_research(state: ResearchState) -> str:
    iteration = state.get("iteration", 0)
    max_iterations = state.get("max_iterations", 3)
    analysis = state.get("analysis", "")
    if iteration >= max_iterations:
        return "write_report"
    analysis_lower = analysis.lower()
    if "low" in analysis_lower and "confidence" in analysis_lower:
        return "generate_queries"
    if "significant gaps" in analysis_lower or "need more" in analysis_lower:
        return "generate_queries"
    return "write_report"

# --- Graph ---

workflow = StateGraph(ResearchState)
workflow.add_node("generate_queries", generate_queries)
workflow.add_node("search_web", search_web)
workflow.add_node("analyze_results", analyze_results)
workflow.add_node("write_report", write_report)

workflow.add_edge(START, "generate_queries")
workflow.add_edge("generate_queries", "search_web")
workflow.add_edge("search_web", "analyze_results")
workflow.add_conditional_edges(
    "analyze_results",
    should_continue_research,
    {"generate_queries": "generate_queries", "write_report": "write_report"},
)
workflow.add_edge("write_report", END)

memory = MemorySaver()
agent = workflow.compile(checkpointer=memory)

# --- Main ---

if __name__ == "__main__":
    print("=" * 60)
    print("LangGraph Research Agent")
    print("=" * 60)

    topic = input("\nEnter a research topic: ").strip()
    if not topic:
        topic = "How are companies using AI agents in production in 2026?"

    print(f"\nResearching: {topic}")
    print("-" * 60)

    initial_state = {
        "research_topic": topic,
        "messages": [],
        "search_queries": [],
        "sources": [],
        "analysis": "",
        "final_report": "",
        "iteration": 0,
        "max_iterations": 3,
    }

    config = {"configurable": {"thread_id": "session-001"}}

    for event in agent.stream(initial_state, config=config):
        for node_name, output in event.items():
            print(f"\n>> Node: {node_name}")
            if node_name == "generate_queries" and "search_queries" in output:
                print(f"   Queries: {output['search_queries'][-3:]}")
            elif node_name == "search_web" and "sources" in output:
                print(f"   Found {len(output['sources'])} total sources")
            elif node_name == "analyze_results" and "analysis" in output:
                print(f"   Iteration: {output.get('iteration', '?')}")
                print(f"   Analysis preview: {output['analysis'][:200]}...")
            elif node_name == "write_report" and "final_report" in output:
                print(f"\n{'=' * 60}")
                print("RESEARCH REPORT")
                print("=" * 60)
                print(output["final_report"])

    print(f"\n{'=' * 60}")
    print("Research complete.")

---

Key Takeaways

1. LangGraph models agents as graphs — nodes are functions, edges are transitions, conditional edges are decisions. This makes complex agent behavior natural to express and debug.

2. State is everything. Design your state schema carefully. It is the agent's working memory and determines what the agent can know and track.

3. Always add safety limits. Any agent that loops needs a max_iterations guard. Infinite loops waste money and time.

4. Start simple, add complexity later. Our research agent started with 4 nodes. You can add tool nodes, approval gates, and sub-agents incrementally.

5. LangGraph is production-ready. With v1.0, durable execution, and adoption by major companies, it is no longer experimental. Build with confidence.

The AI agent space is moving fast. But the fundamentals — state management, tool use, decision loops, and safety limits — are stable patterns that will serve you regardless of which framework leads next year. LangGraph gives you the right primitives to build on.

---

Building AI-powered workflows? Check out our guide on Claude Code advanced workflows with subagents for production patterns, or explore free AI coding tools to build your development stack without spending a dime.