DEV Community: Ayi NEDJIMI

Building a production-ready AI chatbot with memory and context

Ayi NEDJIMI — Sun, 20 Sep 2026 10:01:58 +0000

Most chatbot tutorials end after the first response. Production chatbots need to remember what was said three turns ago, survive restarts, and stay within token budgets — that's where the real work is. Skip that layer and you'll be rewriting core logic under pressure at 2am.

The statefulness problem

A language model has no memory between API calls. Every request starts from a blank slate. If you want the bot to remember the user's name from turn one, you have to send that history yourself in every subsequent request.

That works fine in demos. In production, you'll hit the model's context limit fast — and you'll pay for every token you resend. The solution is a memory layer between your application and the model: it stores conversation history, decides what to include in each request, and handles gracefully the case where history grows too large.

Three components make this work:

Session store — persists message history with a TTL (Redis is a natural fit)
Context builder — selects which messages fit within the token budget
API wrapper — calls the model and appends responses to history

Implementing the session store

I'll use Redis because it handles TTL natively, which matters for chatbot sessions: you want idle conversations to expire automatically rather than accumulating forever. tiktoken handles token counting server-side.

import json
import redis
import tiktoken

r = redis.Redis(host="localhost", port=6379, db=0)
enc = tiktoken.get_encoding("cl100k_base")

SESSION_TTL = 1800  # 30 minutes

def load_history(session_id: str) -> list[dict]:
    raw = r.get(f"chat:{session_id}")
    if not raw:
        return []
    return json.loads(raw)

def save_history(session_id: str, messages: list[dict]) -> None:
    r.set(f"chat:{session_id}", json.dumps(messages), ex=SESSION_TTL)

def count_tokens(messages: list[dict]) -> int:
    total = 0
    for msg in messages:
        total += 4  # per-message overhead (role, separators)
        total += len(enc.encode(msg.get("content", "")))
    return total + 2  # reply priming tokens

One detail worth calling out: the +4 per message and +2 at the end are not arbitrary. They match the actual overhead the API charges for message formatting. If you skip this, your token budget will drift and you'll occasionally hit limit errors on the next turn.

Building the context window manager

This is where most tutorials skip ahead and production systems break. If you blindly append messages to each request, you'll eventually exceed the model's context limit and surface an error at the worst possible moment.

The strategy here: keep a hard token budget, always include the system prompt, and trim from the oldest end first. The most recent turns are almost always more relevant than what was said ten exchanges ago.

def build_context(
    system_prompt: str,
    history: list[dict],
    token_budget: int = 3800,
) -> list[dict]:
    system = [{"role": "system", "content": system_prompt}]
    system_tokens = count_tokens(system)
    budget = token_budget - system_tokens

    selected = []
    for msg in reversed(history):
        msg_tokens = count_tokens([msg])
        if budget - msg_tokens < 0:
            break
        selected.insert(0, msg)
        budget -= msg_tokens

    return system + selected

Set token_budget to about 80% of the model's context limit, leaving headroom for the response. For a model with a 16k context, token_budget=12800 with max_tokens=2048 on the response leaves a safe margin.

One edge case to plan for: if a user pastes a large block of text in a single message, that message alone may exceed the budget. The loop above skips it cleanly, but the user's input will be silently dropped. Add a pre-check that rejects user messages over a configurable character limit and returns a friendly error.

Wiring it together

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["LLM_API_KEY"])

SYSTEM_PROMPT = """You are a support assistant for a B2B SaaS product.
Answer questions accurately and concisely. If you don't know the answer, say so."""

def chat(session_id: str, user_message: str) -> str:
    history = load_history(session_id)
    history.append({"role": "user", "content": user_message})

    context = build_context(SYSTEM_PROMPT, history)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=context,
        temperature=0.2,
        max_tokens=512,
    )

    assistant_reply = response.choices[0].message.content
    history.append({"role": "assistant", "content": assistant_reply})
    save_history(session_id, history)

    return assistant_reply

Usage:

sid = "user-42-session-001"
print(chat(sid, "What are your pricing tiers?"))
print(chat(sid, "Which plan includes SSO?"))  # second call sees the first exchange

The second call automatically includes the first exchange. If the user returns after the 30-minute TTL, the session is expired and the conversation starts fresh — no stale context carrying over.

What you'll need before going live

The code above handles the core logic. Before exposing it to real users, a few additions are non-negotiable:

Idempotency: network failures cause clients to retry. If you call the model twice for the same user message, you'll charge twice and return inconsistent replies. Store a request_id (generated client-side) with each exchange and return the cached reply if you've seen that ID before.

Cost tracking per session: log response.usage.prompt_tokens and response.usage.completion_tokens on every call. In practice, you'll find a small fraction of users consuming a disproportionate share of context — usually because they paste large documents mid-conversation. Knowing this before your first billing surprise is useful.

Summarization fallback: when history exceeds the token budget and you're trimming, you lose early context. For most support bots this is fine. For assistants that need to remember facts across long conversations — the user's name, their plan tier, decisions made in turn one — consider a separate summarization step: when history crosses a threshold, summarize the oldest N turns into a single context block and store it alongside the raw history.

Security: chatbot APIs are a common target for prompt injection. A user can attempt to override the system prompt by embedding instructions in their message. This doesn't require a complex mitigation strategy, but it does require thinking about what your system prompt grants and what the model will refuse to do. Reviewing a security hardening checklist before launch is a practical starting point.

The takeaway

Memory in a chatbot is infrastructure, not a feature. Get the session store and context builder right early; retrofitting them after you have users is painful. The token budget logic is the most important piece — models have hard limits, and exceeding them breaks conversations in ways that are hard to diagnose.

The pattern above (Redis + token-aware trimming + latest-first context) covers most production cases. If you need the bot to remember facts across weeks of conversation rather than a single session, semantic memory retrieval becomes relevant — but that's a separate layer on top of this one, not a replacement.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Building an AI-Powered Code Vulnerability Scanner

Ayi NEDJIMI — Sat, 19 Sep 2026 10:01:14 +0000

Static analysis tools like Semgrep, Bandit, and CodeQL catch well-known patterns. They miss logic bugs, insecure data flows that span multiple files, and context-dependent vulnerabilities that require understanding intent. A language model can fill that gap — not as a replacement for traditional SAST, but as a second layer that reasons about what the code actually does.

How It Works

The architecture is straightforward: chunk source files by function or class, send each chunk to a language model with a structured prompt, parse the JSON output, and deduplicate findings across chunks. The key insight is that most real vulnerabilities are local — they live within a single function or a tight call chain. You don't need to send an entire codebase at once.

The pipeline has four stages:

Parse — extract functions/classes from source files using AST
Prompt — send each chunk with a security-focused system prompt
Parse output — extract structured findings (severity, CWE, line, description)
Report — deduplicate and rank by severity

Parsing Source Code into Chunks

Using Python's built-in ast module, we can extract function and class bodies with their line numbers:

import ast
import textwrap
from dataclasses import dataclass
from pathlib import Path

@dataclass
class CodeChunk:
    filepath: str
    name: str
    start_line: int
    source: str

def extract_chunks(filepath: str) -> list[CodeChunk]:
    source = Path(filepath).read_text()
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return []

    lines = source.splitlines()
    chunks = []

    for node in ast.walk(tree):
        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            continue
        start = node.lineno - 1
        end = node.end_lineno
        body = "\n".join(lines[start:end])
        chunks.append(CodeChunk(
            filepath=filepath,
            name=node.name,
            start_line=node.lineno,
            source=textwrap.dedent(body),
        ))

    return chunks

For Go or other languages, tree-sitter is the right tool — it has Python bindings and parsers for 40+ languages. The same pattern applies: walk the AST, extract named blocks, record their line ranges.

The Vulnerability Prompt

The system prompt is the most important part. Vague prompts produce vague findings. You want structured JSON output with a fixed schema so parsing is deterministic:

import json
import httpx

SYSTEM_PROMPT = """You are a security code reviewer. Analyze the provided code for vulnerabilities.

Return ONLY a JSON object with this exact schema:
{
  "findings": [
    {
      "severity": "critical|high|medium|low|info",
      "cwe": "CWE-XXX",
      "line": <integer, relative to chunk start>,
      "title": "<short title>",
      "description": "<what is wrong and why>",
      "remediation": "<concrete fix>"
    }
  ]
}

If no vulnerabilities are found, return {"findings": []}.
Focus on: injection flaws, broken auth, insecure deserialization, hardcoded secrets,
path traversal, SSRF, XXE, missing input validation, race conditions."""

def scan_chunk(chunk: CodeChunk, api_key: str, model: str = "gpt-4.1") -> list[dict]:
    prompt = f"File: {chunk.filepath}\nFunction: {chunk.name} (starts at line {chunk.start_line})\n\n```
{% endraw %}
python\n{chunk.source}\n
{% raw %}
```"

    resp = httpx.post(
        "https://api.openai.com/v1/chat/completions",
        headers={"Authorization": f"Bearer {api_key}"},
        json={
            "model": model,
            "messages": [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": prompt},
            ],
            "response_format": {"type": "json_object"},
            "temperature": 0,
        },
        timeout=30,
    )
    resp.raise_for_status()

    raw = resp.json()["choices"][0]["message"]["content"]
    parsed = json.loads(raw)
    findings = parsed.get("findings", [])

    # Adjust line numbers to absolute positions
    for f in findings:
        f["line"] = chunk.start_line + f.get("line", 1) - 1
        f["filepath"] = chunk.filepath

    return findings

Setting temperature: 0 and response_format: {"type": "json_object"} is non-negotiable for production use. Without them, you'll get markdown fences around the JSON and inconsistent schemas.

Scanning a Repository

Wire everything together with a concurrency-aware runner. Rate limits on LLM APIs are the main bottleneck, so use a semaphore:

import asyncio
from pathlib import Path

async def scan_repo(repo_path: str, api_key: str, concurrency: int = 5) -> list[dict]:
    py_files = list(Path(repo_path).rglob("*.py"))
    sem = asyncio.Semaphore(concurrency)
    all_findings = []

    async def scan_file(filepath: str):
        chunks = extract_chunks(filepath)
        async with sem:
            for chunk in chunks:
                # Run sync HTTP call in thread pool
                findings = await asyncio.get_event_loop().run_in_executor(
                    None, scan_chunk, chunk, api_key
                )
                all_findings.extend(findings)

    await asyncio.gather(*[scan_file(str(f)) for f in py_files])

    # Sort by severity
    severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4}
    all_findings.sort(key=lambda x: severity_order.get(x.get("severity", "info"), 5))

    return all_findings

# Usage
if __name__ == "__main__":
    findings = asyncio.run(scan_repo("./myproject", api_key="sk-..."))
    for f in findings:
        print(f"[{f['severity'].upper()}] {f['filepath']}:{f['line']} — {f['title']}")

On a mid-sized Python codebase (150 files, ~300 functions), this runs in under 3 minutes with concurrency=5 and costs roughly $0.40–$0.80 in API tokens depending on the model.

Reducing False Positives

Raw LLM findings have a false positive rate of 20–40% in practice. Two techniques help significantly:

1. Two-pass validation — after the first scan, send findings back to the model and ask: "Is this actually exploitable given the full context?" This cuts false positives roughly in half.

2. Cross-reference with known patterns — before reporting a finding, check it against a SAST tool. If Bandit also flags the same line, confidence goes up. If only the LLM flags it, mark it as needs_review in your report.

For a security team workflow, integrating this with your existing checklist process makes findings actionable. We use a structured approach to vulnerability triage that's documented in our security hardening checklists — the same classification logic applies here.

Limitations You Need to Know

This approach does not replace traditional SAST for two reasons:

No dataflow tracking — the LLM sees a chunk, not a call graph. Taint analysis across files requires a real AST/IR.
Non-deterministic — even at temperature 0, different runs can produce different findings. Don't rely on it for compliance evidence.

Use it as a complement: run Semgrep or Bandit first for fast, deterministic pattern matching, then run the LLM scanner on files that changed in a PR for deeper reasoning. This hybrid approach gives you the best of both worlds without the cost of scanning everything every time.

The Takeaway

A language model can catch vulnerability classes that pattern-matching tools miss — business logic flaws, incorrect use of cryptography, subtly wrong authentication checks. The implementation is genuinely simple: parse to AST, chunk by function, prompt with a strict JSON schema, deduplicate.

The real work is in the prompt engineering and the validation layer. A model that returns "possible injection" with no line number or remediation is useless in a review workflow. Force structure from the start, and build the two-pass validation early — it's much harder to retrofit.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

How to Build an AI-Powered Log Summarizer for DevOps

Ayi NEDJIMI — Fri, 18 Sep 2026 10:03:02 +0000

Production logs are noisy. A single deployment can generate thousands of lines of output — stack traces buried in INFO messages, warnings you can't parse in under 30 seconds, and errors that appear 200 lines before the actual failure. Most teams either ignore logs until something breaks, or spend 20 minutes hunting through them after an incident.

There's a better approach: feed your logs to a language model and get a structured, human-readable summary in seconds. This isn't a replacement for proper observability — keep your Grafana/Loki stack — but it's a fast triage layer that shaves time off every incident.

This article walks through building a working log summarizer in Python: one that ingests logs, extracts relevant chunks, and returns a concise incident summary.

Why Raw Logs Are Hard to Triage

Logs aren't designed for humans. They're optimized for machines to write and search. A typical 500 lines of output during an incident might contain:

400 lines of normal request handling
10 lines of retries and connection resets
2 lines of the actual error
88 lines of stack trace

A language model can parse all of this, understand the sequence of events, and produce: "Database connection pool exhausted after 3 retries, followed by 502 errors on /api/orders. The root connection failure started at 14:32:01."

That's the output you actually want at 3am.

Step 1: Log Preprocessing and Chunking

Language models have token limits. A 500MB log file won't fit in a single prompt. The solution is to preprocess: filter, deduplicate, and chunk the logs before sending them to the model.

import re
from dataclasses import dataclass

LOG_LEVEL_PATTERN = re.compile(
    r"(?P<level>ERROR|WARN(?:ING)?|FATAL|CRITICAL|PANIC)", re.IGNORECASE
)
TIMESTAMP_PATTERN = re.compile(r"\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}")


@dataclass
class LogChunk:
    lines: list[str]
    has_error: bool
    start_ts: str | None
    end_ts: str | None


def extract_chunks(
    log_lines: list[str],
    window: int = 50,
    max_chunks: int = 8,
) -> list[LogChunk]:
    error_indices = [
        i for i, line in enumerate(log_lines)
        if LOG_LEVEL_PATTERN.search(line)
    ]

    seen: set[int] = set()
    chunks: list[LogChunk] = []

    for idx in error_indices:
        start = max(0, idx - window // 2)
        end = min(len(log_lines), idx + window // 2)
        key = (start // window, end // window)
        if key in seen:
            continue
        seen.add(key)

        window_lines = log_lines[start:end]
        timestamps = [
            m.group() for line in window_lines
            if (m := TIMESTAMP_PATTERN.search(line))
        ]
        chunks.append(LogChunk(
            lines=window_lines,
            has_error=True,
            start_ts=timestamps[0] if timestamps else None,
            end_ts=timestamps[-1] if timestamps else None,
        ))
        if len(chunks) >= max_chunks:
            break

    return chunks

The key idea: instead of summarizing every log line, we find windows around ERROR/FATAL lines, deduplicate overlapping windows, and pass only those to the model. For a typical incident, this produces 3–6 focused excerpts totalling under 3,000 tokens.

Step 2: Building the Summarization Prompt

Prompt engineering matters here. A vague “summarize these logs” prompt gets vague output. You want structured answers: what failed, when, what the impact was, and what to investigate next.

import json

SYSTEM_PROMPT = (
    "You are an expert SRE analyzing application logs.\n"
    "Given log excerpts, return a JSON object with these exact fields:\n"
    "- summary: one-sentence description of the incident\n"
    "- severity: one of [critical, high, medium, low]\n"
    "- root_cause: your best hypothesis based on the evidence\n"
    "- impacted_services: list of service names you can identify\n"
    "- timeline: list of {timestamp, event} objects, chronological\n"
    "- recommended_action: the single most important next diagnostic step\n"
    "Return only valid JSON. No markdown fences, no explanation."
)


def build_prompt(chunks: list) -> str:
    sections = []
    for i, chunk in enumerate(chunks, 1):
        header = f"--- Excerpt {i}"
        if chunk.start_ts:
            header += f" ({chunk.start_ts} to {chunk.end_ts})"
        sections.append(header + "\n" + "\n".join(chunk.lines))
    return "\n\n".join(sections)


def summarize_logs(log_text: str, client) -> dict:
    lines = log_text.splitlines()
    chunks = extract_chunks(lines)

    if not chunks:
        chunks = [LogChunk(
            lines=lines[-100:],
            has_error=False,
            start_ts=None,
            end_ts=None,
        )]

    response = client.messages.create(
        model="your-model-here",
        max_tokens=1024,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": build_prompt(chunks)}],
    )

    return json.loads(response.content[0].text)

The client parameter is your LLM SDK instance. The function is model-agnostic by design — swap in any provider that supports a system prompt.

Step 3: CLI and Pipeline Integration

A summarizer is only useful if it fits into existing workflows. Here's a minimal CLI wrapper that reads from stdin or a file:

#!/usr/bin/env python3
import argparse
import json
import sys


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--file", help="Log file path (default: stdin)")
    parser.add_argument("--json", action="store_true", help="Raw JSON output")
    args = parser.parse_args()

    log_text = open(args.file).read() if args.file else sys.stdin.read()

    if not log_text.strip():
        print("No log input.", file=sys.stderr)
        sys.exit(1)

    client = init_llm_client()  # initialize your SDK client here
    result = summarize_logs(log_text, client)

    if args.json:
        print(json.dumps(result, indent=2))
        return

    print(f"\n[{result['severity'].upper()}] {result['summary']}")
    print(f"Root cause:  {result['root_cause']}")
    print(f"Impacted:    {', '.join(result['impacted_services'])}")
    print(f"Next step:   {result['recommended_action']}")

    if result.get("timeline"):
        print("\nTimeline:")
        for event in result["timeline"]:
            print(f"  {event['timestamp']}  {event['event']}")


if __name__ == "__main__":
    main()

Pipe this with --json into a Slack webhook or PagerDuty API and you get machine-readable summaries that auto-populate incident tickets. It integrates cleanly with Loki alert webhooks, AWS Lambda triggers on CloudWatch filters, or any alerting system that can execute a shell command.

Practical Considerations

Cost. You're sending compressed log excerpts, not full files. A typical incident window is under 3,000 tokens. At current API rates, that's fractions of a cent per summary — negligible compared to the time saved.

Accuracy. The model will occasionally misidentify service names or get timestamps slightly wrong. Treat the summary as a starting hypothesis, not ground truth. It directs your attention; your judgment closes the loop.

Privacy and security. Logs often contain IP addresses, user IDs, session tokens, and internal hostnames. Before sending logs to any external API, either redact PII with a preprocessing pass, or run a local model. If you're deploying this in a regulated environment, review the security hardening checklists relevant to your stack before going live.

Local models. Running a quantized local model (7B–13B parameters) via Ollama works well for this use case. Log summarization doesn't require frontier-model reasoning — it needs pattern recognition and structured output, which smaller models handle reliably.

The Takeaway

This pattern won't replace structured logging or distributed tracing. What it does is cut the first 15 minutes of "what happened?" from every incident. Feed the relevant log window to a language model, get a structured summary, start fixing things.

The full pipeline is about 150 lines of Python. The hardest part is the prompt — spend time getting the output schema right, and everything else is plumbing.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

How to Implement Function Calling with Structured Outputs

Ayi NEDJIMI — Thu, 17 Sep 2026 10:03:56 +0000

Most language models return free-form text. That works for chatbots, but it breaks down when you need your application to actually do something with the response — parse a price, trigger an API call, update a record. Function calling (also called tool use) solves this by making the model return a structured payload instead of prose.

This guide shows you how to implement it in Python, covering both the standard function-calling pattern and structured output mode. By the end, you will have working code for extracting structured data, chaining tool calls, and handling errors gracefully.

What function calling actually does

The term is misleading. The model does not call your functions. What happens is:

You describe one or more tools (functions) in your API request — their names, parameters, and types.
The model decides whether to "call" a tool, and if so, returns a JSON object that matches the schema you described.
Your code parses that JSON and runs the actual function.
You feed the result back into the conversation.

The model is a JSON router, not an executor. Keeping this mental model clear matters when debugging — if the model returns bad JSON, the problem is usually in your schema definition, not your function logic.

Defining tools with JSON Schema

Most LLM providers accept tools in JSON Schema format. Here is a minimal example that extracts invoice data from raw text:

import json
import openai

client = openai.OpenAI()

tools = [
    {
        "type": "function",
        "function": {
            "name": "extract_invoice",
            "description": "Extract structured fields from an invoice text.",
            "parameters": {
                "type": "object",
                "properties": {
                    "vendor": {"type": "string", "description": "Company or person issuing the invoice"},
                    "amount": {"type": "number", "description": "Total amount in the invoice currency"},
                    "currency": {"type": "string", "enum": ["EUR", "USD", "GBP"]},
                    "due_date": {"type": "string", "description": "Due date in ISO 8601 format (YYYY-MM-DD)"},
                },
                "required": ["vendor", "amount", "currency"],
            },
        },
    }
]

def extract_invoice_data(raw_text: str) -> dict:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "user", "content": f"Extract the invoice data from this text:\n\n{raw_text}"}
        ],
        tools=tools,
        tool_choice={"type": "function", "function": {"name": "extract_invoice"}},
    )

    message = response.choices[0].message
    if message.tool_calls:
        return json.loads(message.tool_calls[0].function.arguments)
    return {}

Setting tool_choice to a specific function forces the model to always call it. Use "auto" when you want the language model to decide whether a tool call is appropriate — useful when not every user message requires a structured response.

Chaining tool calls in a conversation loop

Real agents rarely need just one tool call. A more common pattern is a loop where you keep feeding results back until the model stops requesting tools:

def run_agent(user_message: str, tools: list, tool_map: dict) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=tools,
            tool_choice="auto",
        )

        choice = response.choices[0]
        messages.append(choice.message)

        # No tool calls → model is done
        if choice.finish_reason == "stop":
            return choice.message.content

        # Execute each tool call and append results
        for tc in choice.message.tool_calls:
            fn_name = tc.function.name
            fn_args = json.loads(tc.function.arguments)

            if fn_name not in tool_map:
                result = {"error": f"Unknown tool: {fn_name}"}
            else:
                try:
                    result = tool_map<a href="**fn_args">fn_name</a>
                except Exception as e:
                    result = {"error": str(e)}

            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": json.dumps(result),
            })

tool_map is a plain dict mapping function names to callables. This keeps the loop generic — you add new tools by updating the dict, not by touching the loop logic. Always catch exceptions inside the loop and return them as JSON errors; the model can often recover from a tool error on its own.

Validating outputs with Pydantic

The model's JSON output is not guaranteed to match your schema exactly. Enums may come back in unexpected casing, required fields may be missing if the input was ambiguous. Validate before using:

from pydantic import BaseModel, field_validator
from typing import Optional
from datetime import date

class Invoice(BaseModel):
    vendor: str
    amount: float
    currency: str
    due_date: Optional[date] = None

    @field_validator("currency")
    @classmethod
    def normalize_currency(cls, v: str) -> str:
        return v.upper()

def safe_extract(raw_text: str) -> Invoice | None:
    raw = extract_invoice_data(raw_text)
    if not raw:
        return None
    try:
        return Invoice.model_validate(raw)
    except Exception as e:
        print(f"Validation failed: {e}")
        return None

Pydantic coerces types where it can (string "42.5" becomes float 42.5) and raises a clear error when it cannot. This single validation layer catches the majority of production failures.

Structured outputs: the stricter alternative

Some providers now support "structured outputs" — a mode where the model is constrained at the generation level to match a JSON Schema. Unlike regular function calling, this is a hard guarantee: invalid JSON becomes impossible.

schema = {
    "name": "invoice_extraction",
    "strict": True,
    "schema": {
        "type": "object",
        "properties": {
            "vendor": {"type": "string"},
            "amount": {"type": "number"},
            "currency": {"type": "string"},
        },
        "required": ["vendor", "amount", "currency"],
        "additionalProperties": False,
    },
}

response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[{"role": "user", "content": "Extract: Invoice from Acme Corp, $1,250 USD"}],
    response_format={"type": "json_schema", "json_schema": schema},
)

data = json.loads(response.choices[0].message.content)

The strict: True flag enables constrained generation. It requires additionalProperties: false and all properties must be in required. This is the right choice when you know exactly what shape you need and cannot afford to handle malformed output at runtime.

For language models that don't natively support structured outputs, you can approximate this with careful prompting plus Pydantic validation and a retry loop — though you will never get the same guarantee.

Picking the right approach

Scenario	Recommended approach
Extract known fields from text	Function calling with `tool_choice` forced
Agent that may or may not need data	Function calling with `tool_choice: auto`
Critical data pipeline, no retries	Structured outputs (if supported)
Multi-step agentic loop	Conversation loop with tool_map

The security checklists at AYI NEDJIMI Consultants include a section on sandboxing and validating LLM tool outputs — worth reviewing if your tools touch sensitive systems or external APIs.

The takeaway

Function calling gives you a clean contract between your application logic and a language model. Define your schema with the minimum required fields, validate the output with Pydantic, and keep your execution loop generic. Structured outputs remove the validation burden entirely when your provider supports them — but they come with schema restrictions you need to plan around.

The common mistake is over-engineering the schema upfront. Start with the minimum fields your code will break without, and iterate from there.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Prompt Caching Strategies to Cut LLM Costs by 70%

Ayi NEDJIMI — Wed, 16 Sep 2026 10:01:59 +0000

LLM API costs add up fast when you're running a production application. If your system prompt is 2,000 tokens and you're processing 10,000 requests a day, you're paying for those same tokens 10,000 times. Prompt caching lets you pay once and reuse — and the savings can be dramatic.

What Prompt Caching Actually Is

Most LLM providers charge for every token you send in a request — input tokens plus output tokens. Prompt caching allows the provider to store the computed state (the KV cache) of a prompt prefix on their servers, so subsequent requests that share that prefix only pay a fraction of the original input token cost.

The key insight: prompt caching only works on static prefix content. The cached portion must be at the beginning of your prompt and must be identical across requests. If you change anything in that prefix — even a single character — the cache misses and you pay full price.

Typical savings with caching enabled:

Input tokens in the cached prefix: ~90% cheaper
Output tokens: unchanged
Net reduction on token-heavy workloads: 50–75%

Structuring Prompts for Maximum Cache Hit Rate

The single biggest mistake developers make is putting dynamic content early in the prompt. User-specific data, timestamps, or session IDs near the top will break the cache for every single request.

The correct structure is:

[SYSTEM PROMPT — static, long, cached]
[FEW-SHOT EXAMPLES — static, cached]
[RETRIEVED CONTEXT — cacheable if content-addressed]
[USER QUERY — dynamic, never cached]

Here's a quick analyzer to check your prompt layout before deploying:

def analyze_prompt_structure(messages: list[dict]) -> dict:
    """
    Estimate how much of a conversation benefits from caching.
    """
    total_tokens = 0
    static_tokens = 0

    for i, msg in enumerate(messages):
        content = msg.get("content", "")
        estimated = len(content.split()) * 1.3  # rough token estimate
        total_tokens += estimated

        if msg["role"] == "system":
            static_tokens += estimated
        elif i < len(messages) - 2:  # all but the last user turn
            static_tokens += estimated

    cache_ratio = static_tokens / total_tokens if total_tokens > 0 else 0

    return {
        "total_tokens_estimate": int(total_tokens),
        "cacheable_tokens_estimate": int(static_tokens),
        "cache_ratio": round(cache_ratio, 2),
        "recommendation": (
            "Good caching potential"
            if cache_ratio > 0.6
            else "Restructure — dynamic content is too early"
        ),
    }


messages = [
    {"role": "system", "content": "You are an expert security analyst. " * 100},
    {"role": "user", "content": "Analyze this log file..."},
    {"role": "assistant", "content": "I found the following issues..."},
    {"role": "user", "content": "What about this IP: 192.168.1.1?"},
]

print(analyze_prompt_structure(messages))
# {'total_tokens_estimate': 481, 'cacheable_tokens_estimate': 390,
#  'cache_ratio': 0.81, 'recommendation': 'Good caching potential'}

Marking Cache Breakpoints Explicitly

Some providers let you annotate cache boundaries directly in your request payload. This gives fine-grained control over what gets stored versus what flows through uncached.

import httpx

SYSTEM_PROMPT = "You are a security analyst assistant..." * 50  # ~2 000 tokens


def build_cached_request(user_query: str, history: list, api_key: str) -> dict:
    return {
        "model": "your-model-id",
        "max_tokens": 1024,
        "system": [
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"},  # provider-specific marker
            }
        ],
        "messages": history + [{"role": "user", "content": user_query}],
    }


def call_llm(query: str, history: list, api_key: str) -> dict:
    payload = build_cached_request(query, history, api_key)

    resp = httpx.post(
        "https://api.your-llm-provider.com/v1/messages",
        headers={"x-api-key": api_key, "content-type": "application/json"},
        json=payload,
        timeout=30,
    )
    resp.raise_for_status()
    data = resp.json()

    usage = data.get("usage", {})
    cache_read = usage.get("cache_read_input_tokens", 0)
    cache_write = usage.get("cache_creation_input_tokens", 0)
    print(f"Cache read: {cache_read} | Cache write: {cache_write}")

    return data

The cache_read_input_tokens field in the response is your ground truth. On the first request, you'll see cache_creation_input_tokens — you paid to populate the cache. On every subsequent request with the same prefix, cache_read_input_tokens climbs and creation tokens drop to zero.

Content-Addressed Caching for RAG Pipelines

In Retrieval-Augmented Generation (RAG) systems, retrieved documents change per query — which looks like it defeats caching. But if you sort and deduplicate chunks by content hash before injecting them, identical queries retrieve identical chunks in the same order. Same order means the same prompt prefix, which means a cache hit.

import hashlib


def stable_chunk_id(chunk: str) -> str:
    return hashlib.sha256(chunk.encode()).hexdigest()[:16]


def build_rag_prompt(query: str, chunks: list[str]) -> tuple[str, str]:
    """
    Returns (context_section, query_section).
    Deterministic ordering ensures cache hits across equivalent retrieval sets.
    """
    deduped = {stable_chunk_id(c): c for c in chunks}
    sorted_chunks = sorted(deduped.values(), key=stable_chunk_id)
    context = "\n\n---\n\n".join(sorted_chunks)
    return context, query


# Two different queries that retrieve the same three chunks
# will produce byte-for-byte identical context → cache hit
context, query = build_rag_prompt(
    "What is the NIST framework?",
    [
        "NIST CSF overview: identify, protect, detect, respond, recover.",
        "NIST SP 800-53 defines control families for federal systems.",
        "Risk management tiers range from 1 (partial) to 4 (adaptive).",
    ],
)

This technique is especially effective for FAQ-style applications where 20% of unique queries account for 80% of traffic. You can also pre-warm the cache by sending your top 100 most common context sets at startup — the first 100 requests pay to write the cache, and everything after that reads it for a fraction of the cost.

Monitoring Cache Efficiency in Production

Shipping caching without instrumentation is flying blind. Track these three numbers from day one:

Cache hit rate = cache_read_tokens / (cache_read_tokens + cache_creation_tokens)
Cost per request = derived from input/output/cache_read/cache_write token counts and provider pricing
Cache miss patterns — which requests miss and why (dynamic content creeping into the prefix)

from dataclasses import dataclass


@dataclass
class CacheMetrics:
    hits: int = 0
    misses: int = 0
    tokens_saved: int = 0

    @property
    def hit_rate(self) -> float:
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0.0


metrics = CacheMetrics()


def track_usage(usage: dict) -> None:
    cache_read = usage.get("cache_read_input_tokens", 0)
    cache_write = usage.get("cache_creation_input_tokens", 0)

    if cache_read > 0:
        metrics.hits += 1
        metrics.tokens_saved += cache_read
    elif cache_write > 0:
        metrics.misses += 1

    if (metrics.hits + metrics.misses) % 100 == 0:
        print(
            f"Hit rate: {metrics.hit_rate:.1%} | "
            f"Tokens saved: {metrics.tokens_saved:,}"
        )

In production, pipe these numbers into Prometheus or your existing observability stack. If your hit rate drops below 70% on a workload that should be caching well, something dynamic is leaking into the prefix. The LLM security and configuration checklists at AYI NEDJIMI Consultants treat cost controls and access controls in the same review — both are operational risks.

The Takeaway

Prompt caching is one of the few LLM optimizations that requires no model changes, no architecture redesign, and no quality tradeoff. The entire gain comes from reorganizing your prompts and measuring the outcome:

Move all static content to the top — system instructions, examples, knowledge bases
Keep dynamic content (user queries, timestamps, session data) at the very end
Use content-addressing in RAG pipelines to maximize hit rate across semantically similar queries
Track cache_read_input_tokens per request from day one

On a real workload with a 3,000-token system prompt and 100 tokens of user input, caching reduces input costs by ~97% for the static portion. At scale, that is not a rounding error.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Implementing AI Observability: Tracing LLM Calls End-to-End

Ayi NEDJIMI — Mon, 14 Sep 2026 10:02:43 +0000

When a Python microservice misbehaves in production, you reach for logs and distributed traces. When an LLM-powered application misbehaves, most teams are flying blind. Tokens leak, latency spikes, costs explode — and all you have is a user complaint. That gap is what AI observability is designed to close.

What you actually need to trace

Classic application observability covers latency, error rate, and throughput. LLM applications need more:

Token usage (prompt tokens, completion tokens, total) — directly tied to cost
Model and version — a provider rollout can silently degrade quality
Latency breakdown — time-to-first-token vs. total generation time
Prompt fingerprint — which template version triggered this call
Output quality signals — length, structured output validity, guard failures

Omit any of these and you will debug production issues by reading tea leaves.

Building a minimal LLM tracer in Python

You don't need an observability vendor on day one. A decorator-based tracer that writes structured JSON gives you enough to start.

import time
import json
import hashlib
import logging
from functools import wraps
from dataclasses import dataclass, asdict
from typing import Optional

logger = logging.getLogger("llm.trace")

@dataclass
class LLMSpan:
    trace_id: str
    model: str
    prompt_tokens: int
    completion_tokens: int
    latency_ms: float
    cost_usd: float
    error: Optional[str] = None

COST_PER_1K = {
    "gpt-4o": {"input": 0.005, "output": 0.015},
    "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
}

def compute_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    prices = COST_PER_1K.get(model, {"input": 0.001, "output": 0.003})
    return (prompt_tokens / 1000 * prices["input"]) + \
           (completion_tokens / 1000 * prices["output"])

def trace_llm(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        error_msg = None
        result = None
        try:
            result = func(*args, **kwargs)
            return result
        except Exception as e:
            error_msg = str(e)
            raise
        finally:
            elapsed = (time.perf_counter() - start) * 1000
            usage = getattr(result, "usage", None) if result else None
            prompt_tokens = getattr(usage, "prompt_tokens", 0) if usage else 0
            completion_tokens = getattr(usage, "completion_tokens", 0) if usage else 0
            model = kwargs.get("model", "unknown")
            span = LLMSpan(
                trace_id=hashlib.md5(str(start).encode()).hexdigest()[:12],
                model=model,
                prompt_tokens=prompt_tokens,
                completion_tokens=completion_tokens,
                latency_ms=round(elapsed, 2),
                cost_usd=round(compute_cost(model, prompt_tokens, completion_tokens), 6),
                error=error_msg,
            )
            logger.info(json.dumps(asdict(span)))
    return wrapper

Wrap your API call:

@trace_llm
def call_llm(client, messages: list, model: str = "gpt-4o-mini"):
    return client.chat.completions.create(
        model=model,
        messages=messages,
    )

Every call now emits a structured JSON line. Feed that to jq, Loki, or any log aggregation tool you already have. No new infrastructure required.

Connecting to OpenTelemetry

Once your team outgrows flat logs, OpenTelemetry spans let you attach LLM traces to the parent HTTP request that triggered them.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

provider = TracerProvider()
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://otelcol:4317"))
)
trace.set_tracer_provider(provider)
otel_tracer = trace.get_tracer("llm-service")

def call_llm_otel(client, messages: list, model: str = "gpt-4o-mini"):
    with otel_tracer.start_as_current_span("llm.completion") as span:
        start = time.perf_counter()
        response = client.chat.completions.create(model=model, messages=messages)
        elapsed_ms = (time.perf_counter() - start) * 1000

        usage = response.usage
        span.set_attribute("llm.model", model)
        span.set_attribute("llm.prompt_tokens", usage.prompt_tokens)
        span.set_attribute("llm.completion_tokens", usage.completion_tokens)
        span.set_attribute("llm.latency_ms", round(elapsed_ms, 2))
        span.set_attribute(
            "llm.cost_usd",
            round(compute_cost(model, usage.prompt_tokens, usage.completion_tokens), 6)
        )
        return response

Your Grafana or Jaeger instance will now show LLM latency nested inside the request trace. When a specific user flow slows down, you can see exactly which LLM call is responsible — and whether it was a slow model or a bloated prompt.

Aggregating and alerting on LLM metrics

Raw spans are useful for debugging individual requests. Aggregated metrics catch systemic problems before users notice.

A small script that reads your structured logs and summarizes by model:

import json
import sys
from collections import defaultdict

def summarize_llm_logs(log_lines: list[str]) -> dict:
    totals = defaultdict(lambda: {
        "calls": 0, "errors": 0,
        "tokens": 0, "cost_usd": 0.0, "latency_ms": []
    })
    for line in log_lines:
        try:
            span = json.loads(line)
        except json.JSONDecodeError:
            continue
        m = span.get("model", "unknown")
        totals[m]["calls"] += 1
        if span.get("error"):
            totals[m]["errors"] += 1
        totals[m]["tokens"] += span.get("prompt_tokens", 0) + span.get("completion_tokens", 0)
        totals[m]["cost_usd"] += span.get("cost_usd", 0)
        totals[m]["latency_ms"].append(span.get("latency_ms", 0))
    result = {}
    for model, data in totals.items():
        lats = sorted(data["latency_ms"])
        result[model] = {
            "calls": data["calls"],
            "error_rate": round(data["errors"] / max(data["calls"], 1), 4),
            "total_tokens": data["tokens"],
            "total_cost_usd": round(data["cost_usd"], 4),
            "p50_ms": lats[len(lats) // 2] if lats else 0,
            "p99_ms": lats[int(len(lats) * 0.99)] if lats else 0,
        }
    return result

if __name__ == "__main__":
    lines = sys.stdin.readlines()
    summary = summarize_llm_logs(lines)
    print(json.dumps(summary, indent=2))

Run it with:

grep '"model"' /var/log/app/llm.log | python3 summarize.py

Set a threshold: if p99_ms exceeds 10,000 ms or error_rate exceeds 0.02, fire an alert. These numbers beat a vague "the AI is slow today" complaint.

What the data reveals in practice

Once you have a week of traces, patterns emerge quickly:

Cost spikes usually trace to one prompt template that grew during a refactor. A prompt_tokens histogram shows this immediately.
Latency regression often coincides with a provider's infrastructure event. Check your p99 against the provider's status page — if they correlate, the bug is not yours.
Error clusters reveal retry storms: the tracer will show 10x expected calls in a 30-second window, which points to a misconfigured backoff policy.

None of this requires a paid APM product. The security hardening checklists at AYI NEDJIMI include an LLM application security checklist that covers observability requirements alongside prompt injection and output validation controls — worth running through before you go to production.

The takeaway

AI observability is not a nice-to-have — it is the prerequisite for cost control, SLA enforcement, and meaningful debugging. Start with the structured JSON decorator above: it costs nothing, adds under 5 ms of overhead, and gives you actionable data within minutes. Layer in OpenTelemetry when you need to correlate LLM calls with the rest of your service mesh.

The hardest part is not the implementation. It is the habit: every language model call in your codebase should emit a trace span before it reaches production.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

AI Model Versioning and Rollback Strategies for Production

Ayi NEDJIMI — Sun, 13 Sep 2026 10:01:44 +0000

Deploying an AI model to production is the easy part. Knowing which version is running, what changed between versions, and how to roll back when outputs silently degrade — that's where most teams discover they have no plan.

This article covers a practical approach to AI model versioning and rollback, with code you can adapt immediately.

Why AI Model Versioning Differs from Code Versioning

In software, a bad commit breaks tests or throws exceptions. A bad model update is subtler: accuracy drifts, latency spikes, a particular input class starts returning nonsense. You often don't know until users complain or a downstream metric tanks.

This asymmetry means versioning an AI model requires more metadata than a git SHA:

Model weights and architecture: the serialized checkpoint
Training data hash: ensures reproducibility
Hyperparameters: learning rate, batch size, epochs
Evaluation metrics: accuracy, F1, latency p99 at the time of packaging
Input/output schema: what the model expects and what it returns
Dependency pinning: CUDA, PyTorch, transformers versions

Miss any of these and you can't reliably reproduce or compare behavior across versions.

A Practical Versioning Scheme

Use semantic versioning adapted for models: MAJOR.MINOR.PATCH.

MAJOR: architecture change (new backbone, different tokenizer, different task)
MINOR: significant retraining (new dataset, fine-tuning on new domain)
PATCH: weight updates from continued training or prompt adjustments

Pair each version with a manifest file stored alongside the weights:

{
  "version": "2.3.1",
  "model_type": "text-classifier",
  "created_at": "2026-09-13T10:00:00Z",
  "training_data_sha256": "a3f8c91d...",
  "metrics": {
    "accuracy": 0.941,
    "f1_macro": 0.927,
    "latency_p99_ms": 48
  },
  "input_schema": {"type": "string", "max_tokens": 512},
  "dependencies": {
    "transformers": "4.40.0",
    "torch": "2.3.0"
  },
  "promoted_by": "engineer@example.com",
  "promoted_at": "2026-09-13T11:30:00Z"
}

This manifest is the source of truth. Store it in the same bucket or registry as the weights — never separately.

Implementing a Model Registry in Python

You don't need a heavyweight MLOps platform to get started. A registry backed by object storage and a PostgreSQL table covers most small and mid-size teams.

import hashlib
import json
import datetime
from pathlib import Path
import boto3
import psycopg2

class ModelRegistry:
    def __init__(self, bucket: str, db_conn_string: str):
        self.s3 = boto3.client("s3")
        self.bucket = bucket
        self.conn = psycopg2.connect(db_conn_string)

    def push(self, model_path: Path, manifest: dict, version: str) -> str:
        """Upload model weights + manifest, register in DB."""
        weights_key = f"models/{manifest['model_type']}/{version}/weights.pt"
        manifest_key = f"models/{manifest['model_type']}/{version}/manifest.json"

        sha = hashlib.sha256(model_path.read_bytes()).hexdigest()
        manifest["weights_sha256"] = sha
        manifest["version"] = version

        self.s3.upload_file(str(model_path), self.bucket, weights_key)
        self.s3.put_object(
            Bucket=self.bucket,
            Key=manifest_key,
            Body=json.dumps(manifest, indent=2),
        )

        with self.conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO model_versions
                  (model_type, version, weights_key, manifest_key, metrics, created_at, status)
                VALUES (%s, %s, %s, %s, %s, %s, 'staged')
                """,
                (
                    manifest["model_type"],
                    version,
                    weights_key,
                    manifest_key,
                    json.dumps(manifest.get("metrics", {})),
                    datetime.datetime.utcnow(),
                ),
            )
            self.conn.commit()
        return weights_key

    def promote(self, model_type: str, version: str) -> None:
        """Mark a version as production-active."""
        with self.conn.cursor() as cur:
            cur.execute(
                "UPDATE model_versions SET status='retired' WHERE model_type=%s AND status='active'",
                (model_type,),
            )
            cur.execute(
                "UPDATE model_versions SET status='active', promoted_at=%s WHERE model_type=%s AND version=%s",
                (datetime.datetime.utcnow(), model_type, version),
            )
            self.conn.commit()

    def get_active(self, model_type: str) -> dict:
        """Return manifest for the current production version."""
        with self.conn.cursor() as cur:
            cur.execute(
                "SELECT manifest_key FROM model_versions WHERE model_type=%s AND status='active'",
                (model_type,),
            )
            row = cur.fetchone()
        if not row:
            raise ValueError(f"No active version found for {model_type}")
        response = self.s3.get_object(Bucket=self.bucket, Key=row[0])
        return json.loads(response["Body"].read())

The status column (staged → active → retired) gives you a clean audit trail. Rollback is explicit: promote the previous version, the current one becomes retired.

Rollback Patterns

Three patterns, ordered from fast to deliberate:

1. Blue/green with registry pointer

Keep two serving instances — blue (current active) and green (previous active). When monitoring detects degradation, flip the active flag in the registry and redeploy. Time to restore: 30–90 seconds depending on your deployment pipeline.

2. Canary with automatic rollback

Route 5–10% of traffic to the new version. Measure proxy metrics — user correction rate, downstream pipeline reject rate, output confidence distribution — against the baseline. If the error rate diverges beyond a threshold, trigger rollback automatically:

def should_rollback(
    baseline_error_rate: float,
    canary_error_rate: float,
    threshold: float = 0.10
) -> bool:
    """Return True if canary error rate exceeds baseline by more than threshold."""
    return (canary_error_rate - baseline_error_rate) > threshold


def monitor_canary(
    registry: ModelRegistry,
    model_type: str,
    canary_version: str,
    previous_active: str,
    check_interval_s: int = 60,
    max_checks: int = 30,
) -> bool:
    import time

    for i in range(max_checks):
        baseline = get_error_rate(model_type, "active")
        canary   = get_error_rate(model_type, canary_version)

        if should_rollback(baseline, canary):
            print(
                f"[rollback] {canary_version} degraded "
                f"(baseline={baseline:.3f}, canary={canary:.3f}). "
                f"Restoring {previous_active}."
            )
            registry.promote(model_type, previous_active)
            return False

        time.sleep(check_interval_s)

    # Canary survived — promote it
    registry.promote(model_type, canary_version)
    return True

3. Shadow mode before promotion

Run the new model in shadow: it receives all requests, produces outputs, but those outputs are discarded. Compare shadow outputs to active model outputs offline using a statistical test (e.g. Wilcoxon signed-rank test on accuracy). Only promote after a statistically significant comparison over at least 1,000 samples. This slows deployment but eliminates surprise regressions.

For teams building AI pipelines in regulated or security-sensitive environments, a hardened deployment checklist — covering canary thresholds, rollback triggers, observability requirements, and access controls — is as important as the code itself. We publish free security hardening checklists covering AI deployment and adjacent topics.

The Takeaway

Model versioning is not a nice-to-have once you have real users. The key points:

Every model version needs a manifest with metrics, data hash, and dependency pins — not just a filename or timestamp
Use semantic versioning adapted to the nature of the change (architecture vs. data vs. weight update)
Build a simple registry before reaching for heavyweight MLOps platforms
Implement rollback as a first-class operation, tested in staging before you need it in production
Canary deployment with automatic rollback beats manual hotfixes when availability matters

The failure mode to avoid: discovering your rollback procedure on a Friday evening when the on-call engineer has never run it before.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Building a Local LLM API Server with Ollama and FastAPI

Ayi NEDJIMI — Thu, 10 Sep 2026 10:02:51 +0000

Running a language model locally means you control the data, avoid per-token costs, and can tune latency to your use case. The problem is that Ollama's built-in HTTP API is minimal — no auth, no schema validation, no easy integration with your existing Python stack. Wrapping it with FastAPI fixes that in under 150 lines and gives you an endpoint you can hand to teammates or wire into a pipeline without exposing raw inference infrastructure.

Why Ollama + FastAPI

Ollama provides a dead-simple way to pull and serve open models (Llama 3, Mistral, Qwen, Phi, etc.) locally or on a private server. Its REST API is functional but basic: no middleware, no request validation, no usage tracking. FastAPI adds exactly what's missing:

Pydantic models for input validation and clear error messages
Async support compatible with Ollama's SSE streaming
Auth middleware, rate limiting, structured logging
OpenAPI docs out of the box at /docs

The result is a drop-in local replacement for cloud LLM APIs that you fully control. No data leaves your server, no per-token billing, no vendor rate limits during load spikes.

Setting Up Ollama

Install Ollama on Linux or macOS:

curl -fsSL https://ollama.com/install.sh | sh
ollama pull llama3.2
ollama serve  # starts on http://localhost:11434 by default

Verify the bare API responds:

curl http://localhost:11434/api/generate \
  -d '{"model": "llama3.2", "prompt": "Hello", "stream": false}'

Ollama exposes two main endpoints: /api/generate for single-turn completions and /api/chat for multi-turn conversations with message history. We'll proxy and enhance both.

The FastAPI Wrapper

Install dependencies:

pip install fastapi uvicorn httpx pydantic

Here's the full server:

import httpx
import json
import os
import time
import logging
from typing import Optional, List

from fastapi import FastAPI, HTTPException, Depends, Header, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI(title="Local LLM API", version="1.0.0")
logger = logging.getLogger("llm_api")
logging.basicConfig(level=logging.INFO)

OLLAMA_BASE_URL = os.getenv("OLLAMA_URL", "http://localhost:11434")
API_KEY = os.getenv("API_KEY", "")

# --- Auth ---

async def verify_api_key(x_api_key: Optional[str] = Header(None)):
    if API_KEY and x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")

# --- Request/Response models ---

class Message(BaseModel):
    role: str   # "user" or "assistant"
    content: str

class ChatRequest(BaseModel):
    model: str = "llama3.2"
    messages: List[Message]
    stream: bool = False
    temperature: Optional[float] = 0.7
    max_tokens: Optional[int] = None

# --- Middleware for latency logging ---

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.monotonic()
    response = await call_next(request)
    duration = time.monotonic() - start
    logger.info(
        "method=%s path=%s status=%d duration=%.3fs",
        request.method, request.url.path,
        response.status_code, duration,
    )
    return response

# --- Routes ---

@app.get("/health")
async def health():
    async with httpx.AsyncClient() as client:
        try:
            r = await client.get(f"{OLLAMA_BASE_URL}/api/tags", timeout=3.0)
            r.raise_for_status()
            models = [m["name"] for m in r.json().get("models", [])]
            return {"status": "ok", "models": models}
        except Exception:
            raise HTTPException(status_code=503, detail="Ollama unreachable")

@app.post("/v1/chat", dependencies=[Depends(verify_api_key)])
async def chat(req: ChatRequest):
    payload = {
        "model": req.model,
        "messages": [m.model_dump() for m in req.messages],
        "stream": req.stream,
        "options": {"temperature": req.temperature},
    }
    if req.max_tokens:
        payload["options"]["num_predict"] = req.max_tokens

    if req.stream:
        return StreamingResponse(
            _stream_ollama("/api/chat", payload),
            media_type="text/event-stream",
        )

    async with httpx.AsyncClient(timeout=120.0) as client:
        r = await client.post(f"{OLLAMA_BASE_URL}/api/chat", json=payload)
        if r.status_code != 200:
            raise HTTPException(status_code=r.status_code, detail=r.text)
        data = r.json()
        return {
            "model": req.model,
            "content": data["message"]["content"],
            "done": data.get("done", True),
        }

async def _stream_ollama(path: str, payload: dict):
    async with httpx.AsyncClient(timeout=120.0) as client:
        async with client.stream(
            "POST", f"{OLLAMA_BASE_URL}{path}", json=payload
        ) as r:
            async for line in r.aiter_lines():
                if line:
                    yield f"data: {line}\n\n"
    yield "data: [DONE]\n\n"

Run it:

API_KEY=mysecret uvicorn server:app --host 0.0.0.0 --port 8080

The /health endpoint tells you which models are loaded. Hit /docs for the auto-generated OpenAPI UI — useful when teammates are exploring what the server supports.

Consuming the Streaming Endpoint

For chat interfaces and long outputs, streaming matters. First-token latency is everything for perceived responsiveness. Here's a minimal Python client that prints tokens as they arrive:

import httpx, json

def stream_chat(prompt: str, api_key: str = "mysecret"):
    messages = [{"role": "user", "content": prompt}]
    with httpx.Client() as client:
        with client.stream(
            "POST",
            "http://localhost:8080/v1/chat",
            json={"messages": messages, "stream": True},
            headers={"x-api-key": api_key},
            timeout=120.0,
        ) as r:
            for line in r.iter_lines():
                if line.startswith("data: ") and line != "data: [DONE]":
                    chunk = json.loads(line[6:])
                    token = chunk.get("message", {}).get("content", "")
                    if token:
                        print(token, end="", flush=True)
    print()

stream_chat("Explain mTLS in two sentences")

The [DONE] sentinel keeps the client loop clean without relying on connection close detection, and it's easy to adapt for JavaScript EventSource clients if you're building a frontend.

Deploying as a Systemd Service

For a persistent local API, run both Ollama and FastAPI as systemd services. Create /etc/systemd/system/llm-api.service:

[Unit]
Description=Local LLM API (FastAPI)
After=ollama.service

[Service]
User=llm
WorkingDirectory=/opt/llm-api
ExecStart=/opt/llm-api/venv/bin/uvicorn server:app --host 127.0.0.1 --port 8080
Restart=on-failure
Environment=API_KEY=your_secret_here
Environment=OLLAMA_URL=http://127.0.0.1:11434

[Install]
WantedBy=multi-user.target

systemctl daemon-reload
systemctl enable --now llm-api

Add Nginx in front to terminate TLS and you have a production-grade local LLM endpoint that survives reboots. For hardening the API layer — authentication patterns, rate limit configuration, allowed model allowlists — the security hardening checklists we publish cover these patterns with concrete configs.

What to Build Next

A few natural extensions once the base is running:

Request caching: hash the (model, messages, temperature) tuple and store responses in Redis or SQLite. Repeated identical prompts return instantly with zero inference cost — useful for FAQ-style bots.
Model router: inspect prompt length and complexity, route short factual queries to a small fast model (Phi-3 mini), longer reasoning tasks to a larger one. Ollama's /api/tags tells you what's available.
Token usage tracking: write a row to SQLite per request — model name, Ollama's eval_count and prompt_eval_count fields, latency. Gives you cost-equivalent data even with local inference, so you can benchmark models fairly.
Rate limiting per API key: the current implementation has a single key. Add a key→tier mapping in a dict or SQLite table, and use a token-bucket decorator per key.

The Takeaway

Ollama handles model management and inference. FastAPI handles everything around it: validation, auth, streaming, observability. The combination is roughly 150 lines of Python and gives you a private, cost-free, API-compatible LLM endpoint you control end to end.

The bare Ollama API is fine for local tinkering. Once you need to share the endpoint with teammates, add auth, or integrate it into a production pipeline, wrapping it with FastAPI takes an afternoon and pays for itself immediately in clarity and maintainability.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Building a Vector Search Engine from Scratch with HNSW in Python

Ayi NEDJIMI — Wed, 09 Sep 2026 10:03:05 +0000

Every vector database tutorial opens with pip install pinecone and skips the part where you understand why vector search works. That's fine for shipping fast, but it costs you later — when recall drops and you don't know where to look, or when a tuning parameter turns your latency from 10ms to 800ms.

HNSW (Hierarchical Navigable Small World) is the algorithm behind almost every ANN (Approximate Nearest Neighbor) library in production today: FAISS, Qdrant, Weaviate, and hnswlib all use it internally. This post builds a working HNSW index in pure Python — not production-grade, but real enough to give you a concrete mental model.

What HNSW actually does

The naive approach to nearest neighbor search is brute force: compare your query vector to every stored vector, return the k closest. That's O(n·d) per query — for 1M vectors at 1536 dimensions, expect around 6 seconds per lookup.

HNSW solves this with a layered graph. Vectors live across multiple layers:

Higher layers: sparse "highway" connections between distant nodes
Lower layers: dense "local" connections between nearby nodes
Layer 0: the complete graph with the richest connectivity

At query time:

Enter at the top layer via a fixed entry point
Greedily walk toward nodes closer to the query
Drop to the next layer, repeat
At layer 0, collect the k nearest neighbors

Expected complexity drops from O(n) to O(log n). The trade-off: more memory, and approximate results — you may occasionally miss the single true nearest neighbor, but you'll be well within a small margin.

Building the index in Python

Here's a minimal HNSW implementation that captures the core structure:

import numpy as np
import heapq
import math
import random
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class HNSWNode:
    vector: np.ndarray
    neighbors: dict = field(default_factory=dict)  # layer -> list of node ids

class HNSWIndex:
    def __init__(self, dim: int, M: int = 16, ef_construction: int = 200, max_layers: int = 6):
        self.dim = dim
        self.M = M               # max neighbors per layer
        self.M0 = M * 2          # max neighbors at layer 0
        self.ef = ef_construction
        self.max_layers = max_layers
        self.nodes: list[HNSWNode] = []
        self.entry_point: Optional[int] = None
        self.top_layer = 0

    def _distance(self, a: np.ndarray, b: np.ndarray) -> float:
        diff = a - b
        return float(np.dot(diff, diff))  # squared L2

    def _random_level(self) -> int:
        level = 0
        while random.random() < (1.0 / self.M) and level < self.max_layers - 1:
            level += 1
        return level

    def _search_layer(self, query: np.ndarray, entry_id: int, ef: int, layer: int) -> list[int]:
        visited = {entry_id}
        entry_dist = self._distance(query, self.nodes[entry_id].vector)
        candidates = [(entry_dist, entry_id)]
        dynamic_list = [(entry_dist, entry_id)]

        while candidates:
            c_dist, c_id = heapq.heappop(candidates)
            worst_dist = max(d for d, _ in dynamic_list)
            if c_dist > worst_dist:
                break
            for nb_id in self.nodes[c_id].neighbors.get(layer, []):
                if nb_id not in visited:
                    visited.add(nb_id)
                    nb_dist = self._distance(query, self.nodes[nb_id].vector)
                    if nb_dist < worst_dist or len(dynamic_list) < ef:
                        heapq.heappush(candidates, (nb_dist, nb_id))
                        dynamic_list.append((nb_dist, nb_id))
                        if len(dynamic_list) > ef:
                            dynamic_list.remove(max(dynamic_list))

        return [nid for _, nid in sorted(dynamic_list)[:ef]]

    def add(self, vector: np.ndarray) -> int:
        node_id = len(self.nodes)
        node = HNSWNode(vector=vector)
        self.nodes.append(node)
        level = self._random_level()

        if self.entry_point is None:
            self.entry_point = node_id
            self.top_layer = level
            for l in range(level + 1):
                node.neighbors[l] = []
            return node_id

        ep = self.entry_point
        for l in range(self.top_layer, level, -1):
            candidates = self._search_layer(vector, ep, ef=1, layer=l)
            ep = candidates[0]

        for l in range(min(level, self.top_layer) + 1):
            neighbors = self._search_layer(vector, ep, ef=self.ef, layer=l)
            M = self.M0 if l == 0 else self.M
            neighbors = sorted(
                neighbors,
                key=lambda nid: self._distance(vector, self.nodes[nid].vector)
            )[:M]
            node.neighbors[l] = neighbors
            for nb_id in neighbors:
                nb_node = self.nodes[nb_id]
                if l not in nb_node.neighbors:
                    nb_node.neighbors[l] = []
                nb_node.neighbors[l].append(node_id)
                if len(nb_node.neighbors[l]) > M:
                    nb_node.neighbors[l] = sorted(
                        nb_node.neighbors[l],
                        key=lambda nid: self._distance(nb_node.vector, self.nodes[nid].vector)
                    )[:M]
            ep = neighbors[0] if neighbors else ep

        if level > self.top_layer:
            self.top_layer = level
            self.entry_point = node_id
        return node_id

    def search(self, query: np.ndarray, k: int = 10, ef: int = 50) -> list[tuple[float, int]]:
        if self.entry_point is None:
            return []
        ep = self.entry_point
        for l in range(self.top_layer, 0, -1):
            candidates = self._search_layer(query, ep, ef=1, layer=l)
            ep = candidates[0]
        candidates = self._search_layer(query, ep, ef=max(ef, k), layer=0)
        results = sorted(candidates, key=lambda nid: self._distance(query, self.nodes[nid].vector))
        return [(self._distance(query, self.nodes[nid].vector), nid) for nid in results[:k]]

Using the index

import numpy as np
from hnsw import HNSWIndex

# Build an index with 128-dimensional vectors
index = HNSWIndex(dim=128, M=16, ef_construction=200)

# Insert 10k random vectors (documents, embeddings, etc.)
np.random.seed(42)
n = 10_000
vectors = np.random.randn(n, 128).astype(np.float32)

print("Building index...")
for i, vec in enumerate(vectors):
    index.add(vec)
    if (i + 1) % 1000 == 0:
        print(f"  {i + 1}/{n} vectors inserted")

# Query with a random vector
query = np.random.randn(128).astype(np.float32)
results = index.search(query, k=5, ef=50)

print("\nTop 5 nearest neighbors (HNSW):")
for dist, node_id in results:
    print(f"  node_id={node_id}, sq_distance={dist:.4f}")

# Verify against brute force
all_dists = np.sum((vectors - query) ** 2, axis=1)
brute_top5 = sorted(enumerate(all_dists), key=lambda x: x[1])[:5]
print("\nBrute-force top 5:")
for nid, dist in brute_top5:
    print(f"  node_id={nid}, sq_distance={dist:.4f}")

On a 10k-vector dataset, you'll typically see 4 or 5 out of 5 results overlap between HNSW and brute force. That's the "approximate" in ANN: for most real workloads, occasionally missing one true neighbor is an acceptable trade-off for a 50x–100x query speedup.

Key parameters and their effect

M — max connections per node per layer. Higher M means better recall and a denser graph, at the cost of more memory and slower inserts. Start with 16. Go up to 32 if your recall consistently falls below 90%.

ef_construction — candidate pool size during index build. Larger values produce a better-connected graph but slow down insertion. 200 is a solid default; lower it to 100 if insertion throughput matters more than recall.

ef (search-time) — candidate pool size at query time. This is the most important lever for production tuning because you can adjust it without rebuilding the index. Setting ef = k gives the fastest queries; ef = 500 approaches brute-force recall. Plotting ef vs. recall typically reveals a knee around ef = 50–100 for most datasets — that's your sweet spot.

Understanding these parameters matters the moment your AI-powered search starts returning wrong results. Before blaming your embedding model, check ef and whether your index was built with sufficient ef_construction. It's the same diagnostic discipline that applies across security tooling — knowing the internals saves time when something breaks. The security hardening checklists we publish follow the same philosophy: understand the mechanism, then apply the control.

Production considerations

This implementation deliberately omits several things you'd need in production:

Deletion support: HNSW graphs don't support clean node removal without a full rebuild. Libraries like Qdrant implement soft-deletion with tombstones.
Persistence: serialize node vectors and neighbor lists to disk (numpy's np.save for vectors, JSON or msgpack for the neighbor maps).
Thread safety: concurrent inserts will corrupt the neighbor lists. Use a lock or batch inserts single-threaded.
SIMD distance computation: the Python distance function is the bottleneck. hnswlib's C++ core runs distance in vectorized instructions — easily 100x faster.

For any production workload, use hnswlib directly (pip install hnswlib) or a vector database that wraps it. The API is almost identical to what's shown above, and you get persistence, filtering, and proper concurrency for free.

The takeaway

HNSW is not magic — it's a graph with controlled connectivity. The layered structure lets queries skip large sections of the search space, and the ef parameter gives you a runtime dial between speed and accuracy.

Building a stripped-down version reveals what every production vector database is doing under the hood. That understanding pays off when you're debugging recall issues at 3am — and it's faster than re-reading library documentation when you already know what the parameters mean.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

LLM Fine-Tuning with LoRA: A Practical Guide for Developers

Ayi NEDJIMI — Mon, 07 Sep 2026 10:02:23 +0000

Pre-trained language models are powerful, but they're generalists. When you need a model that behaves consistently in a narrow domain — classifying security alerts, extracting structured data from medical records, or generating responses in your product's voice — fine-tuning is the right tool. LoRA (Low-Rank Adaptation) makes this tractable on hardware most developers actually own.

What LoRA Does (and Why the Math Is Worth Understanding)

Full fine-tuning a 7B parameter model requires updating billions of weights, which means tens of gigabytes of GPU memory just for gradients. LoRA sidesteps this by injecting small trainable matrices into the model's attention layers, leaving the original weights frozen.

The core insight: a weight update matrix ΔW can be approximated as a product of two low-rank matrices A and B, where rank r << min(d, k). Instead of updating a d×k matrix with d*k parameters, you update A (d×r) and B (r×k), totaling r*(d+k) parameters. For r=8 and d=k=4096 (typical for a 7B model), that's 65,536 trainable parameters versus 16,777,216 — a 256x reduction.

In practice, this means you can fine-tune a 7B model on a single A100 (or two consumer 3090s) with bfloat16 precision.

Setting Up a Training Run

Let's say you want to fine-tune a base model on security incident reports to extract structured fields. Here's a minimal, working setup using peft and trl.

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, TaskType
from trl import SFTTrainer

MODEL_ID = "mistralai/Mistral-7B-v0.3"

# QLoRA: 4-bit quantization + LoRA = fits on a single GPU
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_use_double_quant=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
)

model = AutoModelForCausalLM.from_pretrained(
    MODEL_ID,
    quantization_config=bnb_config,
    device_map="auto",
)
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
tokenizer.pad_token = tokenizer.eos_token

lora_config = LoraConfig(
    r=16,                 # rank — higher = more capacity, more params
    lora_alpha=32,        # scaling: typically 2*r
    target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type=TaskType.CAUSAL_LM,
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# trainable params: 6,815,744 || all params: 3,759,030,272 || trainable%: 0.18%

Key decisions:

r=16 is a reasonable default. Increase to 32 or 64 for complex domain-specific tasks. Decrease to 4 or 8 if you observe overfitting on a small dataset.
target_modules: q, k, v, and o covers the full attention mechanism. Add up_proj and down_proj for the MLP layers if you need more representational capacity.
QLoRA (4-bit quantization + LoRA) is the practical choice for single-GPU setups. The accuracy cost is small; the memory savings are large.

Data Formatting: The Real Source of Failed Runs

The biggest source of failed fine-tuning runs is poor data formatting — not hyperparameters, not model choice. Your training examples must match exactly the format you'll use at inference time.

For instruction-tuned models, use the model's chat template:

def format_example(example):
    messages = [
        {
            "role": "system",
            "content": (
                "You are a security analyst. Extract the following fields from "
                "the incident report as JSON: severity, affected_systems, "
                "attack_vector, indicators_of_compromise."
            )
        },
        {
            "role": "user",
            "content": example["raw_report"]
        },
        {
            "role": "assistant",
            "content": example["structured_output"]  # must be a valid JSON string
        }
    ]
    return {
        "text": tokenizer.apply_chat_template(
            messages,
            tokenize=False,
            add_generation_prompt=False
        )
    }

from datasets import load_dataset
dataset = load_dataset("your-org/security-incident-dataset", split="train")
dataset = dataset.map(format_example)

Two patterns that silently kill training quality:

Label leakage in the user turn: partial answers embedded in the prompt teach the model to shortcut rather than reason.
Inconsistent output format: if your assistant turns mix JSON, YAML, and prose, the model learns to be inconsistent too. Pick one schema and apply it everywhere.

If you're integrating fine-tuned models into a security workflow — alert classification, indicator tagging, incident correlation — enforcing strict JSON output with a fixed schema is the right call. It composes cleanly with downstream parsers and keeps the pipeline auditable. Our free security hardening checklists cover the broader system design side of AI-integrated security tooling.

Evaluating the Adapter Before You Merge

Don't merge the LoRA adapter into the base model until you've validated it. peft lets you load adapters without touching the base weights, so you can swap and compare:

from peft import PeftModel

# Load base in full precision for accurate evaluation
base_model = AutoModelForCausalLM.from_pretrained(
    MODEL_ID,
    torch_dtype=torch.bfloat16,
    device_map="auto"
)
model_with_adapter = PeftModel.from_pretrained(base_model, "./lora-output")
model_with_adapter.eval()

test_prompt = tokenizer.apply_chat_template(
    [{"role": "user", "content": test_report}],
    tokenize=False,
    add_generation_prompt=True
)
inputs = tokenizer(test_prompt, return_tensors="pt").to("cuda")

with torch.no_grad():
    outputs = model_with_adapter.generate(
        **inputs,
        max_new_tokens=512,
        temperature=0.1,
        do_sample=True,
    )
    # Decode only the generated tokens, not the prompt
    gen_tokens = outputs[0][inputs["input_ids"].shape[1]:]
    response = tokenizer.decode(gen_tokens, skip_special_tokens=True)
    print(response)

# When validated, merge for deployment
merged = model_with_adapter.merge_and_unload()
merged.save_pretrained("./merged-model")
tokenizer.save_pretrained("./merged-model")

Metrics to track during evaluation:

Format compliance rate: percentage of outputs that are valid JSON (or your target format)
Field extraction F1: compare against a held-out labeled set
General capability regression: test on a broader benchmark to confirm the model hasn't degraded — this happens with high-rank LoRA on small datasets

Hyperparameters That Actually Move the Needle

Parameter	Reasonable default	When to change
`r`	16	Increase for complex tasks; decrease to fight overfitting
`lora_alpha`	32 (= 2×r)	Keep at 2×r unless you see training instability
Learning rate	2e-4	Drop to 1e-4 if loss is noisy or oscillates
Batch size	4–8	Use gradient accumulation to simulate larger batches
Epochs	2–3	Fine-tuning rarely benefits from more than 3 epochs

Overfitting is the dominant failure mode: training loss drops steadily while validation loss plateaus or climbs. When that happens, reduce r, increase lora_dropout, cut epochs, or add more labeled data — in that order.

The Takeaway

LoRA gives you meaningful customization of large language models without requiring hyperscaler hardware. The workflow is straightforward: choose a base model already close to your task, format training data with a consistent schema, validate the adapter before merging, and watch for regression on general capabilities.

The pattern scales from 7B to 70B — the memory math changes, but the code structure doesn't. If you're serving in production, look at vLLM's LoRA endpoint, which lets you load and swap adapters at runtime without restarting the inference server. That's significantly easier to operate than maintaining one merged model per customer or use case.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Agentic AI Security: Sandboxing LLM Tool Calls in Production

Ayi NEDJIMI — Sun, 06 Sep 2026 10:01:56 +0000

When you give a language model the ability to call tools — execute code, read files, hit APIs — you create a new attack surface that most teams aren't ready for. A compromised prompt or a crafted user input can now trigger arbitrary commands, leak credentials, or exfiltrate data. Production agentic systems have already been exploited via prompt injection attacks that hijack tool calls. Sandboxing LLM tool calls is not optional in production. Here is how to do it.

Why Tool Calls Are a Different Threat Model

Traditional web API security is well-understood: validate inputs, sanitize outputs, authenticate callers. In an agentic system, the "caller" is a language model that produces natural language instructions converted to structured tool invocations. The threat model shifts in three ways:

Indirect prompt injection: an external document retrieved by the agent contains malicious instructions that redirect tool behavior. The model never knows it was manipulated.

Over-permissioned tools: the agent has access to a delete_file tool it never actually needs — but a crafted prompt can invoke it anyway.

Unconstrained arguments: tool call arguments are model-generated and may contain unexpected values even when the schema type checks pass.

Layer 1 — Tool Schema as the First Defense

The cleanest way to limit what an agent can do is to constrain what it can express. Use strict JSON schemas for every tool, and validate the model's output before execution — not after.

from jsonschema import validate, ValidationError
import json

TOOL_SCHEMAS = {
    "read_file": {
        "type": "object",
        "properties": {
            "path": {
                "type": "string",
                "pattern": "^/data/[a-zA-Z0-9_\\-]+\\.txt$"
            }
        },
        "required": ["path"],
        "additionalProperties": False
    }
}

def validate_tool_call(tool_name: str, arguments: dict) -> None:
    schema = TOOL_SCHEMAS.get(tool_name)
    if schema is None:
        raise ValueError(f"Unknown tool: {tool_name}")
    try:
        validate(instance=arguments, schema=schema)
    except ValidationError as e:
        raise ValueError(f"Invalid tool arguments: {e.message}")

def execute_tool(tool_name: str, raw_arguments: str) -> str:
    arguments = json.loads(raw_arguments)
    validate_tool_call(tool_name, arguments)
    return dispatch_tool(tool_name, arguments)

The pattern on path is the critical detail. Without it, a model producing {"path": "../../etc/passwd"} passes type validation just fine. Regex allowlisting is faster and more reliable than blocklisting traversal sequences.

Layer 2 — Process Isolation for Code Execution

If your agent executes code — Python sandboxes, shell commands, SQL — schema validation is not enough. You need OS-level isolation. Docker is the minimum; gVisor or Firecracker microvms are better for untrusted workloads. Here is a minimal secure code execution wrapper using subprocess with hard limits:

import subprocess
import resource
import tempfile
from pathlib import Path

def run_sandboxed_python(code: str, timeout: int = 5) -> dict:
    """Execute model-generated Python in a restricted subprocess.
    No network, capped CPU and memory, no new processes."""
    with tempfile.NamedTemporaryFile(suffix=".py", mode="w", delete=False) as f:
        f.write(code)
        script_path = f.name

    def set_limits():
        resource.setrlimit(resource.RLIMIT_CPU, (5, 5))           # 5s CPU
        resource.setrlimit(resource.RLIMIT_AS,
                           (128 * 1024 * 1024, 128 * 1024 * 1024))  # 128 MB RAM
        resource.setrlimit(resource.RLIMIT_NPROC, (0, 0))          # no fork

    try:
        result = subprocess.run(
            ["python3", "-u", script_path],
            capture_output=True,
            text=True,
            timeout=timeout,
            preexec_fn=set_limits,
            env={"PATH": "/usr/bin:/bin", "HOME": "/tmp"}
        )
        return {
            "stdout": result.stdout[:4096],
            "stderr": result.stderr[:1024],
            "returncode": result.returncode
        }
    except subprocess.TimeoutExpired:
        return {"error": "timeout", "stdout": "", "stderr": ""}
    finally:
        Path(script_path).unlink(missing_ok=True)

This does not replace containerization — it is one layer in a defence-in-depth stack. In production, combine it with a Docker container that has --network none, a read-only root filesystem, and a seccomp profile.

Layer 3 — Tool Permission Scoping Per Agent Role

Every agent should operate under the minimal tool set its task actually requires. Define tool ACLs per role, not globally.

from enum import Enum
from dataclasses import dataclass, field

class ToolPermission(Enum):
    READ_FILE  = "read_file"
    WRITE_FILE = "write_file"
    RUN_CODE   = "run_code"
    HTTP_GET   = "http_get"
    HTTP_POST  = "http_post"
    DB_READ    = "db_read"
    DB_WRITE   = "db_write"

@dataclass
class AgentRole:
    name: str
    allowed_tools: set[ToolPermission] = field(default_factory=set)

ANALYST_ROLE = AgentRole(
    name="analyst",
    allowed_tools={ToolPermission.READ_FILE, ToolPermission.DB_READ}
)

def check_permission(role: AgentRole, tool: ToolPermission) -> None:
    if tool not in role.allowed_tools:
        raise PermissionError(
            f"Role '{role.name}' cannot call '{tool.value}'"
        )

This pattern maps cleanly onto existing RBAC systems. An analyst agent that only reads data cannot be tricked into writing or posting, regardless of what the model produces. The blast radius of any successful injection is bounded by the role definition — not by hope.

For teams building this from scratch, the LLM deployment section of our security hardening checklists covers role scoping decisions end-to-end, including how to handle multi-agent pipelines where agents spawn sub-agents with inherited permissions.

Layer 4 — Audit Logging Every Tool Call

Every tool invocation should be logged with enough context to reconstruct the full chain of events: session ID, role, tool name, argument hash, result length, and a short preview. Feed these into your SIEM or anomaly detection layer.

import logging
import json
from datetime import datetime, timezone

tool_logger = logging.getLogger("tool_audit")

def audited_tool_call(session_id: str, role: AgentRole,
                      tool_name: str, arguments: dict, result: str) -> None:
    tool_logger.info(json.dumps({
        "ts":           datetime.now(timezone.utc).isoformat(),
        "session_id":   session_id,
        "role":         role.name,
        "tool":         tool_name,
        "args_hash":    hash(json.dumps(arguments, sort_keys=True)),
        "args_preview": str(arguments)[:200],
        "result_len":   len(result),
        "result_preview": result[:100]
    }))

Unusual patterns — an agent calling http_post ten times in a minute, accessing paths outside its normal working set, or suddenly invoking write tools after behaving read-only for a month — are worth alerting on. The log structure above gives you the fields you need for both rule-based and ML-based anomaly detection.

The Takeaway

Agentic AI security is not one control: it is a layered system. Schema validation catches obvious injection attempts at the input boundary. Process isolation contains code execution at the OS level. Tool permission scoping limits blast radius at the role level. Audit logs give you forensic visibility after the fact.

The single most common mistake is giving agents more tools than they need "for flexibility." Before you deploy any agentic system to production, enumerate every tool call it can make and ask: if this call were hijacked, what is the worst case? If the answer is "anything," you have work to do.

Start with the smallest possible tool set. Add tools only when the agent demonstrably needs them. Audit everything. That approach holds regardless of which underlying language model you use.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

How to Evaluate RAG Pipeline Quality: Metrics and Test Harness

Ayi NEDJIMI — Sat, 05 Sep 2026 10:02:53 +0000

Most teams ship a RAG pipeline, run a few manual tests, and call it done. Then users start complaining that answers are wrong, incomplete, or making things up. The problem is almost never the language model itself — it's that you have no systematic way to measure what's failing. This article shows you how to build a test harness that catches retrieval and generation failures before they reach production.

Why RAG evaluation is harder than it looks

RAG pipelines fail in two independent ways: the retriever surfaces the wrong chunks, or the language model ignores the retrieved context and generates text from its training data instead. These failures look identical to the user (a wrong answer) but have completely different fixes.

Standard ML accuracy metrics don't apply here. There's no single ground-truth label — just a pipeline where retrieval quality and generation quality can silently degrade whenever you change your embedding model, index configuration, or prompt. Most teams don't discover the regression until it's already in front of users.

You need three measurements at minimum: context recall (did the retriever surface what was needed?), faithfulness (does the answer stay within the retrieved context?), and answer relevancy (is the answer actually responsive to the question?).

Building the data structures

Start with minimal Python dataclasses to represent a test case and a result:

from dataclasses import dataclass

@dataclass
class RAGTestCase:
    question: str
    expected_answer: str
    ground_truth_contexts: list[str]  # passages that MUST appear in retrieval

@dataclass
class RAGResult:
    question: str
    retrieved_contexts: list[str]
    generated_answer: str

def context_recall(result: RAGResult, test_case: RAGTestCase) -> float:
    """
    Fraction of ground-truth contexts covered by retrieved contexts.
    Uses substring containment; swap for cosine similarity for fuzzy matching.
    """
    if not test_case.ground_truth_contexts:
        return 1.0

    hits = 0
    for gt in test_case.ground_truth_contexts:
        gt_stripped = gt.strip()
        for ret in result.retrieved_contexts:
            ret_stripped = ret.strip()
            if gt_stripped in ret_stripped or ret_stripped in gt_stripped:
                hits += 1
                break

    return hits / len(test_case.ground_truth_contexts)

Context recall measures the retriever in isolation. If this number is low, tuning your prompt won't help — fix the index or the chunking strategy first.

Measuring faithfulness with a judge model

Faithfulness is harder to compute without a model. You need to check whether every claim in the generated answer is actually supported by the retrieved passages. String matching won't catch paraphrasing or subtle hallucinations.

The standard approach is to use a language model as a judge — send it the answer and the contexts, and ask it to flag unsupported claims:

import httpx
import json

def check_faithfulness(
    answer: str,
    contexts: list[str],
    api_url: str,
    api_key: str,
    model: str = "gpt-4o-mini",
) -> dict:
    """
    Returns {"faithful": bool, "score": float, "unsupported_claims": list[str]}
    Score 1.0 = every claim grounded in the provided contexts.
    """
    context_block = "\n\n".join(
        f"[Context {i+1}]\n{ctx}" for i, ctx in enumerate(contexts)
    )

    prompt = f"""You are evaluating whether an AI-generated answer is faithful to its source documents.

SOURCE CONTEXTS:
{context_block}

GENERATED ANSWER:
{answer}

Instructions:
- Identify every factual claim in the GENERATED ANSWER.
- For each claim, check if it is explicitly supported by the SOURCE CONTEXTS.
- Do NOT penalize for rephrasing — only flag claims that add new information not present in the contexts.

Return a JSON object with:
- "faithful": true if all claims are supported, false otherwise
- "score": float 0.0-1.0 (proportion of claims that are supported)
- "unsupported_claims": list of unsupported claim strings (empty list if faithful)

Return only valid JSON, no markdown."""

    resp = httpx.post(
        f"{api_url}/v1/chat/completions",
        headers={"Authorization": f"Bearer {api_key}"},
        json={
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0,
            "response_format": {"type": "json_object"},
        },
        timeout=30,
    )
    resp.raise_for_status()
    return json.loads(resp.json()["choices"][0]["message"]["content"])

One important detail: set temperature=0 for judge calls. You want deterministic verdicts, not creative interpretations. For high-stakes use cases (compliance, regulated environments), run the judge call twice with different prompt phrasings and take the conservative score.

Building a synthetic golden dataset

The bottleneck in RAG evaluation is almost always data — you need question-context-answer triplets to test against, and most teams don't have them. The fastest path is synthetic generation: use a language model to generate questions from your document corpus, then have domain experts review a random sample.

def generate_test_cases(
    documents: list[str],
    api_url: str,
    api_key: str,
    questions_per_doc: int = 3,
    model: str = "gpt-4o-mini",
) -> list[RAGTestCase]:
    """Generate synthetic test cases from your document corpus."""
    test_cases = []

    for doc in documents:
        prompt = f"""Generate {questions_per_doc} questions answerable using ONLY the text below.

DOCUMENT:
{doc[:3000]}

Rules:
- Each question must be answerable from this document alone.
- Avoid yes/no questions; prefer specific factual questions.
- Include the correct answer for each question.

Return a JSON array of {{"question": "...", "answer": "..."}} objects."""

        resp = httpx.post(
            f"{api_url}/v1/chat/completions",
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "model": model,
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.4,
                "response_format": {"type": "json_object"},
            },
            timeout=30,
        )
        resp.raise_for_status()

        raw = json.loads(resp.json()["choices"][0]["message"]["content"])
        items = raw if isinstance(raw, list) else next(iter(raw.values()), [])

        for item in items:
            if "question" in item and "answer" in item:
                test_cases.append(RAGTestCase(
                    question=item["question"],
                    expected_answer=item["answer"],
                    ground_truth_contexts=[doc],
                ))

    return test_cases

Even 50–100 test cases covering your most important document types is enough to catch regressions. Add real user queries to the dataset as you collect them — production signal is always more valuable than synthetic data.

Running the evaluation and setting CI thresholds

Wire everything together into a runner that produces a structured report:

import statistics

def run_evaluation(
    test_cases: list[RAGTestCase],
    rag_fn,        # callable(question: str) -> RAGResult
    api_url: str,
    api_key: str,
) -> dict:
    records = []

    for tc in test_cases:
        result = rag_fn(tc.question)
        recall = context_recall(result, tc)
        faith = check_faithfulness(
            result.generated_answer,
            result.retrieved_contexts,
            api_url,
            api_key,
        )
        records.append({
            "question": tc.question,
            "context_recall": recall,
            "faithfulness_score": faith["score"],
            "faithful": faith["faithful"],
            "unsupported_claims": faith.get("unsupported_claims", []),
        })

    return {
        "n": len(records),
        "avg_context_recall": round(statistics.mean(r["context_recall"] for r in records), 3),
        "avg_faithfulness": round(statistics.mean(r["faithfulness_score"] for r in records), 3),
        "fully_faithful_pct": round(sum(1 for r in records if r["faithful"]) / len(records), 3),
        "failed_cases": [r for r in records if not r["faithful"] or r["context_recall"] < 0.5],
    }

For CI gates, use these thresholds as a starting point:

Metric	Minimum threshold
Context recall	≥ 0.75
Average faithfulness	≥ 0.85
Fully faithful answers	≥ 70%

If retrieval recall drops below 0.75, answers will be systematically incomplete regardless of how good your model is. If faithfulness drops below 0.85, you have a hallucination problem that no amount of prompt tuning will fix — you need better retrieval or a different chunking strategy.

For regulated environments (compliance, security audits), push the faithfulness threshold to ≥ 0.95. An answer that adds unsupported claims in a compliance context is not just wrong — it's a liability. I keep security-focused checklists for AI pipelines that cover evaluation gates alongside other controls for regulated deployments.

The takeaway

RAG evaluation comes down to three things: measure the retriever and generator separately, build a golden dataset before you need it (not after), and gate every model or index change on your metrics.

Libraries like RAGAS and DeepEval implement these patterns with more statistical rigor. But understanding what the metrics actually measure means you can debug a drop in faithfulness score instead of just observing it. That's the difference between a RAG pipeline you can iterate on confidently and one that regresses silently every time someone touches the index.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.