Part 25: LangGraph - Multi-Agent State Machines & Workflows

Master state graphs, nodes, edges, conditional routing, checkpointing, human-in-the-loop, persistence, and agent orchestration with LangGraph.

Part 25: LangGraph - Multi-Agent State Machines & Workflows

← Back to Master Index


1. Why LangGraph in 2026?

LangGraph is the future of agentic AI. Engineers with LangGraph expertise command 50-100% higher salaries in AI engineering roles.

Key Concepts

  • State Graphs: Manage complex agent workflows
  • Nodes: Processing units in the graph
  • Edges: Control flow between nodes
  • Conditional Routing: Dynamic workflow paths
  • Persistence: State management across sessions

2. LangGraph Fundamentals

Installation

pip install langgraph
pip install -U langchain-google-genai  # For Google models

Basic Agent Graph

from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, List

# Define state
class AgentState(TypedDict):
    messages: List[HumanMessage | AIMessage]

# Create state graph
graph = StateGraph(AgentState)

# Define nodes
def agent_node(state: AgentState):
    # Call LLM with messages
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Add nodes
graph.add_node("agent", agent_node)

# Add edges
graph.add_edge(START, "agent")
graph.add_edge("agent", END)

# Compile graph
agent_graph = graph.compile()

3. Multi-Agent Workflows

Research Agent Example

# Define specialized agents
class ResearchState(TypedDict):
    query: str
    research_results: List[str]
    summary: str

def researcher_agent(state: ResearchState):
    # Research using web search
    results = web_search(state["query"])
    return {"research_results": results}

def summarizer_agent(state: ResearchState):
    # Summarize research
    prompt = f"Summarize: {state['research_results']}"
    summary = llm.invoke(prompt)
    return {"summary": summary}

# Build multi-agent graph
graph = StateGraph(ResearchState)

graph.add_node("researcher", researcher_agent)
graph.add_node("summarizer", summarizer_agent)

graph.add_edge("researcher", "summarizer")
graph.add_edge(START, "researcher")
graph.add_edge("summarizer", END)

multi_agent_graph = graph.compile()

Agent Orchestration

# Orchestrator pattern
def orchestrator(state: AgentState):
    query = state["query"]
    
    # Route to appropriate agent
    if "code" in query.lower():
        return "code_agent"
    elif "research" in query.lower():
        return "research_agent"
    else:
        return "general_agent"

# Conditional routing
graph.add_conditional_edges(
    "router",
    orchestrator,
    {
        "code_agent": "code_agent",
        "research_agent": "research_agent",
        "general_agent": "general_agent"
    }
)

4. Checkpointing and Persistence

Memory Persistence

from langgraph.checkpoint.memory import MemorySaver

# Create memory saver
memory = MemorySaver()

# Compile with persistence
graph_with_memory = graph.compile(checkpointer=memory)

# Invoke with thread ID
config = {"configurable": {"thread_id": "user_123"}}
result = graph_with_memory.invoke(state, config)

# Resume conversation
next_state = {"messages": [HumanMessage(content="Continue")]}
result = graph_with_memory.invoke(next_state, config)

Database Persistence

from langgraph.checkpoint.postgres import PostgresChecker

# PostgreSQL checkpoint store
checkpointer = PostgresChecker(
    conn_string="postgresql://user:pass@host:5432/db"
)

graph = graph.compile(checkpointer=checkpointer)

5. Human-in-the-Loop

Interruptible Agents

def human_review_node(state: AgentState):
    # Pause for human review
    human_input = input("Review this output: ")
    return {"human_review": human_input}

# Add interrupt point
graph.add_node("human_review", human_review_node)
graph.add_edge("agent", "human_review")
graph.add_conditional_edges("human_review", should_continue, {END})

# Run with interrupts
config = {
    "configurable": {"thread_id": "user_123"},
    "interrupt_after": ["agent"]  # Interrupt after agent node
}

Approval Workflows

def approval_node(state: AgentState):
    # Check if approval needed
    if state.get("needs_approval"):
        # Send for approval
        send_approval_request(state["output"])
        return {"status": "pending_approval"}
    return {"status": "approved"}

# Conditional approval flow
graph.add_conditional_edges(
    "process",
    lambda x: "approve" if x.get("needs_approval") else "complete",
    {"approve": "approval", "complete": END}
)

6. Tool-Augmented Agents

Tool Integration

from langgraph.prebuilt import ToolNode, create_agent

# Define tools
tools = [
    web_search_tool,
    calculator_tool,
    database_query_tool
]

# Create agent with tools
agent = create_agent(llm, tools)

# Add tool node
graph.add_node("tools", ToolNode(tools))

# Add conditional edge for tool use
graph.add_conditional_edges(
    "agent",
    lambda x: "tools" if should_use_tools(x) else END,
    {"tools": "tools", END: END}
)

Tool Schema Definition

from langchain_core.tools import Tool

web_search = Tool(
    name="web_search",
    description="Search the web for current information",
    func=lambda query: search_web(query)
)

calculator = Tool(
    name="calculator",
    description="Perform mathematical calculations",
    func=lambda expr: eval(expr)
)

7. Resource Directory: LangGraph

Best Books

BookAuthorPriceKey Topics
Building Agentic SystemsO'ReillyPaidAgent design
LangGraph DocumentationOfficialFreeOfficial guide
Multi-Agent SystemsManningPaidAgent coordination
AI Agents HandbookO'ReillyPaidAgent development

Best Udemy Courses

CourseInstructorPrice (INR)Key Topics
LangGraph & AgentsInstructor₹1,999-2,999LangGraph mastery
Multi-Agent SystemsInstructor₹1,499-2,299Agent teams
Agentic AIInstructor₹1,499-2,299Autonomous agents
Building AI AgentsInstructor₹999-1,499Agent development

Best O'Reilly Resources

ResourceTopicAccess
Building Agentic SystemsO'ReillyPaid
LangGraph in ProductionO'ReillyPaid
Multi-Agent AIO'ReillyPaid

Best LinkedIn Learning Courses

CourseInstructorAccess
LangGraph Essential TrainingInstructorPaid
Agentic AI SystemsInstructorPaid
Multi-Agent WorkflowsInstructorPaid

Free Resources

PlatformResourceLink
LangGraph DocsOfficial docspython.langchain.com/docs/langgraph
LangGraph TutorialsGitHubgithub.com/langchain-ai/langgraph
Awesome LangGraphGitHubgithub.com/ranahaani/awesome-langgraph
LangSmithMonitoringsmith.langchain.com

8. Common LangGraph Interview Questions

QuestionAnswer
What is a state graph?A directed graph that manages state transitions in agent workflows.
Difference between Runnable and Node?Nodes are graph components, Runnables are LangChain's execution interface.
How to handle long-running agents?Use checkpointing and persistence for state management.
What is conditional routing?Dynamic path selection based on state or conditions.
How to implement human-in-the-loop?Use interrupt points and approval nodes in the graph.

9. Part Navigation

Previous Parts

Part 24: LangChain Foundations

Next Parts

Part 26: Multi-Agent Systems · Part 27: Tool-Augmented Agents


Proceed to Part 26: Multi-Agent Systems →

Comments

Comments are powered by giscus. Set PUBLIC_GISCUS_REPO_ID and PUBLIC_GISCUS_CATEGORY_ID in your environment to enable them.