LLM Frameworks

LangGraph — Stateful, Multi-Step AI Agents

LangGraph is a framework for building stateful, multi-actor applications with LLMs. It models agent workflows as graphs with nodes (functions) and edges (conditional routing), enabling complex, cyclical agent behaviors.

Graph-Based Workflows — Model complex agent logic as state machines
Persistent State — Resume conversations and workflows across sessions
Human-in-the-Loop — Pause, inspect, and modify agent state at any point
Production Ready — Streaming, checkpointing, and error recovery built in

"LangGraph turns agent development from art to engineering."

LangGraph Stateful Agents

LangGraph extends LangChain by modeling agent workflows as cyclic graphs. Unlike linear chains, LangGraph supports loops, conditional branching, and persistent state — essential for real-world agent applications.

Core Concepts

1. Your First LangGraph Agent

from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# --- Define State ---
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    next_action: str

# --- Define Tools ---
@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"Weather in {city}: 72F, sunny"

tools = [calculator, get_weather]

# --- Create LLM with Tools ---
llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)

# --- Define Nodes ---
def call_model(state: AgentState):
    """Call the LLM with current messages."""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState):
    """Decide: use tool or end."""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

# --- Build Graph ---
graph = StateGraph(AgentState)

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

# Add edges
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")  # After tools, go back to agent

# Compile
app = graph.compile()

# --- Run ---
result = app.invoke({
    "messages": [HumanMessage(content="What is 2 + 2? Also, what's the weather in NYC?")]
})
print(result["messages"][-1].content)

2. State Machine with Conditional Routing

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

class ResearchState(TypedDict):
    topic: str
    sources: list[str]
    summary: str
    quality_score: float
    iteration: int

def gather_sources(state: ResearchState):
    """Gather research sources."""
    # In production: call search APIs
    return {
        "sources": [f"source1_{state['topic']}", f"source2_{state['topic']}"],
        "iteration": state.get("iteration", 0) + 1
    }

def summarize(state: ResearchState):
    """Summarize findings."""
    return {"summary": f"Summary of {state['topic']} from {len(state['sources'])} sources"}

def evaluate_quality(state: ResearchState):
    """Evaluate if summary is good enough."""
    score = min(1.0, len(state["sources"]) / 5)
    return {"quality_score": score}

def route_after_eval(state: ResearchState) -> Literal["gather_sources", "end"]:
    """Route based on quality score."""
    if state["quality_score"] >= 0.8:
        return "end"
    if state["iteration"] >= 3:
        return "end"
    return "gather_sources"

# --- Build the Graph ---
workflow = StateGraph(ResearchState)

workflow.add_node("gather_sources", gather_sources)
workflow.add_node("summarize", summarize)
workflow.add_node("evaluate", evaluate_quality)

workflow.set_entry_point("gather_sources")
workflow.add_edge("gather_sources", "summarize")
workflow.add_edge("summarize", "evaluate")
workflow.add_conditional_edges("evaluate", route_after_eval, {
    "gather_sources": "gather_sources",
    "end": END
})

app = workflow.compile()

# Run with iterative improvement
result = app.invoke({"topic": "LangGraph", "sources": [], "summary": "", "quality_score": 0, "iteration": 0})

3. Human-in-the-Loop Agent

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages

class ApprovalState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    proposal: str
    approved: bool

def generate_proposal(state: ApprovalState):
    """Generate a proposal for human review."""
    # In production: LLM generates proposal
    return {"proposal": "I recommend investing in index funds for long-term growth."}

def human_review(state: ApprovalState):
    """Pause for human approval."""
    # This node will be interrupted for human input
    pass

def execute_if_approved(state: ApprovalState):
    """Execute based on approval."""
    if state.get("approved"):
        return {"messages": [AIMessage(content=f"Executing: {state['proposal']}")]}
    return {"messages": [AIMessage(content="Proposal rejected. Please provide feedback.")]}

# Build with checkpointing for human-in-the-loop
checkpointer = MemorySaver()

workflow = StateGraph(ApprovalState)
workflow.add_node("generate", generate_proposal)
workflow.add_node("review", human_review)
workflow.add_node("execute", execute_if_approved)

workflow.set_entry_point("generate")
workflow.add_edge("generate", "review")
workflow.add_conditional_edges("execute", lambda s: "end" if s.get("approved") else "generate", {"generate": "generate", "end": END})

app = workflow.compile(checkpointer=checkpointer)

# Run and interrupt for human input
config = {"configurable": {"thread_id": "approval-1"}}
result = app.invoke({"messages": [HumanMessage(content="Invest my savings")]}, config)

# Pause for human review...
# Human reviews and approves:
app.update_state(config, {"approved": True})
result = app.invoke(None, config)

Human-in-the-Loop SVG

4. Multi-Agent Research System

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI

class ResearchState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    topic: str
    search_results: list[str]
    analysis: str
    report: str
    phase: str

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def researcher_node(state: ResearchState):
    """Search for information on the topic."""
    # In production: call search API
    results = [
        f"Research finding 1 about {state['topic']}",
        f"Research finding 2 about {state['topic']}",
        f"Research finding 3 about {state['topic']}",
    ]
    return {"search_results": results, "phase": "research"}

def analyst_node(state: ResearchState):
    """Analyze research findings."""
    prompt = f"""Analyze these research findings about {state['topic']}:
    
Findings: {state['search_results']}

Provide a structured analysis."""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"analysis": response.content, "phase": "analysis"}

def writer_node(state: ResearchState):
    """Write final report."""
    prompt = f"""Write a comprehensive report on {state['topic']}.

Analysis: {state['analysis']}

Write a professional report."""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"report": response.content, "phase": "writing"}

def route_after_research(state: ResearchState) -> str:
    """After research, decide next step."""
    if len(state.get("search_results", [])) >= 3:
        return "analyst"
    return "researcher"

# Build the graph
workflow = StateGraph(ResearchState)
workflow.add_node("researcher", researcher_node)
workflow.add_node("analyst", analyst_node)
workflow.add_node("writer", writer_node)

workflow.set_entry_point("researcher")
workflow.add_conditional_edges("researcher", route_after_research, {
    "researcher": "researcher",
    "analyst": "analyst"
})
workflow.add_edge("analyst", "writer")
workflow.add_edge("writer", END)

app = workflow.compile()
result = app.invoke({"messages": [], "topic": "LangGraph", "search_results": [], "analysis": "", "report": "", "phase": ""})

5. ReAct Agent with LangGraph

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver

@tool
def search(query: str) -> str:
    """Search the web for information."""
    return f"Results for '{query}': LangGraph is a framework for building stateful agent applications."

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

# Create a ReAct agent with tools and memory
llm = ChatOpenAI(model="gpt-4o")
memory = MemorySaver()

agent = create_react_agent(
    model=llm,
    tools=[search, calculate],
    checkpointer=memory,
)

# Run with thread for persistent conversation
config = {"configurable": {"thread_id": "user-123"}}

# First turn
result1 = agent.invoke(
    {"messages": [HumanMessage(content="Search for info about LangGraph")]},
    config
)

# Continue conversation (state persists)
result2 = agent.invoke(
    {"messages": [HumanMessage(content="Now calculate 15 * 23")]},
    config
)
# Agent remembers the previous search context

6. Streaming and Real-Time Output

from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage

# Build graph (reuse from above)
app = workflow.compile()

# Stream tokens as they're generated
for event in app.stream(
    {"messages": [HumanMessage(content="Research quantum computing")]},
    stream_mode="updates",  # Stream node updates
):
    for node_name, output in event.items():
        print(f"[{node_name}]", end=" ")
        if "messages" in output:
            for msg in output["messages"]:
                if hasattr(msg, "content"):
                    print(msg.content[:100], end=" ")
        print()

# Stream with full state snapshots
for event in app.stream(
    {"messages": [HumanMessage(content="Research AI safety")]},
    stream_mode="values",  # Full state each step
):
    print(f"Phase: {event.get('phase', 'unknown')}")
    print(f"Messages: {len(event.get('messages', []))}")

7. Deployment with LangGraph Cloud

# langgraph.json configuration
config = {
    "dependencies": ["."],
    "graphs": {
        "research_agent": "./agent.py:app"
    },
    "env": ".env"
}

# Deploy with CLI
# $ langgraph deploy --config langgraph.json

# Use the deployed agent
import requests

response = requests.post(
    "https://api.langchain.com/v1/threads/thread-123/runs",
    json={
        "assistant_id": "research_agent",
        "input": {"messages": [{"role": "user", "content": "Research AI safety"}]},
    },
    headers={"Authorization": "Bearer your-api-key"}
)

Interview Prep

Q: When would you use LangGraph over plain LangChain agents?

Answer:

Complex control flow — Loops, conditionals, parallel execution
State management — Persistent state across turns, checkpointing
Human-in-the-loop — Need to pause for approval/feedback
Multi-agent systems — Multiple agents coordinating
Production reliability — Error recovery, streaming, observability

Q: How does LangGraph handle state persistence?

Answer: LangGraph uses checkpointers to save state after each node execution:

MemorySaver — In-memory (development)
SqliteSaver — SQLite file (testing)
PostgresSaver — PostgreSQL (production)
Each conversation gets a thread_id for state isolation
Supports time travel — replay from any checkpoint

Q: What is the difference between a node and an edge in LangGraph?

Answer:

Node: A function that receives state, performs work, and returns state updates
Edge: A connection between nodes that defines execution flow
Conditional edge: Routes to different nodes based on state
Cyclic edges: Enable loops (essential for agent reasoning)

# Node: function that modifies state
def my_node(state: dict) -> dict:
    return {"field": "new_value"}

# Edge: connects nodes
graph.add_edge("node_a", "node_b")

# Conditional edge: routes based on state
graph.add_conditional_edges("node_a", routing_function, {"path_a": "node_b", "path_b": "node_c"})

Q: How do you test LangGraph agents?

Answer: Use deterministic testing with mocked LLMs:

from langchain_core.messages import AIMessage

def test_agent_research_flow():
    # Mock the LLM
    mock_llm = FakeLLM(responses=["I need to search", "Here's the report"])
    
    # Build graph with mock
    app = build_graph(llm=mock_llm)
    
    # Run and assert
    result = app.invoke({"messages": [HumanMessage(content="test")]})
    assert result["phase"] == "writing"
    assert len(result["report"]) > 0

LangGraph Stateful Agents

LangGraph — Stateful, Multi-Step AI Agents

LangGraph Stateful Agents

Core Concepts

1. Your First LangGraph Agent

2. State Machine with Conditional Routing

3. Human-in-the-Loop Agent

Human-in-the-Loop SVG

4. Multi-Agent Research System

5. ReAct Agent with LangGraph

6. Streaming and Real-Time Output

7. Deployment with LangGraph Cloud

Interview Prep

Q: When would you use LangGraph over plain LangChain agents?

Q: How does LangGraph handle state persistence?

Q: What is the difference between a node and an edge in LangGraph?

Q: How do you test LangGraph agents?

Need Expert LLM Help?