langgraph-workflows

LangGraph Workflows (1.0)

Build production-ready agent workflows with LangGraph's state machine architecture.

LangGraph 1.0 Overview

LangGraph is the standard for building stateful, multi-step agent applications:

• Durable execution: Survive failures and restarts
• Human-in-the-loop: Pause for approval, resume later
• Streaming: First-class support for token streaming
• Debugging: Full execution traces and replay

Core Concepts

┌─────────────────────────────────────────────────────────────┐
│                        StateGraph                            │
│                                                              │
│   ┌─────────┐    ┌─────────┐    ┌─────────┐                │
│   │  Node   │───▶│  Node   │───▶│  Node   │                │
│   └─────────┘    └─────────┘    └─────────┘                │
│       │              │              │                       │
│       │         Conditional         │                       │
│       │           Edge              │                       │
│       │              │              │                       │
│       │              ▼              │                       │
│       │         ┌─────────┐        │                       │
│       └────────▶│  Node   │◀───────┘                       │
│                 └─────────┘                                 │
│                                                              │
│   State flows through nodes, edges control routing          │
└─────────────────────────────────────────────────────────────┘

Pattern 1: Basic State Graph

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

# Define state schema
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # Append messages
    current_step: str
    result: str

# Define nodes (functions that transform state)
def process_input(state: AgentState) -> dict:
    """First node: process user input."""
    user_message = state["messages"][-1]
    return {
        "current_step": "processed",
        "messages": [{"role": "system", "content": f"Processing: {user_message}"}]
    }

def generate_response(state: AgentState) -> dict:
    """Second node: generate response."""
    # Call LLM here
    response = llm.invoke(state["messages"])
    return {
        "result": response.content,
        "messages": [{"role": "assistant", "content": response.content}]
    }

# Build graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("process", process_input)
workflow.add_node("generate", generate_response)

# Add edges
workflow.set_entry_point("process")
workflow.add_edge("process", "generate")
workflow.add_edge("generate", END)

# Compile
app = workflow.compile()

# Run
result = app.invoke({
    "messages": [{"role": "user", "content": "Hello!"}],
    "current_step": "",
    "result": ""
})