xbe.at

LLMs and generative AI / Agents and LangChain

Basics of LangChain's LangGraph

6 min read, 14 sections

Contents
  1. Introduction to LangGraph Core Components
  2. State Management in LangGraph
  3. Creating Custom Nodes
  4. Implementing Conditional Branching
  5. Tool Integration with LangGraph
  6. Parallel Processing in LangGraph
  7. Event-Driven Workflows
  8. Memory Management in LangGraph
  9. Error Handling and Recovery
  10. Real-World Example - Customer Support Workflow
  11. Advanced Agent Coordination
  12. Real-World Example - Document Processing Pipeline
  13. Performance Optimization Techniques
  14. Additional Resources

Introduction to LangGraph Core Components

LangGraph serves as an extension of LangChain, enabling the creation of multi-agent systems through graph-based workflows. It provides a structured approach to designing complex agent interactions using nodes and edges, where each node represents a distinct processing state and edges define valid state transitions.

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

# Define the state structure
class AgentState(TypedDict):
    messages: list
    next_step: str
    
# Initialize the graph
graph = StateGraph(AgentState)

# Add nodes and edges
graph.add_node("start", start_node_fn)
graph.add_node("process", process_node_fn)
graph.add_edge("start", "process")

State Management in LangGraph

LangGraph implements state management through TypedDict classes, ensuring type safety and clear data structures. The state object maintains consistency across node transitions and enables data persistence throughout the workflow execution cycle.

from typing import Annotated, Sequence, TypedDict, Union
from langgraph.prebuilt import ToolExecutor

class WorkflowState(TypedDict):
    messages: Sequence[str]
    current_status: str
    tools_output: dict
    
def state_manager(state: WorkflowState):
    return {
        "messages": state["messages"],
        "status": state["current_status"],
        "tools": state["tools_output"]
    }

Creating Custom Nodes

Custom nodes in LangGraph function as processing units that can perform specific tasks, interact with language models, or execute tool operations. Each node receives the current state and must return a modified state according to the workflow requirements.

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

def create_processing_node():
    chat = ChatOpenAI()
    prompt = ChatPromptTemplate.from_template(
        "Process the following input: {input}"
    )
    
    def node_function(state: AgentState):
        response = chat.invoke(
            prompt.format_messages(input=state["messages"][-1])
        )
        state["messages"].append(response.content)
        return state
    
    return node_function

Implementing Conditional Branching

The power of LangGraph lies in its ability to create dynamic workflows through conditional branching. This allows for complex decision-making processes where the next state is determined by the output of the current node's processing.

def branch_function(state: AgentState) -> str:
    last_message = state["messages"][-1]
    
    if "error" in last_message.lower():
        return "error_handler"
    elif "complete" in last_message.lower():
        return "final_state"
    else:
        return "continue_processing"

# Add conditional branching to graph
graph.add_node("decision", branch_function)
graph.add_conditional_edges(
    "decision",
    branch_function,
    {
        "error_handler": "error_handling_node",
        "final_state": "completion_node",
        "continue_processing": "process_node"
    }
)

Tool Integration with LangGraph

LangGraph seamlessly integrates with LangChain's tool ecosystem, allowing agents to interact with external systems and APIs. The ToolExecutor class manages tool execution and maintains the workflow state.

from langchain.tools import BaseTool
from langchain.pydantic_v1 import BaseModel

class CustomTool(BaseTool):
    name = "custom_processor"
    description = "Processes specific data formats"
    
    def _run(self, input_data: str) -> str:
        # Tool implementation
        processed_result = f"Processed: {input_data}"
        return processed_result
    
tool_executor = ToolExecutor([CustomTool()])
graph.add_node("tool_execution", tool_executor)

Parallel Processing in LangGraph

LangGraph supports concurrent execution of multiple nodes through its parallel processing capabilities. This feature enables efficient handling of independent tasks and improves overall workflow performance by executing compatible nodes simultaneously.

from langgraph.graph import END, StateGraph
from typing import List
import asyncio

async def parallel_processor(states: List[AgentState]) -> List[AgentState]:
    async def process_single(state):
        # Simulate processing
        await asyncio.sleep(1)
        state["messages"].append(f"Processed state {state['next_step']}")
        return state
    
    tasks = [process_single(state) for state in states]
    results = await asyncio.gather(*tasks)
    return results

graph.add_parallel_nodes(
    ["processor_1", "processor_2", "processor_3"],
    parallel_processor
)

Event-Driven Workflows

LangGraph enables the creation of event-driven workflows where state transitions are triggered by specific events or conditions. This pattern is particularly useful for implementing reactive systems and handling asynchronous operations.

from typing import Any, Dict

class EventState(TypedDict):
    events: List[str]
    handlers: Dict[str, Any]

def create_event_handler():
    def handle_event(state: EventState) -> EventState:
        current_event = state["events"][-1]
        if current_event in state["handlers"]:
            handler = state["handlers"][current_event]
            result = handler(state)
            state["events"].append(f"Handled: {current_event}")
        return state
    return handle_event

event_graph = StateGraph(EventState)
event_graph.add_node("event_processor", create_event_handler())

Memory Management in LangGraph

LangGraph implements sophisticated memory management through state persistence and retrieval mechanisms. This enables long-term storage of workflow data and facilitates complex decision-making based on historical interactions.

class MemoryState(TypedDict):
    short_term: List[str]
    long_term: Dict[str, Any]
    context: Dict[str, Any]

def memory_manager(state: MemoryState) -> MemoryState:
    # Implement memory retention logic
    current_context = state["short_term"][-1]
    
    if len(state["short_term"]) > 5:
        # Move older items to long-term memory
        key = f"memory_{len(state['long_term'])}"
        state["long_term"][key] = state["short_term"].pop(0)
        
    # Update context
    state["context"].update({
        "recent_memory": state["short_term"],
        "memory_size": len(state["long_term"])
    })
    
    return state

Error Handling and Recovery

Robust error handling in LangGraph involves implementing recovery mechanisms and fallback strategies. This ensures workflow resilience and maintains system stability even when encountering unexpected situations.

from typing import Optional

class ErrorState(TypedDict):
    error: Optional[str]
    retry_count: int
    max_retries: int

def create_error_handler():
    def handle_error(state: ErrorState) -> Union[ErrorState, END]:
        if state["error"]:
            if state["retry_count"] < state["max_retries"]:
                state["retry_count"] += 1
                state["error"] = None
                return state
            else:
                # Maximum retries reached, terminate workflow
                return END
        return state
    
    return handle_error

# Implementation in graph
graph.add_node("error_handler", create_error_handler())
graph.add_edge("error_handler", "process")

Real-World Example - Customer Support Workflow

This implementation demonstrates a practical customer support system using LangGraph, incorporating natural language processing, sentiment analysis, and automated response generation with fallback to human operators.

from langchain.prompts import ChatPromptTemplate
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS

class SupportState(TypedDict):
    query: str
    sentiment: float
    response: Optional[str]
    requires_human: bool

def create_support_workflow():
    embeddings = OpenAIEmbeddings()
    knowledge_base = FAISS.from_texts(
        ["FAQ 1", "FAQ 2", "FAQ 3"],
        embeddings
    )
    
    def process_query(state: SupportState) -> SupportState:
        # Find relevant response
        docs = knowledge_base.similarity_search(state["query"])
        response = docs[0].page_content
        state["response"] = response
        state["requires_human"] = state["sentiment"] < -0.5
        return state
    
    return process_query

support_graph = StateGraph(SupportState)
support_graph.add_node("support", create_support_workflow())

Advanced Agent Coordination

LangGraph enables sophisticated agent coordination through hierarchical control structures and message passing protocols. This architecture allows for complex multi-agent systems where agents can collaborate, compete, or operate independently based on workflow requirements.

from typing import List, Dict, Any

class AgentCoordinator(TypedDict):
    agents: Dict[str, Any]
    messages: List[Dict[str, str]]
    priorities: Dict[str, int]

def create_coordinator():
    def coordinate(state: AgentCoordinator) -> AgentCoordinator:
        # Sort agents by priority
        active_agents = sorted(
            state["agents"].items(),
            key=lambda x: state["priorities"][x[0]],
            reverse=True
        )
        
        for agent_id, agent in active_agents:
            response = agent.process(state["messages"][-1])
            state["messages"].append({
                "agent": agent_id,
                "content": response
            })
            
            if response.get("final", False):
                break
                
        return state
    
    return coordinate

Real-World Example - Document Processing Pipeline

This implementation showcases a complete document processing system utilizing LangGraph for handling multiple document types, extraction, and validation with error recovery mechanisms.

from typing import List, Dict, Optional
import re

class DocumentState(TypedDict):
    raw_text: str
    extracted_data: Dict[str, Any]
    validation_errors: List[str]
    processing_stage: str

def create_document_pipeline():
    def extract_information(text: str) -> Dict[str, Any]:
        patterns = {
            'email': r'[\w\.-]+@[\w\.-]+\.\w+',
            'phone': r'\+?[\d\-\(\)]{10,}',
            'date': r'\d{2,4}[-/]\d{2}[-/]\d{2,4}'
        }
        
        return {
            field: re.findall(pattern, text)
            for field, pattern in patterns.items()
        }
    
    def process_document(state: DocumentState) -> DocumentState:
        try:
            # Extract information
            state["extracted_data"] = extract_information(state["raw_text"])
            
            # Validate extracted data
            for field, values in state["extracted_data"].items():
                if not values:
                    state["validation_errors"].append(
                        f"Missing {field} information"
                    )
            
            state["processing_stage"] = "completed" if not state["validation_errors"] else "failed"
            
        except Exception as e:
            state["validation_errors"].append(str(e))
            state["processing_stage"] = "error"
            
        return state
    
    return process_document

doc_graph = StateGraph(DocumentState)
doc_graph.add_node("processor", create_document_pipeline())

Performance Optimization Techniques

LangGraph workflows can be optimized through various techniques including caching, batching, and selective execution. These optimizations improve throughput and reduce latency in complex multi-agent systems.

from functools import lru_cache
from typing import Optional, List

class OptimizedState(TypedDict):
    batch_size: int
    cache_enabled: bool
    execution_stats: Dict[str, float]

@lru_cache(maxsize=1000)
def cached_processor(input_data: str) -> str:
    # Simulate expensive processing
    return f"Processed: {input_data}"

def create_optimized_workflow():
    def process_batch(state: OptimizedState) -> OptimizedState:
        batch = state.get("current_batch", [])
        
        if len(batch) >= state["batch_size"]:
            results = []
            for item in batch:
                if state["cache_enabled"]:
                    result = cached_processor(str(item))
                else:
                    result = f"Processed: {item}"
                results.append(result)
                
            state["execution_stats"].update({
                "batch_size": len(batch),
                "processed_items": len(results)
            })
            
            state["current_batch"] = []
            state["results"] = results
            
        return state
    
    return process_batch

Additional Resources

arXiv Papers:

More in Agents and LangChain

  1. Building a Content Summarizer with OpenAI LangChain and Gradio10 min

    The foundation of our YouTube and website summarizer requires specific Python packages. We'll use OpenAI for the language model, LangChain for chain...

  2. Automated Design of Intelligent Agents with Python10 min

    Automated design of agentic systems refers to the process of creating intelligent agents that can make decisions and take actions autonomously. This field...

  3. Building a Custom LangChain QnA Agent with Wikipedia Using Python4 min

    LangChain is a framework for building applications with large language models (LLMs). It provides abstractions for working with LLMs, data retrieval, and...

  4. Applications of LLM-Based Agents in Business8 min

    LLM-Based Agents are systems that leverage large language models to plan, reason, and act in real-world scenarios. These agents combine the powerful natural...

  5. Building a LangGraph Agent with Graph Memory8 min

    A LangGraph Agent combines natural language processing with graph neural networks to create an intelligent system capable of maintaining contextual memory...

  6. Building a Voice-Enabled AI Sales Agent11 min

    This slide covers the essential environment configuration required for building a voice-enabled AI sales agent. The setup includes installing necessary...