# How to Create Custom Analyzers for Proprietary Log Formats in LogSentinelAI

> Learn to create custom analyzers for proprietary log formats in LogSentinelAI. Extend support by implementing Python modules and Pydantic schemas for structured log analysis.

- Repository: [JungJungIn/logsentinelai](https://github.com/call518/logsentinelai)
- Tags: how-to-guide
- Published: 2026-02-26

---

**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`](https://github.com/call518/logsentinelai/blob/main/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`](https://github.com/call518/logsentinelai/blob/main/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`](https://github.com/call518/logsentinelai/blob/main/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`](https://github.com/call518/logsentinelai/blob/main/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`](https://github.com/call518/logsentinelai/blob/main/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`](https://github.com/call518/logsentinelai/blob/main/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`](https://github.com/call518/logsentinelai/blob/main/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`](https://github.com/call518/logsentinelai/blob/main/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`](https://github.com/call518/logsentinelai/blob/main/src/logsentinelai/analyzers/myapp_log.py):

```python

# 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`](https://github.com/call518/logsentinelai/blob/main/src/logsentinelai/core/prompts.py):

```python
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`](https://github.com/call518/logsentinelai/blob/main/src/logsentinelai/analyzers/__init__.py):

```python

# 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`](https://github.com/call518/logsentinelai/blob/main/src/logsentinelai/cli.py):

```python
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:

```bash
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`](https://github.com/call518/logsentinelai/blob/main/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`](https://github.com/call518/logsentinelai/blob/main/src/logsentinelai/core/commons.py) to inherit SSH support, GeoIP enrichment, and error handling.
- **Registration**: Export classes in [`__init__.py`](https://github.com/call518/logsentinelai/blob/main/__init__.py) and optionally wire CLI commands in [`cli.py`](https://github.com/call518/logsentinelai/blob/main/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`](https://github.com/call518/logsentinelai/blob/main/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.