Building a Log Analysis Agent with Pydantic AI: Complete Guide

Introduction to Log Analysis Agents

Modern applications generate enormous volumes of logs across distributed services, containers, and cloud infrastructure. Manually sifting through these logs to detect anomalies, diagnose failures, and surface actionable insights is tedious and error-prone. A log analysis agent is an AI-powered system that reads log entries, reasons about their meaning, classifies severity, identifies root causes, and produces structured recommendations.

Pydantic AI is a Python framework that brings type safety, structured outputs, and dependency injection to LLM-powered agents. By combining Pydantic AI with your log pipeline, you can build an agent that returns strongly typed analysis results you can directly integrate into dashboards, alerting systems, or incident response workflows.

Why Use an Agent Instead of Plain Regex or Rules?

• Contextual reasoning: Agents understand the semantic meaning of log lines, not just patterns.
• Adaptive classification: They handle novel error formats without rewriting rules.
• Structured outputs: Pydantic models guarantee the agent returns data your code can consume safely.
• Dependency injection: Pydantic AI lets you inject databases, APIs, or retrieval tools cleanly.

Prerequisites and Setup

Before building the agent, ensure you have Python 3.10 or later. Create a virtual environment and install the required packages:

python -m venv .venv
source .venv/bin/activate
pip install pydantic-ai pydantic python-dotenv

You will also need an API key for an LLM provider. Pydantic AI supports OpenAI, Anthropic, Gemini, Groq, Ollama, and others. For this tutorial we will use OpenAI, but the patterns apply to any provider.

# .env
OPENAI_API_KEY=sk-your-key-here

Load the environment variables in your entry script:

from dotenv import load_dotenv
load_dotenv()

Designing the Structured Output Models

The strength of Pydantic AI lies in forcing the LLM to produce outputs that conform to a schema. For a log analysis agent, we want the model to return a structured report containing severity, identified issues, root cause hypotheses, and recommended actions.

from pydantic import BaseModel, Field
from typing import Literal, Optional
from datetime import datetime


class LogIssue(BaseModel):
    """A single issue identified in the logs."""
    title: str = Field(description="Short descriptive title of the issue")
    severity: Literal["info", "warning", "error", "critical"] = Field(
        description="Severity level of the issue"
    )
    affected_service: str = Field(description="Name of the service or component affected")
    log_excerpt: str = Field(description="Relevant snippet from the logs")
    explanation: str = Field(description="Why this is considered an issue")


class RootCauseHypothesis(BaseModel):
    """A hypothesis about the underlying cause of observed issues."""
    hypothesis: str = Field(description="Description of the suspected root cause")
    confidence: float = Field(
        ge=0.0, le=1.0,
        description="Confidence score between 0 and 1"
    )
    supporting_evidence: list[str] = Field(
        description="Log evidence supporting this hypothesis"
    )


class RecommendedAction(BaseModel):
    """An actionable recommendation to resolve or investigate issues."""
    action: str = Field(description="The recommended action to take")
    priority: Literal["low", "medium", "high"] = Field(
        description="Priority of the action"
    )
    owner: Optional[str] = Field(
        default=None,
        description="Suggested team or role responsible"
    )


class LogAnalysisReport(BaseModel):
    """The complete structured analysis report returned by the agent."""
    summary: str = Field(description="High-level summary of log analysis")
    overall_health: Literal["healthy", "degraded", "critical"] = Field(
        description="Overall system health assessment"
    )
    issues: list[LogIssue] = Field(description="List of identified issues")
    root_cause_hypotheses: list[RootCauseHypothesis] = Field(
        description="Hypotheses about root causes"
    )
    recommended_actions: list[RecommendedAction] = Field(
        description="Recommended next steps"
    )
    analyzed_at: datetime = Field(
        default_factory=datetime.utcnow,
        description="Timestamp of analysis"
    )

These models serve as the contract between the LLM and your application. Any output that does not conform will be rejected and retried automatically by Pydantic AI.

Defining Dependencies

Real log analysis often requires access to external systems: a log database, a metrics API, or a knowledge base of past incidents. Pydantic AI uses a dataclass to define dependencies that are injected into agent tools.

from dataclasses import dataclass


@dataclass
class LogDeps:
    """Dependencies injected into the log analysis agent."""
    service_catalog: dict[str, str]
    known_incidents: list[dict]
    environment: str

The service_catalog maps service names to descriptions, helping the LLM understand what each component does. The known_incidents list provides historical context so the agent can correlate current logs with past outages.

Building the Agent

Now we create the agent itself. We define a system prompt that establishes the agent's role, and we register tools that the agent can call to enrich its analysis.

from pydantic_ai import Agent, RunContext

log_agent = Agent(
    model="openai:gpt-4o",
    deps_type=LogDeps,
    output_type=LogAnalysisReport,
    system_prompt=(
        "You are a senior Site Reliability Engineer specialized in log analysis. "
        "Given a batch of application logs, your job is to:\n"
        "1. Identify errors, warnings, and anomalies.\n"
        "2. Group related log lines into distinct issues.\n"
        "3. Formulate root cause hypotheses with supporting evidence.\n"
        "4. Recommend concrete, prioritized actions.\n"
        "Be precise. Quote exact log excerpts when citing evidence. "
        "If logs are healthy, return an empty issues list and set overall_health to 'healthy'."
    ),
)


@log_agent.tool
async def get_service_description(ctx: RunContext[LogDeps], service_name: str) -> str:
    """Look up a description for a given service name."""
    catalog = ctx.deps.service_catalog
    return catalog.get(
        service_name,
        f"Unknown service '{service_name}'. No description available."
    )


@log_agent.tool
async def find_similar_incidents(ctx: RunContext[LogDeps], keyword: str) -> list[dict]:
    """Search known past incidents for entries matching a keyword."""
    keyword_lower = keyword.lower()
    matches = [
        inc for inc in ctx.deps.known_incidents
        if keyword_lower in inc.get("summary", "").lower()
        or keyword_lower in inc.get("resolution", "").lower()
    ]
    return matches[:5]

Notice how each tool receives a RunContext typed with our LogDeps. This gives the tool type-safe access to injected dependencies. The agent decides when to call these tools based on the logs it receives.

Running the Agent on Real Logs

Let us assemble a sample log batch and run the agent. In production, you would pull these from Elasticsearch, Loki, CloudWatch, or a file.

import asyncio
import json


sample_logs = """
2024-11-15T10:23:01Z [INFO]  auth-service: User login successful for user_id=4821
2024-11-15T10:23:45Z [WARN]  payment-service: Retry attempt 2/3 for gateway timeout
2024-11-15T10:24:02Z [ERROR] payment-service: Gateway timeout after 30000ms, upstream=stripe-api
2024-11-15T10:24:03Z [ERROR] payment-service: Transaction failed for order_id=99321, reason=upstream_timeout
2024-11-15T10:24:10Z [WARN]  order-service: Order 99321 marked as PAYMENT_FAILED
2024-11-15T10:25:00Z [ERROR] notification-service: Failed to send email, SMTP connection refused
2024-11-15T10:25:12Z [INFO]  order-service: Compensation workflow started for order 99321
2024-11-15T10:26:30Z [CRITICAL] payment-service: Circuit breaker opened, all payment requests failing
2024-11-15T10:27:00Z [ERROR] order-service: 14 orders stuck in PENDING_PAYMENT state
"""

deps = LogDeps(
    service_catalog={
        "auth-service": "Handles user authentication and session management.",
        "payment-service": "Processes payments via external gateway providers.",
        "order-service": "Manages order lifecycle and state transitions.",
        "notification-service": "Sends emails, SMS, and push notifications.",
    },
    known_incidents=[
        {
            "summary": "Stripe API outage caused payment failures",
            "resolution": "Enabled failover to secondary gateway and notified Stripe support.",
            "date": "2024-08-03",
        },
        {
            "summary": "SMTP server unreachable due to firewall change",
            "resolution": "Reverted firewall rules and added monitoring for port 587.",
            "date": "2024-09-21",
        },
    ],
    environment="production",
)


async def analyze_logs(log_text: str, dependencies: LogDeps) -> LogAnalysisReport:
    prompt = (
        f"Analyze the following production logs from the '{dependencies.environment}' "
        f"environment. Use the available tools to look up service descriptions and "
        f"search for similar past incidents.\n\nLOGS:\n{log_text}"
    )
    result = await log_agent.run(prompt, deps=dependencies)
    return result.output


async def main():
    report = await analyze_logs(sample_logs, deps)
    print(json.dumps(report.model_dump(), indent=2, default=str))


if __name__ == "__main__":
    asyncio.run(main())

When you run this script, the agent will reason through the logs, potentially call get_service_description to understand the payment service, call find_similar_incidents to check for prior Stripe outages, and return a fully validated LogAnalysisReport.

Example Output Structure

{
  "summary": "Payment service is experiencing upstream timeouts to Stripe, triggering a circuit breaker and leaving 14 orders in a pending state. Notification service is also failing due to SMTP connectivity issues.",
  "overall_health": "critical",
  "issues": [
    {
      "title": "Stripe gateway timeout causing payment failures",
      "severity": "critical",
      "affected_service": "payment-service",
      "log_excerpt": "Gateway timeout after 30000ms, upstream=stripe-api",
      "explanation": "The payment service cannot reach Stripe, causing transactions to fail and the circuit breaker to open."
    },
    {
      "title": "SMTP connection refused in notification service",
      "severity": "error",
      "affected_service": "notification-service",
      "log_excerpt": "Failed to send email, SMTP connection refused",
      "explanation": "The notification service cannot connect to the SMTP server, preventing email delivery."
    }
  ],
  "root_cause_hypotheses": [
    {
      "hypothesis": "Stripe API is experiencing an outage or network partition between the cluster and Stripe.",
      "confidence": 0.85,
      "supporting_evidence": [
        "Gateway timeout after 30000ms, upstream=stripe-api",
        "Circuit breaker opened, all payment requests failing",
        "Similar incident on 2024-08-03 resolved by enabling failover gateway"
      ]
    }
  ],
  "recommended_actions": [
    {
      "action": "Enable failover to secondary payment gateway immediately.",
      "priority": "high",
      "owner": "payments-team"
    },
    {
      "action": "Investigate SMTP server connectivity and firewall rules on port 587.",
      "priority": "high",
      "owner": "infra-team"
    },
    {
      "action": "Run compensation workflow for the 14 stuck orders once payments are restored.",
      "priority": "medium",
      "owner": "orders-team"
    }
  ],
  "analyzed_at": "2024-11-15T10:30:00Z"
}

Integrating with a Live Log Stream

To make the agent useful in production, you need to feed it logs continuously. A common pattern is to batch logs over a sliding time window and run analysis periodically.

import time
from collections import deque


class LogBatcher:
    """Collects logs and emits batches for analysis."""

    def __init__(self, window_seconds: int = 60, max_batch_size: int = 200):
        self.window_seconds = window_seconds
        self.max_batch_size = max_batch_size
        self.buffer: deque = deque()
        self.last_flush = time.monotonic()

    def add(self, log_line: str) -> None:
        self.buffer.append(log_line)
        if len(self.buffer) >= self.max_batch_size:
            self.flush()

    def should_flush(self) -> bool:
        elapsed = time.monotonic() - self.last_flush
        return elapsed >= self.window_seconds and len(self.buffer) > 0

    def flush(self) -> str:
        if not self.buffer:
            return ""
        batch = "\n".join(self.buffer)
        self.buffer.clear()
        self.last_flush = time.monotonic()
        return batch


async def monitoring_loop(batcher: LogBatcher, dependencies: LogDeps):
    while True:
        await asyncio.sleep(5)
        if batcher.should_flush():
            batch = batcher.flush()
            report = await analyze_logs(batch, dependencies)
            if report.overall_health != "healthy":
                await send_alert(report)


async def send_alert(report: LogAnalysisReport):
    """Forward critical reports to an alerting channel."""
    print(f"[ALERT] Health={report.overall_health}: {report.summary}")
    for action in report.recommended_actions:
        if action.priority == "high":
            print(f"  -> {action.action} (owner: {action.owner})")

This pattern lets you plug the agent into any log ingestion pipeline. Whether you read from a Kafka topic, a Redis stream, or tail a file, you simply push lines into the batcher and let the agent handle the rest.

Adding a Dynamic System Prompt

Sometimes the system prompt needs to incorporate runtime information, such as the current environment or the time window being analyzed. Pydantic AI supports dynamic system prompts via decorated functions.

@log_agent.system_prompt
async def add_context(ctx: RunContext[LogDeps]) -> str:
    return (
        f"You are analyzing logs from the '{ctx.deps.environment}' environment. "
        f"The current time is {datetime.utcnow().isoformat()}. "
        f"Available services: {', '.join(ctx.deps.service_catalog.keys())}. "
        f"Prioritize issues that could cause customer-facing impact."
    )

Dynamic prompts are evaluated on every run, ensuring the agent always has fresh context.

Best Practices

• Keep log batches bounded. Sending thousands of lines in a single prompt wastes tokens and degrades reasoning quality. Aim for 50 to 300 lines per batch, focusing on errors and warnings.
• Pre-filter when possible. Use traditional log tools to pre-filter by severity or service before passing logs to the agent. This reduces cost and latency.
• Use retries wisely. Pydantic AI retries automatically when output validation fails, but set a sensible retries limit on the agent to avoid runaway costs.
• Cache service descriptions. If your service catalog rarely changes, consider embedding it directly in the system prompt instead of exposing it as a tool to reduce round trips.
• Log the agent's reasoning. Use the result.all_messages() method to inspect tool calls and intermediate steps for debugging.
• Validate confidence scores. Treat LLM confidence scores as rough signals, not precise probabilities. Use them for ranking, not for automated decision-making without human review.
• Test with golden log sets. Maintain a curated set of known log scenarios with expected analyses. Run the agent against these in CI to catch regressions.
• Handle rate limits. Wrap agent calls with exponential backoff and circuit breakers so a slow LLM provider does not stall your monitoring pipeline.

Extending the Agent

Once the core agent is working, you can extend it in several directions. You might add a tool that queries a metrics database like Prometheus to correlate log spikes with CPU or memory anomalies. You could add a retrieval tool that searches a vector database of runbooks, letting the agent cite specific remediation steps. You could also chain multiple agents: one for triage, one for root cause analysis, and one for drafting incident reports.

Here is an example of adding a metrics correlation tool:

@log_agent.tool
async def query_metrics(
    ctx: RunContext[LogDeps],
    service_name: str,
    metric_name: str
) -> dict:
    """Query recent metric values for a service. Simulated here."""
    # In production, call Prometheus, Datadog, or CloudWatch APIs.
    mock_data = {
        "payment-service": {
            "error_rate": 0.92,
            "p99_latency_ms": 31000,
            "cpu_usage": 0.45,
        },
        "notification-service": {
            "error_rate": 0.78,
            "p99_latency_ms": 1200,
            "cpu_usage": 0.30,
        },
    }
    return mock_data.get(service_name, {}).get(metric_name, "metric not found")

With this tool available, the agent can verify whether a log-reported error correlates with a spike in error rate or latency, strengthening its root cause hypotheses with quantitative evidence.

Conclusion

Building a log analysis agent with Pydantic AI gives you a powerful, type-safe way to turn raw log data into structured, actionable intelligence. By defining clear output models, injecting real-world dependencies through tools, and feeding the agent bounded log batches, you create a system that reasons about failures much like a seasoned SRE would. The structured outputs integrate cleanly with alerting pipelines, incident management platforms, and dashboards, bridging the gap between unstructured log text and engineering action. Start with the core agent described here, iterate on your prompts and tools against real production logs, and gradually extend it with metrics correlation, runbook retrieval, and multi-agent orchestration to build a robust automated observability assistant.