LangGraph Basics for CitadelMesh

LangGraph powers CitadelMesh agents with state machines that orchestrate complex building operations. This guide introduces LangGraph concepts and how they're used in CitadelMesh.

What is LangGraph?

LangGraph is a framework for building stateful, multi-step agents using state machines. In CitadelMesh:

• Agents are state machines that process events and make decisions
• States represent stages in building control workflows
• Edges define transitions between states based on conditions
• Nodes execute logic for each state (API calls, policy checks, etc.)

Why LangGraph for Building Automation?

✅ Deterministic - Predictable state transitions ✅ Debuggable - Clear execution flow ✅ Testable - Mock states and transitions ✅ Visualizable - Generate state machine diagrams ✅ Resumable - Persist and resume long-running processes

Core Concepts

1. State

State is the data that flows through your agent:

from dataclasses import dataclass, field
from typing import List, Dict, Any
from enum import Enum

class ThreatLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class SecurityState:
    """State for security agent"""
    events: List[Dict] = field(default_factory=list)
    threat_level: ThreatLevel = ThreatLevel.LOW
    response_plan: List[str] = field(default_factory=list)
    context: Dict[str, Any] = field(default_factory=dict)
    next_action: str = None

State is immutable - each node returns a new state.

2. Nodes

Nodes are functions that process state:

async def analyze_threat(state: SecurityState) -> SecurityState:
    """Analyze security events to determine threat level"""

    # Process events
    threat_score = 0
    for event in state.events:
        if event["severity"] == "high":
            threat_score += 10

    # Update threat level
    if threat_score >= 20:
        state.threat_level = ThreatLevel.CRITICAL
    elif threat_score >= 10:
        state.threat_level = ThreatLevel.HIGH
    else:
        state.threat_level = ThreatLevel.MEDIUM

    return state

3. Edges

Edges connect nodes and define transitions:

from langgraph.graph import StateGraph, END

# Create graph
workflow = StateGraph(SecurityState)

# Add nodes
workflow.add_node("monitor", monitor_events)
workflow.add_node("analyze", analyze_threat)
workflow.add_node("respond", execute_response)

# Add edges
workflow.add_edge("monitor", "analyze")  # Always go to analyze
workflow.add_edge("respond", END)        # End after response

4. Conditional Edges

Route based on state:

def determine_response(state: SecurityState) -> str:
    """Decide next action based on threat level"""
    if state.threat_level == ThreatLevel.CRITICAL:
        return "escalate"
    elif state.threat_level == ThreatLevel.HIGH:
        return "respond"
    else:
        return "monitor"

# Add conditional routing
workflow.add_conditional_edges(
    "analyze",                    # From node
    determine_response,           # Decision function
    {
        "escalate": "escalate",   # Route to escalate node
        "respond": "respond",     # Route to respond node
        "monitor": "monitor"      # Route back to monitor
    }
)

5. Entry Point

Define where execution starts:

# Set entry point
workflow.set_entry_point("monitor")

# Compile graph
graph = workflow.compile()

Building a Simple Agent

Let's build a temperature control agent step-by-step.

Step 1: Define State

from dataclasses import dataclass
from typing import Optional

@dataclass
class TemperatureState:
    """State for temperature control agent"""
    zone_id: str
    current_temp: float
    target_temp: float
    action: Optional[str] = None
    error: Optional[str] = None

Step 2: Create Nodes

async def read_temperature(state: TemperatureState) -> TemperatureState:
    """Read current temperature from sensor"""
    # Simulate reading from sensor
    state.current_temp = 72.0
    print(f"Current temperature: {state.current_temp}°F")
    return state

async def calculate_action(state: TemperatureState) -> TemperatureState:
    """Determine heating/cooling action"""
    diff = state.target_temp - state.current_temp

    if abs(diff) < 1.0:
        state.action = "maintain"
    elif diff > 0:
        state.action = "heat"
    else:
        state.action = "cool"

    print(f"Action: {state.action}")
    return state

async def execute_action(state: TemperatureState) -> TemperatureState:
    """Execute HVAC command"""
    print(f"Executing: {state.action} in {state.zone_id}")
    # In production: call MCP adapter
    return state

Step 3: Build Graph

from langgraph.graph import StateGraph, END

def build_temperature_agent():
    """Build temperature control state machine"""

    # Create graph
    workflow = StateGraph(TemperatureState)

    # Add nodes
    workflow.add_node("read_temp", read_temperature)
    workflow.add_node("calculate", calculate_action)
    workflow.add_node("execute", execute_action)

    # Define flow
    workflow.add_edge("read_temp", "calculate")

    # Conditional routing after calculation
    workflow.add_conditional_edges(
        "calculate",
        lambda state: "execute" if state.action != "maintain" else "done",
        {
            "execute": "execute",
            "done": END
        }
    )

    workflow.add_edge("execute", END)

    # Set entry point
    workflow.set_entry_point("read_temp")

    return workflow.compile()

Step 4: Run Agent

import asyncio

async def main():
    # Build agent
    agent = build_temperature_agent()

    # Initial state
    initial_state = TemperatureState(
        zone_id="zone-lobby",
        current_temp=0.0,  # Will be read
        target_temp=72.0
    )

    # Run agent
    final_state = await agent.ainvoke(initial_state)

    print(f"\nFinal state: {final_state}")

# Run
asyncio.run(main())

Output:

Current temperature: 72.0°F
Action: maintain

Final state: TemperatureState(
    zone_id='zone-lobby',
    current_temp=72.0,
    target_temp=72.0,
    action='maintain',
    error=None
)

CitadelMesh Agent Pattern

All CitadelMesh agents follow this pattern:

from src.agents.runtime.base_agent import BaseAgent, AgentConfig, AgentState
from langgraph.graph import StateGraph, END

class MyBuildingAgent(BaseAgent):
    """Custom building control agent"""

    def __init__(self, config: AgentConfig):
        super().__init__(config)

    def build_graph(self) -> StateGraph:
        """Build the agent state machine"""

        workflow = StateGraph(AgentState)

        # Add your nodes
        workflow.add_node("monitor", self._monitor)
        workflow.add_node("analyze", self._analyze)
        workflow.add_node("execute", self._execute)

        # Define transitions
        workflow.add_edge("monitor", "analyze")
        workflow.add_conditional_edges(
            "analyze",
            self._decide_action,
            {
                "execute": "execute",
                "wait": "monitor"
            }
        )
        workflow.add_edge("execute", END)

        workflow.set_entry_point("monitor")

        return workflow.compile()

    async def _monitor(self, state: AgentState) -> AgentState:
        """Monitor building systems"""
        # Your logic here
        return state

    async def _analyze(self, state: AgentState) -> AgentState:
        """Analyze data and make decisions"""
        # Your logic here
        return state

    async def _execute(self, state: AgentState) -> AgentState:
        """Execute building control actions"""
        # Your logic here
        return state

    def _decide_action(self, state: AgentState) -> str:
        """Routing logic"""
        # Your decision logic
        return "execute"

    async def process_event(self, event):
        """Handle incoming events"""
        # Implement event processing
        pass

Debugging State Machines

1. Print State Transitions

async def debug_node(state: SecurityState) -> SecurityState:
    """Node with debug output"""
    print(f"[DEBUG] Entering node with state: {state}")
    # Process...
    print(f"[DEBUG] Exiting node with state: {state}")
    return state

2. Visualize Graph

from langgraph.graph import StateGraph

# Build graph
graph = build_temperature_agent()

# Generate mermaid diagram
print(graph.get_graph().draw_mermaid())

Output:

graph TD
    read_temp --> calculate
    calculate --> execute
    calculate --> END
    execute --> END

3. Step Through Execution

# Run step-by-step
agent = build_temperature_agent()
state = initial_state

# Execute one step at a time
for step in agent.stream(state):
    print(f"Step: {step}")
    input("Press Enter for next step...")

4. Add Breakpoints

async def analyze_with_breakpoint(state: SecurityState) -> SecurityState:
    """Node with breakpoint for debugging"""
    import pdb; pdb.set_trace()  # Breakpoint here
    # Continue debugging...
    return state

Advanced Patterns

1. Parallel Execution

Execute multiple nodes in parallel:

from langgraph.graph import StateGraph

workflow = StateGraph(SecurityState)

# These can run in parallel
workflow.add_node("check_cameras", check_cameras)
workflow.add_node("check_doors", check_doors)
workflow.add_node("check_sensors", check_sensors)

# Converge results
workflow.add_node("merge_results", merge_results)

# Parallel edges
workflow.add_edge("start", "check_cameras")
workflow.add_edge("start", "check_doors")
workflow.add_edge("start", "check_sensors")

# Wait for all to complete
workflow.add_edge("check_cameras", "merge_results")
workflow.add_edge("check_doors", "merge_results")
workflow.add_edge("check_sensors", "merge_results")

2. Looping

Create feedback loops:

async def retry_logic(state: MyState) -> str:
    """Retry up to 3 times"""
    if state.retry_count < 3:
        return "retry"
    else:
        return "failed"

workflow.add_conditional_edges(
    "execute",
    retry_logic,
    {
        "retry": "execute",  # Loop back
        "failed": "handle_failure"
    }
)

3. Subgraphs

Compose complex agents from smaller graphs:

# Sub-agent for door control
door_agent = build_door_control_graph()

# Main security agent
workflow = StateGraph(SecurityState)
workflow.add_node("door_control", door_agent)
workflow.add_edge("analyze", "door_control")

Common Pitfalls

Issue: State not updating

Problem: Mutations don't propagate

# ❌ Wrong - mutating state
async def bad_node(state: MyState) -> MyState:
    state.value = 10  # Mutation might not persist
    return state

Solution: Return new state or use dataclasses properly

# ✅ Correct
from dataclasses import replace

async def good_node(state: MyState) -> MyState:
    return replace(state, value=10)

Issue: Infinite loops

Problem: No exit condition

# ❌ Wrong - can loop forever
workflow.add_edge("process", "process")

Solution: Add loop counter and exit condition

def check_exit(state: MyState) -> str:
    if state.iterations >= 10:
        return "exit"
    return "continue"

workflow.add_conditional_edges("process", check_exit, {
    "continue": "process",
    "exit": END
})

Testing State Machines

import pytest

@pytest.mark.asyncio
async def test_temperature_agent():
    """Test temperature control agent"""
    agent = build_temperature_agent()

    # Test heating scenario
    state = TemperatureState(
        zone_id="test-zone",
        current_temp=65.0,
        target_temp=72.0
    )

    result = await agent.ainvoke(state)

    assert result.action == "heat"
    assert result.error is None

Next Steps

• Agent Template - Create your first agent
• Policy Integration - Connect to OPA
• Testing Agents - Comprehensive testing

Resources

• LangGraph Documentation
• State Machine Patterns

---

Master state machines and build intelligent building agents! Continue to Agent Template.