agents-libagents-lib
AgentsMCP ServersWorkflowsTutorialsComparisonsTemplatesNewsAbout

LangGraph Workflow Template

Workflow

Pre-built LangGraph workflow with state management.

LangGraph Workflow Template

Overview

Pre-built LangGraph workflow template with state management, checkpointer support, human-in-the-loop capabilities, and production-ready patterns for building sophisticated AI workflows.

What is LangGraph?

LangGraph is a library for building stateful, multi-actor applications with LLMs. It extends LangChain by providing:

  • Stateful workflows: Define workflows as graphs with explicit state
  • Cyclic graphs: Support for loops and conditional branching
  • Persistence: Built-in checkpointer for state persistence and time-travel
  • Human-in-the-loop: Pause workflows for human review and intervention
  • Streaming: Real-time output streaming for long-running workflows
  • Multi-agent: Native support for multi-agent conversations

Template Structure

langgraph-template/
├── src/
│   ├── state.py           # State definition with TypedDict
│   ├── nodes.py           # Node functions (workflow steps)
│   ├── edges.py           # Conditional edge logic
│   ├── graph.py           # Graph construction
│   ├── config.py          # Configuration and settings
│   ├── types.py           # Type definitions
│   └── utils.py           # Utility functions
├── checkpoints/
│   ├── memory.py          # In-memory checkpointer
│   ├── sqlite.py          # SQLite checkpointer
│   └── postgres.py        # PostgreSQL checkpointer
├── tests/
│   ├── test_workflow.py   # Unit tests
│   └── test_nodes.py      # Node tests
├── main.py                # Entry point
├── requirements.txt
└── README.md

Installation

# Clone template
git clone https://github.com/langchain-ai/langgraph-examples.git
cd langgraph-examples/template

# Install dependencies
pip install langgraph langchain-openai langchain-community

# Or create fresh
pip install langgraph langchain langchain-openai

Core Concepts

State Management

# state.py
from typing import TypedDict, Annotated, List, Dict, Any
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

class WorkflowState(TypedDict):
    """Main workflow state."""
    # Messages history (with automatic concatenation)
    messages: Annotated[List[BaseMessage], add_messages]
    
    # Task-specific state
    topic: str
    research_results: List[Dict[str, Any]]
    draft_content: str
    final_content: str
    
    # Control flow
    current_step: str
    iteration_count: int
    max_iterations: int
    
    # Metadata
    created_at: str
    updated_at: str
    status: str  # 'pending', 'running', 'completed', 'failed', 'paused'

Node Functions

# nodes.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage
from typing import TypedDict, List, Dict, Any

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

async def research_node(state: WorkflowState) -> Dict[str, Any]:
    """Research step: gather information on the topic."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a research assistant. Gather comprehensive information."),
        ("human", "Research the following topic: {topic}"),
    ])
    
    chain = prompt | llm
    response = await chain.ainvoke({"topic": state["topic"]})
    
    return {
        "research_results": [{
            "source": "LLM Research",
            "content": response.content,
            "timestamp": datetime.now().isoformat()
        }],
        "current_step": "research_complete",
        "updated_at": datetime.now().isoformat(),
    }

async def draft_node(state: WorkflowState) -> Dict[str, Any]:
    """Drafting step: create initial content."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a content writer. Create engaging content."),
        ("human", "Write content about: {topic}\n\nResearch findings:\n{research}"),
    ])
    
    research_text = "\n".join([r["content"] for r in state.get("research_results", [])])
    chain = prompt | llm
    response = await chain.ainvoke({
        "topic": state["topic"],
        "research": research_text
    })
    
    return {
        "draft_content": response.content,
        "current_step": "draft_complete",
        "iteration_count": state.get("iteration_count", 0) + 1,
        "updated_at": datetime.now().isoformat(),
    }

async def review_node(state: WorkflowState) -> Dict[str, Any]:
    """Review step: evaluate and provide feedback."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an editor. Review content for quality and accuracy."),
        ("human", "Review the following content:\n\n{content}\n\nProvide feedback and suggestions."),
    ])
    
    chain = prompt | llm
    response = await chain.ainvoke({"content": state.get("draft_content", "")})
    
    return {
        "review_feedback": response.content,
        "current_step": "review_complete",
        "updated_at": datetime.now().isoformat(),
    }

async def finalize_node(state: WorkflowState) -> Dict[str, Any]:
    """Finalize step: produce final output."""
    return {
        "final_content": state.get("draft_content", ""),
        "status": "completed",
        "current_step": "finalize_complete",
        "updated_at": datetime.now().isoformat(),
    }

Conditional Edges

# edges.py
from typing import Literal

def should_continue(state: WorkflowState) -> Literal["draft", "finalize"]:
    """Determine next step based on current state."""
    if state.get("iteration_count", 0) >= state.get("max_iterations", 3):
        return "finalize"
    return "draft"

def should_review(state: WorkflowState) -> Literal["review", "finalize"]:
    """Decide whether to review or finalize."""
    # Example logic: review if draft is short
    draft_length = len(state.get("draft_content", ""))
    if draft_length < 500:
        return "review"
    return "finalize"

def has_feedback(state: WorkflowState) -> Literal["draft", "finalize"]:
    """Continue drafting if there's feedback, otherwise finalize."""
    if state.get("review_feedback"):
        return "draft"
    return "finalize"

Graph Construction

# graph.py
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from nodes import research_node, draft_node, review_node, finalize_node
from edges import should_continue, should_review, has_feedback
from state import WorkflowState

def create_workflow(checkpointer=None):
    """Create and compile the workflow graph."""
    
    # Initialize graph
    builder = StateGraph(WorkflowState)
    
    # Add nodes
    builder.add_node("research", research_node)
    builder.add_node("draft", draft_node)
    builder.add_node("review", review_node)
    builder.add_node("finalize", finalize_node)
    
    # Define edges
    builder.add_edge("research", "draft")
    builder.add_conditional_edges(
        "draft",
        should_continue,
        {
            "draft": "draft",
            "finalize": "finalize",
        }
    )
    builder.add_conditional_edges(
        "review",
        has_feedback,
        {
            "draft": "draft",
            "finalize": "finalize",
        }
    )
    builder.add_edge("finalize", END)
    
    # Set entry point
    builder.set_entry_point("research")
    
    # Compile with optional checkpointer
    return builder.compile(checkpointer=checkpointer)

# Create workflow with memory checkpointer
checkpointer = MemorySaver()
workflow = create_workflow(checkpointer=checkpointer)

Checkpointers

In-Memory Checkpointer

# checkpoints/memory.py
from langgraph.checkpoint.memory import MemorySaver

# Simple in-memory storage (for development/testing)
checkpointer = MemorySaver()

# Thread-specific state
config = {"configurable": {"thread_id": "session-123"}}

SQLite Checkpointer

# checkpoints/sqlite.py
from langgraph.checkpoint.sqlite import SqliteSaver

# Persistent SQLite storage
checkpointer = SqliteSaver.from_conn_string("checkpoints.sqlite")

# For async usage
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
async_checkpointer = await AsyncSqliteSaver.afrom_conn_string("checkpoints.sqlite")

PostgreSQL Checkpointer

# checkpoints/postgres.py
from langgraph.checkpoint.postgres import PostgresSaver
import psycopg2

# Persistent PostgreSQL storage
conn = psycopg2.connect("dbname=langgraph user=postgres password=secret")
checkpointer = PostgresSaver(conn)

# For async usage
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
async_checkpointer = await AsyncPostgresSaver.from_conn_string(
    "postgresql://user:password@localhost:5432/langgraph"
)

Human-in-the-Loop

Pause for Review

# main.py
from graph import workflow
from langgraph.types import Command

async def run_with_human_review(input_data: dict):
    """Run workflow with human review checkpoints."""
    
    config = {"configurable": {"thread_id": "review-session"}}
    
    # Start workflow
    async for event in workflow.astream(
        input_data,
        config=config,
        stream_mode="values",
    ):
        print(f"Step completed: {event.get('current_step')}")
        
        # Pause at review point
        if event.get("current_step") == "draft_complete":
            print("\n--- DRAFT READY FOR REVIEW ---")
            print(event.get("draft_content"))
            print("\n--- PAUSED FOR HUMAN INPUT ---")
            
            # Wait for human approval
            approval = await ask_human_approval()
            
            if approval == "approve":
                # Continue with finalize
                continue
            elif approval == "revise":
                # Provide feedback and continue
                feedback = await get_human_feedback()
                yield Command(
                    update={"review_feedback": feedback},
                    go_to="draft",  # Go back to draft node
                )
            else:
                # Cancel workflow
                break

async def ask_human_approval() -> str:
    """Ask human for approval decision."""
    response = input("Approve draft? (approve/revise/cancel): ")
    return response.lower()

async def get_human_feedback() -> str:
    """Get human feedback for revision."""
    return input("Enter feedback: ")

Interrupt Before Node

from langgraph.graph import StateGraph

builder = StateGraph(WorkflowState)
builder.add_node("draft", draft_node)
builder.add_node("review", review_node)

# Interrupt before review node
builder.add_edge("draft", "review")
builder.interrupt_before("review")

workflow = builder.compile(checkpointer=MemorySaver())

# Resume after human intervention
snapshot = workflow.get_state(config)
if snapshot.next:
    # Human has reviewed, resume
    workflow.invoke(None, config=config)

Streaming Output

Stream Updates

async def stream_workflow(input_data: dict):
    """Stream workflow updates in real-time."""
    
    config = {"configurable": {"thread_id": "stream-session"}}
    
    async for event in workflow.astream(
        input_data,
        config=config,
        stream_mode="updates",
    ):
        for node_name, node_output in event.items():
            print(f"[{node_name}] {node_output}")
            yield node_output

Stream Values

async def stream_state(input_data: dict):
    """Stream complete state after each step."""
    
    config = {"configurable": {"thread_id": "state-session"}}
    
    async for state in workflow.astream(
        input_data,
        config=config,
        stream_mode="values",
    ):
        print(f"Current step: {state['current_step']}")
        print(f"Iteration: {state['iteration_count']}")
        yield state

Stream Custom Modes

async def stream_custom(input_data: dict):
    """Stream custom output format."""
    
    config = {"configurable": {"thread_id": "custom-session"}}
    
    async for event in workflow.astream(
        input_data,
        config=config,
        stream_mode="custom",
    ):
        # Custom streaming logic
        yield event

Advanced Patterns

Subgraphs

from langgraph.graph import StateGraph, MessagesState

# Create subgraph for research phase
research_builder = StateGraph(MessagesState)
research_builder.add_node("search", search_node)
research_builder.add_node("analyze", analyze_node)
research_builder.add_edge("search", "analyze")
research_builder.set_entry_point("search")
research_builder.add_edge("analyze", END)

research_graph = research_builder.compile()

# Main graph with subgraph
main_builder = StateGraph(WorkflowState)
main_builder.add_node("research", research_graph)
main_builder.add_node("write", write_node)
main_builder.add_edge("research", "write")
main_builder.set_entry_point("research")

workflow = main_builder.compile()

Parallel Execution

from langgraph.graph import START

builder = StateGraph(WorkflowState)

# Parallel research branches
builder.add_node("research_web", web_research_node)
builder.add_node("research_docs", docs_research_node)
builder.add_node("research_books", books_research_node)

# Merge results
def merge_results(state: WorkflowState) -> Dict[str, Any]:
    all_results = (
        state.get("web_results", []) +
        state.get("docs_results", []) +
        state.get("books_results", [])
    )
    return {"research_results": all_results}

builder.add_node("merge", merge_results)

# Parallel edges from start
builder.add_edge(START, "research_web")
builder.add_edge(START, "research_docs")
builder.add_edge(START, "research_books")

# All parallel nodes feed into merge
builder.add_edge("research_web", "merge")
builder.add_edge("research_docs", "merge")
builder.add_edge("research_books", "merge")

workflow = builder.compile()

Conditional Parallelism

from typing import Literal

def select_research_sources(state: WorkflowState) -> Literal["web", "docs", "both"]:
    """Determine which research sources to use."""
    if state["topic"] == "technical":
        return "docs"
    elif state["topic"] == "current_events":
        return "web"
    else:
        return "both"

builder.add_conditional_edges(
    START,
    select_research_sources,
    {
        "web": "research_web",
        "docs": "research_docs",
        "both": ["research_web", "research_docs"],
    }
)

Error Handling

from langgraph.errors import NodeInterrupt

async def resilient_research_node(state: WorkflowState) -> Dict[str, Any]:
    """Research node with error handling and retries."""
    max_retries = 3
    attempt = 0
    
    while attempt < max_retries:
        try:
            result = await perform_research(state["topic"])
            return {
                "research_results": result,
                "current_step": "research_complete",
            }
        except Exception as e:
            attempt += 1
            if attempt == max_retries:
                # Log error and continue with partial results
                return {
                    "research_results": [],
                    "error": f"Research failed after {max_retries} attempts: {str(e)}",
                    "current_step": "research_failed",
                }
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

Time Travel

# Get state at specific checkpoint
snapshot = workflow.get_state(config)
print(f"Current state: {snapshot.values}")

# Get history of states
history = workflow.get_state_history(config)
for checkpoint in history:
    print(f"Checkpoint {checkpoint.config}: {checkpoint.values}")

# Jump to previous state
previous_config = history[1].config
workflow.update_state(previous_config, {"draft_content": "Updated content"})

Configuration

Environment Variables

# .env
OPENAI_API_KEY=sk-...
LANGCHAIN_API_KEY=lsv2_...
LANGCHAIN_TRACING_V2=true
LANGCHAIN_ENDPOINT=https://api.smith.langchain.com
DATABASE_URL=postgresql://user:pass@localhost:5432/langgraph

Workflow Configuration

# config.py
from pydantic import BaseModel

class WorkflowConfig(BaseModel):
    max_iterations: int = 3
    temperature: float = 0.7
    model: str = "gpt-4o"
    timeout_seconds: int = 300
    enable_persistence: bool = True
    enable_streaming: bool = True
    human_review_points: list[str] = ["draft_complete"]

config = WorkflowConfig()

Testing

Unit Tests

# tests/test_nodes.py
import pytest
from nodes import research_node, draft_node
from state import WorkflowState

@pytest.mark.asyncio
async def test_research_node():
    state = WorkflowState(
        topic="AI agents",
        messages=[],
        current_step="start",
    )
    result = await research_node(state)
    
    assert "research_results" in result
    assert len(result["research_results"]) > 0
    assert result["current_step"] == "research_complete"

@pytest.mark.asyncio
async def test_draft_node():
    state = WorkflowState(
        topic="AI agents",
        research_results=[{"content": "AI agents are tools"}],
        messages=[],
    )
    result = await draft_node(state)
    
    assert "draft_content" in result
    assert len(result["draft_content"]) > 0

Integration Tests

# tests/test_workflow.py
import pytest
from langgraph.checkpoint.memory import MemorySaver
from graph import create_workflow

@pytest.mark.asyncio
async def test_full_workflow():
    checkpointer = MemorySaver()
    workflow = create_workflow(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "test-1"}}
    input_data = {"topic": "Test topic", "messages": []}
    
    result = await workflow.ainvoke(input_data, config=config)
    
    assert result["status"] == "completed"
    assert "final_content" in result
    assert len(result["final_content"]) > 0

Deployment

Docker

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
CMD ["python", "main.py"]

FastAPI Server

# api/server.py
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from graph import workflow
from langgraph.checkpoint.sqlite import SqliteSaver

app = FastAPI()

class WorkflowInput(BaseModel):
    topic: str
    max_iterations: int = 3

@app.post("/run")
async def run_workflow(input: WorkflowInput):
    thread_id = f"thread-{uuid.uuid4()}"
    config = {"configurable": {"thread_id": thread_id}}
    
    # Run workflow asynchronously
    result = await workflow.ainvoke(
        {"topic": input.topic, "messages": []},
        config=config
    )
    
    return {"thread_id": thread_id, "result": result}

@app.get("/status/{thread_id}")
async def get_status(thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    state = workflow.get_state(config)
    return {"status": state.values.get("status"), "step": state.values.get("current_step")}

Troubleshooting

Common Issues

State not persisting:

  • Ensure checkpointer is passed to compile()
  • Verify thread_id is consistent across invocations
  • Check database connectivity for persistent checkpointer

Infinite loops:

  • Add max_iterations limit in state
  • Implement proper termination conditions in edges
  • Use NodeInterrupt to break cycles

Memory errors:

  • Use streaming for long-running workflows
  • Implement state pruning for large histories
  • Consider using persistent checkpointer with cleanup

Serialization errors:

  • Ensure all state values are JSON-serializable
  • Use Annotated with reducer functions for complex types
  • Avoid storing non-serializable objects in state

Debugging

# Enable verbose logging
import langgraph
langgraph.logger.setLevel(logging.DEBUG)

# Trace execution
async for event in workflow.astream(input, config=config, stream_mode="debug"):
    print(event)

Resources

Last updated: May 2026

View on GitHub

Related Templates

CrewAI Agent Template

Starter template for building CrewAI multi-agent systems.

View →

MCP Server Boilerplate

Quick start template for building custom MCP servers.

View →

AutoGen Conversation Template

Template for multi-agent conversation flows.

View →