How to Create Custom Analyzers for Proprietary Log Formats in LogSentinelAI

You can extend LogSentinelAI to support any proprietary log format by implementing a Python module in src/logsentinelai/analyzers/ that defines Pydantic schemas for structured output, registers a prompt template in core.prompts, and delegates execution to the generic analysis drivers in core.commons.

LogSentinelAI is built around a plug-in architecture where each log type is handled by an independent analyzer module. This design allows security teams to integrate proprietary or internal log formats without modifying the core codebase. By following the established patterns in the call518/logsentinelai repository, you inherit the complete LLM processing pipeline including GeoIP enrichment, JSON schema validation, and Elasticsearch ingestion.

Understanding the Analyzer Architecture

Every analyzer follows a standardized execution flow implemented in src/logsentinelai/core/commons.py. This generic pipeline handles the heavy lifting so you only need to supply the data model and prompt logic.

1. Argument parsing via create_argument_parser for standardized CLI interfaces.
2. SSH handling through handle_ssh_arguments for remote log access.
3. Driver selection between run_generic_batch_analysis and run_generic_realtime_analysis.
4. Prompt generation using Jinja-style templates from src/logsentinelai/core/prompts.py populated with log chunks and Pydantic schemas.
5. LLM processing via process_log_chunk, which validates responses against your schema, enriches data with GeoIP, and ships results to Elasticsearch.

Concrete implementations reside in src/logsentinelai/analyzers/. The general_log.py file demonstrates the canonical structure that all custom analyzers should replicate.

Step-by-Step Implementation Guide

Step 1: Create the Analyzer Module

Create a new Python file under src/logsentinelai/analyzers/<mylog>.py. This module must import the core execution helpers and define the entry point function main().

Step 2: Define Pydantic Data Models

Use Pydantic models to enforce the JSON structure the LLM must return. Define enums for SeverityLevel and EventCategory, then create event and analysis classes that inherit from BaseModel. The generic drivers call analysis_schema_class.model_json_schema() to inject the schema into prompts and validate LLM outputs.

Step 3: Add a Prompt Template

In src/logsentinelai/core/prompts.py, add a constant PROMPT_TEMPLATE_<MYLOG> and a getter function get_<mylog>_prompt(). The template must include the {model_schema} placeholder, which the drivers replace at runtime with your Pydantic schema.

Step 4: Implement the Driver Logic

The main() function orchestrates the workflow using helpers from src/logsentinelai/core/commons.py:

• Build the argument parser using create_argument_parser.
• Resolve SSH configuration via handle_ssh_arguments.
• Select batch or real-time mode based on CLI arguments.
• Invoke run_generic_batch_analysis or run_generic_realtime_analysis with your schema class and prompt getter.

Step 5: Register the Analyzer

Export your analyzer in src/logsentinelai/analyzers/__init__.py by importing the analysis class and adding it to the __all__ list. This makes the module discoverable to the rest of the package.

Step 6: Wire the CLI Sub-Command

To make the analyzer accessible via the unified logsentinelai command, extend src/logsentinelai/cli.py. Add a new sub-parser that forwards arguments to your analyzer's main() function.

Step 7: Configure Defaults (Optional)

For convenience, add default log paths and chunk sizes to LOG_PATHS and LOG_CHUNK_SIZES in src/logsentinelai/core/config.py. The generic drivers use get_analysis_config() to resolve these settings.

Complete Working Example: MyApp Log Analyzer

The following implementation demonstrates a production-ready analyzer for a hypothetical proprietary "MyApp" format. Save this as src/logsentinelai/analyzers/myapp_log.py:

# src/logsentinelai/analyzers/myapp_log.py

from pydantic import BaseModel, Field
from enum import Enum
from typing import Optional, List

# Prompt import

from ..core.prompts import get_myapp_log_prompt

# Core helpers

from ..core.commons import (
    run_generic_batch_analysis,
    run_generic_realtime_analysis,
    create_argument_parser,
    handle_ssh_arguments,
)

# ── Severity & Event definitions ────────────────────────────────────────

class SeverityLevel(str, Enum):
    CRITICAL = "CRITICAL"
    HIGH = "HIGH"
    MEDIUM = "MEDIUM"
    LOW = "LOW"
    INFO = "INFO"


class EventCategory(str, Enum):
    AUTH = "AUTH"
    DATA = "DATA"
    SYSTEM = "SYSTEM"
    UNKNOWN = "UNKNOWN"


class MyAppEvent(BaseModel):
    category: EventCategory
    severity: SeverityLevel
    related_logs: List[str] = Field(
        min_length=1,
        description="Exact log lines that triggered this event"
    )
    description: str
    confidence_score: float = Field(ge=0.0, le=1.0)
    source_ips: List[str] = Field(description="All source IPs in the chunk")
    recommended_actions: List[str]
    requires_human_review: bool


class MyAppAnalysis(BaseModel):
    summary: str
    events: List[MyAppEvent] = Field(
        min_length=1,
        description="At least one event must be present"
    )
    # Optional statistics block

    total_events: int = Field(description="Total events detected")
    unique_ips: int = Field(description="Number of distinct source IPs")


def main() -> int:
    parser = create_argument_parser("MyApp Log Analysis")
    args = parser.parse_args()

    # SSH handling (optional)

    ssh_cfg = handle_ssh_arguments(args)
    remote_mode = "ssh" if ssh_cfg else "local"

    log_type = "myapp_log"
    analysis_title = "MyApp Log Analysis"

    if args.mode == "realtime":
        run_generic_realtime_analysis(
            log_type=log_type,
            analysis_schema_class=MyAppAnalysis,
            prompt_template=get_myapp_log_prompt(),
            analysis_title=analysis_title,
            chunk_size=args.chunk_size,
            log_path=args.log_path,
            only_sampling_mode=args.only_sampling_mode,
            sampling_threshold=args.sampling_threshold,
            remote_mode=remote_mode,
            ssh_config=ssh_cfg,
        )
    else:
        run_generic_batch_analysis(
            log_type=log_type,
            analysis_schema_class=MyAppAnalysis,
            prompt_template=get_myapp_log_prompt(),
            analysis_title=analysis_title,
            log_path=args.log_path,
            remote_mode=remote_mode,
            ssh_config=ssh_cfg,
        )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Add the corresponding prompt template to src/logsentinelai/core/prompts.py:

PROMPT_TEMPLATE_MYAPP_LOG = """
Expert analyst for the proprietary MyApp log format. Detect security events,
operational anomalies and business‑logic incidents.

LOG PARSING CRITICAL:
- Identify individual log entries (they may span multiple lines).
- Extract timestamps, user identifiers, IP addresses, and any structured key‑value pairs.

SEVERITY:
- CRITICAL: Data exfiltration, privilege escalation, authentication bypass.
- HIGH: Repeated failed logins, suspicious API calls.
- MEDIUM: Unusual configuration changes.
- LOW: Minor warnings.
- INFO: Normal activity (e.g., scheduled jobs, health checks).

MANDATORY EVENT RULES:
- Never leave the `events` array empty.
- Always include a complete `source_ips` list.
- If no security concerns, create an INFO event summarising traffic.

STATISTICS:
- total_events, unique_ips

LANGUAGE: {response_language}

JSON schema: {model_schema}

<LOGS BEGIN>
{logs}
<LOGS END>
"""

def get_myapp_log_prompt() -> str:
    """Return the MyApp log analysis prompt."""
    return PROMPT_TEMPLATE_MYAPP_LOG

Register the analyzer in src/logsentinelai/analyzers/__init__.py:

# src/logsentinelai/analyzers/__init__.py

from .myapp_log import LogAnalysis as MyAppLogAnalysis

__all__ = [
    "HTTPDAccessAnalysis",
    "HTTPDServerAnalysis",
    "LinuxSystemAnalysis",
    "MyAppLogAnalysis",
]

Finally, expose it via the CLI in src/logsentinelai/cli.py:

myapp_parser = subparsers.add_parser(
    "myapp-log",
    help="Analyze proprietary MyApp logs"
)
myapp_parser.add_argument("--log-path", help="Path to MyApp log file")
myapp_parser.add_argument(
    "--mode",
    choices=["batch", "realtime"],
    default="batch",
    help="Analysis mode (default: batch)",
)

# Later in the command router:

elif args.command == "myapp-log":
    from .analyzers.myapp_log import main as myapp_main
    sys.argv = ["logsentinelai-myapp-log"]
    if args.log_path:
        sys.argv.extend(["--log-path", args.log_path])
    if args.mode:
        sys.argv.extend(["--mode", args.mode])
    myapp_main()

Testing Your Custom Analyzer

Validate the implementation by running the analyzer against sample data:

python -m logsentinelai.analyzers.myapp_log --log-path /var/log/myapp/test.log --mode batch

Verify that the output produces valid JSON matching your Pydantic schema and that documents appear in Elasticsearch with the correct enrichment fields.

Summary

• Plug-in architecture: LogSentinelAI isolates log-type logic in src/logsentinelai/analyzers/, allowing zero-core-code modifications for new formats.
• Schema-first design: Define your output structure using Pydantic models; the generic drivers handle LLM validation and indexing automatically.
• Centralized prompts: Store Jinja-style templates in src/logsentinelai/core/prompts.py with the mandatory {model_schema} placeholder.
• Generic drivers: Delegate execution to run_generic_batch_analysis or run_generic_realtime_analysis in src/logsentinelai/core/commons.py to inherit SSH support, GeoIP enrichment, and error handling.
• Registration: Export classes in __init__.py and optionally wire CLI commands in cli.py for unified access.

Frequently Asked Questions

What Python version does LogSentinelAI require for custom analyzers?

LogSentinelAI requires Python 3.9 or higher to support the Pydantic v2 features and type hints used throughout the analyzer modules. Ensure your virtual environment matches the version specified in the repository's pyproject.toml.

Can I use an external API instead of the built-in LLM client?

Yes. While the generic drivers in core.commons use the internal LLM client configured via environment variables, you can override the process_log_chunk call within your analyzer's main() function. However, you must still return data matching your Pydantic schema to maintain compatibility with the Elasticsearch export pipeline.

How do I handle multi-line log entries in proprietary formats?

Define explicit parsing logic in your prompt template's "LOG PARSING CRITICAL" section. Instruct the LLM to identify continuation patterns (e.g., indented lines, timestamps, or stack traces). The analyzer feeds raw text chunks to the LLM; the prompt engineering determines how the model boundaries individual entries within those chunks.

Is real-time mode supported for all custom analyzers?

Yes. Any analyzer implementing the standard main() structure with argument parsing can invoke run_generic_realtime_analysis instead of the batch driver. The real-time driver uses the same schema validation and prompt generation, but streams chunks continuously from the log file tail rather than processing the entire file at once.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}