DEV Community: Ayi NEDJIMI

Building a Production-Ready AI Chatbot with Memory and Context

Ayi NEDJIMI — Fri, 09 Oct 2026 10:01:55 +0000

Building a chatbot that holds a coherent conversation for more than a few turns is harder than it looks. Language models are stateless — every API call starts fresh, with no memory of anything said before. The "memory" in polished chatbot products is maintained entirely by the application layer. Getting that right for production means making a set of architectural decisions up front that compound as the codebase grows.

The Core Problem: Stateless Models and Token Cost

A language model only knows what you put in the prompt. If you want it to remember that a user said "I prefer Go examples" four messages ago, you have to include that exchange in the current request.

The naive approach is full conversation replay: append every past turn to the messages array and send the whole thing on every call.

import os
import anthropic

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

def chat(history: list[dict], user_message: str) -> str:
    history.append({"role": "user", "content": user_message})
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system="You are a helpful technical assistant.",
        messages=history,
    )
    reply = response.content[0].text
    history.append({"role": "assistant", "content": reply})
    return reply

This works until the conversation grows past the context window, or until your token bill does. A 50-turn conversation at 200 tokens per turn costs 10,000 input tokens on every single request — most of it repeated context the model already processed in the previous call.

Strategy 1: Sliding Window

The simplest fix is a sliding window: keep only the last N turns verbatim.

MAX_TURNS = 20  # each turn = 1 user + 1 assistant message

def trim_history(history: list[dict]) -> list[dict]:
    max_entries = MAX_TURNS * 2
    if len(history) > max_entries:
        return history[-max_entries:]
    return history

Call trim_history before every LLM request. Input tokens stay bounded regardless of session length.

The downside is abrupt amnesia: the model stops knowing anything said before turn 21. For task-oriented chatbots — support flows, form filling, one-session Q&A — this is usually fine. For anything requiring continuity across a longer conversation, it breaks the experience.

Strategy 2: Summarization-Based Memory

Instead of discarding old context, compress it. When history crosses a threshold, summarize everything before the last N turns and inject that summary into the system prompt.

def summarize_old_turns(old_history: list[dict], client) -> str:
    prompt = (
        "Summarize the key facts, preferences, and decisions from this conversation "
        "in 3-5 bullet points. Be specific and concise.\n\n"
        + "\n".join(
            f"{m['role'].upper()}: {m['content']}"
            for m in old_history
        )
    )
    response = client.messages.create(
        model="claude-3-haiku-20240307",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.content[0].text


def chat_with_memory(
    history: list[dict],
    warm_summary: str,
    user_message: str,
    client,
) -> tuple[str, str]:
    KEEP_VERBATIM = 10  # turns to keep verbatim

    if len(history) > KEEP_VERBATIM * 2:
        old = history[: -KEEP_VERBATIM * 2]
        new_summary = summarize_old_turns(old, client)
        warm_summary = f"{warm_summary}\n{new_summary}".strip() if warm_summary else new_summary
        history = history[-KEEP_VERBATIM * 2 :]

    system = "You are a helpful technical assistant."
    if warm_summary:
        system += f"\n\nConversation context:\n{warm_summary}"

    history.append({"role": "user", "content": user_message})
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system=system,
        messages=history,
    )
    reply = response.content[0].text
    history.append({"role": "assistant", "content": reply})
    return reply, warm_summary

The summarization call uses a faster, cheaper model — it is pure compression, not generation. In practice this approach preserves the semantics that matter while keeping prompt size predictable.

Strategy 3: Semantic Memory with Vector Retrieval

For persistent sessions — where users come back days later and expect continuity — a vector store gives you addressable long-term memory. Instead of replaying the whole session, you retrieve only the turns relevant to the current query.

from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticMemory:
    def __init__(self):
        self.encoder = SentenceTransformer("all-MiniLM-L6-v2")
        self._texts: list[str] = []
        self._embeddings: list[np.ndarray] = []

    def store(self, text: str) -> None:
        vec = self.encoder.encode(text, normalize_embeddings=True)
        self._texts.append(text)
        self._embeddings.append(vec)

    def retrieve(self, query: str, top_k: int = 3) -> list[str]:
        if not self._embeddings:
            return []
        q_vec = self.encoder.encode(query, normalize_embeddings=True)
        scores = [float(np.dot(q_vec, e)) for e in self._embeddings]
        ranked = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)
        return [self._texts[i] for i in ranked[:top_k]]

In production, swap the in-memory list for pgvector, Qdrant, or any other vector database — the interface stays the same, only the backend changes. Store one entry per significant exchange, not every single turn; store facts, not filler.

The Production Architecture

The pattern that holds up under load layers all three strategies:

Hot memory: the last N turns, verbatim, always in the prompt
Warm memory: a rolling summary of earlier turns, injected as a system note
Cold memory: vector-retrieved facts from persistent storage, pulled per query

At each turn: retrieve relevant cold memories, prepend them to the system prompt, append the warm summary, append hot history, append the user message. Total input tokens stay bounded regardless of session length or how long the user has been a customer.

One operational detail teams miss: summarization and embedding writes should be asynchronous. If either happens synchronously in the request path, users feel the latency. Offload them to a background task queue and write through a cache.

For the storage layer, securing these memory backends matters — especially when they contain personal data or proprietary context. Our security hardening checklists cover encryption at rest and access control patterns for both relational and vector databases.

The Takeaway

A chatbot without memory management is a prototype. The model is stateless by design — your application layer owns the continuity. The three strategies (window trimming, summarization, semantic retrieval) are not alternatives; mature systems layer all three.

Start with the sliding window. Add summarization when context loss hurts output quality. Add semantic retrieval when sessions go persistent or cross-session. Do not add complexity before the problem is real — but do design the interface to support it from the start.

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

Implementing graceful shutdown in Go HTTP servers

Ayi NEDJIMI — Wed, 07 Oct 2026 10:03:19 +0000

When you send SIGTERM to a Go HTTP server that's handling 50 concurrent requests, all 50 die instantly. No cleanup, no drain, no responses. In production this means failed API calls, half-written database rows, and clients retrying on errors that shouldn't have happened. Graceful shutdown is the difference between a clean deployment and an incident.

Why Abrupt Shutdown Breaks Things

The default net/http server has no concept of "wait for active connections." When you kill the process, the OS closes all file descriptors, and every in-flight HTTP connection gets a TCP RST. Your clients see connection errors. Your database sees abandoned transactions.

Three scenarios where this matters most:

Zero-downtime deploys with rolling restarts (Kubernetes, ECS, Nomad) — the old pod receives a SIGTERM while the load balancer is still routing traffic to it
Long-running requests like file uploads, batch processing, or SSE streams
Background workers triggered by HTTP handlers that need to finish before the process exits

The Basics: Catching SIGTERM and Draining Connections

Go's net/http package has had Server.Shutdown since Go 1.8. Here's the minimal pattern:

package main

import (
    "context"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"
)

func main() {
    mux := http.NewServeMux()
    mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
        w.WriteHeader(http.StatusOK)
        w.Write([]byte("ok"))
    })

    srv := &http.Server{
        Addr:         ":8080",
        Handler:      mux,
        ReadTimeout:  10 * time.Second,
        WriteTimeout: 30 * time.Second,
        IdleTimeout:  60 * time.Second,
    }

    // Start server in a goroutine so it doesn't block
    go func() {
        log.Println("Server listening on :8080")
        if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Fatalf("ListenAndServe: %v", err)
        }
    }()

    // Block until we receive SIGTERM or SIGINT
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGTERM, syscall.SIGINT)
    <-quit

    log.Println("Shutdown signal received")

    // Give in-flight requests 30 seconds to complete
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    if err := srv.Shutdown(ctx); err != nil {
        log.Fatalf("Server forced to shutdown: %v", err)
    }

    log.Println("Server exited cleanly")
}

srv.Shutdown(ctx) stops accepting new connections, then waits for active connections to finish or for the context to expire — whichever comes first. http.ErrServerClosed is the expected error from ListenAndServe after Shutdown is called; you need to ignore it.

Handling Long-Running Requests With Context Propagation

The simple pattern above has a blind spot: a handler that ignores its context will block until the shutdown timeout expires, even if the work is already futile. Handlers need to respect r.Context() cancellation signals.

Here's a handler that respects cancellation:

func longRunningHandler(w http.ResponseWriter, r *http.Request) {
    workDone := make(chan struct{})

    go func() {
        defer close(workDone)
        // Simulate work: in production this is a DB query, external API call, etc.
        time.Sleep(10 * time.Second)
    }()

    select {
    case <-workDone:
        w.WriteHeader(http.StatusOK)
        w.Write([]byte(`{"status":"done"}`))
    case <-r.Context().Done():
        log.Printf("Request cancelled: %v", r.Context().Err())
        // Don't write — the connection is already closing
        return
    }
}

When Shutdown is called, Go cancels the context of all active requests. Handlers that select on r.Context().Done() can exit early instead of holding up the drain window.

For database calls, pass the request context directly:

func userHandler(db *sql.DB) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        row := db.QueryRowContext(r.Context(),
            "SELECT name FROM users WHERE id = $1",
            r.URL.Query().Get("id"),
        )
        var name string
        if err := row.Scan(&name); err != nil {
            if r.Context().Err() != nil {
                return // shutdown in progress, don't write error response
            }
            http.Error(w, "not found", http.StatusNotFound)
            return
        }
        w.Write([]byte(name))
    }
}

Coordinating Background Workers

HTTP handlers often enqueue work — sending emails, processing files, updating search indexes. If those workers run as goroutines, they can outlive the HTTP server or get cut off mid-job. The right tool is sync.WaitGroup:

type App struct {
    wg sync.WaitGroup
}

func (a *App) submitJob(ctx context.Context, payload []byte) {
    a.wg.Add(1) // increment before spawning the goroutine
    go func() {
        defer a.wg.Done()
        if err := processPayload(ctx, payload); err != nil {
            log.Printf("job failed: %v", err)
        }
    }()
}

func (a *App) shutdown(srv *http.Server, timeout time.Duration) {
    ctx, cancel := context.WithTimeout(context.Background(), timeout)
    defer cancel()

    // Stop accepting HTTP requests first
    srv.Shutdown(ctx)

    // Then wait for background jobs within the same deadline
    done := make(chan struct{})
    go func() {
        a.wg.Wait()
        close(done)
    }()

    select {
    case <-done:
        log.Println("All background jobs completed")
    case <-ctx.Done():
        log.Println("Shutdown timeout: some jobs may be incomplete")
    }
}

Two details worth calling out: the wg.Add(1) must happen before the goroutine starts — incrementing inside the goroutine is a race condition if wg.Wait() is called concurrently. And sharing the same context deadline between HTTP drain and worker drain keeps the total shutdown window predictable.

Kubernetes and the SIGTERM → SIGKILL Gap

In Kubernetes, pod termination follows this sequence:

Pod gets removed from Endpoints (stops receiving new traffic)
SIGTERM is sent to the container
After terminationGracePeriodSeconds (default: 30s), SIGKILL arrives

The catch: steps 1 and 2 happen in parallel, not sequentially. iptables rules may not have propagated before your server receives SIGTERM, which means requests can still arrive for a few seconds after shutdown starts.

The practical fix is a pre-shutdown sleep:

<-quit
log.Println("SIGTERM received, waiting for load balancer drain")
time.Sleep(5 * time.Second) // wait for iptables/IPVS propagation
srv.Shutdown(ctx)

Five seconds covers most clusters. Pair this with returning 503 from your readiness probe immediately after SIGTERM — that makes the load balancer stop routing to you faster. For a full deployment readiness checklist covering signal handling and zero-downtime releases, the AYI NEDJIMI hardening checklists have a dedicated section.

The Takeaway

Graceful shutdown in Go requires three things working together: catching SIGTERM, calling srv.Shutdown(ctx) with a realistic timeout, and making sure every handler and background worker respects context cancellation. Skip any one of these and you either drop requests or get goroutine leaks at shutdown time.

The patterns above cover the 95% case. The remaining 5% — server-sent events with long-lived connections, Unix domain sockets, streaming responses — follow the same principles but need explicit stream termination logic. Start with this foundation and add edge case handling as your specific workload demands it.

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 — Tue, 06 Oct 2026 10:01:29 +0000

Debugging a production incident means combing through hundreds of thousands of log lines. By the time you find the root cause, the outage has already cost you an hour. An LLM-backed log summarizer doesn't replace your observability stack — but it gives your on-call engineer a head start that matters.

The Problem with Raw Logs

Log aggregation tools like Loki, Elasticsearch, and Datadog are excellent at storing and searching logs, but they leave interpretation entirely to you. You run a query, you get 10,000 lines, and you start scrolling.

The gap isn't storage or retrieval — it's comprehension. You need someone (or something) to read a window of logs and say: "Between 14:23 and 14:31, the payment service returned 503 errors on every POST to /checkout. Three upstream timeout messages from the Redis connection pool preceded each failure."

A language model can produce exactly that — if you feed it the right data and give it the right prompt.

Architecture: What We're Building

The tool has three parts:

A log collector — reads from a file, stdin, or a log query API
A preprocessor — deduplicates noise, truncates to stay within token limits
An LLM summarizer — sends the cleaned logs and returns a structured summary

We'll keep it as a Python CLI tool with no framework dependencies beyond httpx for the API call. One script, one responsibility.

Collecting and Preprocessing Logs

Raw logs have two problems: volume and repetition. Before sending anything to a language model, you need to reduce both.

import re
from collections import Counter
from typing import Iterator

MAX_LINES = 200
NOISE_PATTERNS = [
    re.compile(r'health_?check', re.IGNORECASE),
    re.compile(r'GET /ping'),
    re.compile(r'keepalive'),
]

def is_noise(line: str) -> bool:
    return any(p.search(line) for p in NOISE_PATTERNS)

def deduplicate(lines: list[str], threshold: int = 5) -> list[str]:
    """Keep first N occurrences of lines that repeat more than threshold times."""
    counts: Counter = Counter()
    result = []
    for line in lines:
        key = re.sub(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}[^ ]*', '<ts>', line)
        key = re.sub(r'\b\d+\b', '<n>', key)
        counts[key] += 1
        if counts[key] <= threshold:
            result.append(line)
    return result

def preprocess(lines: Iterator[str]) -> str:
    cleaned = [l.rstrip() for l in lines if l.strip() and not is_noise(l)]
    deduped = deduplicate(cleaned)
    truncated = deduped[-MAX_LINES:]  # keep most recent
    return "\n".join(truncated)

The deduplicate function normalizes timestamps and numbers before counting, so "connection refused at 14:23:01" and "connection refused at 14:23:07" count as the same pattern. You keep the first five occurrences, then stop — the language model doesn't need to see the same error 400 times to understand it's a pattern.

The MAX_LINES = 200 cap is conservative. With typical log verbosity, 200 lines fits in 3–4k tokens, leaving plenty of room for the model's response within a 16k context window.

Summarizing with an LLM

The summarization call is straightforward. The key is the prompt: you want the model to identify errors, correlate events, and surface the probable root cause — not paraphrase the logs.

import httpx
import os

API_URL = "https://api.openai.com/v1/chat/completions"  # any OpenAI-compatible endpoint

SYSTEM_PROMPT = """You are a DevOps incident assistant. Given a block of application logs,
produce a structured summary with these sections:
- Timeline: key events with approximate timestamps
- Errors: list of distinct error types and their count
- Root cause hypothesis: your best guess at what caused the issue
- Recommended next steps: 2-3 concrete investigation actions

Be concise. If you are uncertain, say so. Do not invent details not present in the logs."""

def summarize_logs(log_text: str, model: str = "gpt-4o-mini") -> str:
    token = os.environ["OPENAI_API_KEY"]
    payload = {
        "model": model,
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"<logs>\n{log_text}\n</logs>"},
        ],
        "temperature": 0.2,
        "max_tokens": 800,
    }
    with httpx.Client(timeout=30) as client:
        resp = client.post(
            API_URL,
            headers={"Authorization": f"Bearer {token}"},
            json=payload,
        )
        resp.raise_for_status()
        return resp.json()["choices"][0]["message"]["content"]

A few things worth noting:

temperature=0.2 keeps output factual. Higher values produce more varied but less reliable summaries.
The <logs> XML-style wrapper helps the model separate log content from instructions.
max_tokens=800 is enough for a useful summary without wasting budget on padding.

This function works with any OpenAI-compatible API, including local models served via Ollama or vLLM. Swap the API_URL and remove the auth header for local inference.

Putting It Together: A CLI Tool

#!/usr/bin/env python3
import sys
import argparse
from pathlib import Path

def main():
    parser = argparse.ArgumentParser(description="Summarize logs with an LLM")
    parser.add_argument("file", nargs="?", help="Log file (default: stdin)")
    parser.add_argument("--model", default="gpt-4o-mini")
    args = parser.parse_args()

    if args.file:
        lines = Path(args.file).read_text().splitlines(keepends=True)
    else:
        lines = sys.stdin

    log_text = preprocess(lines)
    if not log_text.strip():
        print("No meaningful log content after preprocessing.", file=sys.stderr)
        sys.exit(1)

    summary = summarize_logs(log_text, model=args.model)
    print(summary)

if __name__ == "__main__":
    main()

Run it against a file or pipe a live stream directly:

# From a file
python log_summarizer.py /var/log/app/production.log

# From a live Kubernetes tail
kubectl logs -n production deploy/api-server --since=10m | python log_summarizer.py

The kubectl pipe pattern is where this shines. During an active incident, you pipe the last 10 minutes of logs and get a structured summary in about 3 seconds — faster than reading the same window yourself, with output you can paste directly into your incident channel.

Practical Limits to Know Before You Ship

Log sensitivity. Application logs frequently contain PII, session tokens, or internal IP addresses. Before sending logs to a hosted language model, run a redaction pass. A basic regex sweep for email patterns, bearer tokens, and internal hostnames is enough for most cases. For a checklist covering what to scrub before sending logs to any external service, the security hardening checklists at AYI NEDJIMI Consultants include a data-before-LLM section.

Context window exhaustion. The MAX_LINES cap is a safety rail, but measure token count before sending if you're dealing with verbose services. Libraries like tiktoken (for OpenAI models) let you count tokens before the API call and bail early if the window would overflow.

Hallucination on sparse data. If your preprocessed log window is only 15 lines, the model has almost nothing to work with and will often produce a plausible-sounding but fabricated root cause. Add a minimum threshold: if fewer than 20 lines survive preprocessing, output them raw and skip the LLM call entirely.

Cost. With gpt-4o-mini at roughly $0.15 per 1M input tokens, 200 lines of logs (about 4k tokens) costs less than a tenth of a cent per invocation. Running this on every deploy or on-call page is fine. Running it in a hot loop on every HTTP error is not.

The Takeaway

The hard part isn't the code — it's getting the system prompt to reliably produce structured output. Start with a fixed log source, a single service, and iterate on the prompt until the model consistently delivers the timeline/errors/hypothesis format you want. Then generalize.

This also works as an onboarding tool: new engineers can pipe an unfamiliar service's logs and get a plain-language explanation of what's happening without needing to know every log format on day one. The 80 lines of Python takes an afternoon; the prompt takes longer — but once it's tuned, it earns its keep in the first real incident.

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 — Sun, 04 Oct 2026 10:02:42 +0000

Your LLM-powered app is in production. Users are hitting it. Something is slow — or wrong — and you have no idea what. You can't reproduce it locally, and your generic APM shows... a single HTTP call to an external API. That's the problem with LLM observability today: the gap between "a request happened" and "here's what the model did, at what cost, with what latency" is usually empty.

This article shows how to close that gap with a tracing layer you can build in an afternoon.

What You Need to Trace

Before writing any code, define what actually matters for LLM calls:

Latency breakdown: time to first token (TTFT) vs. total completion time
Token usage: prompt tokens, completion tokens, cached tokens (if your provider supports it)
Cost: calculated per call, not just per month
Model and version: which model handled the request
Input/output sampling: truncated copies of the prompt and response for debugging
Errors: rate limits, timeouts, content policy rejections

This is different from what a generic APM captures. OpenTelemetry traces give you spans and HTTP status codes. They won't tell you "prompt tokens cost $0.0023 and the model refused because of the system prompt." You need a layer that understands LLM semantics.

Building a Wrapper Tracer in Python

The cleanest approach is a thin wrapper around your LLM client. You keep all existing call sites, add one decorator, and tracing is transparent.

import time
import json
import uuid
import logging
from dataclasses import dataclass, field
from typing import Any

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

@dataclass
class LLMTrace:
    trace_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    model: str = ""
    prompt_tokens: int = 0
    completion_tokens: int = 0
    cached_tokens: int = 0
    latency_ms: float = 0.0
    ttft_ms: float = 0.0
    cost_usd: float = 0.0
    error: str | None = None
    response_preview: str = ""

# Cost per 1M tokens (input / output)
COST_TABLE = {
    "claude-opus-5":      (15.0, 75.0),
    "claude-sonnet-5":    (3.0, 15.0),
    "claude-haiku-4-5":   (0.8, 4.0),
    "gpt-4o":             (2.5, 10.0),
    "gpt-4o-mini":        (0.15, 0.6),
}

def calculate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    costs = COST_TABLE.get(model, (0.0, 0.0))
    return (prompt_tokens / 1_000_000 * costs[0]) + (completion_tokens / 1_000_000 * costs[1])


def trace_llm_call(func):
    import functools

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        trace = LLMTrace()
        start = time.monotonic()
        try:
            result = func(*args, **kwargs)
            trace.latency_ms = (time.monotonic() - start) * 1000
            _extract_from_response(trace, result)
            return result
        except Exception as e:
            trace.latency_ms = (time.monotonic() - start) * 1000
            trace.error = type(e).__name__ + ": " + str(e)[:200]
            raise
        finally:
            _emit(trace)

    return wrapper


def _extract_from_response(trace: LLMTrace, response: Any) -> None:
    usage = getattr(response, "usage", None)
    if usage:
        trace.prompt_tokens = (
            getattr(usage, "input_tokens", 0) or
            getattr(usage, "prompt_tokens", 0)
        )
        trace.completion_tokens = (
            getattr(usage, "output_tokens", 0) or
            getattr(usage, "completion_tokens", 0)
        )

    model = getattr(response, "model", "")
    trace.model = model
    trace.cost_usd = calculate_cost(model, trace.prompt_tokens, trace.completion_tokens)

    content = getattr(response, "content", None) or getattr(response, "choices", None)
    if content and isinstance(content, list):
        if hasattr(content[0], "text"):
            trace.response_preview = content[0].text[:300]
        elif hasattr(content[0], "message"):
            trace.response_preview = content[0].message.content[:300]


def _emit(trace: LLMTrace) -> None:
    logger.info(json.dumps({
        "trace_id": trace.trace_id,
        "model": trace.model,
        "prompt_tokens": trace.prompt_tokens,
        "completion_tokens": trace.completion_tokens,
        "latency_ms": round(trace.latency_ms, 1),
        "cost_usd": round(trace.cost_usd, 6),
        "error": trace.error,
    }))

You use it like this:

@trace_llm_call
def classify_text(text: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=256,
        messages=[{"role": "user", "content": f"Classify as spam or not: {text}"}],
    )
    return response.content[0].text

Every call now emits a structured JSON log with token counts, cost, and latency. No external SDK required.

Adding Time-to-First-Token for Streaming

For streaming responses, total latency is misleading. A call that takes 3s total but starts streaming after 200ms feels fast. One that stalls for 2.8s before the first token feels broken, even if total time is identical. Measure both separately.

def trace_streaming_call(client, model: str, messages: list[dict]) -> str:
    trace = LLMTrace(model=model)
    start = time.monotonic()
    first_token_time: float | None = None
    chunks: list[str] = []

    with client.messages.stream(
        model=model,
        max_tokens=512,
        messages=messages,
    ) as stream:
        for event in stream:
            if (first_token_time is None
                    and hasattr(event, "type")
                    and event.type == "content_block_delta"):
                first_token_time = time.monotonic()
                trace.ttft_ms = (first_token_time - start) * 1000

            if hasattr(event, "delta") and hasattr(event.delta, "text"):
                chunks.append(event.delta.text)

        final = stream.get_final_message()

    trace.latency_ms = (time.monotonic() - start) * 1000
    _extract_from_response(trace, final)
    trace.response_preview = "".join(chunks)[:300]
    _emit(trace)
    return "".join(chunks)

With TTFT tracked independently, you can alert on provider-side slowdowns separate from model generation time — a much more actionable signal.

Shipping Traces to a Backend

Structured JSON logs are a solid foundation, but for dashboards and alerting you want something queryable. Two practical paths:

OpenTelemetry (OTLP): convert each trace to an OTEL span and ship to any compatible collector (Jaeger, Grafana Tempo, Honeycomb). Vendor-neutral, but OTEL's data model wasn't designed for LLM-specific fields — cost_usd and cached_tokens end up as plain span attributes without native UI treatment.

Dedicated LLM observability tools (Langfuse, Phoenix, Helicone): they understand LLM semantics natively — token cost timelines, model comparison, prompt diffing. Most are open-source and self-hostable. If your prompts contain sensitive data, self-hosting is the only responsible option.

The pragmatic path for most teams: structured logs first (zero dependencies, works today), then add a dedicated platform once you know which metrics you actually query. Getting any visibility now beats waiting for perfect infrastructure.

For teams building secure AI systems, pairing observability with input validation and output controls is worth structuring from the start. The security hardening checklists at ayinedjimi-consultants.fr/checklists include an LLM section covering rate limiting, prompt injection detection, and output schema validation — all things that interlock naturally with the tracing layer above.

What to Alert On

Once traces are flowing, these thresholds are worth configuring first:

Metric	Alert condition
Error rate	> 2% over 5 minutes
P95 total latency	> 5s
TTFT P50	> 1.5s
Hourly cost	> budget threshold
Cached token ratio	< 30%

The cost alert is the most overlooked. A loop bug in an agent pipeline can burn hundreds of dollars in minutes. An hourly budget cap with a hard circuit breaker is not optional in production.

The Takeaway

LLM observability requires instrumenting at the right semantic level — not just "an HTTP call happened" but "which model, how many tokens, was it cached, what did it cost, and how long to first token." A thin wrapper around your LLM client gives you all of that in under 100 lines of Python, with no external dependencies. Add streaming TTFT measurement early — it surfaces provider-side issues that total latency hides. Ship structured logs first, then evaluate dedicated platforms once you know which questions you actually need to answer.

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 — Sat, 03 Oct 2026 10:01:50 +0000

You deployed a new model version on Friday afternoon. By Monday morning, your error rate tripled and you have no clean way to roll back because the deployment script overwrote the previous weights. This scenario happens more often than teams admit — not because engineers are careless, but because model versioning is treated as an afterthought until it isn't.

This article covers practical versioning patterns and rollback strategies for production AI systems, with working Python code you can adapt today.

Why model versioning is different from code versioning

Git handles source code well. But a production AI model is more than code — it's a combination of model weights (gigabytes, binary files), hyperparameters, training data version, evaluation metrics at training time, and runtime dependencies.

Git LFS can store weights, but querying "what was the F1 score of the model deployed on Oct 3rd?" requires something more structured. The minimum viable versioning scheme stores a version manifest alongside the model artifact.

import json
import hashlib
import datetime
from pathlib import Path

def save_model_with_manifest(model, save_path: str, metadata: dict):
    """Save model artifact with a version manifest."""
    path = Path(save_path)
    path.mkdir(parents=True, exist_ok=True)

    import pickle
    weights_path = path / "model.pkl"
    with open(weights_path, "wb") as f:
        pickle.dump(model, f)

    with open(weights_path, "rb") as f:
        artifact_hash = hashlib.sha256(f.read()).hexdigest()

    manifest = {
        "version": metadata.get("version", "0.0.0"),
        "created_at": datetime.datetime.utcnow().isoformat() + "Z",
        "artifact_hash": artifact_hash,
        "metrics": metadata.get("metrics", {}),
        "training_data_hash": metadata.get("training_data_hash"),
        "dependencies": metadata.get("dependencies", {}),
        "description": metadata.get("description", ""),
    }

    with open(path / "manifest.json", "w") as f:
        json.dump(manifest, f, indent=2)

    print(f"Saved model v{manifest['version']} to {path}")
    return manifest

# Usage
metadata = {
    "version": "1.3.0",
    "metrics": {"f1": 0.923, "precision": 0.941, "recall": 0.906},
    "training_data_hash": "d3f8a2b1c9e4...",
    "dependencies": {"scikit-learn": "1.4.2", "numpy": "1.26.4"},
    "description": "Retrained with Q3 data, improved recall on edge cases",
}

This manifest becomes your audit trail. Store it in a database or object storage alongside the weights.

A versioned model registry in Python

A local model registry is the next step — an API to list, load, promote, and retire versions. For production teams, tools like MLflow or BentoML solve this at scale. But understanding the underlying pattern helps when integrating with existing infrastructure.

import json
import shutil
from pathlib import Path
from typing import Optional

class ModelRegistry:
    def __init__(self, registry_path: str):
        self.root = Path(registry_path)
        self.root.mkdir(parents=True, exist_ok=True)
        self.index_path = self.root / "registry.json"
        self._load_index()

    def _load_index(self):
        if self.index_path.exists():
            with open(self.index_path) as f:
                self.index = json.load(f)
        else:
            self.index = {"models": [], "active_version": None}

    def _save_index(self):
        with open(self.index_path, "w") as f:
            json.dump(self.index, f, indent=2)

    def register(self, model_path: str, version: str, metrics: dict):
        dest = self.root / version
        shutil.copytree(model_path, dest, dirs_exist_ok=True)
        entry = {
            "version": version,
            "path": str(dest),
            "metrics": metrics,
            "status": "candidate",  # candidate -> active -> retired
        }
        self.index["models"].append(entry)
        self._save_index()
        print(f"Registered v{version} (status: candidate)")

    def promote(self, version: str):
        """Promote a version to active, retiring the current one."""
        for m in self.index["models"]:
            if m["status"] == "active":
                m["status"] = "retired"
            if m["version"] == version:
                m["status"] = "active"
                self.index["active_version"] = version
        self._save_index()
        print(f"Promoted v{version} to active")

    def rollback(self) -> Optional[str]:
        """Reactivate the most recent retired version."""
        retired = [m for m in self.index["models"] if m["status"] == "retired"]
        if not retired:
            print("No retired versions to roll back to")
            return None
        for m in self.index["models"]:
            if m["status"] == "active":
                m["status"] = "retired"
        previous = retired[-1]
        previous["status"] = "active"
        self.index["active_version"] = previous["version"]
        self._save_index()
        print(f"Rolled back to v{previous['version']}")
        return previous["version"]

    def get_active(self) -> Optional[dict]:
        for m in self.index["models"]:
            if m["status"] == "active":
                return m
        return None

The status field (candidate → active → retired) mirrors the promotion workflow most teams already have mentally — but rarely encode explicitly.

Rollback strategies in production

Having versioned artifacts is necessary but not sufficient. The rollback strategy depends on your serving architecture.

Direct model replacement is the simplest and highest risk approach. The serving process loads new weights in-place. Rollback means restarting with the previous artifact path. Works for batch inference or low-traffic APIs where a brief restart is acceptable.

Blue-green deployment runs two identical serving environments: one live, one staging. You promote staging to live by updating a load balancer rule. Rollback is a one-line config change. The tradeoff: double infrastructure cost while both environments run.

Canary deployment routes N% of traffic to the new model version while the rest hits the current stable version. You monitor production metrics and either ramp up or abort. This is the right approach for models where the failure mode is silent — a new model that's technically healthy but produces worse predictions.

For monitoring canaries, instrument your serving layer to tag each response with its model version:

from fastapi import FastAPI, Request
import hashlib
import time

app = FastAPI()

CANARY_VERSION = "v1.4.0"
STABLE_VERSION = "v1.3.0"
CANARY_TRAFFIC_PCT = 10  # send 10% to canary

def get_model_version(request_id: str) -> str:
    """Deterministic routing: same request_id always maps to same version."""
    bucket = int(hashlib.md5(request_id.encode()).hexdigest(), 16) % 100
    return CANARY_VERSION if bucket < CANARY_TRAFFIC_PCT else STABLE_VERSION

@app.post("/predict")
async def predict(request: Request, payload: dict):
    request_id = request.headers.get("X-Request-ID", str(time.time()))
    version = get_model_version(request_id)

    # result = models[version].predict(payload["features"])

    return {
        "prediction": "...",
        "model_version": version,  # Always return this for monitoring
        "request_id": request_id,
    }

Returning model_version in every response lets you filter your metrics dashboards by version and catch regressions before they affect all traffic. Pair this with a security hardening checklist that covers your inference endpoints — TLS, auth headers, and rate limiting apply to model servers the same way they apply to any production API.

When to automate rollback decisions

Manual rollback decisions are slow. Consider automated rollback when you can define a quantitative threshold — for example, "if error rate on the canary exceeds 5% for more than 10 minutes, revert."

This requires three things: a metric you trust (not just HTTP 5xx — include model-specific errors like empty outputs or latency P99 spikes), a stable baseline from the previous version, and an automated action that calls rollback() or updates the load balancer rule.

The common mistake is rolling back on noisy metrics. If your baseline had a 2% error rate and you trigger at 5%, you'll get false positives during traffic spikes. Use a burn-rate approach: track the error budget consumed in a sliding window rather than a point-in-time rate. This is the same principle behind SLO-based alerting — applied to model quality instead of infrastructure.

The takeaway

Versioning a model means more than tagging the weights file. It means capturing the manifest (metrics, data provenance, dependencies), encoding a promotion workflow (candidate → active → retired), and deciding in advance what rollback looks like for your serving architecture.

The code here is intentionally minimal — enough to understand the pattern, not tied to a specific ML platform. Once you outgrow it, MLflow, BentoML, or a cloud-native registry slot in cleanly because the underlying concepts are the same.

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 — Fri, 02 Oct 2026 10:02:29 +0000

Every vector database you have ever used — Qdrant, Weaviate, Milvus — relies on the same core algorithm for fast approximate nearest neighbor search: HNSW (Hierarchical Navigable Small World). Most developers treat it as a black box. This post tears it open, implements it in pure Python, and shows you exactly what levers you pull when you tune a production vector index.

Why Not Just Use Brute Force?

For small datasets (under 10k vectors), brute-force k-NN works fine: compute the distance from your query to every stored vector, sort, return top-k. At 1M+ vectors it breaks down. A single 1536-dimensional embedding query requires roughly 1.5 billion float multiplications against the full dataset. HNSW reduces that to a few thousand operations at 95–99% recall.

The accuracy cost is measured by recall@k: the fraction of the true top-k nearest neighbors that the approximate index returns. Production HNSW configurations typically achieve recall@10 > 0.97 at 50–200x the speed of brute force.

How HNSW Works

HNSW builds a layered proximity graph — think of it as a hierarchy of skip lists in high-dimensional space.

Layer 0 contains every inserted vector, each connected to its M nearest neighbors.
Layer 1 contains a random ~1/e subset of layer 0, similarly connected.
Higher layers contain progressively sparser subsets, up to a single entry point.

Search starts at the entry point in the highest layer, descends greedily to the closest node, drops one layer, and repeats until layer 0. At layer 0, a beam search with width ef collects the final candidates.

Insert mirrors search: find the neighborhood of the new vector at each layer, then add bidirectional edges.

Two parameters control the quality-speed tradeoff:

M — connections per node. Higher = better recall, more memory and build time.
ef_construction — beam width at index build. Higher = better graph quality, slower inserts.

The Implementation

import heapq
import math
import random
from typing import Optional


class HNSWIndex:
    def __init__(self, dim: int, M: int = 16, ef_construction: int = 200):
        self.dim = dim
        self.M = M
        self.M_max0 = 2 * M
        self.ef_construction = ef_construction
        self.ml = 1.0 / math.log(M)

        self.vectors: list[list[float]] = []
        self.graphs: list[dict[int, list[int]]] = []
        self.entry_point: Optional[int] = None
        self.max_layer = -1

    def _dist(self, a: list[float], b: list[float]) -> float:
        return sum((x - y) ** 2 for x, y in zip(a, b)) ** 0.5

    def _random_level(self) -> int:
        return int(-math.log(random.random()) * self.ml)

    def _search_layer(
        self, query: list[float], ep: int, ef: int, layer: int
    ) -> list[tuple[float, int]]:
        visited = {ep}
        d = self._dist(query, self.vectors[ep])
        candidates = [(d, ep)]   # min-heap
        found = [(-d, ep)]       # max-heap (negated distances)

        while candidates:
            dist, node = heapq.heappop(candidates)
            worst = -found[0][0]
            if dist > worst:
                break
            for nb in self.graphs[layer].get(node, []):
                if nb not in visited:
                    visited.add(nb)
                    nd = self._dist(query, self.vectors[nb])
                    if nd < worst or len(found) < ef:
                        heapq.heappush(candidates, (nd, nb))
                        heapq.heappush(found, (-nd, nb))
                        if len(found) > ef:
                            heapq.heappop(found)

        return sorted((-neg_d, node) for neg_d, node in found)

    def add(self, vector: list[float]) -> int:
        idx = len(self.vectors)
        self.vectors.append(vector)
        level = self._random_level()

        while len(self.graphs) <= level:
            self.graphs.append({})
        for lyr in range(len(self.graphs)):
            self.graphs[lyr].setdefault(idx, [])

        if self.entry_point is None:
            self.entry_point = idx
            self.max_layer = level
            return idx

        ep = self.entry_point

        for lyr in range(self.max_layer, level, -1):
            results = self._search_layer(vector, ep, 1, lyr)
            if results:
                ep = results[0][1]

        for lyr in range(min(level, self.max_layer), -1, -1):
            neighbors = self._search_layer(vector, ep, self.ef_construction, lyr)
            M_max = self.M_max0 if lyr == 0 else self.M
            selected = [i for _, i in neighbors[:M_max]]
            self.graphs[lyr][idx] = selected

            for nb in selected:
                self.graphs[lyr][nb].append(idx)
                if len(self.graphs[lyr][nb]) > M_max:
                    nb_vec = self.vectors[nb]
                    pruned = sorted(
                        (self._dist(nb_vec, self.vectors[n]), n)
                        for n in self.graphs[lyr][nb]
                    )[:M_max]
                    self.graphs[lyr][nb] = [n for _, n in pruned]

            if neighbors:
                ep = neighbors[0][1]

        if level > self.max_layer:
            self.max_layer = level
            self.entry_point = idx

        return idx

    def search(
        self, query: list[float], k: int = 10, ef: int = 50
    ) -> list[tuple[float, int]]:
        if self.entry_point is None:
            return []
        ep = self.entry_point
        for lyr in range(self.max_layer, 0, -1):
            results = self._search_layer(query, ep, 1, lyr)
            if results:
                ep = results[0][1]
        return self._search_layer(query, ep, max(ef, k), 0)[:k]

The _search_layer function is the heart of the algorithm. It maintains two heaps in parallel: a min-heap of candidates to explore and a max-heap of the ef best results found so far. The early-exit condition (dist > worst) is what makes HNSW fast — once the closest unvisited candidate is farther than the worst result in your best-ef set, you stop.

Measuring Recall Against Brute Force

Never deploy an ANN index without first measuring its actual recall on your data distribution. Here is a simple benchmark harness:

import random
import time


def brute_force_knn(
    vectors: list[list[float]], query: list[float], k: int
) -> set[int]:
    dists = [
        (sum((a - b) ** 2 for a, b in zip(query, v)) ** 0.5, i)
        for i, v in enumerate(vectors)
    ]
    return {i for _, i in sorted(dists)[:k]}


def benchmark(n: int = 5_000, dim: int = 64, k: int = 10):
    rng = random.Random(42)
    data = [[rng.gauss(0, 1) for _ in range(dim)] for _ in range(n)]

    index = HNSWIndex(dim=dim, M=16, ef_construction=100)
    t0 = time.perf_counter()
    for v in data:
        index.add(v)
    print(f"Build: {time.perf_counter() - t0:.2f}s for {n} vectors")

    queries = [[rng.gauss(0, 1) for _ in range(dim)] for _ in range(200)]
    recall_total = 0.0
    t0 = time.perf_counter()
    for q in queries:
        hnsw_hits = {i for _, i in index.search(q, k=k, ef=50)}
        true_hits = brute_force_knn(data, q, k)
        recall_total += len(hnsw_hits & true_hits) / k
    avg_ms = (time.perf_counter() - t0) / len(queries) * 1000
    print(f"Recall@{k}: {recall_total / len(queries):.3f}")
    print(f"Avg query: {avg_ms:.2f}ms")


benchmark()

On a modern laptop with 5k vectors at dim=64, expect ~0.92 recall@10 at under 2ms per query with ef=50. Increase ef to 200 to push recall past 0.98 at 4–6x the latency cost. The tradeoff is tunable at query time without rebuilding the index.

Production Notes

The pure-Python implementation above is for understanding, not production throughput. For real workloads, use hnswlib (C++ backend, Python bindings, ~100x faster) or a dedicated vector database.

Before you commit to a hosted vector search service, review the security posture of your embedding pipeline — embedding user data and storing it in a third-party index has data residency and access control implications. We have a concise checklist covering this in our security hardening resources.

A few practical tuning notes:

Memory: at M=16, expect ~800 bytes per vector for the graph structure alone. Budget accordingly.
Dimensionality: embeddings above 512 dimensions benefit from PCA reduction to 256 before indexing — roughly 6x speedup at ~2% recall loss.
Concurrent access: HNSW search is read-safe for parallelism; inserts require exclusive write locking.
Persistence: serialize vectors and graphs with msgpack or pickle. hnswlib has built-in save_index / load_index.

The Takeaway

HNSW is not magic — it is a hierarchical proximity graph where you exchange a small fraction of accuracy for orders-of-magnitude speed gains. Implementing it from scratch makes the parameters M, ef_construction, and ef concrete rather than abstract knobs. When your vector similarity queries return unexpected results in production, you will know immediately whether to increase ef, rebuild with higher M, check your distance metric, or look at the data quality upstream.

The pure-Python code here is a learning artifact. The mental model it gives you is a production asset.

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

How to build a job queue in Go without external dependencies

Ayi NEDJIMI — Thu, 01 Oct 2026 10:03:04 +0000

Most job queue tutorials start with Redis, RabbitMQ, or some managed queue service. That's fine for teams with existing infrastructure — but if you're building an internal tool, a side project, or a microservice that needs to process tasks asynchronously, pulling in a message broker is overkill. Go's standard library gives you everything you need: goroutines, channels, sync primitives, and context for cancellation.

What we're building

A concurrent job queue with:

A fixed pool of workers pulling from a buffered channel
Non-blocking enqueue — full queue returns an error, no caller hangs
Per-job retry with a configurable limit
Graceful shutdown on SIGINT/SIGTERM — no jobs dropped mid-flight
An error channel so worker failures don't crash the process

Around 150 lines of Go. No external libraries.

The core structure

package queue

import (
    "context"
    "fmt"
    "sync"
)

type Job struct {
    ID       string
    Payload  any
    Retries  int
    MaxRetry int
}

type Handler func(ctx context.Context, job Job) error

type Queue struct {
    jobs    chan Job
    wg      sync.WaitGroup
    handler Handler
    errors  chan error
}

func New(bufferSize, workers int, handler Handler) *Queue {
    q := &Queue{
        jobs:    make(chan Job, bufferSize),
        handler: handler,
        errors:  make(chan error, bufferSize),
    }
    for i := 0; i < workers; i++ {
        q.wg.Add(1)
        go q.worker(context.Background())
    }
    return q
}

func (q *Queue) Enqueue(job Job) error {
    select {
    case q.jobs <- job:
        return nil
    default:
        return fmt.Errorf("queue full: dropped job %s", job.ID)
    }
}

func (q *Queue) worker(ctx context.Context) {
    defer q.wg.Done()
    for job := range q.jobs {
        err := q.handler(ctx, job)
        if err != nil && job.Retries < job.MaxRetry {
            job.Retries++
            q.jobs <- job // re-queue
            continue
        }
        if err != nil {
            q.errors <- fmt.Errorf("job %s failed after %d retries: %w", job.ID, job.Retries, err)
        }
    }
}

func (q *Queue) Stop() {
    close(q.jobs)
    q.wg.Wait()
    close(q.errors)
}

func (q *Queue) Errors() <-chan error {
    return q.errors
}

A few design decisions worth noting:

Buffered channel: Enqueue uses a non-blocking select. The queue either accepts the job immediately or returns an error — the caller decides how to handle backpressure.
No global state: the queue is a plain struct. You can run multiple independent queues in the same process without interference.
Retry loop: re-queuing happens inside the worker rather than a separate scheduler. A job with MaxRetry: 3 occupies a worker slot for all attempts before giving up — acceptable for low-volume queues, worth revisiting if workers become a bottleneck.

Graceful shutdown with OS signals

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    "yourmodule/queue"
)

func main() {
    q := queue.New(100, 4, func(ctx context.Context, job queue.Job) error {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(200 * time.Millisecond):
            log.Printf("processed job %s: %v", job.ID, job.Payload)
            return nil
        }
    })

    go func() {
        for err := range q.Errors() {
            log.Println("worker error:", err)
        }
    }()

    for i := 0; i < 20; i++ {
        _ = q.Enqueue(queue.Job{
            ID:       fmt.Sprintf("job-%d", i),
            Payload:  i * 10,
            MaxRetry: 2,
        })
    }

    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
    <-sig

    log.Println("shutting down...")
    q.Stop()
    log.Println("all jobs completed")
}

When SIGINT arrives, new enqueues stop and workers drain the remaining jobs before exiting. The sync.WaitGroup in Stop() blocks until every goroutine returns — no job is silently dropped.

Adding priorities with two channels

Standard channels have no priority concept. The practical solution is two channels — one for high-priority, one for normal — with a worker select that checks high-priority first.

type PriorityQueue struct {
    high   chan Job
    normal chan Job
    wg     sync.WaitGroup
}

func (pq *PriorityQueue) worker(ctx context.Context, handler Handler, errors chan<- error) {
    defer pq.wg.Done()
    for {
        // Always drain high-priority before picking normal
        select {
        case job, ok := <-pq.high:
            if !ok {
                return
            }
            if err := handler(ctx, job); err != nil {
                errors <- err
            }
            continue
        default:
        }

        select {
        case job, ok := <-pq.high:
            if !ok {
                return
            }
            if err := handler(ctx, job); err != nil {
                errors <- err
            }
        case job, ok := <-pq.normal:
            if !ok {
                return
            }
            if err := handler(ctx, job); err != nil {
                errors <- err
            }
        }
    }
}

The double select — first checking high alone with a default, then both — is the canonical Go pattern for priority queues. Without the first pass, a select with two ready channels picks randomly, so high-priority jobs can be starved under sustained normal load.

What you trade away vs. a real broker

Be clear-eyed about the limitations:

No persistence: if the process crashes, queued jobs are gone. For anything financial or critical, write pending jobs to Postgres or SQLite before enqueuing.
No distributed consumers: this queue lives in one process. Horizontal scaling requires distributing work at the application level before it reaches the queue.
No built-in dead-letter queue: exhausted retries emit an error to the error channel. Logging or storing those failed jobs is your responsibility.
Coarse backpressure: the buffer either accepts or rejects. Real brokers give you consumer acknowledgement, flow control, and visibility into queue depth across replicas.

For background jobs within a single service — sending emails, generating reports, processing uploads — this design handles thousands of jobs per second with zero operational overhead. When you need durability or cross-process coordination, move to a proper broker. If your pipeline handles sensitive data, review access controls before shipping — our free security checklists cover API and service hardening in detail.

The takeaway

Go's standard library is sufficient for the majority of job queue use cases. A buffered channel with a worker pool, a WaitGroup for shutdown, and an error channel for failures gives you a solid foundation without Redis, RabbitMQ, or any external dependency to operate.

Build this first. Add a broker when you can demonstrate — with real production data — that you actually need one.

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 — Wed, 30 Sep 2026 10:01:45 +0000

Fine-tuning a large language model on your own data used to require serious GPU budgets and weeks of infrastructure work. LoRA (Low-Rank Adaptation) changed the math: you can adapt a 7B-parameter model to a specific domain in a few hours on a single consumer GPU, without touching most of the original weights. This guide covers how LoRA works, when to use it, and the exact steps to go from a base model to a specialized one that actually behaves differently.

How LoRA works under the hood

A transformer model has billions of weight matrices. Full fine-tuning updates all of them, which is expensive and prone to catastrophic forgetting. LoRA takes a different approach: instead of modifying the original weight matrix W directly, it learns two small matrices A and B such that their product AB has the same dimensions as W, and adds ΔW = BA to the original weights during the forward pass.

The rank r controls the trade-off. With r=8 and a model with hidden_dim=4096, you go from 4096×4096=16M parameters per layer to 2×4096×8=65K — a 250x compression. You train only A and B; W stays frozen.

In practice this means:

Training time: 3–5x faster than full fine-tuning
VRAM: a 7B model that needs 80GB for full fine-tuning needs ~12GB with LoRA+4-bit quantization
Storage: your adapter is 50–200MB, not 14GB

The catch: LoRA performs poorly on tasks that require fundamentally new factual knowledge. It excels at behavior change — tone, output format, domain-specific vocabulary, instruction-following style.

Setting up LoRA fine-tuning with PEFT

The peft library makes this straightforward. Here is a minimal training setup for a causal language model:

from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments, Trainer
from peft import LoraConfig, get_peft_model, TaskType
from datasets import load_dataset

model_name = "mistralai/Mistral-7B-v0.1"

tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token

model = AutoModelForCausalLM.from_pretrained(
    model_name,
    load_in_4bit=True,   # QLoRA: quantize base model to 4-bit
    device_map="auto",
)

lora_config = LoraConfig(
    r=8,                 # rank — start low, increase if underfitting
    lora_alpha=16,       # scaling factor (alpha/r = effective LR scale)
    target_modules=["q_proj", "v_proj"],  # attention layers to adapt
    lora_dropout=0.05,
    bias="none",
    task_type=TaskType.CAUSAL_LM,
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# trainable params: 4,194,304 || all params: 3,752,071,168 || trainable%: 0.1118

dataset = load_dataset("json", data_files={"train": "data/train.jsonl"})

def tokenize(example):
    prompt = f"### Instruction:\n{example['instruction']}\n\n### Response:\n{example['output']}"
    tokens = tokenizer(prompt, truncation=True, max_length=512, padding="max_length")
    tokens["labels"] = tokens["input_ids"].copy()
    return tokens

tokenized = dataset.map(tokenize, remove_columns=dataset["train"].column_names)

training_args = TrainingArguments(
    output_dir="./lora-output",
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    num_train_epochs=3,
    learning_rate=2e-4,
    fp16=True,
    logging_steps=50,
    save_strategy="epoch",
    warmup_ratio=0.05,
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized["train"],
)

trainer.train()
model.save_pretrained("./lora-adapter")

load_in_4bit=True activates QLoRA — the base model is quantized to 4-bit, which cuts VRAM further at a small accuracy cost. The target_modules parameter is architecture-specific: for Mistral and LLaMA models, q_proj and v_proj are the standard targets; for Falcon, it's query_key_value.

Your training data matters more than your hyperparameters

LoRA doesn't need a lot of data, but it needs clean, representative data. 500–2000 high-quality examples in instruction format often outperform 10,000 noisy ones.

Format your data as JSONL with instruction and output fields:

{"instruction": "Summarize this security incident report in 3 bullet points.", "output": "- Attacker gained initial access via phishing on 2026-09-12\n- Lateral movement reached the billing database within 4 hours\n- No customer PII was exfiltrated; access was revoked at 03:14 UTC"}
{"instruction": "Convert this CVE description to a CVSS v3.1 severity rating.", "output": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N — Score: 7.5 (High)"}
{"instruction": "Write a Python function to validate a JWT token without a library.", "output": "import base64, json, hmac, hashlib\n..."}

Common mistakes that will hurt your fine-tune:

Mixed quality: if 20% of your examples are malformed, the model learns to produce malformed output
Length imbalance: outputs averaging 20 tokens mixed with outputs averaging 800 tokens — the model learns to truncate or pad randomly
Template leakage: not stripping HTML, markdown artifacts, or system prompts from scraped data
Wrong tokenization: not setting pad_token = eos_token for models without a dedicated padding token causes silent training failures

Before training, always inspect token length distribution. If p95 of your outputs exceeds 600 tokens but you set max_length=512, you're silently truncating nearly half your training signal.

Merging and deploying your adapter

After training you have two options: keep the adapter separate (fast iteration, smaller artifact) or merge it permanently into the base weights (better inference performance, single artifact).

from peft import PeftModel
from transformers import AutoModelForCausalLM

base = AutoModelForCausalLM.from_pretrained("mistralai/Mistral-7B-v0.1")
model = PeftModel.from_pretrained(base, "./lora-adapter")

merged = model.merge_and_unload()
merged.save_pretrained("./mistral-7b-finetuned")

The merged model is a standard HuggingFace model. It loads in vLLM, Ollama, or any inference server without adapter-specific configuration.

If you keep the adapter separate, inference adds 2–5ms on the first forward pass for adapter loading. That's negligible for most workloads, but it matters when hot-swapping adapters at runtime — for example, routing different tasks to different specialized adapters on a shared base model to reduce memory footprint.

When LoRA is not the right tool

LoRA changes how the model responds, not what it knows. If you need the model to know new facts — a new codebase, a new product catalog, recent events — retrieval-augmented generation will outperform fine-tuning every time. Fine-tuning doesn't reliably inject new knowledge; it biases behavior.

Other situations where LoRA underperforms:

Very small datasets (< 100 examples): the model mostly memorizes rather than generalizes; few-shot prompting will produce better results with less effort
Raw pretrained base models: use an instruction-following variant (Llama-3-Instruct, Mistral-Instruct) as your starting point; fine-tuning a raw pretrained model on 1,000 examples is fighting the pre-training distribution with a small dataset
Safety-sensitive deployments: LoRA fine-tuning can inadvertently degrade safety guardrails baked into the base model. Always run adversarial red-team prompts against your adapted model before deployment — the security hardening checklists at AYI NEDJIMI Consultants include an LLM safety evaluation section you can use as a baseline

The takeaway

LoRA is practical and accessible: start with an instruction-following base model, collect 500–1000 clean examples, train with r=8 and lora_alpha=16, evaluate on a held-out set, and merge only when the adapter behavior satisfies your test cases.

The failure mode I see most often is treating a decreasing loss curve as a green light to ship. Loss going down doesn't mean the model does what you want — it means the model is getting better at predicting your training tokens. Always test on real prompts from your target use case before deploying, and maintain a regression set that covers your critical behaviors.

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 — Tue, 29 Sep 2026 10:04:25 +0000

Most teams building RAG systems spend 90% of their time on retrieval and generation, and 10% on evaluation. That ratio is backwards. Without a rigorous test harness, you're shipping a black box — and you'll only find out it's broken when a user does.

What You're Actually Measuring

A RAG pipeline has two distinct failure modes: retrieval failures (the right context isn't fetched) and generation failures (the model hallucinates or ignores the retrieved context). Your evaluation framework needs to catch both independently.

The four metrics that matter in practice:

Context Recall: Did the retriever find the documents needed to answer the question?
Context Precision: Of what was retrieved, how much was actually relevant?
Faithfulness: Does the generated answer stick to the retrieved context, or does it introduce invented facts?
Answer Relevance: Does the answer actually address the question asked?

You can measure these with a reference dataset (ground-truth question/answer/context triplets) or, more scalably, with a language model acting as a judge. Both approaches are worth understanding.

Building a Reference Dataset

The foundation of any test harness is a dataset of question/answer/context triplets. For each item you need: the question, the correct answer, and the document chunks that should be retrieved.

import json
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class RAGSample:
    question: str
    ground_truth: str
    ground_truth_contexts: List[str]
    generated_answer: Optional[str] = None
    retrieved_contexts: Optional[List[str]] = None

def load_test_set(path: str) -> List[RAGSample]:
    with open(path) as f:
        data = json.load(f)
    return [RAGSample(**item) for item in data]

Build this dataset by exporting real user queries paired with your best-known answers and the source chunks. Even 50 well-curated samples will catch most regressions. Seed it with edge cases: short questions, ambiguous phrasings, questions that span multiple documents.

Computing Faithfulness with an LLM Judge

Faithfulness is hard to measure with string matching — you need semantic understanding. The standard approach is using a language model as a judge. The judge receives the retrieved context and the generated answer, then returns a score from 0 to 1.

import json
from openai import OpenAI

client = OpenAI()

FAITHFULNESS_PROMPT = (
    "You are evaluating whether an answer is faithful to the given context.\n"
    "Faithful means: every factual claim in the answer is supported by the context.\n"
    "Score from 0.0 (completely unfaithful) to 1.0 (fully faithful).\n\n"
    "Context:\n{context}\n\nAnswer:\n{answer}\n\n"
    'Respond with JSON: {"score": float, "reason": string}'
)

def score_faithfulness(context: str, answer: str) -> float:
    prompt = FAITHFULNESS_PROMPT.format(context=context, answer=answer)
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0,
    )
    result = json.loads(resp.choices[0].message.content)
    return float(result["score"])

A faithfulness score below 0.8 typically means the retriever is fetching irrelevant chunks and the generator is hallucinating from them — a retrieval bug that looks like a model bug until you measure it properly.

Computing Context Recall with Token Overlap

For context recall — "did we retrieve what we needed?" — a simple token overlap heuristic works well as a fast baseline before reaching for a semantic similarity model:

from collections import Counter
import re
from typing import List

def tokenize(text: str) -> Counter:
    tokens = re.findall(r'\b\w+\b', text.lower())
    return Counter(tokens)

def token_recall(retrieved_chunks: List[str], ground_truth_chunks: List[str]) -> float:
    retrieved_text = " ".join(retrieved_chunks)
    retrieved_tokens = tokenize(retrieved_text)

    total, found = 0, 0
    for chunk in ground_truth_chunks:
        for token, count in tokenize(chunk).items():
            total += count
            found += min(count, retrieved_tokens[token])

    return found / total if total > 0 else 0.0

This is fast, zero-cost, and good enough to catch retrieval regressions in CI. For production monitoring, replace it with a semantic similarity model such as a bi-encoder from sentence-transformers — token overlap misses synonyms and paraphrases.

The Test Harness: Putting It Together

import statistics
from typing import Callable, Tuple, List

def run_rag_evaluation(
    test_set: List[RAGSample],
    rag_pipeline: Callable[[str], Tuple[str, List[str]]],
) -> dict:
    faithfulness_scores = []
    recall_scores = []

    for sample in test_set:
        answer, retrieved = rag_pipeline(sample.question)
        sample.generated_answer = answer
        sample.retrieved_contexts = retrieved

        f_score = score_faithfulness(" ".join(retrieved), answer)
        r_score = token_recall(retrieved, sample.ground_truth_contexts)

        faithfulness_scores.append(f_score)
        recall_scores.append(r_score)

        print(f"Q: {sample.question[:60]}...")
        print(f"  Faithfulness: {f_score:.2f}  Recall: {r_score:.2f}")

    sorted_f = sorted(faithfulness_scores)
    results = {
        "faithfulness_mean": statistics.mean(faithfulness_scores),
        "faithfulness_p10": sorted_f[max(0, len(sorted_f) // 10)],
        "recall_mean": statistics.mean(recall_scores),
        "n_samples": len(test_set),
    }
    return results

The p10 faithfulness score — the 10th percentile — matters more than the mean. Your pipeline's worst cases are what end up in user complaints.

Integrating into CI

Evaluation should run automatically on every PR that touches the retriever or the prompt template. Gate on a hard threshold:

- name: RAG evaluation
  run: |
    python evaluate_rag.py \
      --test-set tests/rag_test_set.json \
      --faithfulness-threshold 0.85 \
      --recall-threshold 0.80
  env:
    OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}

If faithfulness drops below 0.85, the PR fails. This forces anyone changing retrieval logic to justify the tradeoff explicitly. The friction is intentional: silently degrading quality is worse than a blocked PR.

For security-sensitive RAG applications — customer support bots with access to internal docs, compliance tools, incident response assistants — you also need a factual grounding check to prevent the model from leaking information outside the retrieved context. This type of control is part of what we cover in our security hardening checklists.

Scaling Beyond Toy Datasets

Once your harness works on 50 samples, three problems appear at scale:

Evaluation drift: the language model judge you use today may score differently after a provider model update. Pin the judge model version and store raw scores as artifacts, not just pass/fail flags.

Dataset staleness: as your knowledge base evolves, ground-truth contexts become outdated. Tag each sample with the document version it was created from, and rebuild affected samples when source documents change.

Cost: at 1000 samples x 2 LLM calls (faithfulness + answer relevance), each evaluation run costs roughly $0.50-$1.00 with gpt-4o-mini. That is acceptable for weekly regression runs but expensive for every commit. Use token overlap metrics in fast CI, LLM judges in nightly runs.

The Takeaway

RAG evaluation is not optional — it is what separates a demo from a system you would put in production. Start with a small, curated test set, compute faithfulness and recall on every change, and gate your CI pipeline on minimum thresholds.

The typical failure mode in production RAG is not hallucination at the generation step. It is over-retrieval at the search step: fetching 10 chunks when 2 are relevant actively degrades faithfulness, because the model gets confused by noise. The metrics above will surface this clearly. Measure first, then tune.

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

Go error handling patterns that scale: wrapping, unwrapping, sentinels

Ayi NEDJIMI — Mon, 28 Sep 2026 10:05:47 +0000

Error handling in Go is explicit by design. Every function that can fail returns an error as its last value, and callers must deal with it. That explicitness is valuable — but as your codebase grows, return err stops being enough. You lose context, callers can't distinguish failure modes, and debugging production incidents becomes guesswork. This article covers three patterns that hold up at scale: error wrapping, sentinel errors, and custom error types.

Why `return err` Isn't Enough

Consider a function that reads a config file, parses it, then initializes a database connection. Each step can fail. If you bubble up raw errors untouched, the caller might see:

permission denied

Was that the file read? The DB connection? No idea. The fix is adding context at each layer, so by the time the error reaches the top, it reads like a breadcrumb trail.

Error Wrapping with fmt.Errorf and %w

Go 1.13 introduced the %w verb in fmt.Errorf. It wraps an error so the original remains inspectable via errors.Is and errors.As, while still prepending human-readable context.

package main

import (
    "errors"
    "fmt"
    "os"
)

func readConfig(path string) ([]byte, error) {
    data, err := os.ReadFile(path)
    if err != nil {
        return nil, fmt.Errorf("readConfig %q: %w", path, err)
    }
    return data, nil
}

func loadApp(configPath string) error {
    _, err := readConfig(configPath)
    if err != nil {
        return fmt.Errorf("loadApp: %w", err)
    }
    return nil
}

func main() {
    err := loadApp("/etc/myapp/config.yaml")
    if err != nil {
        fmt.Println(err)
        // loadApp: readConfig "/etc/myapp/config.yaml": open /etc/myapp/config.yaml: no such file or directory

        var pathErr *os.PathError
        if errors.As(err, &pathErr) {
            fmt.Println("path that failed:", pathErr.Path)
        }
    }
}

The error message reads like an English stack trace. Each layer adds its prefix, and the underlying *os.PathError remains accessible through the chain.

One practical rule: use %w only once per fmt.Errorf call. If you find yourself wanting to wrap two errors at once, the function is probably doing too much.

Sentinel Errors

Sentinel errors are package-level variables that callers compare against to branch on specific failure modes. They're the right tool when the caller needs to change behavior — returning a 404 instead of a 500, retrying instead of aborting.

package store

import (
    "errors"
    "fmt"
)

var (
    ErrNotFound   = errors.New("record not found")
    ErrConflict   = errors.New("record already exists")
    ErrPermission = errors.New("permission denied")
)

type User struct {
    ID   int
    Name string
}

var db = map[int]*User{1: {ID: 1, Name: "alice"}}

func GetUser(id int) (*User, error) {
    user, ok := db[id]
    if !ok {
        return nil, fmt.Errorf("GetUser %d: %w", id, ErrNotFound)
    }
    return user, nil
}

The HTTP handler can now make clean decisions:

func handleGetUser(w http.ResponseWriter, r *http.Request) {
    id := parseID(r)
    user, err := store.GetUser(id)
    if err != nil {
        switch {
        case errors.Is(err, store.ErrNotFound):
            http.Error(w, "user not found", http.StatusNotFound)
        case errors.Is(err, store.ErrPermission):
            http.Error(w, "forbidden", http.StatusForbidden)
        default:
            log.Printf("GetUser: %v", err)
            http.Error(w, "internal server error", http.StatusInternalServerError)
        }
        return
    }
    json.NewEncoder(w).Encode(user)
}

errors.Is traverses the full wrapping chain, so even if GetUser wrapped ErrNotFound with added context, the check still resolves correctly. Never compare errors with == unless you want pointer equality — errors.Is is always the right call.

When not to create sentinels: if no caller branches on a particular error, a well-wrapped descriptive error is enough. Sentinel proliferation is as bad as the absence of them.

Custom Error Types

When you need structured data in an error — field names, HTTP status codes, retry-after durations — a custom type is cleaner than encoding that data into a string.

type ValidationError struct {
    Field   string
    Message string
}

func (e *ValidationError) Error() string {
    return fmt.Sprintf("validation failed on %q: %s", e.Field, e.Message)
}

func ValidateAge(age int) error {
    if age < 0 {
        return &ValidationError{Field: "age", Message: "must be non-negative"}
    }
    if age > 150 {
        return &ValidationError{Field: "age", Message: "outside realistic range"}
    }
    return nil
}

Callers extract the structured data with errors.As:

if err := ValidateAge(-5); err != nil {
    var valErr *ValidationError
    if errors.As(err, &valErr) {
        fmt.Printf("field=%s message=%s\n", valErr.Field, valErr.Message)
    }
}

This pattern works well for API validation pipelines that return field-level errors to clients. It also ties into broader security posture: typed errors prevent accidentally leaking internal error strings to end users — something worth checking in your security hardening checklists.

If your custom type wraps another error, implement Unwrap() so the chain stays navigable:

type AppError struct {
    Op  string
    Err error
}

func (e *AppError) Error() string { return fmt.Sprintf("%s: %v", e.Op, e.Err) }
func (e *AppError) Unwrap() error  { return e.Err }

Without Unwrap(), errors.Is and errors.As stop at your type and cannot inspect what's underneath.

errors.Is vs errors.As: Choosing the Right Tool

Both functions walk the unwrapping chain recursively. The difference is what they look for:

errors.Is(err, target) — checks whether target appears anywhere in the chain. Use with sentinel values.
errors.As(err, &target) — checks whether any error in the chain can be assigned to the type of target. Use with custom error types.

Using errors.Is with a custom type pointer will always return false — it compares values, not types. Using errors.As with a sentinel makes no sense either. Pick the function that matches what you're inspecting.

Anti-Patterns to Avoid

Silent discards: _ = someFunc() is almost never right. If an error genuinely doesn't matter, add a comment explaining why.

String matching: strings.Contains(err.Error(), "not found") is fragile. Error messages change; sentinel comparisons don't.

Panic for runtime failures: panic belongs in startup code for invariant violations, not in request handlers. A panicking HTTP server is not a well-behaved HTTP server.

Wrapping with %v instead of %w: fmt.Errorf("ctx: %v", err) bakes the error into a string and breaks the unwrapping chain. Use %w unless you specifically want to hide the underlying type from callers.

The Takeaway

The three patterns compose naturally. Wrap at every layer to preserve context. Define sentinels only for errors that callers need to branch on. Use custom types when you need structured data. Implement Unwrap() whenever you embed an error in a struct.

The discipline pays off at 2am when a production alert fires and the error log tells you exactly which operation failed, on which input, in which layer — without requiring you to read the code.

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

Building a Go CLI Tool with Cobra and Configuration Management

Ayi NEDJIMI — Sun, 27 Sep 2026 10:02:34 +0000

Most Go CLI tools start as a single main.go with a handful of flag.Parse() calls. That works fine until you need subcommands, config file support, environment variable overrides, and sensible defaults. Cobbling that together manually is tedious and brittle. Cobra + Viper solves this in a composable way that actually scales.

Project Structure

Install the dependencies first:

go get github.com/spf13/cobra@latest
go get github.com/spf13/viper@latest

The standard layout for a Cobra project:

mycli/
  cmd/
    root.go
    scan.go
    version.go
  main.go
  go.mod

main.go is deliberately thin — it just hands off to the cmd package:

package main

import "mycli/cmd"

func main() {
    cmd.Execute()
}

This keeps the entry point free of logic and makes commands testable without spawning a subprocess.

Wiring the Root Command

cmd/root.go is where you initialize the root command, declare persistent flags, and load configuration. The key insight is that cobra.OnInitialize registers a function to run before any subcommand executes — that is where Viper reads its config.

package cmd

import (
    "fmt"
    "os"
    "strings"

    "github.com/spf13/cobra"
    "github.com/spf13/viper"
)

var cfgFile string

var rootCmd = &cobra.Command{
    Use:   "mycli",
    Short: "A security audit CLI",
    Long:  `mycli performs network and configuration security checks.`,
}

func Execute() {
    if err := rootCmd.Execute(); err != nil {
        fmt.Fprintln(os.Stderr, err)
        os.Exit(1)
    }
}

func init() {
    cobra.OnInitialize(initConfig)

    rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file (default $HOME/.mycli.yaml)")
    rootCmd.PersistentFlags().String("log-level", "info", "log level: debug, info, warn, error")

    viper.BindPFlag("log_level", rootCmd.PersistentFlags().Lookup("log-level"))
    viper.SetDefault("log_level", "info")
}

func initConfig() {
    if cfgFile != "" {
        viper.SetConfigFile(cfgFile)
    } else {
        home, err := os.UserHomeDir()
        if err != nil {
            fmt.Fprintln(os.Stderr, err)
            os.Exit(1)
        }
        viper.AddConfigPath(home)
        viper.AddConfigPath(".")
        viper.SetConfigType("yaml")
        viper.SetConfigName(".mycli")
    }

    viper.SetEnvPrefix("MYCLI")
    viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
    viper.AutomaticEnv()

    if err := viper.ReadInConfig(); err == nil {
        fmt.Fprintln(os.Stderr, "Using config:", viper.ConfigFileUsed())
    }
}

viper.AutomaticEnv() with a prefix means MYCLI_LOG_LEVEL automatically overrides log_level — no manual env parsing needed. The SetEnvKeyReplacer call maps dotted keys like scan.timeout to MYCLI_SCAN_TIMEOUT. Config file lookup is purely additive: if no file exists, Viper falls back to flags, then env vars, then defaults. The tool works correctly even in a container with no config file at all.

Adding Subcommands

Each subcommand lives in its own file. Declare the command, register it on rootCmd in init(), and bind flags to Viper keys immediately after declaring them — this is the step people forget.

// cmd/scan.go
package cmd

import (
    "fmt"

    "github.com/spf13/cobra"
    "github.com/spf13/viper"
)

var scanCmd = &cobra.Command{
    Use:   "scan [target]",
    Short: "Scan a target host for open ports and TLS issues",
    Args:  cobra.ExactArgs(1),
    RunE: func(cmd *cobra.Command, args []string) error {
        target := args[0]
        timeout := viper.GetInt("scan.timeout")
        verbose := viper.GetBool("scan.verbose")
        format := viper.GetString("output.format")

        fmt.Printf("Scanning %s (timeout=%ds, verbose=%v, format=%s)\n",
            target, timeout, verbose, format)
        return runScan(target, timeout, verbose, format)
    },
}

func init() {
    rootCmd.AddCommand(scanCmd)

    scanCmd.Flags().Int("timeout", 30, "scan timeout in seconds")
    scanCmd.Flags().Bool("verbose", false, "verbose output")

    viper.BindPFlag("scan.timeout", scanCmd.Flags().Lookup("timeout"))
    viper.BindPFlag("scan.verbose", scanCmd.Flags().Lookup("verbose"))

    viper.SetDefault("scan.timeout", 30)
    viper.SetDefault("output.format", "text")
}

Always use RunE (not Run) when a command can fail — the error propagates cleanly and Cobra handles the exit code. cobra.ExactArgs(1) validates argument count before your function runs; Cobra prints a useful error and usage hint automatically, so you do not write that boilerplate yourself.

Configuration Layering

Viper resolves configuration in strict priority order, from highest to lowest:

Explicit viper.Set() calls
CLI flags
Environment variables
Config file
Defaults

A sample ~/.mycli.yaml:

log_level: debug

scan:
  timeout: 60
  verbose: true

output:
  format: json
  file: /tmp/results.json

The layering means that in CI/CD you can override any value via environment variables without touching the config file. A developer running locally gets the file-based config. Both use the same binary and the same code path — no special-casing needed.

Set defaults in init() immediately after declaring flags so the tool works without any config file present:

viper.SetDefault("scan.timeout", 30)
viper.SetDefault("output.format", "text")
viper.SetDefault("log_level", "info")

This means viper.GetInt("scan.timeout") is always safe to call; you never get zero values silently.

Testing Commands Without a Subprocess

Cobra commands are directly testable in unit tests. Set the output writer, pass arguments, execute, and assert against the buffer:

func TestScanCommand(t *testing.T) {
    t.Cleanup(func() { viper.Reset() })

    buf := new(bytes.Buffer)
    rootCmd.SetOut(buf)
    rootCmd.SetErr(buf)
    rootCmd.SetArgs([]string{"scan", "192.168.1.1", "--timeout", "10"})

    if err := rootCmd.Execute(); err != nil {
        t.Fatalf("unexpected error: %v", err)
    }

    output := buf.String()
    if !strings.Contains(output, "192.168.1.1") {
        t.Errorf("expected target in output, got: %s", output)
    }
}

The viper.Reset() call in t.Cleanup is critical. Viper is a global singleton by default, so config set in one test leaks into the next without it. For larger projects, inject a *viper.Viper instance into command structs rather than relying on the package-level functions — that makes tests fully isolated without any cleanup ceremony.

Exit Codes and Distribution

Exit codes matter for scripting. Cobra defaults to exit code 1 on error, which is usually fine, but you sometimes need specific codes — 2 for usage errors, 3 for scan-found-issues vs scan-failed. Handle this explicitly:

RunE: func(cmd *cobra.Command, args []string) error {
    findings, err := runScan(args[0])
    if err != nil {
        return fmt.Errorf("scan failed: %w", err)
    }
    if len(findings) > 0 {
        os.Exit(2) // findings present — useful for CI gating
    }
    return nil
},

For distribution, goreleaser integrates cleanly with Cobra projects and handles cross-compilation, checksums, and GitHub releases in one YAML config. Pair it with a version subcommand that reads from a build-time ldflags variable and you have a complete, professional CLI package with minimal overhead.

The Takeaway

The Cobra + Viper combination is not magic — it is disciplined wiring. The value comes from getting the resolution chain right once (flags → env → file → defaults) and never revisiting it as you add subcommands. Each new command inherits the entire config system for free.

For security tooling specifically, the environment variable override path matters a lot: operators running audits in CI/CD pipelines need to inject credentials and targets without modifying files on disk. Getting this right from the start avoids a painful retrofit when your tool graduates from local use to automated pipelines. For hardening the environments these tools run in, the free security hardening checklists at AYI NEDJIMI Consultants are a useful companion reference.

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

Implementing Distributed Tracing in Go with OpenTelemetry

Ayi NEDJIMI — Sat, 26 Sep 2026 10:06:42 +0000

Distributed tracing is one of those things that's easy to skip when a project is small and genuinely painful to retrofit when it's not. A request hits your API gateway, fans out across three services, spawns a database query and two external calls, then returns — and when something is slow or broken, you're left grepping through logs hoping timestamps line up. OpenTelemetry fixes this, and integrating it into a Go service takes less time than you think.

What OpenTelemetry actually is (and isn't)

OpenTelemetry (OTel) is a vendor-neutral observability framework — it handles the instrumentation layer so you're not locked to Datadog, Jaeger, Zipkin, or any other backend. You instrument once, then ship traces anywhere that speaks OTLP (OpenTelemetry Protocol). The Go SDK is stable and actively maintained.

OTel is not a tracing backend. You still need something to store and visualize traces. For local development, Jaeger is the fastest option to spin up. In production, most teams ship to Grafana Tempo, Honeycomb, or a commercial APM that supports OTLP ingest. The exporter endpoint is a one-line config change — so the backend decision is reversible.

Setting up the tracer provider

Install the required packages first:

go get go.opentelemetry.io/otel \
    go.opentelemetry.io/otel/sdk/trace \
    go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc \
    go.opentelemetry.io/otel/propagation \
    go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp \
    google.golang.org/grpc

Then initialize the tracer provider at startup. The key decisions here are which exporter to use, which sampler, and how to propagate context across service boundaries.

package telemetry

import (
    "context"
    "fmt"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
)

func InitTracer(ctx context.Context, serviceName, collectorAddr string) (*sdktrace.TracerProvider, error) {
    conn, err := grpc.NewClient(
        collectorAddr,
        grpc.WithTransportCredentials(insecure.NewCredentials()),
    )
    if err != nil {
        return nil, fmt.Errorf("failed to connect to collector: %w", err)
    }

    exporter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithGRPCConn(conn))
    if err != nil {
        return nil, fmt.Errorf("failed to create exporter: %w", err)
    }

    res := resource.NewWithAttributes(
        semconv.SchemaURL,
        semconv.ServiceName(serviceName),
        semconv.ServiceVersion("1.0.0"),
    )

    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exporter),
        sdktrace.WithResource(res),
        sdktrace.WithSampler(sdktrace.ParentBased(sdktrace.TraceIDRatioBased(0.1))),
    )

    otel.SetTracerProvider(tp)
    otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
        propagation.TraceContext{},
        propagation.Baggage{},
    ))

    return tp, nil
}

ParentBased(TraceIDRatioBased(0.1)) means sample 10% of root spans, but always follow the sampling decision of an upstream parent. If an upstream service decides to trace a request, you trace it too. This is almost always what you want in a multi-service setup — a gateway controls the sampling decision and all downstream services defer to it.

Call tp.Shutdown(ctx) in your graceful shutdown path — it flushes buffered spans before the process exits.

Instrumenting HTTP handlers

The fastest way to instrument HTTP handlers is through the otelhttp middleware. For standard net/http:

package middleware

import (
    "net/http"

    "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

func TraceMiddleware(next http.Handler) http.Handler {
    return otelhttp.NewHandler(next, "http.request",
        otelhttp.WithMessageEvents(otelhttp.ReadEvents, otelhttp.WriteEvents),
    )
}

For Gin, use go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin. For Fiber, otelfiber does the same. All of them:

Extract trace context from incoming traceparent headers (W3C Trace Context standard)
Create a server span for each request
Record HTTP method, route, and status code as span attributes
Inject span context into the request's context.Context for downstream propagation

Propagating context to downstream calls

The most common mistake is creating spans correctly but not propagating the context when making outbound HTTP calls. This severs the trace — you get disconnected spans instead of a single trace tree. Distributed tracing only works when every hop in a call chain forwards the traceparent header.

import (
    "context"
    "net/http"

    "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/propagation"
)

func callDownstream(ctx context.Context, url string) (*http.Response, error) {
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return nil, err
    }

    // Inject trace context into outbound request headers
    otel.GetTextMapPropagator().Inject(ctx, propagation.HeaderCarrier(req.Header))

    client := &http.Client{
        Transport: otelhttp.NewTransport(http.DefaultTransport),
    }
    return client.Do(req)
}

Using otelhttp.NewTransport handles injection and creates a client span for the outbound call. The downstream service receives traceparent and links its spans to the same trace. For database calls, go.opentelemetry.io/contrib/instrumentation/database/sql/otelsql wraps database/sql and traces every query with no additional code changes.

Adding custom spans to business logic

Auto-instrumentation captures HTTP and DB calls, but interesting latency usually lives in your business logic. Add custom spans to the code paths that matter:

import (
    "context"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/codes"
)

func processOrder(ctx context.Context, orderID string) error {
    tracer := otel.Tracer("order-service")
    ctx, span := tracer.Start(ctx, "processOrder")
    defer span.End()

    span.SetAttributes(
        attribute.String("order.id", orderID),
        attribute.String("order.channel", "api"),
    )

    if err := validateInventory(ctx, orderID); err != nil {
        span.RecordError(err)
        span.SetStatus(codes.Error, err.Error())
        return err
    }

    return chargePayment(ctx, orderID)
}

Keep span names stable and low-cardinality. processOrder is good; processOrder-550e8400-e29b-41d4-a716 is not — high-cardinality names break trace aggregation in every backend.

Sampling strategy: the decision that matters most

At 10 req/s, sampling 100% is fine. At 10k req/s, you're looking at real storage and egress costs. The sampler you configure on the tracer provider is the main lever.

Two practical patterns:

Head-based sampling (decided at trace start): use TraceIDRatioBased for uniform sampling, or implement the Sampler interface for rule-based decisions — always trace requests containing auth failures, never trace health checks.

Tail-based sampling (decided after the trace completes): run the OpenTelemetry Collector with tailsamplingprocessor. More infrastructure to maintain, but you can keep all traces containing errors and drop the healthy majority. This is the right choice for security-sensitive services where you need full fidelity on anomalies.

For security-focused services — access control layers, authentication endpoints, audit pipelines — trace retention requirements often align with compliance requirements. The security hardening checklists at AYI NEDJIMI Consultants cover observability retention policies alongside the usual hardening items.

The takeaway

Getting OTel running in a Go service takes under two hours, including a local Jaeger instance for validation. The hard part is getting the team to propagate context correctly through every outbound call — that discipline is what makes distributed tracing useful rather than a collection of disconnected spans. Start with otelhttp middleware and the OTLP gRPC exporter pointed at Jaeger locally. Add custom spans where business logic lives. Once the structure is right, swapping the collector endpoint for production is a one-line change.

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

DEV Community: Ayi NEDJIMI

Building a Production-Ready AI Chatbot with Memory and Context

The Core Problem: Stateless Models and Token Cost

Strategy 1: Sliding Window

Strategy 2: Summarization-Based Memory

Strategy 3: Semantic Memory with Vector Retrieval

The Production Architecture

The Takeaway

Implementing graceful shutdown in Go HTTP servers

Why Abrupt Shutdown Breaks Things

The Basics: Catching SIGTERM and Draining Connections

Handling Long-Running Requests With Context Propagation

Coordinating Background Workers

Kubernetes and the SIGTERM → SIGKILL Gap

The Takeaway

How to Build an AI-Powered Log Summarizer for DevOps

The Problem with Raw Logs

Architecture: What We're Building

Collecting and Preprocessing Logs

Summarizing with an LLM

Putting It Together: A CLI Tool

Practical Limits to Know Before You Ship

The Takeaway

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

What You Need to Trace

Building a Wrapper Tracer in Python

Adding Time-to-First-Token for Streaming

Shipping Traces to a Backend

What to Alert On

The Takeaway

AI Model Versioning and Rollback Strategies for Production

Why model versioning is different from code versioning

A versioned model registry in Python

Rollback strategies in production

When to automate rollback decisions

The takeaway

Building a Vector Search Engine from Scratch with HNSW in Python

Why Not Just Use Brute Force?

How HNSW Works

The Implementation

Measuring Recall Against Brute Force

Production Notes

The Takeaway

How to build a job queue in Go without external dependencies

What we're building

The core structure

Graceful shutdown with OS signals

Adding priorities with two channels

What you trade away vs. a real broker

The takeaway

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

How LoRA works under the hood

Setting up LoRA fine-tuning with PEFT

Your training data matters more than your hyperparameters

Merging and deploying your adapter

When LoRA is not the right tool

The takeaway

How to Evaluate RAG Pipeline Quality: Metrics and Test Harness

What You're Actually Measuring

Building a Reference Dataset

Computing Faithfulness with an LLM Judge

Computing Context Recall with Token Overlap

The Test Harness: Putting It Together

Integrating into CI

Scaling Beyond Toy Datasets

The Takeaway

Go error handling patterns that scale: wrapping, unwrapping, sentinels

Why return err Isn't Enough

Error Wrapping with fmt.Errorf and %w

Sentinel Errors

Custom Error Types

errors.Is vs errors.As: Choosing the Right Tool

Anti-Patterns to Avoid

The Takeaway

Building a Go CLI Tool with Cobra and Configuration Management

Project Structure

Wiring the Root Command

Adding Subcommands

Configuration Layering

Testing Commands Without a Subprocess

Why `return err` Isn't Enough