	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Generative AI > Google Gemini API > Log Anomaly Detection Agent

Log Anomaly Detection Agent

Author: Venkata Sudhakar

Application logs are the first line of defence for detecting production issues and security threats. ShopMax India's platform generates millions of log entries daily. An intelligent agent that summarises logs and surfaces anomalies - sudden error spikes, unusual login patterns, or API abuse - enables the ops team to act before customers are impacted.

This tutorial builds a Gemini ADK agent that analyses an hourly log summary, compares error rates and request volumes against baseline thresholds, and raises anomaly alerts with recommended investigation steps.

The below example shows a log anomaly detection agent for ShopMax India in a business context.

import asyncio
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Baseline thresholds (hourly averages)
BASELINE = {
    "error_rate_pct": 0.5,    # normal: <0.5% error rate
    "p99_latency_ms": 800,    # normal: p99 below 800ms
    "failed_logins_per_hr": 20,  # normal: <20 failed logins/hr
    "api_requests_per_hr": 50000,  # normal: ~50k req/hr
}

# Current hour log summary
CURRENT_HOUR = {
    "hour": "2026-04-08T09:00",
    "error_rate_pct": 4.2,
    "p99_latency_ms": 2100,
    "failed_logins_per_hr": 312,
    "api_requests_per_hr": 48500,
    "top_errors": [
        "DB connection timeout (892 occurrences)",
        "Payment API 503 (128 occurrences)",
    ],
}

def analyse_logs() -> dict:
    anomalies = []
    curr = CURRENT_HOUR
    b = BASELINE
    # Error rate check
    if curr["error_rate_pct"] > b["error_rate_pct"] * 3:
        anomalies.append({
            "type": "ERROR_SPIKE",
            "detail": f"Error rate {curr['error_rate_pct']}% vs baseline {b['error_rate_pct']}%",
            "severity": "HIGH",
            "action": "Check DB connections and downstream API health",
        })
    # Latency check
    if curr["p99_latency_ms"] > b["p99_latency_ms"] * 2:
        anomalies.append({
            "type": "LATENCY_SPIKE",
            "detail": f"p99 latency {curr['p99_latency_ms']}ms vs baseline {b['p99_latency_ms']}ms",
            "severity": "HIGH",
            "action": "Investigate slow queries or resource contention",
        })
    # Failed login check
    if curr["failed_logins_per_hr"] > b["failed_logins_per_hr"] * 5:
        anomalies.append({
            "type": "BRUTE_FORCE_RISK",
            "detail": f"{curr['failed_logins_per_hr']} failed logins vs baseline {b['failed_logins_per_hr']}",
            "severity": "CRITICAL",
            "action": "Enable rate limiting on login endpoint. Review IP blocklist.",
        })
    return {
        "hour": curr["hour"],
        "anomalies_found": len(anomalies),
        "anomalies": anomalies,
        "top_errors": curr["top_errors"],
        "status": "ACTION REQUIRED" if anomalies else "NORMAL",
    }

log_tool = FunctionTool(func=analyse_logs)

agent = LlmAgent(
    name="log_anomaly_agent",
    model="gemini-2.0-flash",
    description="Detects anomalies in ShopMax India application logs and recommends investigation steps.",
    instruction=(
        "You are a log analysis agent for ShopMax India. "
        "When asked to analyse logs, call analyse_logs. "
        "Present anomalies by severity (CRITICAL first), with clear recommended actions. "
        "State the overall system health status."
    ),
    tools=[log_tool],
)

async def main():
    session_service = InMemorySessionService()
    runner = Runner(agent=agent, app_name="log_app", session_service=session_service)
    session = await session_service.create_session(app_name="log_app", user_id="ops1")
    msg = types.Content(role="user", parts=[types.Part(
        text="Analyse the 9 AM log summary and flag any anomalies"
    )])
    async for event in runner.run_async(user_id="ops1", session_id=session.id, new_message=msg):
        if event.is_final_response() and event.content:
            print(event.content.parts[0].text)

asyncio.run(main())

It gives the following output,

Log Anomaly Report - ShopMax India | 09:00 AM

Status: ACTION REQUIRED | 3 Anomalies Detected

[CRITICAL] BRUTE_FORCE_RISK
  312 failed logins vs baseline 20/hr (15.6x spike)
  Action: Enable rate limiting on login endpoint. Review IP blocklist.

[HIGH] ERROR_SPIKE
  Error rate 4.2% vs baseline 0.5% (8.4x normal)
  Top errors: DB connection timeout (892), Payment API 503 (128)
  Action: Check DB connection pool and Payment API health immediately.

[HIGH] LATENCY_SPIKE
  p99 latency 2100ms vs baseline 800ms (2.6x normal)
  Action: Investigate slow queries or DB resource contention.

The DB connection timeout errors and latency spike are likely related.
Resolve DB health first - this may fix both HIGH anomalies simultaneously.

ShopMax India can run this agent on an automated schedule every 15 minutes, feeding it aggregated log metrics from CloudWatch, Datadog, or Elastic. When CRITICAL anomalies are detected, the agent triggers a PagerDuty alert and posts a summary to the #incidents Slack channel, ensuring the on-call engineer has context before they even open their laptop.

Send your comments, suggestions or queries regarding this site to [email protected].