Building a Log Analysis Agent with LangGraph: Complete Guide

Building a Log Analysis Agent with LangGraph: Complete Guide

Modern applications generate enormous volumes of log data every minute. Manually sifting through thousands of log lines to find anomalies, root causes, or security incidents is tedious and error-prone. By combining the orchestration power of LangGraph with the reasoning capabilities of large language models, you can build an autonomous agent that ingests logs, classifies them, investigates anomalies, and produces actionable reports. This guide walks you through the entire process from concept to a working implementation.

What Is a Log Analysis Agent?

A log analysis agent is an LLM-powered system that autonomously processes log streams or log files. Rather than running fixed regex rules, the agent reasons about the content of each log entry, decides whether it represents a normal event, a warning, or a critical failure, and can take follow-up actions such as querying related metrics, summarizing incidents, or escalating alerts. LangGraph provides the graph-based execution framework that lets you model these steps as nodes and edges, giving you fine-grained control over the agent's workflow.

Why LangGraph for Log Analysis?

• Stateful workflows: Logs often require multi-step investigation. LangGraph's shared state object lets each node pass context to the next.
• Conditional routing: Different log severities warrant different handling paths. LangGraph supports conditional edges natively.
• Looping and re-evaluation: The agent can re-inspect logs or request additional context when initial analysis is inconclusive.
• Human-in-the-loop: Critical incidents can pause execution and wait for operator confirmation before escalating.
• Persistence: Checkpointing allows long-running analysis jobs to resume after interruptions.

Prerequisites and Setup

Before building the agent, install the required packages and ensure you have an OpenAI API key (or another LLM provider) configured in your environment.

pip install langgraph langchain langchain-openai python-dotenv

Create a .env file with your API key:

OPENAI_API_KEY=sk-your-key-here

Designing the Agent Workflow

Our log analysis agent will follow a structured pipeline. Each stage is a node in the LangGraph, and the state flows between them. The workflow consists of the following stages:

• Ingest: Load raw log entries into state.
• Classify: Use an LLM to label each log entry with a severity level.
• Route: Send critical or error-level logs to the investigation node; send info-level logs directly to summarization.
• Investigate: For anomalies, the agent reasons about possible root causes and correlates related entries.
• Summarize: Produce a final human-readable report.

Defining the State

LangGraph uses a typed state object that every node reads from and writes to. For our log agent, the state holds the raw logs, classified results, investigation findings, and the final summary.

from typing import TypedDict, List, Optional, Annotated
from langgraph.graph import MessagesState

class LogEntry(TypedDict):
    timestamp: str
    source: str
    message: str
    severity: Optional[str]
    analysis: Optional[str]

class LogAnalysisState(TypedDict):
    raw_logs: List[str]
    parsed_logs: List[LogEntry]
    critical_entries: List[LogEntry]
    investigation_notes: List[str]
    summary: Optional[str]
    iterations: int

Building the Nodes

Each node is a plain Python function that accepts the state and returns a partial state update. Let us start with the ingestion node, which converts raw log strings into structured entries.

import re
from datetime import datetime

def ingest_node(state: LogAnalysisState) -> dict:
    parsed = []
    pattern = re.compile(
        r'(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+'
        r'(?P<source>\w+)\s+(?P<message>.*)'
    )
    for line in state["raw_logs"]:
        match = pattern.match(line.strip())
        if match:
            parsed.append({
                "timestamp": match.group("timestamp"),
                "source": match.group("source"),
                "message": match.group("message"),
                "severity": None,
                "analysis": None,
            })
        else:
            parsed.append({
                "timestamp": "unknown",
                "source": "unknown",
                "message": line.strip(),
                "severity": None,
                "analysis": None,
            })
    return {"parsed_logs": parsed, "iterations": 0}

Next, the classification node uses an LLM to assign a severity level to each log entry. We batch the entries to reduce token usage and latency.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def classify_node(state: LogAnalysisState) -> dict:
    logs = state["parsed_logs"]
    formatted = "\n".join(
        f"{i}: [{e['timestamp']}] {e['source']}: {e['message']}"
        for i, e in enumerate(logs)
    )

    prompt = f"""You are a log classification engine. For each log entry below,
assign one of these severity levels: INFO, WARN, ERROR, CRITICAL.
Return ONLY a JSON array of objects with keys "index" and "severity".

Logs:
{formatted}
"""
    response = llm.invoke(prompt)
    import json
    try:
        results = json.loads(response.content)
    except json.JSONDecodeError:
        results = []

    severity_map = {r["index"]: r["severity"] for r in results if "index" in r}
    for i, entry in enumerate(logs):
        entry["severity"] = severity_map.get(i, "INFO")

    return {"parsed_logs": logs}

The routing logic determines which entries need deeper investigation. We implement this as a conditional edge function rather than a node.

def route_after_classify(state: LogAnalysisState) -> str:
    critical = [
        e for e in state["parsed_logs"]
        if e["severity"] in ("ERROR", "CRITICAL")
    ]
    if critical:
        return "investigate"
    return "summarize"

The investigation node asks the LLM to reason about root causes and correlate related log entries. This is where the agent adds the most value beyond simple rule-based filtering.

def investigate_node(state: LogAnalysisState) -> dict:
    critical = [
        e for e in state["parsed_logs"]
        if e["severity"] in ("ERROR", "CRITICAL")
    ]

    formatted = "\n".join(
        f"- [{e['timestamp']}] {e['source']}: {e['message']}"
        for e in critical
    )

    prompt = f"""You are a senior site reliability engineer investigating
production incidents. Analyze the following error and critical log entries.
For each, identify:
1. Likely root cause
2. Related log entries that may be correlated
3. Recommended remediation action

Error logs:
{formatted}
"""
    response = llm.invoke(prompt)
    notes = response.content

    for entry in critical:
        entry["analysis"] = notes

    return {
        "critical_entries": critical,
        "investigation_notes": [notes],
        "iterations": state["iterations"] + 1,
    }

Finally, the summarize node produces a concise report that an on-call engineer can read in seconds.

def summarize_node(state: LogAnalysisState) -> dict:
    total = len(state["parsed_logs"])
    by_severity = {}
    for e in state["parsed_logs"]:
        by_severity[e["severity"]] = by_severity.get(e["severity"], 0) + 1

    notes = "\n".join(state.get("investigation_notes", []))

    prompt = f"""Write a concise incident report based on the following log analysis.

Total log entries: {total}
Breakdown by severity: {by_severity}

Investigation notes:
{notes if notes else "No critical issues detected."}

Format the report with sections: Overview, Key Findings, Recommendations.
"""
    response = llm.invoke(prompt)
    return {"summary": response.content}

Assembling the Graph

With all nodes defined, we now wire them together using LangGraph's StateGraph API. The graph starts at ingestion, moves to classification, then conditionally routes to investigation or directly to summarization, and finally ends.

from langgraph.graph import StateGraph, START, END

graph_builder = StateGraph(LogAnalysisState)

graph_builder.add_node("ingest", ingest_node)
graph_builder.add_node("classify", classify_node)
graph_builder.add_node("investigate", investigate_node)
graph_builder.add_node("summarize", summarize_node)

graph_builder.add_edge(START, "ingest")
graph_builder.add_edge("ingest", "classify")
graph_builder.add_conditional_edges(
    "classify",
    route_after_classify,
    {
        "investigate": "investigate",
        "summarize": "summarize",
    },
)
graph_builder.add_edge("investigate", "summarize")
graph_builder.add_edge("summarize", END)

log_agent = graph_builder.compile()

Running the Agent

To execute the agent, prepare a list of raw log strings and invoke the compiled graph. The graph returns the final state, including the generated summary.

sample_logs = [
    "2024-01-15 10:23:01 app User login successful for user_id=42",
    "2024-01-15 10:23:05 app Database query took 3400ms on table users",
    "2024-01-15 10:23:06 db Connection pool exhausted, rejecting new connections",
    "2024-01-15 10:23:07 app Failed to process request: timeout connecting to db",
    "2024-01-15 10:23:10 app Retrying request (attempt 2/3)",
    "2024-01-15 10:23:12 app Request failed permanently after 3 retries",
    "2024-01-15 10:23:15 monitor Health check endpoint returning 503",
]

result = log_agent.invoke({"raw_logs": sample_logs})

print(result["summary"])

When you run this, the agent will classify the database connection pool exhaustion and timeout entries as ERROR or CRITICAL, route them to the investigation node, reason about the root cause (likely a slow query saturating the pool), and produce a structured report with recommendations such as increasing pool size or optimizing the slow query.

Adding a Self-Correction Loop

One powerful feature of LangGraph is the ability to loop. Suppose the investigation is inconclusive and the agent should retry with more context. We can add a conditional edge after investigation that checks whether the analysis is sufficient.

def should_reinvestigate(state: LogAnalysisState) -> str:
    if state["iterations"] >= 2:
        return "summarize"
    notes = state.get("investigation_notes", [""])[-1].lower()
    if "inconclusive" in notes or "insufficient" in notes:
        return "investigate"
    return "summarize"

# Rebuild the graph with the loop
graph_builder2 = StateGraph(LogAnalysisState)
graph_builder2.add_node("ingest", ingest_node)
graph_builder2.add_node("classify", classify_node)
graph_builder2.add_node("investigate", investigate_node)
graph_builder2.add_node("summarize", summarize_node)

graph_builder2.add_edge(START, "ingest")
graph_builder2.add_edge("ingest", "classify")
graph_builder2.add_conditional_edges(
    "classify",
    route_after_classify,
    {"investigate": "investigate", "summarize": "summarize"},
)
graph_builder2.add_conditional_edges(
    "investigate",
    should_reinvestigate,
    {"investigate": "investigate", "summarize": "summarize"},
)
graph_builder2.add_edge("summarize", END)

log_agent_v2 = graph_builder2.compile()

This loop ensures the agent can refine its analysis when initial findings are incomplete, while the iteration counter prevents infinite cycles.

Integrating Human-in-the-Loop

For critical incidents, you may want a human operator to review the investigation before the report is finalized. LangGraph supports this through its interrupt mechanism. Compile the graph with a checkpointer and interrupt before the summarize node.

from langgraph.checkpoint.memory import MemorySaver

graph_builder3 = StateGraph(LogAnalysisState)
graph_builder3.add_node("ingest", ingest_node)
graph_builder3.add_node("classify", classify_node)
graph_builder3.add_node("investigate", investigate_node)
graph_builder3.add_node("summarize", summarize_node)

graph_builder3.add_edge(START, "ingest")
graph_builder3.add_edge("ingest", "classify")
graph_builder3.add_conditional_edges(
    "classify",
    route_after_classify,
    {"investigate": "investigate", "summarize": "summarize"},
)
graph_builder3.add_edge("investigate", "summarize")
graph_builder3.add_edge("summarize", END)

checkpointer = MemorySaver()
log_agent_v3 = graph_builder3.compile(
    checkpointer=checkpointer,
    interrupt_before=["summarize"],
)

config = {"configurable": {"thread_id": "thread-1"}}
result = log_agent_v3.invoke({"raw_logs": sample_logs}, config=config)

# At this point execution pauses before summarize.
# A human can inspect the state and optionally modify it.
current_state = log_agent_v3.get_state(config)
print("Paused for review. Critical entries found:")
for entry in current_state.values.get("critical_entries", []):
    print(f"  [{entry['severity']}] {entry['message']}")

# Resume execution
final_result = log_agent_v3.invoke(None, config=config)
print(final_result["summary"])

Best Practices

• Batch log entries: Sending hundreds of individual LLM calls is slow and expensive. Group logs into batches of 20 to 50 entries per classification call.
• Use structured output: Prefer models and prompts that return JSON. This makes parsing reliable and reduces post-processing errors.
• Limit context windows: For large log files, pre-filter using cheap heuristics like keyword matching before sending entries to the LLM.
• Cache classification results: Repeated log patterns do not need reclassification. Use a hash of the message as a cache key.
• Set iteration caps: Always include a maximum iteration counter on loops to prevent runaway execution costs.
• Log the agent's own actions: Record which nodes executed and how long each took. This helps with debugging and cost tracking.
• Validate LLM output: Never trust LLM output blindly. Validate severity labels against an allowed set and handle malformed JSON gracefully.
• Use streaming for real-time logs: For live log streams, consider using LangGraph's streaming mode to process entries as they arrive rather than batching everything upfront.

Extending the Agent

The basic agent can be extended in several directions. You can add a tool-calling node that lets the agent query external systems such as Prometheus, Grafana, or a ticketing system. You can add a deduplication node that groups similar log entries together before classification. You can also integrate vector search to compare current errors against historical incident databases, giving the agent memory of past outages.

For production deployments, wrap the compiled graph in a FastAPI service that accepts log payloads via HTTP, processes them asynchronously, and returns reports through a webhook or WebSocket connection. Pair this with a message queue like Redis or Kafka to handle high-throughput log streams.

Conclusion

Building a log analysis agent with LangGraph gives you a flexible, stateful, and extensible framework for turning raw log data into actionable insight. By modeling the workflow as a graph of focused nodes, you gain precise control over how logs are classified, investigated, and reported, while the LLM handles the reasoning that traditional rule-based systems cannot. Start with the basic pipeline shown here, then iteratively add self-correction loops, human review checkpoints, and tool integrations as your operational needs grow. With careful attention to batching, output validation, and iteration limits, this pattern scales from a development prototype to a production-grade observability assistant.