LangGraph: Graph-Based Agent Workflows

Overview

LangGraph is a framework for building agent workflows as directed graphs. Perfect for complex, multi-step agent systems with loops and conditional logic.

---

StateGraph Basics

from langgraph.graph import StateGraph
from typing import TypedDict

class AgentState(TypedDict):
    """Shared agent state"""
    messages: list
    next: str
    final_answer: str = None

class ResearchAgent:
    """Build agent with LangGraph"""

    def __init__(self):
        self.workflow = StateGraph(AgentState)
        self.setup_workflow()

    def setup_workflow(self):
        """Define graph structure"""

        # Add nodes
        self.workflow.add_node("research", self.research_node)
        self.workflow.add_node("analyze", self.analyze_node)
        self.workflow.add_node("synthesize", self.synthesize_node)

        # Add edges
        self.workflow.add_edge("research", "analyze")
        self.workflow.add_edge("analyze", "synthesize")

        # Set entry point
        self.workflow.set_entry_point("research")

        # Compile
        self.graph = self.workflow.compile()

    def research_node(self, state: AgentState):
        """Research phase"""

        query = state["messages"][-1]
        results = self.search(query)

        return {
            "messages": state["messages"] + [f"Found: {results}"],
            "next": "analyze"
        }

    def analyze_node(self, state: AgentState):
        """Analysis phase"""

        findings = state["messages"][-1]
        analysis = self.analyze(findings)

        return {
            "messages": state["messages"] + [f"Analysis: {analysis}"],
            "next": "synthesize"
        }

    def synthesize_node(self, state: AgentState):
        """Final synthesis"""

        analysis = state["messages"][-1]
        final = self.synthesize(analysis)

        return {
            "messages": state["messages"],
            "final_answer": final
        }

    def run(self, query: str):
        """Execute workflow"""

        initial_state = {
            "messages": [query],
            "next": "research"
        }

        result = self.graph.invoke(initial_state)
        return result["final_answer"]

---

Conditional Routing

Dynamic Decision Making

class ConditionalWorkflow:
    """Route based on conditions"""

    def __init__(self):
        self.workflow = StateGraph(AgentState)
        self.setup_conditional_routing()

    def setup_conditional_routing(self):
        """Add conditional edges"""

        self.workflow.add_node("classify", self.classify_node)
        self.workflow.add_node("simple_response", self.simple_node)
        self.workflow.add_node("complex_analysis", self.complex_node)

        # Conditional routing
        self.workflow.add_conditional_edges(
            "classify",
            self.route_based_on_complexity,  # Routing function
            {
                "simple": "simple_response",
                "complex": "complex_analysis"
            }
        )

        self.workflow.set_entry_point("classify")
        self.graph = self.workflow.compile()

    def classify_node(self, state: AgentState):
        """Classify query difficulty"""

        query = state["messages"][-1]
        complexity = self.estimate_complexity(query)

        return {
            "messages": state["messages"],
            "complexity": complexity
        }

    def route_based_on_complexity(self, state: AgentState):
        """Route based on complexity"""

        if state["complexity"] < 0.5:
            return "simple"
        else:
            return "complex"

    def simple_node(self, state):
        """Handle simple queries"""
        answer = self.quick_answer(state["messages"][-1])
        return {"messages": state["messages"], "final": answer}

    def complex_node(self, state):
        """Handle complex queries"""
        answer = self.deep_analysis(state["messages"][-1])
        return {"messages": state["messages"], "final": answer}

---

Loops & Reflection

Self-Improvement Cycles

class ReflectiveWorkflow:
    """Agent that reflects on its work"""

    def __init__(self):
        self.workflow = StateGraph(AgentState)
        self.setup_reflection_loop()

    def setup_reflection_loop(self):
        """Create loop with reflection"""

        self.workflow.add_node("generate", self.generate_node)
        self.workflow.add_node("critique", self.critique_node)
        self.workflow.add_node("revise", self.revise_node)

        # Create loop
        self.workflow.add_edge("generate", "critique")

        # Conditional: should we revise?
        self.workflow.add_conditional_edges(
            "critique",
            self.should_revise,
            {
                "yes": "revise",
                "no": "end"
            }
        )

        self.workflow.add_edge("revise", "generate")  # Loop back
        self.workflow.set_entry_point("generate")
        self.graph = self.workflow.compile()

    def generate_node(self, state):
        """Generate solution"""
        solution = self.generate_solution(state["messages"][-1])
        return {"messages": state["messages"] + [solution]}

    def critique_node(self, state):
        """Critique solution"""
        critique = self.critique(state["messages"][-1])
        return {
            "messages": state["messages"] + [critique],
            "critique_score": self.score_critique(critique)
        }

    def should_revise(self, state):
        """Decide if we should revise"""
        score = state["critique_score"]
        return "yes" if score < 0.7 else "no"

    def revise_node(self, state):
        """Improve solution based on critique"""
        revised = self.improve(state["messages"])
        return {"messages": state["messages"] + [revised]}

---

Streaming & Monitoring

Real-Time Output

class MonitoredWorkflow:
    """Monitor workflow execution"""

    def run_with_streaming(self, query):
        """Execute with streaming"""

        initial_state = {"messages": [query], "next": None}

        # Stream events
        for event in self.graph.stream(initial_state):
            node_name = list(event.keys())[0]
            node_output = event[node_name]

            print(f"Node: {node_name}")
            print(f"Output: {node_output}")

            yield node_output

---

3 Warnings ⚠️

Warning 1: Infinite Loops

# ❌ WRONG
# Loop without termination condition
workflow.add_edge("revise", "generate")  # Always loops

# ✅ RIGHT
# Use conditional edge to exit
workflow.add_conditional_edges(
    "revise",
    should_continue,
    {
        "continue": "generate",
        "end": END
    }
)

Warning 2: State Explosion

# ❌ WRONG
# State grows unbounded
state["messages"].append(every_output)
# Memory usage explodes

# ✅ RIGHT
# Clean up old state
if len(state["messages"]) > 20:
    state["messages"] = state["messages"][-20:]

Warning 3: Overcomplex Graphs

# ❌ WRONG
# Graph with too many nodes
# Hard to understand and debug

# ✅ RIGHT
# Start simple, add complexity if needed
# 3-5 nodes initially
# Add more only when necessary

---

Last Updated: August 9, 2026