Skip to main content

Documentation Index

Fetch the complete documentation index at: https://resources.devweekends.com/llms.txt

Use this file to discover all available pages before exploring further.

​
What is LangGraph?

LangGraph is a framework for building stateful, multi-step agent workflows. It models agents as graphs where:
  • Nodes = Processing steps (LLM calls, tool use, logic)
  • Edges = Transitions between steps
  • State = Data passed between nodes
Why LangGraph? Simple agent loops break down with complex logic. LangGraph gives you explicit control over agent flow, branching, and state management.

​
Core Concepts

        ┌─────────────┐
        │   START     │
        └──────┬──────┘
               │
        ┌──────▼──────┐
        │    Agent    │◄────────────┐
        └──────┬──────┘             │
               │                    │
        ┌──────▼──────┐             │
        │  Should Use │             │
        │   Tools?    │             │
        └──────┬──────┘             │
               │                    │
      ┌────────┼────────┐           │
      │ Yes    │        │ No        │
      ▼        │        ▼           │
┌─────────┐    │   ┌─────────┐      │
│  Tools  │────┘   │   END   │      │
└────┬────┘        └─────────┘      │
     │                              │
     └──────────────────────────────┘

​
Installation

pip install langgraph langchain-openai

​
Basic Agent Graph

from typing import TypedDict, Annotated, Sequence
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import operator

# Define state schema
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

# Initialize LLM
llm = ChatOpenAI(model="gpt-4o")

# Define tools
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Search the web for information."""
    return f"Results for '{query}': [mock search results]"

@tool
def calculator(expression: str) -> str:
    """Calculate a mathematical expression."""
    return str(eval(expression))

tools = [search, calculator]
llm_with_tools = llm.bind_tools(tools)

# Define nodes
def agent(state: AgentState) -> dict:
    """Main agent node - decides what to do"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    """Determine next step based on last message"""
    last_message = state["messages"][-1]
    
    if last_message.tool_calls:
        return "tools"
    return END

# Build graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("agent", agent)
workflow.add_node("tools", ToolNode(tools))

# Add edges
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        END: END
    }
)
workflow.add_edge("tools", "agent")

# Compile
app = workflow.compile()

# Run
result = app.invoke({
    "messages": [HumanMessage(content="What is 25 * 4 + 10?")]
})

print(result["messages"][-1].content)

​
Multi-Step Workflow

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END

class WorkflowState(TypedDict):
    task: str
    plan: list[str]
    current_step: int
    results: list[str]
    final_output: str

def planner(state: WorkflowState) -> dict:
    """Create a plan for the task"""
    response = llm.invoke(f"""
    Create a step-by-step plan for: {state['task']}
    Return as a numbered list.
    """)
    
    # Parse steps
    lines = response.content.strip().split('\n')
    steps = [line.strip() for line in lines if line.strip()]
    
    return {"plan": steps, "current_step": 0, "results": []}

def executor(state: WorkflowState) -> dict:
    """Execute current step"""
    step = state["plan"][state["current_step"]]
    
    response = llm.invoke(f"""
    Execute this step: {step}
    Previous results: {state['results']}
    """)
    
    return {
        "results": state["results"] + [response.content],
        "current_step": state["current_step"] + 1
    }

def should_continue_execution(state: WorkflowState) -> Literal["executor", "synthesizer"]:
    """Check if more steps remain"""
    if state["current_step"] < len(state["plan"]):
        return "executor"
    return "synthesizer"

def synthesizer(state: WorkflowState) -> dict:
    """Combine results into final output"""
    response = llm.invoke(f"""
    Task: {state['task']}
    Steps completed: {state['plan']}
    Results: {state['results']}
    
    Synthesize these into a final comprehensive answer.
    """)
    
    return {"final_output": response.content}

# Build workflow
workflow = StateGraph(WorkflowState)

workflow.add_node("planner", planner)
workflow.add_node("executor", executor)
workflow.add_node("synthesizer", synthesizer)

workflow.set_entry_point("planner")
workflow.add_edge("planner", "executor")
workflow.add_conditional_edges("executor", should_continue_execution)
workflow.add_edge("synthesizer", END)

app = workflow.compile()

# Run
result = app.invoke({"task": "Research and summarize the latest AI trends"})
print(result["final_output"])

​
Human-in-the-Loop

from langgraph.checkpoint.memory import MemorySaver

class ApprovalState(TypedDict):
    request: str
    analysis: str
    approved: bool | None
    final_response: str

def analyze(state: ApprovalState) -> dict:
    response = llm.invoke(f"Analyze this request: {state['request']}")
    return {"analysis": response.content}

def human_approval(state: ApprovalState) -> dict:
    """Pause here for human approval"""
    # This node just passes through - approval happens externally
    return {}

def execute_approved(state: ApprovalState) -> dict:
    response = llm.invoke(f"""
    Execute this approved request: {state['request']}
    Analysis: {state['analysis']}
    """)
    return {"final_response": response.content}

def check_approval(state: ApprovalState) -> Literal["execute", "reject"]:
    if state.get("approved"):
        return "execute"
    return "reject"

def reject(state: ApprovalState) -> dict:
    return {"final_response": "Request was not approved."}

# Build with checkpointing
workflow = StateGraph(ApprovalState)
workflow.add_node("analyze", analyze)
workflow.add_node("human_approval", human_approval)
workflow.add_node("execute", execute_approved)
workflow.add_node("reject", reject)

workflow.set_entry_point("analyze")
workflow.add_edge("analyze", "human_approval")
workflow.add_conditional_edges("human_approval", check_approval)
workflow.add_edge("execute", END)
workflow.add_edge("reject", END)

# Compile with memory for persistence
memory = MemorySaver()
app = workflow.compile(checkpointer=memory, interrupt_before=["human_approval"])

# Start workflow
config = {"configurable": {"thread_id": "request-123"}}
result = app.invoke({"request": "Delete all user data"}, config)

# At this point, workflow is paused at human_approval
print("Analysis:", result["analysis"])
print("Waiting for approval...")

# Later: resume with approval
app.update_state(config, {"approved": True})
final_result = app.invoke(None, config)
print("Final:", final_result["final_response"])

​
Parallel Execution

from langgraph.graph import StateGraph, END
from typing import TypedDict

class ParallelState(TypedDict):
    query: str
    web_results: str
    db_results: str
    combined: str

def search_web(state: ParallelState) -> dict:
    # Simulate web search
    return {"web_results": f"Web results for: {state['query']}"}

def search_database(state: ParallelState) -> dict:
    # Simulate database search
    return {"db_results": f"Database results for: {state['query']}"}

def combine_results(state: ParallelState) -> dict:
    combined = f"""
    Web: {state['web_results']}
    Database: {state['db_results']}
    """
    return {"combined": combined}

workflow = StateGraph(ParallelState)

workflow.add_node("web_search", search_web)
workflow.add_node("db_search", search_database)
workflow.add_node("combine", combine_results)

# Fan-out: Start both searches in parallel
workflow.set_entry_point("web_search")
workflow.set_entry_point("db_search")  # Both are entry points

# Fan-in: Both must complete before combine
workflow.add_edge("web_search", "combine")
workflow.add_edge("db_search", "combine")
workflow.add_edge("combine", END)

app = workflow.compile()
result = app.invoke({"query": "LangGraph tutorials"})

​
Subgraphs

# Define a reusable subgraph
def create_research_subgraph():
    class ResearchState(TypedDict):
        topic: str
        sources: list[str]
        summary: str
    
    def gather_sources(state):
        return {"sources": [f"Source about {state['topic']}"]}
    
    def summarize(state):
        return {"summary": f"Summary of {len(state['sources'])} sources"}
    
    subgraph = StateGraph(ResearchState)
    subgraph.add_node("gather", gather_sources)
    subgraph.add_node("summarize", summarize)
    subgraph.set_entry_point("gather")
    subgraph.add_edge("gather", "summarize")
    subgraph.add_edge("summarize", END)
    
    return subgraph.compile()

# Use in parent graph
class MainState(TypedDict):
    question: str
    research: str
    answer: str

research_graph = create_research_subgraph()

def do_research(state: MainState) -> dict:
    result = research_graph.invoke({"topic": state["question"]})
    return {"research": result["summary"]}

def generate_answer(state: MainState) -> dict:
    return {"answer": f"Based on {state['research']}: [answer]"}

main_workflow = StateGraph(MainState)
main_workflow.add_node("research", do_research)
main_workflow.add_node("answer", generate_answer)
main_workflow.set_entry_point("research")
main_workflow.add_edge("research", "answer")
main_workflow.add_edge("answer", END)

​
Visualization

# Generate graph visualization
from IPython.display import Image, display

display(Image(app.get_graph().draw_mermaid_png()))

# Or as Mermaid text
print(app.get_graph().draw_mermaid())

​
Common Patterns

Agent Executor

LLM decides actions, tools execute, loop until done

Plan-Execute

Create plan first, then execute each step

Reflection

Execute, evaluate, improve, repeat

Multi-Agent

Multiple specialized agents in a workflow

​
Best Practices

Only store what’s needed between nodes. Large state = slower execution.
Enable persistence for long-running workflows and human-in-the-loop.
Each node should handle its own errors gracefully.
Unit test nodes before assembling the graph.

​
Next Steps

MCP Protocol

Learn the Model Context Protocol for tool integration

​
Interview Deep-Dive

Strong Answer:
  • A simple while-loop agent works well for straightforward tool-calling patterns: loop until the LLM says “done.” But production agent workflows quickly outgrow this. The moment you need conditional branching (if the user wants a refund, go to the refund flow; if they want a product question, go to the knowledge base flow), parallel execution (search the web AND query the database simultaneously), human-in-the-loop approval gates, or stateful multi-step workflows with persistence, the while loop becomes spaghetti code with nested ifs and global state mutations.
  • LangGraph solves this by modeling the workflow as a directed graph where nodes are processing steps and edges are transitions. The graph gives you three things you cannot easily get from a loop. First, explicit control flow: conditional edges make branching logic visible and testable. I can look at the graph definition and immediately understand every possible path the agent can take, which is impossible with deeply nested conditional loops. Second, state management: the TypedDict state schema is passed between nodes, and each node only modifies its slice of the state. This prevents the “one node accidentally overwrites another node’s data” bugs that plague global-state agent loops. Third, persistence via checkpointing: the graph can be paused at any node, serialized to a database, and resumed later. This is essential for human-in-the-loop workflows where the agent pauses for approval and the user might not respond for hours.
  • In practice, I use a simple loop for agents with 1-2 tools and no branching. I reach for LangGraph when the workflow has 3+ distinct phases, needs any form of human approval, requires parallel execution, or needs to survive process restarts.
Red Flags: Candidate cannot articulate when the graph abstraction adds value versus unnecessary complexity, does not mention state management or persistence benefits, or thinks LangGraph is just “LangChain but newer.”Follow-up: How does LangGraph’s checkpointing work, and why is it critical for production agent systems?Checkpointing serializes the entire graph state after every node execution and stores it in a persistent backend (MemorySaver for development, or a database-backed saver for production). Each checkpoint is keyed by a thread ID and includes the full state, the current node position, and the execution history. This enables three production-critical capabilities. First, human-in-the-loop: I use interrupt_before to pause execution before a sensitive node (like approving a payment), persist the state, and resume when the human approves. Without checkpointing, the entire agent context would be lost between the request and the approval. Second, crash recovery: if the server restarts mid-workflow, the agent resumes from the last checkpoint instead of restarting from scratch. For a workflow that has already made 5 API calls and is on step 6, this avoids redundant work and cost. Third, debugging and audit trails: the checkpoint history is a complete execution trace showing every state transition, which is invaluable for debugging why an agent made a particular decision three steps ago.
Strong Answer:
  • I would model this as a graph with four nodes: a planner, a research agent, a writer agent, and a reviewer. The state schema carries the shared context: the original task, the research plan, collected sources, drafted sections, and review feedback.
  • The planner node takes the user’s request and produces a structured research plan: a list of topics to investigate and an outline for the final report. The research agent node iterates through the plan, using tools (web search, database queries, document retrieval) to gather information for each topic. It writes its findings into the state as structured research notes with source citations. The writer node takes the research notes and the outline and drafts each section of the report. The reviewer node evaluates the draft against the original request and the research notes, checking for accuracy, completeness, and coherence.
  • The handoff design is the critical part. I do not pass raw state between agents — I use a structured interface. The research agent writes to a research_notes field in state with a specific schema (topic, findings, sources, confidence). The writer reads from this field, not from the research agent’s internal tool call history. This decoupling means I can swap out the research agent implementation (replace web search with a domain-specific database) without changing the writer at all.
  • For the review loop, I add a conditional edge from reviewer back to either writer or research: if the reviewer finds factual gaps, it routes back to research with specific questions. If it finds writing quality issues, it routes back to writer with feedback. I cap this loop at 2 iterations to prevent infinite revision cycles.
  • I implement each agent as a subgraph so they can be developed and tested independently. The parent graph orchestrates the handoff and manages the shared state.
Red Flags: Candidate describes agents communicating through unstructured text rather than structured state, does not consider the feedback loop, or does not mention how to prevent infinite loops between agents.Follow-up: How do you test individual nodes in a LangGraph workflow without running the entire graph?Each node is just a function that takes a TypedDict state and returns a state update dictionary. I test them as pure functions: construct a mock state, call the node function directly, and assert on the returned state update. For the research agent node, I mock the tool calls (web search returns canned results) and verify that the node writes properly structured research notes to the state. For the writer node, I provide pre-built research notes in the input state and verify the draft output. For conditional edge functions, I test that they return the correct routing string for different state configurations (reviewer says “needs more research” returns “research,” reviewer says “approved” returns END). This unit-level testing catches 80% of bugs before I ever run the full graph. For integration tests, I compile the graph and run it end-to-end with a controlled input, but I mock external API calls (LLM responses, web search) to keep tests deterministic and fast. The full graph integration test verifies that state flows correctly between nodes and that conditional routing works as expected.
Strong Answer:
  • I have seen five main failure modes in production LangGraph deployments.
  • First, infinite loops. The agent gets stuck cycling between two nodes because the conditional edge never resolves to a terminal state. For example, the agent calls a tool, gets an unhelpful result, calls it again with a slightly different query, gets another unhelpful result, and repeats forever. The fix is a hard iteration limit on every cycle in the graph. I set recursion_limit on the compiled graph (default is 25 in LangGraph) and add per-cycle counters in the state that conditional edges check.
  • Second, state bloat. Each tool call result gets appended to the messages list in state, and after 20+ tool calls the state exceeds the LLM’s context window. The fix is implementing a state summarization node that compresses old messages into a summary when the state exceeds a token threshold. I keep the last 5 messages verbatim and summarize everything older.
  • Third, node failures. An external API call in a tool node times out or returns an error. Each node should have its own try-except with graceful degradation: write an error message to state rather than crashing the entire graph. The conditional edge after the node should check for errors and route to a recovery path (retry with different parameters, skip to next step, or exit gracefully with a partial result).
  • Fourth, checkpoint corruption. If the serialized state becomes invalid (schema change between deployments, or a non-serializable object in state), resuming from a checkpoint fails. I version my state schema and include a migration path. When loading a checkpoint, I check the schema version and migrate if needed.
  • Fifth, LLM decision quality degradation as state grows. The more context the LLM processes, the more likely it is to make poor routing or tool selection decisions. I keep the decision-relevant context small and focused by summarizing irrelevant prior steps and only surfacing the information the current node needs.
Red Flags: Candidate does not mention infinite loops as a risk, assumes LangGraph handles all error cases automatically, or does not consider state size management.Follow-up: How do you monitor and observe a LangGraph agent in production?I instrument at three levels. First, graph-level metrics: total execution time, number of nodes visited per run, which terminal node was reached (success, error, timeout), and whether any fallback paths were taken. Second, node-level metrics: latency per node, failure rate per node, and for LLM nodes specifically, token counts and costs. Third, trace-level observability: I integrate LangSmith tracing so every graph execution produces a full trace showing the state at every transition, the LLM prompts and responses at each node, and the routing decisions at every conditional edge. I set up alerts for three conditions: graph execution time exceeding 2x the p95 baseline (suggests an infinite loop or extremely slow external call), a node failure rate exceeding 5% (suggests an upstream API issue), and the graph reaching the recursion limit (suggests the agent cannot converge). The LangSmith traces are the primary debugging tool — when a user reports a bad outcome, I can pull up the exact trace and see every decision the agent made.