Build LLM Multi-Agent DevOps Workflows with LangGraph in 2026

pip install langgraph langchain-anthropic anthropic

from typing import Annotated, TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
import operator
 
 
class IncidentState(TypedDict):
    alert: dict                     # Original alert data
    severity: str                   # critical/high/medium/low
    is_real: bool                   # Is this a real incident?
    root_cause: str                 # RCA findings
    remediation_plan: list[str]     # Ordered fix steps
    approved: bool                  # Human approved the plan
    executed_steps: list[str]       # Steps completed
    messages: Annotated[list, operator.add]    # Full message history
 
 
llm = ChatAnthropic(model="claude-sonnet-5", temperature=0)

import json
 
 
def triage_agent(state: IncidentState) -> dict:
    """Determine if alert is real and its severity."""
    alert = state["alert"]
 
    response = llm.invoke([
        SystemMessage(content="""You are a DevOps triage agent. Your job is to:
1. Determine if the alert is a real incident or a false positive
2. Assign severity: critical/high/medium/low
3. Classify the type: availability/performance/security/capacity
 
Be conservative — when in doubt, treat as real."""),
        HumanMessage(content=f"""Triage this alert:
{json.dumps(alert, indent=2)}
 
Respond as JSON:
{{"is_real": true/false, "severity": "critical|high|medium|low", "type": "...", "reasoning": "..."}}""")
    ])
 
    try:
        triage = json.loads(response.content)
        return {
            "severity": triage["severity"],
            "is_real": triage["is_real"],
            "messages": [response]
        }
    except json.JSONDecodeError:
        return {"severity": "high", "is_real": True, "messages": [response]}
 
 
def rca_agent(state: IncidentState) -> dict:
    """Perform root cause analysis."""
    alert = state["alert"]
 
    # In production: call real monitoring APIs here
    # Mocked context for this example
    mock_metrics = {
        "pod_restarts": 5,
        "memory_usage_mb": 498,
        "memory_limit_mb": 512,
        "last_deployment": "2026-07-20T14:30:00Z",
        "error_log_sample": "FATAL: Out of memory — Killed process 1234",
    }
 
    response = llm.invoke([
        SystemMessage(content="You are a senior SRE performing root cause analysis. Be specific about what is failing and why."),
        HumanMessage(content=f"""Alert: {json.dumps(alert, indent=2)}
 
Metrics context:
{json.dumps(mock_metrics, indent=2)}
 
Provide root cause analysis:
1. Primary root cause (specific, not vague)
2. Contributing factors
3. Timeline of events
4. Confidence level (0-100%)""")
    ])
 
    return {
        "root_cause": response.content,
        "messages": [response]
    }
 
 
def remediation_agent(state: IncidentState) -> dict:
    """Generate remediation plan."""
    response = llm.invoke([
        SystemMessage(content="""You are a DevOps remediation specialist. Generate an ordered action plan.
Mark each step as SAFE (can auto-execute) or REQUIRES_APPROVAL (destructive/risky).
Format each step as: [SAFE|REQUIRES_APPROVAL] <command or action>"""),
        HumanMessage(content=f"""Root cause: {state['root_cause']}
 
Generate a remediation plan with exact commands. For Kubernetes issues, use kubectl commands.""")
    ])
 
    # Parse the plan into steps
    steps = []
    for line in response.content.split("\n"):
        if line.strip() and ("[SAFE]" in line or "[REQUIRES_APPROVAL]" in line):
            steps.append(line.strip())
 
    return {
        "remediation_plan": steps,
        "messages": [response]
    }
 
 
def human_review_node(state: IncidentState) -> dict:
    """Present plan to human for approval (in production: send to Slack)."""
    print("\n" + "="*60)
    print("INCIDENT REMEDIATION PLAN — REQUIRES APPROVAL")
    print("="*60)
    print(f"Severity: {state['severity']}")
    print(f"Root Cause: {state['root_cause'][:200]}")
    print("\nProposed Actions:")
    for i, step in enumerate(state["remediation_plan"], 1):
        print(f"  {i}. {step}")
 
    # In production: this would send a Slack message with approve/deny buttons
    # Here: simple CLI approval
    approval = input("\nApprove remediation plan? (yes/no): ").strip().lower()
 
    return {"approved": approval == "yes"}
 
 
def execution_agent(state: IncidentState) -> dict:
    """Execute approved safe steps."""
    if not state["approved"]:
        return {"executed_steps": ["Plan rejected by operator — no actions taken"]}
 
    executed = []
    for step in state["remediation_plan"]:
        if "[SAFE]" in step:
            # Extract command after [SAFE]
            command = step.replace("[SAFE]", "").strip()
            # In production: actually run subprocess.run(command.split())
            print(f"Executing: {command}")
            executed.append(f"EXECUTED: {command}")
        elif "[REQUIRES_APPROVAL]" in step:
            executed.append(f"SKIPPED (requires manual execution): {step}")
 
    return {"executed_steps": executed}
 
 
def notification_agent(state: IncidentState) -> dict:
    """Send incident summary to Slack/PagerDuty."""
    summary = f"""
*Incident Response Summary*
Severity: {state['severity']}
Root Cause: {state['root_cause'][:300]}
Actions Taken:
{chr(10).join(f'  - {s}' for s in state['executed_steps'])}
    """.strip()
 
    # In production: requests.post(SLACK_WEBHOOK, json={"text": summary})
    print(f"\nNotification sent:\n{summary}")
 
    return {"messages": [AIMessage(content=f"Incident handled. Summary: {summary}")]}

def should_investigate(state: IncidentState) -> Literal["investigate", "close"]:
    if state["is_real"]:
        return "investigate"
    return "close"
 
 
def severity_gate(state: IncidentState) -> Literal["auto_approve", "human_review"]:
    # Low severity: auto-approve safe-only steps
    if state["severity"] in ("low", "medium"):
        return "auto_approve"
    return "human_review"
 
 
def auto_approve(state: IncidentState) -> dict:
    return {"approved": True}

from langgraph.checkpoint.memory import MemorySaver
 
workflow = StateGraph(IncidentState)
 
# Add nodes
workflow.add_node("triage", triage_agent)
workflow.add_node("rca", rca_agent)
workflow.add_node("remediation", remediation_agent)
workflow.add_node("human_review", human_review_node)
workflow.add_node("auto_approve", auto_approve)
workflow.add_node("execution", execution_agent)
workflow.add_node("notification", notification_agent)
 
# Entry point
workflow.set_entry_point("triage")
 
# Routing
workflow.add_conditional_edges(
    "triage",
    should_investigate,
    {"investigate": "rca", "close": END}
)
 
workflow.add_edge("rca", "remediation")
 
workflow.add_conditional_edges(
    "remediation",
    severity_gate,
    {"human_review": "human_review", "auto_approve": "auto_approve"}
)
 
workflow.add_edge("human_review", "execution")
workflow.add_edge("auto_approve", "execution")
workflow.add_edge("execution", "notification")
workflow.add_edge("notification", END)
 
# Compile with memory for state persistence
memory = MemorySaver()
graph = workflow.compile(checkpointer=memory)

alert_data = {
    "alert_name": "PodCrashLoopBackOff",
    "namespace": "production",
    "pod": "api-deployment-xyz-abc",
    "message": "Container has been restarting for 15 minutes",
    "started_at": "2026-07-20T15:00:00Z",
    "cluster": "prod-eks-cluster"
}
 
initial_state = {
    "alert": alert_data,
    "severity": "",
    "is_real": False,
    "root_cause": "",
    "remediation_plan": [],
    "approved": False,
    "executed_steps": [],
    "messages": []
}
 
config = {"configurable": {"thread_id": "incident-001"}}
result = graph.invoke(initial_state, config)
 
print("\nFinal State:")
print(f"Steps executed: {result['executed_steps']}")