Building Autonomous AI Workflows with LangGraph: A Practical Guide

Building AI agents that can reason, plan, and execute complex tasks is challenging. Traditional chains are linear and inflexible. LangGraph changes this by introducing graphs with cycles, state management, and persistent memory — enabling truly autonomous AI workflows.

This guide covers building production-ready AI workflows with LangGraph: state machines, cyclic execution, human-in-the-loop patterns, and memory persistence.

Why LangGraph?

Traditional LangChain chains are linear: input → processing → output. But real-world AI tasks often require:

┌─────────────────────────────────────────────────────────┐
│  Traditional Chain (Linear)                             │
│                                                         │
│  Input → Step 1 → Step 2 → Step 3 → Output             │
│                                                         │
└─────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────┐
│  LangGraph (Cyclic with State)                          │
│                                                         │
│        ┌──────────┐                                     │
│        │  Input   │                                     │
│        └────┬─────┘                                     │
│             │                                           │
│             ▼                                           │
│        ┌──────────┐                                     │
│   ┌───►│  Agent   │◄───┐                               │
│   │    └────┬─────┘    │                               │
│   │         │          │                               │
│   │    ┌────▼─────┐    │                               │
│   │    │  Tools   │────┘                               │
│   │    └──────────┘                                    │
│   │         │                                          │
│   │    ┌────▼─────┐                                    │
│   └────│  Retry   │                                    │
│        └────┬─────┘                                    │
│             │                                          │
│             ▼                                          │
│        ┌──────────┐                                    │
│        │  Output  │                                    │
│        └──────────┘                                    │
│                                                         │
└─────────────────────────────────────────────────────────┘

LangGraph provides:

• Cycles: Agents can loop, retry, and iterate
• State: Shared state across all nodes
• Persistence: Save and resume workflows
• Human-in-the-loop: Pause for human approval

Core Concepts

State

State is the shared memory that flows through your graph:

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

class AgentState(TypedDict):
    messages: list[dict]
    current_task: str
    attempts: int
    max_attempts: int
    context: dict
    errors: list[str]
    result: str | None

Nodes

Nodes are functions that process state:

def agent_node(state: AgentState) -> AgentState:
    """The reasoning agent that decides what to do."""
    messages = state["messages"]
    
    response = llm.invoke(messages)
    
    return {
        **state,
        "messages": messages + [response],
    }

def tool_node(state: AgentState) -> AgentState:
    """Executes tool calls from the agent."""
    last_message = state["messages"][-1]
    
    if not last_message.get("tool_calls"):
        return state
    
    results = []
    for tool_call in last_message["tool_calls"]:
        result = execute_tool(tool_call["name"], tool_call["arguments"])
        results.append(result)
    
    return {
        **state,
        "messages": state["messages"] + [
            {"role": "tool", "content": str(results)}
        ],
    }

def error_handler(state: AgentState) -> AgentState:
    """Handle errors and decide whether to retry."""
    return {
        **state,
        "attempts": state["attempts"] + 1,
        "errors": state["errors"] + [state.get("last_error", "Unknown error")],
    }

Edges

Edges define the flow between nodes:

def should_continue(state: AgentState) -> str:
    """Determine the next node based on state."""
    last_message = state["messages"][-1]
    
    if state["attempts"] >= state["max_attempts"]:
        return "end"
    
    if last_message.get("tool_calls"):
        return "tools"
    
    if state.get("last_error"):
        return "error_handler"
    
    return "end"

Building a Research Agent

Let’s build a practical research agent that can search, analyze, and synthesize information:

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

llm = ChatOpenAI(model="gpt-4-turbo")

class ResearchState(TypedDict):
    query: str
    search_results: list[dict]
    analysis: str
    sources: list[str]
    iterations: int
    max_iterations: int
    final_report: str | None

@tool
def web_search(query: str) -> list[dict]:
    """Search the web for information."""
    # Implement with your search API
    return [{"title": "Result 1", "url": "...", "snippet": "..."}]

@tool
def extract_content(url: str) -> str:
    """Extract content from a URL."""
    # Implement content extraction
    return "Extracted content..."

def researcher_node(state: ResearchState) -> ResearchState:
    """Agent that searches and gathers information."""
    query = state["query"]
    
    response = llm.invoke([
        {"role": "system", "content": "You are a research assistant. Use tools to search for information."},
        {"role": "user", "content": f"Research: {query}"},
    ])
    
    tool_calls = response.tool_calls if hasattr(response, "tool_calls") else []
    
    if tool_calls:
        results = []
        for tc in tool_calls:
            if tc["name"] == "web_search":
                results.extend(web_search.invoke(tc["args"]))
        
        return {
            **state,
            search_results: state.get("search_results", []) + results,
            iterations: state["iterations"] + 1,
        }
    
    return state

def analyst_node(state: ResearchState) -> ResearchState:
    """Analyzes gathered information."""
    if not state.get("search_results"):
        return state
    
    results_text = "\n".join([
        f"- {r['title']}: {r['snippet']}"
        for r in state["search_results"]
    ])
    
    analysis = llm.invoke([
        {"role": "system", "content": "You are an analyst. Synthesize information."},
        {"role": "user", "content": f"Analyze these results:\n{results_text}"},
    ]).content
    
    return {
        **state,
        "analysis": analysis,
    }

def writer_node(state: ResearchState) -> ResearchState:
    """Writes the final report."""
    report = llm.invoke([
        {"role": "system", "content": "You are a technical writer."},
        {"role": "user", "content": f"Write a report on: {state['query']}\n\nAnalysis:\n{state.get('analysis', '')}"},
    ]).content
    
    return {
        **state,
        "final_report": report,
    }

def should_continue_research(state: ResearchState) -> str:
    """Decide if more research is needed."""
    if state.get("final_report"):
        return "end"
    
    if state["iterations"] >= state["max_iterations"]:
        return "write"
    
    if len(state.get("search_results", [])) >= 5:
        return "analyze"
    
    return "research"

# Build the graph
graph = StateGraph(ResearchState)

graph.add_node("researcher", researcher_node)
graph.add_node("analyst", analyst_node)
graph.add_node("writer", writer_node)

graph.set_entry_point("researcher")

graph.add_conditional_edges(
    "researcher",
    should_continue_research,
    {
        "research": "researcher",
        "analyze": "analyst",
        "write": "writer",
        "end": END,
    }
)

graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)

app = graph.compile()

Human-in-the-Loop Pattern

For critical decisions, pause the workflow for human approval:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import interrupt

class ApprovalState(TypedDict):
    action: str
    proposal: str
    approved: bool | None
    human_feedback: str | None
    result: str | None

def propose_action(state: ApprovalState) -> ApprovalState:
    """Agent proposes an action."""
    proposal = llm.invoke([
        {"role": "user", "content": f"What action should we take for: {state['action']}?"},
    ]).content
    
    return {**state, "proposal": proposal}

def human_approval(state: ApprovalState) -> ApprovalState:
    """Pause for human approval."""
    approved = interrupt({
        "question": f"Do you approve this action?\n\n{state['proposal']}",
        "options": ["approve", "reject", "modify"],
    })
    
    return {
        **state,
        "approved": approved == "approve",
        "human_feedback": approved if approved != "approve" else None,
    }

def execute_action(state: ApprovalState) -> ApprovalState:
    """Execute the approved action."""
    if not state["approved"]:
        return {**state, "result": "Action rejected"}
    
    # Execute the action
    result = f"Executed: {state['proposal']}"
    return {**state, "result": result}

# Build with checkpointing for persistence
checkpointer = MemorySaver()

graph = StateGraph(ApprovalState)
graph.add_node("propose", propose_action)
graph.add_node("approve", human_approval)
graph.add_node("execute", execute_action)

graph.set_entry_point("propose")
graph.add_edge("propose", "approve")
graph.add_edge("approve", "execute")
graph.add_edge("execute", END)

app = graph.compile(checkpointer=checkpointer)

# Usage
config = {"configurable": {"thread_id": "user-123"}}

# First run - pauses at approval
result = app.invoke({"action": "delete database"}, config)

# Human approves
app.update_state(config, {"approved": True})

# Continue execution
final = app.invoke(None, config)

Persistent Memory

Save workflow state to resume later:

from langgraph.checkpoint.sqlite import SqliteSaver

# Use SQLite for persistence
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)
    
    # Start a workflow
    result = app.invoke(
        {"query": "Research AI agents", "iterations": 0, "max_iterations": 3},
        {"configurable": {"thread_id": "session-456"}},
    )
    
    # Later, resume from where it left off
    state = app.get_state({"configurable": {"thread_id": "session-456"}})
    print(state.values)  # Current state
    print(state.next)    # Next nodes to execute
    
    # Continue
    result = app.invoke(None, {"configurable": {"thread_id": "session-456"}})

Parallel Execution

Execute multiple nodes concurrently:

from langgraph.graph import StateGraph

class ParallelState(TypedDict):
    query: str
    web_results: list[dict] | None
    db_results: list[dict] | None
    api_results: list[dict] | None
    combined: str | None

def search_web(state: ParallelState) -> ParallelState:
    return {**state, "web_results": [...]}

def search_database(state: ParallelState) -> ParallelState:
    return {**state, "db_results": [...]}

def call_api(state: ParallelState) -> ParallelState:
    return {**state, "api_results": [...]}

def combine_results(state: ParallelState) -> ParallelState:
    all_results = []
    if state.get("web_results"):
        all_results.extend(state["web_results"])
    if state.get("db_results"):
        all_results.extend(state["db_results"])
    if state.get("api_results"):
        all_results.extend(state["api_results"])
    
    combined = llm.invoke([
        {"role": "user", "content": f"Combine: {all_results}"},
    ]).content
    
    return {**state, "combined": combined}

graph = StateGraph(ParallelState)

graph.add_node("web", search_web)
graph.add_node("database", search_database)
graph.add_node("api", call_api)
graph.add_node("combine", combine_results)

# Parallel branches from start
graph.set_entry_point("web")
graph.set_entry_point("database")
graph.set_entry_point("api")

# All converge to combine
graph.add_edge("web", "combine")
graph.add_edge("database", "combine")
graph.add_edge("api", "combine")
graph.add_edge("combine", END)

Error Handling and Retry

Robust error handling with exponential backoff:

import asyncio
from typing import Literal

class ResilientState(TypedDict):
    task: str
    attempts: int
    max_attempts: int
    last_error: str | None
    result: str | None
    status: Literal["pending", "success", "failed"]

async def resilient_node(state: ResilientState) -> ResilientState:
    """Node with retry logic."""
    if state["attempts"] >= state["max_attempts"]:
        return {**state, "status": "failed"}
    
    try:
        # Exponential backoff
        await asyncio.sleep(2 ** state["attempts"])
        
        result = await execute_with_timeout(state["task"], timeout=30)
        
        return {
            **state,
            "result": result,
            "status": "success",
        }
    except Exception as e:
        return {
            **state,
            "attempts": state["attempts"] + 1,
            "last_error": str(e),
        }

def should_retry(state: ResilientState) -> str:
    """Decide whether to retry or fail."""
    if state["status"] == "success":
        return "end"
    
    if state["attempts"] >= state["max_attempts"]:
        return "fail"
    
    return "retry"

graph = StateGraph(ResilientState)
graph.add_node("execute", resilient_node)
graph.add_node("failure_handler", lambda s: {**s, "result": "Fallback result"})

graph.set_entry_point("execute")

graph.add_conditional_edges(
    "execute",
    should_retry,
    {
        "retry": "execute",
        "fail": "failure_handler",
        "end": END,
    }
)
graph.add_edge("failure_handler", END)

Streaming Responses

Stream intermediate results to users:

async def stream_workflow(app, initial_state, config):
    """Stream workflow execution."""
    async for event in app.astream_events(initial_state, config):
        if event["event"] == "on_chain_start":
            print(f"Starting: {event['name']}")
        
        elif event["event"] == "on_chain_end":
            print(f"Completed: {event['name']}")
            yield f"data: {event['data']}\n\n"
        
        elif event["event"] == "on_tool_start":
            print(f"Tool call: {event['name']}")
        
        elif event["event"] == "on_tool_end":
            print(f"Tool result: {event['data']['output']}")

# Usage with FastAPI
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app_api = FastAPI()

@app_api.get("/research/stream")
async def research_stream(query: str):
    async def generate():
        async for chunk in stream_workflow(
            app,
            {"query": query, "iterations": 0, "max_iterations": 3},
            {"configurable": {"thread_id": "stream-session"}},
        ):
            yield chunk
    
    return StreamingResponse(generate(), media_type="text/event-stream")

Production Considerations

Monitoring

from langsmith import Client

client = Client()

# Enable tracing
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "my-langgraph-app"

# Custom callback for metrics
from langchain.callbacks.base import BaseCallbackHandler

class MetricsCallback(BaseCallbackHandler):
    def on_chain_start(self, serialized, inputs, **kwargs):
        self.start_time = time.time()
    
    def on_chain_end(self, outputs, **kwargs):
        duration = time.time() - self.start_time
        metrics.track("workflow_duration", duration)
        metrics.track("workflow_success", 1)

Rate Limiting

from datetime import datetime, timedelta
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_calls: int, period_seconds: int):
        self.max_calls = max_calls
        self.period = timedelta(seconds=period_seconds)
        self.calls = defaultdict(list)
    
    def allow(self, user_id: str) -> bool:
        now = datetime.now()
        calls = self.calls[user_id]
        
        # Remove old calls
        self.calls[user_id] = [
            t for t in calls 
            if now - t < self.period
        ]
        
        if len(self.calls[user_id]) >= self.max_calls:
            return False
        
        self.calls[user_id].append(now)
        return True

limiter = RateLimiter(max_calls=10, period_seconds=60)

def rate_limited_node(state):
    user_id = state.get("user_id", "anonymous")
    
    if not limiter.allow(user_id):
        return {**state, "error": "Rate limit exceeded"}
    
    return process_node(state)

Cost Tracking

import tiktoken

def estimate_tokens(text: str) -> int:
    encoder = tiktoken.encoding_for_model("gpt-4")
    return len(encoder.encode(text))

class CostTracker:
    def __init__(self):
        self.total_tokens = 0
        self.total_cost = 0.0
    
    # GPT-4 Turbo pricing (as of 2026)
    COST_PER_1K_TOKENS = {
        "input": 0.01,
        "output": 0.03,
    }
    
    def track(self, input_tokens: int, output_tokens: int):
        cost = (
            (input_tokens / 1000) * self.COST_PER_1K_TOKENS["input"] +
            (output_tokens / 1000) * self.COST_PER_1K_TOKENS["output"]
        )
        self.total_cost += cost
        self.total_tokens += input_tokens + output_tokens
        return cost

Real-World Example: Customer Support Agent

class SupportState(TypedDict):
    customer_id: str
    message: str
    history: list[dict]
    intent: str | None
    resolution: str | None
    escalation_needed: bool
    human_agent_id: str | None

def classify_intent(state: SupportState) -> SupportState:
    """Classify customer intent."""
    response = llm.invoke([
        {"role": "system", "content": "Classify intent: billing, technical, general, complaint"},
        {"role": "user", "content": state["message"]},
    ])
    
    return {**state, "intent": response.content.lower()}

def handle_billing(state: SupportState) -> SupportState:
    """Handle billing inquiries."""
    # Look up customer, process refund, etc.
    return {**state, "resolution": "Billing issue resolved"}

def handle_technical(state: SupportState) -> SupportState:
    """Handle technical issues."""
    # Run diagnostics, provide solutions
    return {**state, "resolution": "Technical issue resolved"}

def handle_complaint(state: SupportState) -> SupportState:
    """Handle complaints - may escalate."""
    severity = llm.invoke([
        {"role": "user", "content": f"Rate severity 1-5: {state['message']}"},
    ]).content
    
    if int(severity) >= 4:
        return {**state, "escalation_needed": True}
    
    return {**state, "resolution": "Complaint addressed"}

def escalate_to_human(state: SupportState) -> SupportState:
    """Escalate to human agent."""
    return {**state, "human_agent_id": "agent-123"}

def route_intent(state: SupportState) -> str:
    """Route based on intent."""
    if state.get("escalation_needed"):
        return "escalate"
    
    intent = state.get("intent", "general")
    
    routing = {
        "billing": "billing_handler",
        "technical": "technical_handler",
        "complaint": "complaint_handler",
    }
    
    return routing.get(intent, "general_handler")

graph = StateGraph(SupportState)

graph.add_node("classify", classify_intent)
graph.add_node("billing_handler", handle_billing)
graph.add_node("technical_handler", handle_technical)
graph.add_node("complaint_handler", handle_complaint)
graph.add_node("escalate", escalate_to_human)

graph.set_entry_point("classify")

graph.add_conditional_edges("classify", route_intent, {
    "billing_handler": "billing_handler",
    "technical_handler": "technical_handler",
    "complaint_handler": "complaint_handler",
    "escalate": "escalate",
    "general_handler": END,
})

for handler in ["billing_handler", "technical_handler", "complaint_handler"]:
    graph.add_edge(handler, END)

graph.add_edge("escalate", END)

support_app = graph.compile()

For more on building AI-powered systems, check out my guides on AI agent architecture patterns and production-ready LLM integration.

Conclusion

LangGraph enables building AI workflows that are cyclic, stateful, and persistent — essential for autonomous agents. The graph abstraction makes complex multi-step workflows manageable, while built-in support for human-in-the-loop and checkpointing makes them production-ready.

Start with simple linear flows, then add cycles and branching as needed. Monitor token usage and costs, implement proper error handling, and always have a fallback for when things go wrong.

The future of AI is not just chatbots — it’s autonomous systems that can reason, plan, and execute complex tasks. LangGraph provides the foundation for building that future.

Build autonomous, stay in control! 🚀