How to Build AI Agents with LangGraph and LangChain in 2026

Table of Contents

• How to Build AI Agents with LangGraph and LangChain in 2026
  • Understanding the Agent Architecture and State Machine
    • Why Graph-Based Agents Matter in Production
    • Real-World Use Case: Multi-Source Research Agent
  • Prerequisites and Environment Setup
• Create a virtual environment
• Install core dependencies
  • Building the Core Agent with State Management
    • Step 1: Define the State Schema
    • Step 2: Create the Graph Nodes

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

---

The landscape of AI agent development has evolved significantly. While the recent advancements in agent technology are interesting, they don't represent a paradigm shift—rather, they're incremental improvements in reliability, memory management, and tool orchestration. What matters for production engineers is building agents that actually work at scale, handle edge cases gracefully, and maintain state across complex workflows.

In this tutorial, we'll build a production-ready AI agent using LangGraph (v0.3.x) and LangChain [7] (v0.3.x) that can execute multi-step research tasks, manage conversation memory, and gracefully recover from failures. This isn't a toy demo—we'll cover state management, error boundaries, rate limiting, and observability.

Understanding the Agent Architecture and State Machine

Before writing code, let's understand why LangGraph represents a meaningful improvement over traditional agent frameworks. According to the LangChain documentation, LangGraph models agent workflows as directed graphs where nodes represent computation steps and edges represent state transitions. This is fundamentally different from the linear chain-of-thought approach used in earlier frameworks.

Why Graph-Based Agents Matter in Production

Traditional agent frameworks (like the original LangChain AgentExecutor) use a loop: observe → think → act → observe. This works for simple tasks but fails when you need:

• Parallel execution of tools
• Conditional branching based on intermediate results
• Human-in-the-loop approval gates
• Persistent state across multiple conversation turns

LangGraph solves these by treating agent execution as a state machine. Each node in the graph can access and modify a shared state object, enabling complex workflows without the spaghetti code of nested callbacks.

Real-World Use Case: Multi-Source Research Agent

We'll build an agent that:

1. Takes a research question
2. Searches multiple sources (web, database, API)
3. Synthesizes findings
4. Generates a structured report
5. Handles rate limits and API failures gracefully

This mirrors production use cases at companies like Anthropic [9] and LangChain, where agents must coordinate multiple tools while maintaining reliability.

Prerequisites and Environment Setup

Let's set up our environment with the exact package versions that work together. As of May 2026, these are the stable releases:

# Create a virtual environment
python -m venv agent_env
source agent_env/bin/activate  # On Windows: agent_env\Scripts\activate

# Install core dependencies
pip install langchain==0.3.14 langgraph==0.3.2 langchain-openai [8]==0.3.5
pip install langchain-community==0.3.16 langchain-core==0.3.33
pip install tavily-python==0.5.1  # For web search
pip install pydantic==2.10.4     # For state management
pip install httpx==0.28.1        # For async HTTP
pip install python-dotenv==1.0.1 # For environment variables

Important: LangGraph 0.3.x introduced breaking changes from 0.2.x. The StateGraph class now requires explicit state schema definition using Pydantic models. If you're migrating from an older version, you'll need to refactor your state management.

Create a .env file with your API keys:

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

Building the Core Agent with State Management

Now we'll implement the agent step by step. This is where the rubber meets the road—we're writing production code that handles real-world constraints.

Step 1: Define the State Schema

The state is the backbone of our agent. Every node in the graph reads from and writes to this state. We use Pydantic for validation and serialization:

from typing import TypedDict, Annotated, List, Optional, Literal
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    """State schema for our research agent.

    Using TypedDict for type safety while maintaining LangGraph compatibility.
    The 'messages' key uses special annotation for message accumulation.
    """
    messages: Annotated[List[BaseMessage], add_messages]
    research_question: str
    search_results: Annotated[List[dict], operator.add]
    current_step: str
    errors: Annotated[List[str], operator.add]
    final_report: Optional[str]
    retry_count: int

Key design decisions:

• add_messages reducer: LangGraph uses reducers to handle state updates. add_messages appends new messages to the list, preserving conversation history.
• operator.add for lists: This allows multiple nodes to contribute to search_results and errors without overwriting.
• Optional for final_report: The report starts as None and gets populated at the end.

Step 2: Create the Graph Nodes

Each node is a function that takes the state and returns a partial state update. Let's build the research workflow:

from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langgraph.graph import StateGraph, END

# Initialize our LLM and tools
llm = ChatOpenAI(
    model="gpt [6]-4o",  # Latest stable model as of May 2026
    temperature=0.7,
    max_tokens=4096,
)

web_search = TavilySearchResults(
    max_results=5,
    api_key="tvly-your-key-here",  # In production, use environment variables
)

def analyze_question(state: AgentState) -> dict:
    """Node 1: Analyze the research question and plan the search strategy."""
    messages = [
        SystemMessage(content="""You are a research analyst. Analyze the given question and:
1. Identify key concepts to search for
2. Determine the type of information needed (facts, statistics, opinions)
3. Suggest search queries (max 3)
Return your analysis as structured text."""),
        HumanMessage(content=state["research_question"])
    ]

    response = llm.invoke(messages)

    return {
        "messages": [response],
        "current_step": "searching",
        "search_results": []  # Reset search results for new analysis
    }

def execute_searches(state: AgentState) -> dict:
    """Node 2: Execute web searches based on the analysis.

    Handles rate limiting and partial failures gracefully.
    """
    # Extract search queries from the analysis
    analysis = state["messages"][-1].content

    # In production, you'd parse this more carefully
    # For now, we'll use the original question as a fallback
    search_queries = [state["research_question"]]

    results = []
    errors = []

    for query in search_queries:
        try:
            # Tavily has rate limits - 1000 requests/month on free tier
            # Source: https://docs.tavily.com/docs/rate-limits
            search_result = web_search.invoke({"query": query})
            results.extend(search_result)
        except Exception as e:
            errors.append(f"Search failed for query '{query}': {str(e)}")
            # Don't fail the entire workflow - continue with other queries

    return {
        "search_results": results,
        "errors": errors,
        "current_step": "synthesizing"
    }

def synthesize_findings(state: AgentState) -> dict:
    """Node 3: Synthesize search results into a coherent report."""
    # Prepare context from search results
    context = "\n\n".join([
        f"Source {i+1}: {result['content'][:500]}"
        for i, result in enumerate(state["search_results"][:10])  # Limit to 10 sources
    ])

    messages = [
        SystemMessage(content=f"""You are a research synthesizer. Based on the following search results, create a comprehensive report that:
1. Addresses the original question: {state['research_question']}
2. Synthesizes information from multiple sources
3. Notes any contradictions or uncertainties
4. Provides a balanced conclusion

Search Results:
{context}"""),
        HumanMessage(content="Please synthesize these findings into a structured report.")
    ]

    response = llm.invoke(messages)

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

def should_retry(state: AgentState) -> Literal["analyze_question", "execute_searches", "__end__"]:
    """Router function: Decide next step based on state.

    This implements error recovery logic - if we had errors, retry up to 2 times.
    """
    if state["errors"] and state["retry_count"] < 2:
        return "analyze_question"  # Retry from the beginning
    elif state["final_report"]:
        return "__end__"  # We're done
    else:
        return "execute_searches"  # Continue with more searches

Step 3: Compose the Graph

Now we wire everything together into a state machine:

def build_research_agent() -> StateGraph:
    """Build the complete research agent graph."""

    # Initialize the graph with our state schema
    workflow = StateGraph(AgentState)

    # Add nodes
    workflow.add_node("analyze_question", analyze_question)
    workflow.add_node("execute_searches", execute_searches)
    workflow.add_node("synthesize_findings", synthesize_findings)

    # Set the entry point
    workflow.set_entry_point("analyze_question")

    # Add edges with conditional routing
    workflow.add_conditional_edges(
        "analyze_question",
        lambda state: "execute_searches",  # Always go to search after analysis
        {"execute_searches": "execute_searches"}
    )

    workflow.add_conditional_edges(
        "execute_searches",
        lambda state: "synthesize_findings",  # Always synthesize after search
        {"synthesize_findings": "synthesize_findings"}
    )

    workflow.add_conditional_edges(
        "synthesize_findings",
        should_retry,  # Use our router function
        {
            "analyze_question": "analyze_question",
            "execute_searches": "execute_searches",
            "__end__": END
        }
    )

    # Compile the graph
    return workflow.compile()

# Create the agent
research_agent = build_research_agent()

Step 4: Add Error Handling and Observability

Production agents need visibility into their execution. Let's add logging and error boundaries:

import logging
from datetime import datetime
from typing import Dict, Any

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class AgentExecutor:
    """Wrapper around our graph with observability and error handling."""

    def __init__(self, graph: StateGraph):
        self.graph = graph
        self.execution_history = []

    async def run(self, question: str, max_retries: int = 3) -> Dict[str, Any]:
        """Execute the agent with monitoring."""

        initial_state = {
            "messages": [],
            "research_question": question,
            "search_results": [],
            "current_step": "starting",
            "errors": [],
            "final_report": None,
            "retry_count": 0
        }

        start_time = datetime.now()
        logger.info(f"Starting research for: {question[:50]}..")

        try:
            # Execute the graph
            final_state = await self.graph.ainvoke(initial_state)

            execution_time = (datetime.now() - start_time).total_seconds()

            result = {
                "success": True,
                "report": final_state.get("final_report"),
                "execution_time_seconds": execution_time,
                "steps_executed": final_state.get("current_step"),
                "errors_encountered": final_state.get("errors", []),
                "sources_used": len(final_state.get("search_results", []))
            }

            logger.info(f"Research completed in {execution_time:.2f}s")

            # Store for observability
            self.execution_history.append({
                "timestamp": start_time.isoformat(),
                "question": question,
                "result": result
            })

            return result

        except Exception as e:
            logger.error(f"Agent execution failed: {str(e)}")
            return {
                "success": False,
                "error": str(e),
                "execution_time_seconds": (datetime.now() - start_time).total_seconds()
            }

# Usage
async def main():
    executor = AgentExecutor(research_agent)

    result = await executor.run(
        "What are the latest developments in AI agent frameworks as of 2026?"
    )

    if result["success"]:
        print(f"Report generated in {result['execution_time_seconds']:.2f}s")
        print(f"Sources used: {result['sources_used']}")
        print(f"Report preview: {result['report'][:200]}..")
    else:
        print(f"Failed: {result['error']}")

# Run with: asyncio.run(main())

Edge Cases and Production Considerations

Memory Management

LangGraph state can grow large with conversation history. According to the LangGraph documentation, state is stored in memory by default. For production:

# Use checkpointing for long-running agents
from langgraph.checkpoint.sqlite import SqliteSaver

# Persist state to SQLite
memory = SqliteSaver.from_conn_string("checkpoints.db")
workflow = workflow.compile(checkpointer=memory)

Rate Limiting and API Costs

The Tavily search API has rate limits: 1000 requests/month on the free tier, 5000 on the growth tier ($49/month) Source: Tavily Pricing. Implement a token bucket:

import time
from collections import deque

class RateLimiter:
    def __init__(self, max_calls: int, period: float):
        self.max_calls = max_calls
        self.period = period
        self.calls = deque()

    def acquire(self):
        now = time.time()
        # Remove old calls
        while self.calls and self.calls[0] < now - self.period:
            self.calls.popleft()

        if len(self.calls) >= self.max_calls:
            sleep_time = self.calls[0] + self.period - now
            if sleep_time > 0:
                time.sleep(sleep_time)

        self.calls.append(now)

Handling LLM Hallucinations

Our synthesis node can produce inaccurate information. Add a verification step:

def verify_facts(state: AgentState) -> dict:
    """Node to verify claims against original sources."""
    report = state["final_report"]

    verification_prompt = f"""Review this report for factual accuracy:

Report: {report}

Original sources: {state['search_results'][:3]}

Identify any claims that are not supported by the sources. List them as:
- UNSUPPORTED: [claim]
- SUPPORTED: [claim]"""

    verification = llm.invoke([
        SystemMessage(content="You are a fact-checker."),
        HumanMessage(content=verification_prompt)
    ])

    return {
        "messages": [verification],
        "verification_result": verification.content
    }

Conclusion

We've built a production-ready AI agent using LangGraph and LangChain that handles multi-step research, error recovery, and state persistence. The key takeaways:

1. Graph-based agents provide better control flow than linear chains
2. State management with Pydantic ensures type safety and serialization
3. Error boundaries prevent single failures from crashing the entire workflow
4. Observability is critical for debugging and monitoring

This architecture is used in production at companies like Elastic and Replit for complex agent workflows. The code we've written handles real-world constraints like rate limits, partial failures, and memory management.

What's Next

To extend this agent for production use:

1. Add human-in-the-loop approval using LangGraph's interrupt functionality
2. Implement streaming responses for real-time user feedback
3. Add tool validation with Pydantic schemas for each tool
4. Set up monitoring with OpenTelemetry for distributed tracing

The complete code is available on GitHub. For more on agent architectures, check out our guide on building reliable AI systems.

---