LangGraph Tutorial — The 3 AM State Corruption That Took Down Our Agent Pipeline

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

# Define state schema
class AgentState(TypedDict):
    messages: list[str]
    next_step: str

def node_a(state: AgentState) -> dict:
    # Always return a new dict, never mutate state in-place
    return {"messages": state["messages"] + ["Node A processed"]}

def node_b(state: AgentState) -> dict:
    return {"messages": state["messages"] + ["Node B processed"]}

def route_after_a(state: AgentState) -> Literal["node_b", END]:
    # Conditional routing based on state
    if len(state["messages"]) < 3:
        return "node_b"
    return END

# Build graph
builder = StateGraph(AgentState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.set_entry_point("node_a")
builder.add_conditional_edges("node_a", route_after_a)
builder.add_edge("node_b", END)

# Compile with checkpointer for state persistence
graph = builder.compile()

# Safe invocation with deep copy to prevent mutation
initial_state = {"messages": ["Hello"], "next_step": "start"}
result = graph.invoke(copy.deepcopy(initial_state))
print(result["messages"])  # ['Hello', 'Node A processed', 'Node B processed']

from typing import TypedDict, Literal, Optional
from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver
from langchain_openai import ChatOpenAI
import os

# State schema
class SupportState(TypedDict):
    email: str
    intent: Optional[str]
    account_id: Optional[str]
    draft: Optional[str]
    escalated: bool

def classify_intent(state: SupportState) -> dict:
    llm = ChatOpenAI(model="gpt-4", api_key=os.getenv("OPENAI_API_KEY"))
    response = llm.invoke(f"Classify this email: {state['email']}. Reply with 'billing', 'technical', or 'other'.")
    intent = response.content.strip().lower()
    return {"intent": intent}

def lookup_account(state: SupportState) -> dict:
    # Simulate account lookup
    account_id = "acc_12345"  # In production, query your DB
    return {"account_id": account_id}

def draft_reply(state: SupportState) -> dict:
    llm = ChatOpenAI(model="gpt-4", api_key=os.getenv("OPENAI_API_KEY"))
    prompt = f"Draft a reply for account {state['account_id']} regarding {state['intent']} issue: {state['email']}"
    response = llm.invoke(prompt)
    return {"draft": response.content}

def escalate(state: SupportState) -> dict:
    return {"escalated": True, "draft": "Escalated to human agent."}

def route_after_classify(state: SupportState) -> Literal["lookup_account", "escalate"]:
    if state["intent"] == "other":
        return "escalate"
    return "lookup_account"

# Build graph
builder = StateGraph(SupportState)
builder.add_node("classify_intent", classify_intent)
builder.add_node("lookup_account", lookup_account)
builder.add_node("draft_reply", draft_reply)
builder.add_node("escalate", escalate)
builder.set_entry_point("classify_intent")
builder.add_conditional_edges("classify_intent", route_after_classify)
builder.add_edge("lookup_account", "draft_reply")
builder.add_edge("draft_reply", END)
builder.add_edge("escalate", END)

# Compile with checkpointer and interrupt for escalation
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer, interrupt_after=["escalate"])

# Invoke
result = graph.invoke({"email": "My bill is wrong", "intent": None, "account_id": None, "draft": None, "escalated": False})
print(result["draft"])

# BAD: Using LangGraph for a simple linear pipeline
from langgraph.graph import StateGraph, END
from typing import TypedDict

class SimpleState(TypedDict):
    text: str

def uppercase(state: SimpleState) -> dict:
    return {"text": state["text"].upper()}

def add_exclamation(state: SimpleState) -> dict:
    return {"text": state["text"] + "!"}

# This is overkill for two steps
builder = StateGraph(SimpleState)
builder.add_node("uppercase", uppercase)
builder.add_node("add_exclamation", add_exclamation)
builder.set_entry_point("uppercase")
builder.add_edge("uppercase", "add_exclamation")
builder.add_edge("add_exclamation", END)
graph = builder.compile()

# Better: just use functions
def process_text(text: str) -> str:
    return add_exclamation({"text": uppercase({"text": text})["text"]})["text"]

print(process_text("hello"))  # HELLO!

import asyncio
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.checkpoint import RedisSaver
from copy import deepcopy

class ProdState(TypedDict):
    data: str
    retries: int

async def fetch_data(state: ProdState) -> dict:
    try:
        # Simulate async API call
        await asyncio.sleep(0.1)
        return {"data": state["data"].upper()}
    except Exception as e:
        state["retries"] += 1
        if state["retries"] < 3:
            return await fetch_data(state)  # Retry
        raise

def process(state: ProdState) -> dict:
    return {"data": state["data"] + "_processed"}

# Build graph
builder = StateGraph(ProdState)
builder.add_node("fetch", fetch_data)
builder.add_node("process", process)
builder.set_entry_point("fetch")
builder.add_edge("fetch", "process")
builder.add_edge("process", END)

# Use RedisSaver for production
checkpointer = RedisSaver(host="localhost", port=6379)
graph = builder.compile(checkpointer=checkpointer)

# Invoke with deep copy for isolation
async def run():
    state = deepcopy({"data": "hello", "retries": 0})
    result = await graph.ainvoke(state)
    print(result["data"])  # HELLO_processed

asyncio.run(run())

# Mistake 4: Not handling node failures
# Bad: node raises exception, graph fails
def risky_node(state):
    result = external_api_call()  # May raise
    return {"data": result}

# Good: wrap in try-except and return fallback
def safe_node(state):
    try:
        result = external_api_call()
        return {"data": result}
    except Exception as e:
        return {"error": str(e), "fallback": True}

# Mistake 5: Non-serializable state
# Bad: storing a DB connection
class BadState(TypedDict):
    db_conn: object  # Not serializable

# Good: store only serializable data
class GoodState(TypedDict):
    db_connection_string: str  # Serializable

# LangChain Chain (simple linear)
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

prompt = PromptTemplate.from_template("Translate {text} to French")
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run({"text": "Hello"})
print(result)  # Bonjour

# LangGraph (complex with cycles)
from langgraph.graph import StateGraph, END
# ... build graph with cycles and conditional edges

import logging
import json
from opentelemetry import trace

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(json.dumps({
    "time": "%(asctime)s",
    "level": "%(levelname)s",
    "message": "%(message)s",
    "request_id": "%(request_id)s"
})))
logger.addHandler(handler)

def monitored_node(state):
    tracer = trace.get_tracer(__name__)
    with tracer.start_as_current_span("monitored_node") as span:
        logger.info("Node started", extra={"request_id": state.get("request_id")})
        try:
            result = do_work(state)
            span.set_attribute("success", True)
            return result
        except Exception as e:
            span.set_attribute("error", str(e))
            logger.error(f"Node failed: {e}", extra={"request_id": state.get("request_id")})
            raise