DEV Community: Alain Airom (Ayrom)

Book Review: Build Applications with Local AI Models on a Mac

Alain Airom (Ayrom) — Thu, 23 Jul 2026 09:20:43 +0000

Feedback and synthesis on of my latest readings: “Build Applications with Local AI Models on a Mac”

Introduction

As cloud-based AI infrastructure and API subscription costs continue to rise, running Large Language Models (LLMs) locally on consumer-grade hardware has transitioned from a niche hobby to a compelling architectural shift. The primary drivers behind this movement are privacy and economy. From a privacy perspective, processing data locally guarantees that confidential text, proprietary source code, internal business records, and personally identifiable information (PII) never leave your physical device. There are no remote logging pipelines, third-party data retention policies, or potential cloud leaks to worry about. On the economic front, local execution removes pay-per-token API metering entirely. Once you own the hardware, running inference across millions of tokens carries zero incremental API cost, making local models extremely cost-effective for high-volume tasks, continuous agent loops, embedding generation, and rapid local prototyping.

To be clear, one cannot do all with local LLMs in my opinion, but it helps for some tasks…

I, myself use Ollama on a regular basis with different models for agents, embeddings, etc. I was curious to read this book and figure out whether I would learn something new or not.

Disclaimer: I have no ties, commercial relationships, or affiliations with Manning Publications (manning.com) or the author of this book, Keiji Kamigusa. This review and breakdown are purely independent.

All images are from the book’s author.

Part 1: First Steps — Entering the World of the Terminal

Chapters 1 to 6 are really basic, destinated for beginners! You can ignore big parts of them.

Chapter 1: Getting Started with Local AI

This introductory chapter establishes the conceptual framework for local AI development on Apple Silicon and Intel Macs. It introduces how local inference works on modern hardware — highlighting the unified memory architecture (UMA) of Apple’s M-series chips — and compares local runtime environments against traditional cloud APIs in terms of latency, privacy, and cost predictability.

Chapter 2: Installing and Using Homebrew

Focusing on developer tooling, this chapter guides us through setting up Homebrew, the standard package manager for macOS. It covers terminal basics, environment path setup, and how command-line utilities form the foundational stack for managing dependencies in local machine learning projects.

Chapter 3: Installing and Setting Up Ollama

Here, the book dives into Ollama, the central engine used throughout the book to serve local models. It covers installation, service startup, verifying running instances via terminal commands, and managing local model storage.

# Pulling and running a lightweight model locally via Ollama terminal CLI
# for example...
ollama run granite3.2

Part 2: Core Skills — Building Your First AI Chatbot

Chapter 4: Downloading an LLM and Your First Conversation

This chapter demonstrates basic CLI interactions with quantized open-weights models (such as Granite, Llama, Gemma, and Qwen). We learn how to pull models from the registry, run interactive shell sessions, and monitor memory usage during inference.

Chapter 5: Setting Up VS Code and Your Python Development Environment

Moving from CLI usage to application development, this chapter covers configuring Visual Studio Code, installing Python, and structuring a clean directory workspace for building custom AI tools.

Chapter 6: Creating and Managing Python Virtual Environments

To avoid package dependency conflicts, this chapter explains Python virtual environments (venv). It covers environment creation, activation, and managing dependency lists with requirements.txt.

# Creating and activating a clean Python virtual environment
python3 -m venv .venv
source .venv/bin/activate
pip install ollama streamlit

Chapter 7: Controlling LLMs via the Ollama Python API

This chapter introduces programatically driving local models using the official ollama Python client. It demonstrates synchronous text generation, chat completions, and real-time response streaming.

import ollama

# Querying a local model programmatically with streaming enabled
response = ollama.chat(
    model="gemma2:2b",
    messages=[{"role": "user", "content": "Explain vector embeddings in one sentence."}],
    stream=True,
)

for chunk in response:
    print(chunk["message"]["content"], end="", flush=True)

Chapter 8: Building a Web UI with Streamlit

Demonstrating how to wrap Python backend code into an interactive graphical web interface using Streamlit. The chapter demonstrates rapid prototyping of UI elements like text inputs, buttons, and message display blocks.

Chapter 9: Recording Audio and Transcribing Speech with MLX Whisper

Expanding beyond text input, this chapter integrates Apple Silicon-optimized speech recognition using MLX Whisper. It details audio input capture and converting spoken voice into clean text transcripts locally. (I liked this chapter 🙂 and the following one)

Chapter 10: Building a Voice-Enabled AI Chat Application

Combining speech recognition with LLM generation, this chapter walks through building a hands-free, voice-driven AI conversational agent that listens, transcribes, processes queries, and outputs responses.

Chapter 11: Managing Session State and Chat History

Personally I really appreciated this part of the book!

To handle multi-turn conversations, this chapter addresses managing state inside Streamlit applications using st.session_state. It demonstrates how to maintain chat history, trim context windows, and calculate token counters.

import streamlit as st

# Managing chat state across Streamlit reruns
if "messages" not in st.session_state:
    st.session_state.messages = []

# Displaying chat history
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

Part 3: Advanced Topics — Deeper Understanding and Customization

Chapters 12 and 13 (part 3 of the book) interested me personally the most.

Chapter 12: Comparing and Selecting LLM Models

This chapter provides a comparative framework for evaluating open-weight models (e.g., Granite4, Gemma 3, Qwen 2.5, Phi-3). It contrasts parameter counts, context window sizes, quantization impact, and inference speed benchmarks across different hardware setups.

import time
import ollama

# Benchmarking inference speed across multiple local models
models = ["gemma3:4b", "qwen2.5:3b", "phi3:mini"]
prompt = "What are three key benefits of local AI inference?"

for model_name in models:
    start_time = time.time()
    response = ollama.chat(
        model=model_name,
        messages=[{"role": "user", "content": prompt}],
    )
    elapsed = time.time() - start_time
    print(f"Model: {model_name} | Response Time: {elapsed:.2f}s")

Chapter 13: System Prompts and Parameter Tuning

Focusing on output control, this chapter details how to shape model behavior using system prompts (assigning personas or strict rules) and adjusting hyperparameters like temperature to control randomness.

import ollama

# System prompt persona assignment and parameter tuning
response = ollama.chat(
    model="gemma2:2b",
    messages=[
        {"role": "system", "content": "You are a concise expert. Answer strictly in bullet points."},
        {"role": "user", "content": "Summarize the benefits of virtual environments."},
    ],
    options={"temperature": 0.2},
)
print(response["message"]["content"])

Chapter 14: Offline LLMs and Their Benefits

This chapter focuses on air-gapped operations, demonstrating how locally hosted models function entirely without active internet connectivity — ensuring high security and complete system resilience against remote network failures.

Chapter 15: Common Errors and Troubleshooting

A practical reference chapter covering common execution issues, including missing package dependencies, port binding errors, out-of-memory (OOM) GPU allocation failures, and Ollama daemon connectivity fixes.

Chapter 16: Advancing to RAG, Web UI, and Fine-Tuning

This chapter introduces advanced application design patterns, primarily focusing on Retrieval-Augmented Generation (RAG) using vector stores like ChromaDB, agent tool integrations with LangChain, and high-level interfaces like Open WebUI.

Personally, I prefer other vector databases such as AstraDB or Milvus, but the book focuses on ChromaDB.

import chromadb
import ollama

# RAG setup: Injecting context into local LLM query
client = chromadb.Client()
collection = client.create_collection("my_knowledge")

collection.add(
    documents=["Ollama runs LLMs locally on macOS, Linux, and Windows."],
    ids=["doc1"],
)

results = collection.query(query_texts=["Where can Ollama run?"], n_results=1)
context = "\n".join(results["documents"][0])

prompt = f"Context:\n{context}\n\nQuestion: Where can Ollama run?\nAnswer:"
response = ollama.chat(
    model="gemma3:4b",
    messages=[{"role": "user", "content": prompt}],
)
print(response["message"]["content"])

Chapter 17: The Open Model Revolution of 2026

Taking a broader architectural view, this chapter explores the landscape of open-weight artificial intelligence models, discussing open ecosystem standards, model safety guardrails, and hardware developments shaping modern on-device execution.

Chapter 18: Where to Go from Here

The concluding chapter provides a roadmap for further developer exploration, suggesting paths toward custom agent orchestration frameworks, fine-tuning techniques, and deployment options beyond local development environments.

Conclusion & Personal Opinion

Build Applications with Local AI Models on a Mac provides a practical, step-by-step introduction to running LLMs on consumer hardware and integrating them into functional Python applications. By guiding the readers through terminal setup, Ollama configuration, Streamlit web UI creation, speech transcription via MLX Whisper, and foundational RAG patterns using ChromaDB and LangChain, the book serves as a clear starting guide for on-device AI integration.

Personal Opinion

Overall, this was mostly an enjoyable read . It is particularly well-suited for novices, beginners, or developers who are new to running models locally using Ollama and want a structured, practical path from terminal installation to building functional UI wrappers and voice agents. While experienced developers already familiar with agentic loops, custom embedding pipelines, and low-level quantization mechanics might find much of the initial material elementary, it remains a clean, cohesive refresher. The accompanying code repository provided with the book is well-organized, making it easy to follow along, test code snippets, and quickly adapt the examples for personal experiments.

Thanks for reading 📔

Building an AI-Native Document Platform with Doclang

Alain Airom (Ayrom) — Wed, 22 Jul 2026 08:38:15 +0000

A technical deep dive into architecting an end-to-end playground for DocLang — the AI-native markup language built for LLMs.

🚀 Introduction & Overview

As Large Language Model (LLM) applications mature, a fundamental friction point remains: legacy document formats were never designed for tokenizers.

PDF was built for precise visual print layout.
DOCX was built for human desktop editors.
HTML is bloated with display tags, costing up to 5× more tokens for structured elements like tables.

DocLang is an open document standard governed by *Linux Foundation AI & Data *(backed by IBM, NVIDIA, Red Hat, ABBYY, and HumanSignal). DocLang maps structured documents cleanly to LLM tokens while preserving precise geometry, semantic hierarchies, code blocks, and formulas in an unambiguous XML markup representation.

What is Doclang?

Excerpt from Doclang’s site.

DocLang is the AI-native markup format for unstructured content — documents, images, audio, and video alike. It is engineered from the ground up around a single principle: mapping cleanly to LLM tokens. That makes DocLang the most efficient way to read, write, and reason about real-world content with modern AI.

The world’s knowledge lives in formats — PDF, HTML, Word, LaTeX, MP3, MP4 — that were designed for rendering or playback, not for understanding. The moment you hand them to an AI system, you lose structure, semantics, layout, timing, or all of them at once. And the formats AI does speak fluently fall short too: Markdown can’t describe complex content, HTML wastes tokens, LaTeX is ambiguous, and none of them speak audio or video at all.

DocLang fixes this:

AI-native — a controlled vocabulary that aligns 1-to-1 with LLM tokenizers, keeping prompts and outputs short and predictable
Lossless — preserves structure, semantics, layout, and geometry in a single representation
Expressive — first-class support for tables, formulas, code, charts, nested lists, forms, and multimodal content with visual grounding
Beyond documents — extends naturally to audio and video, with native primitives for transcripts, speakers, timestamps, scenes, and audio-visual grounding — so an interview, a lecture recording, or a film script lives in the same representation as a PDF
Unambiguous — every tag has one job, so the same content always serializes the same way
Open — an ISO standard, developed in the open, governed by industry leaders

If you build with LLMs and VLMs on real-world content, DocLang is the substrate you’ve been missing.

Implementation

To demonstrate the full lifecycle of ingesting, converting, validating, and querying DocLang, Bob authored and wired together the DocLang Demonstration Application.

┌─────────────────────────────────────────────────────────────────┐
│                        Browser (UI)                             │
│  ┌───────────┐ ┌───────────┐ ┌──────────┐ ┌─────────────────┐  │
│  │ Markdown  │ │  DocLang  │ │  Upload  │ │  Ask Ollama     │  │
│  │ → DocLang │ │  Editor   │ │ Document │ │  (Q&A on docs)  │  │
│  └─────┬─────┘ └─────┬─────┘ └────┬─────┘ └────────┬────────┘  │
└────────│─────────────│────────────│───────────────────│─────────┘
         │             │            │                   │
         ▼             ▼            ▼                   ▼
┌─────────────────────────────────────────────────────────────────┐
│                    Flask API (app.py)                            │
│                                                                 │
│  /api/convert/markdown  → markdown_to_doclang()                 │
│  /api/convert/doclang   → doclang_to_html()                     │
│  /api/validate          → validate_doclang()                    │
│  /api/upload/submit     → Docling job started (returns job_id)  │
│  /api/upload/status/<id>→ poll job → doclang + preview          │
│  /api/ask               → Ollama /api/generate                  │
│  /api/save              → output/{name}_{timestamp}.dclg.xml    │
│  /api/samples           → input/**/*.dclg.xml                   │
└──────────┬──────────────────────────┬───────────────────────────┘
           │                          │
           ▼                          ▼
  ┌─────────────────┐      ┌──────────────────────┐
  │  Docling Engine │      │   Ollama (local LLM) │
  │  (PDF/DOCX/HTML │      │   granite / any      │
  │  → DocLang XML) │      │   model installed    │
  └─────────────────┘      └──────────────────────┘

🏗️ System Architecture & Workflow Schemas

The application is structured as a lightweight, fast Flask API backing a multi-tab Single-Page Application (SPA) interface. It seamlessly bridges client-side interactions with heavy-duty document parsing engines and local LLMs.

High-Level System Architecture

The DocLang Demonstration Application is architected as a decoupled, multi-tier system engineered for low latency and smooth document processing. At its core, a lightweight Flask REST API coordinates requests between an interactive Single-Page Application (SPA) frontend, local file storage, and specialized downstream service workers. To keep the UI highly responsive during heavy document transformations, long-running extraction jobs are offloaded to an asynchronous thread pool worker powered by IBM’s Docling engine, while grounded natural language querying is driven directly by a local Ollama LLM instance.

Document Processing & Ingestion Pipeline

Under the hood, the application balances real-time user feedback with heavy computational tasks. The architecture is based on a fast Flask backend running on Python 3.10+, serving an intuitive multi-tab Single-Page Application. Rather than locking up the user interface when parsing massive PDFs, the application decouples frontend UI events from heavy background jobs — handing off document extraction to background threads and local AI inference to Ollama via local API calls.

🛠️ Key Implementation Highlights

Translating the DocLang specification from concept to a production-ready Web application required addressing critical real-world edge cases across multi-threading, schema conversion, and local LLM orchestration. The core backend implementation in app.py bridges low-level system hardware constraints with high-level parsing logic. Below, we break down three foundational components of the application's implementation, detailing how we tackled macOS hardware threading crashes, authored an efficient client-side Markdown converter, and connected DocLang directly to local Ollama inference.

Solving the macOS Metal / PyTorch MPS Threading Crash

When executing deep learning models (such as Docling’s layout and OCR models) inside a background worker thread on macOS, PyTorch attempts to dispatch ops through Apple Metal (MPS). In non-main threads, this throws an unrecoverable C-level abort: MTLCompiler: compileRequest ... Crashing instead.

Bob solved this cleanly in app.py by forcing PyTorch MPS fallback before any imports occur, and pinning Docling explicitly to CPU thread pools:

import os

# MUST be configured prior to importing torch or docling
os.environ.setdefault("PYTORCH_ENABLE_MPS_FALLBACK", "1")

from docling.document_converter import DocumentConverter, PdfFormatOption
from docling.datamodel.pipeline_options import PdfPipelineOptions, AcceleratorOptions, AcceleratorDevice

def _make_converter():
    """Build a DocumentConverter pinned to CPU to ensure thread-safe execution on macOS."""
    accel = AcceleratorOptions(num_threads=4, device=AcceleratorDevice.CPU)

    opts = PdfPipelineOptions()
    opts.accelerator_options = accel

    return DocumentConverter(
        format_options={
            "pdf": PdfFormatOption(pipeline_options=opts),
        }
    )

Hand-Rolled Markdown to DocLang XML Parser

For fast client-side transformations without invoking heavy neural models, app.py features a custom Markdown-to-DocLang parser. It transforms standard Markdown elements into native DocLang tags:

def markdown_to_doclang(md_text: str) -> str:
    """Convert simplified Markdown into valid DocLang XML markup."""
    lines = md_text.strip().splitlines()
    elements: list[str] = []
    i = 0
    while i < len(lines):
        line = lines[i]
        stripped = line.strip()

        if stripped.startswith("### "):
            elements.append(f'  <heading level="3">{stripped[4:]}</heading>')
        elif stripped.startswith("## "):
            elements.append(f'  <heading level="2">{stripped[3:]}</heading>')
        elif stripped.startswith("# "):
            elements.append(f'  <heading level="1">{stripped[2:]}</heading>')
        elif stripped.startswith("```

"):
            lang = stripped[3:].strip() or "text"
            code_lines: list[str] = []
            i += 1
            while i < len(lines) and not lines[i].strip().startswith("

```"):
                code_lines.append(lines[i])
                i += 1
            code_body = "\n".join(code_lines)
            elements.append(
                f'  <code>\n    <label value="{lang}"/>\n'
                f'    <content><![CDATA[\n{code_body}\n    ]]></content>\n  </code>'
            )
        elif stripped.startswith("- "):
            list_items: list[str] = []
            while i < len(lines) and lines[i].strip().startswith("- "):
                list_items.append(f"    <item><text>{lines[i].strip()[2:]}</text></item>")
                i += 1
            elements.append("  <list>\n" + "\n".join(list_items) + "\n  </list>")
            continue
        elif stripped:
            elements.append(f"  <text>{stripped}</text>")

        i += 1

    return "<doclang>\n" + "\n".join(elements) + "\n</doclang>"

Local LLM Q&A Integration with Ollama

DocLang’s concise XML representation makes it optimal for direct inclusion in system prompts. The application routes user queries and DocLang markup directly to local Ollama models:

def ask_ollama(question: str, context_dclg: str) -> str:
    """Perform grounded Q&A over parsed DocLang XML using local Ollama instance."""
    prompt = textwrap.dedent(f"""
        You are a helpful assistant. The following is a document in DocLang format
        (an AI-native XML document standard). Answer the user's question using only
        the information present in the document.

        <document>
        {context_dclg}
        </document>

        Question: {question}
        Answer:
    """).strip()

    resp = requests.post(
        f"{OLLAMA_BASE_URL}/api/generate",
        json={"model": OLLAMA_MODEL, "prompt": prompt, "stream": False},
        timeout=60,
    )
    return resp.json().get("response", "").strip()

📊 Token Efficiency Benchmark

DocLang slashes context window consumption compared to legacy HTML representations — achieving ~64.2% token savings on standard documents containing complex tables, headings, and code samples.

🧪 Automated Testing Suite

To ensure reliability across background task queues, file IO, and API validation, Bob engineered a unit test suite using pytest in tests/test_app.py:

class TestAPIRoutes:
    def test_api_convert_markdown(self, client):
        resp = client.post("/api/convert/markdown", json={"markdown": "# Hello\n\nWorld."})
        assert resp.status_code == 200
        data = resp.get_json()
        assert "doclang" in data
        assert "html_preview" in data
        assert "tokens" in data

    def test_api_upload_submit_creates_job(self, client):
        from io import BytesIO
        data = {"file": (BytesIO(b"<html><body>Hello</body></html>"), "test.html")}
        resp = client.post("/api/upload/submit", data=data, content_type="multipart/form-data")
        assert resp.status_code == 200
        body = resp.get_json()
        assert "job_id" in body

        # Poll async status
        job_resp = client.get(f"/api/upload/status/{body['job_id']}")
        assert job_resp.status_code == 200

💡 Conclusion

The DocLang Demo Application designed by Bob provides a complete blueprint for integrating next-generation AI-native document formats into enterprise software workflows.

By pairing IBM Docling’s multi-format extraction with DocLang’s token-optimized XML schema and Ollama’s local LLM reasoning, developers can build privacy-first, lightning-fast document intelligence pipelines.

Thanks for reading ⭐

Vaulting Beyond Defaults: The Evolution of Secrets Management on Red Hat OpenShift

Alain Airom (Ayrom) — Sun, 19 Jul 2026 12:25:12 +0000

Why using Vault is better to manage your secrets?

In modern containerized infrastructures, managing sensitive data like database passwords, API tokens, and private keys is one of the most critical aspects of day-to-day operations. When moving to Kubernetes and Red Hat OpenShift, understanding the default mechanisms — and where they fall short for enterprise security — is the first step toward building a truly secure by design architecture.

💡 Note from the Field: Before we jump into the post, all code, deployment manifests, and architectural implementation logic featured throughout this guide were designed, orchestrated, and automated by Bob (IBM Bob SDLC).

🤓

The Baseline: Secrets Management in Kubernetes

In vanilla Kubernetes, the native Secret object provides a way to separate sensitive data from application code or standard configurations (ConfigMaps). However, the word "Secret" can be misleading for newcomers:

Base64 Encoding is Not Encryption: By default, Kubernetes secrets are stored as flat, base64-encoded strings within the cluster data store. Base64 is a data transformation format designed to handle binary data over text-based networks; it provides zero obfuscation or security. Anyone with access to the YAML declaration or read permissions to the API can immediately reverse the encoding (echo "bXktcGFzc3dvcmQ=" | base64 -d).
The etcd Vulnerability: In a standard Kubernetes control plane, unless specifically configured otherwise, these base64 strings are stored in plain text inside the etcd database. If an attacker gains access to the underlying master host storage or backups of etcd, the entire cluster's credentials are compromised.
Coarse-Grained Access Control: Kubernetes Roles and RoleBindings restrict access at the resource level, but managing fine-grained programmatic access to specific keys within a single namespace becomes complex and error-prone.

The Upgraded Standard: How OpenShift Secures Native Secrets

Red Hat OpenShift takes the fundamental concepts of Kubernetes and hardens them to meet enterprise compliance and security baselines out-of-the-box.

Mandatory Multi-Tenancy via Projects: OpenShift extends Kubernetes namespaces into Projects. Projects enforce strict multi-tenancy from the start, isolating network traffic, service accounts, and resource limits (ResourceQuotas) so that an application team cannot accidentally view or cross-contaminate another team's secrets.
Automated etcd Encryption at Rest: Unlike vanilla Kubernetes where encryption must be manually configured via an encryption configuration file, OpenShift enables cluster-wide etcd encryption at rest smoothly through its cluster operators. When enabled, secret resources are automatically encrypted using robust cryptographic algorithms (such as AES-256-GCM) before hit-to-disk inside etcd.
Strict Security Context Constraints (SCCs): OpenShift mitigates runtime risks by ensuring pods run under rigorous security rules. Applications cannot blindly run as root or mount forbidden host directories, preventing compromised containers from harvesting cluster tokens or local secrets files directly from the underlying node storage.

While OpenShift native secrets are significantly more robust than vanilla Kubernetes, enterprise security teams often demand zero-trust paradigms, live credential rotation without downtime, centralized audit trails, and elimination of sensitive data from Git repositories entirely. This is where external secrets managers like HashiCorp Vault come into play.

Chapter 1: The Classical Approach — OpenShift Native Secrets (Approach A)

To truly appreciate the advantages of a dynamic secrets manager like HashiCorp Vault, we must first analyze the reference architecture, implementation, and operational lifecycle of the classical, platform-native approach.

High-Level Architecture

In this architecture, the cluster relies entirely on OpenShift’s built-in control plane. Secrets are submitted via the OpenShift API, committed encrypted into etcd, and then bound to the application pod via environment variables or read-only volume mounts.

Step-by-Step Reference Implementation

Declaring the Hardened Project and RBAC Boundaries

First, we establish a clean, production-ready namespace coupled with a dedicated ServiceAccount and explicit Role bindings utilizing the principle of least privilege.

# 00-namespace.yaml
apiVersion: v1
kind: Namespace
metadata:
  name: demo
  labels:
    environment: production
    secrets-approach: native
---
apiVersion: v1
kind: ResourceQuota
metadata:
  name: demo-quota
  namespace: demo
spec:
  hard:
    requests.cpu: "4"
    requests.memory: 8Gi
    limits.cpu: "8"
    limits.memory: 16Gi
    pods: "20"

# 01-rbac.yaml
apiVersion: v1
kind: ServiceAccount
metadata:
  name: demo-sa
  namespace: demo
automountServiceAccountToken: false
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: secret-reader
  namespace: demo
rules:
  - apiGroups: [""]
    resources: ["secrets"]
    verbs: ["get", "list"]
  - apiGroups: [""]
    resources: ["configmaps"]
    verbs: ["get", "list"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: demo-sa-secret-reader
  namespace: demo
subjects:
  - kind: ServiceAccount
    name: demo-sa
    namespace: demo
roleRef:
  kind: Role
  name: secret-reader
  apiGroup: rbac.authorization.k8s.io

Defining the Platform Secrets

Next, we define our credentials using native Secret configurations. Notice that values are raw base64 placeholders.

# 02-secrets.yaml (Excerpt)
apiVersion: v1
kind: Secret
metadata:
  name: demo-db-secret
  namespace: demo
  annotations:
    secrets.management/rotation-interval: "90d"
    secrets.management/last-rotated: "2026-01-01"
type: Opaque
data:
  DB_USER: ZGVtb191c2Vy                         # demo_user
  DB_PASSWORD: CHANGEME_BASE64_ENCODED_PASSWORD==
  DB_HOST: cG9zdGdyZXMuZGVtby5zdmMuY2x1c3Rlci5sb2NhbA==
  DB_PORT: NTQzMg==                             # 5432
  DB_NAME: ZGVtb2Ri                             # demodb

Injecting Secrets into the Application Deployment

The application deployment can consume these secrets in two distinct formats: as environment variables mapped straight into the container’s environment via envFrom, or mounted explicitly as read-only files through standard volumes.

# 04-deployment.yaml (Configuration Excerpt)
spec:
  template:
    spec:
      serviceAccountName: demo-sa
      containers:
        - name: demo-app
          image: image-registry.openshift-image-registry.svc:5000/demo/demo-app:latest
          envFrom:
            - configMapRef:
                name: demo-config
            - secretRef:
                name: demo-db-secret
            - secretRef:
                name: demo-admin-secret
          volumeMounts:
            - name: ldap-secret-vol
              mountPath: /secrets/ldap
              readOnly: true
      volumes:
        - name: ldap-secret-vol
          secret:
            secretName: demo-ldap-secret

How the Application Consumes Native Secrets?

From an application development perspective, using environment-injected native secrets is incredibly simple. There are no specialized APIs, external packages, or external client initializations required. The application merely reads from its operating system context.

# app.py (Excerpt)
import os
from flask import Flask, jsonify

app = Flask(__name__)

class Config:
    SECRET_SOURCE = "environment variables (OpenShift native Secret → envFrom)"

    # The platform injected the value into the process environment
    # before the application process started execution.
    DB_USER     = os.environ.get("DB_USER",     "not-set")
    DB_PASSWORD = os.environ.get("DB_PASSWORD", "not-set")
    DB_HOST     = os.environ.get("DB_HOST",     "not-set")
    DB_PORT     = os.environ.get("DB_PORT",     "5432")
    DB_NAME     = os.environ.get("DB_NAME",     "not-set")

@app.route("/config")
def config():
    return jsonify({
        "approach": "A — OpenShift Native Secrets",
        "secret_source": Config.SECRET_SOURCE,
        "loaded_at": "process start (env vars set by Kubelet before exec)",
        "rotation_note": "Pod RESTART required to pick up a new secret value"
    })

The Operational Reality of Native Secrets

While this appraoch is easy to grasp, has no external system requirements, and operates out-of-the-box in air-gapped data centers, it creates massive friction under enterprise conditions:

The Rotation Problem

When security regulations require quarterly or emergency credential changes, the operations team updates the database engine’s backend password, updates the target Kubernetes Secret object, and must perform a mandatory rolling restart of the application deployments. Because environment variables are bound to the process execution stack at launch time, the containers cannot dynamically fetch the changes, risking service disruption during the rollout window.

The GitOps Paradox

In contemporary workflows, configurations are version-controlled using GitOps engines (like ArgoCD). However, storing raw OpenShift native secret manifests inside Git exposes plaintext credentials to unauthorized auditors or pipeline engineers. Teams must rely on secondary toolsets like Bitnami SealedSecrets. While kubeseal securely encrypts the manifests using asymmetric key pairs before pushing them to public repositories, it introduces an administrative burden—managing cryptographic keys, handling cluster reinstallation decryption failures, and onboarding third-party configurations.

In the next chapter, we will look at A*pproach B: HashiCorp Vault + Vault Agent Injector*, exploring how a specialized identity-driven architecture resolves these operational limitations entirely without rewriting a single line of application code.

Chapter 2: The Modern Alternative — Identity-Driven Secrets with HashiCorp Vault (Approach B)

To overcome the operational limitations of static, platform-bound credentials, modern enterprise architectures transition to an identity-driven security model using HashiCorp Vault.

In this setup, secrets no longer live inside OpenShift’s native etcd storage database, nor are they ever checked into a Git repository. Instead, OpenShift becomes a platform consumer, leveraging its internal cryptographic identities (ServiceAccounts) to request short-lived tokens from a centralized, highly auditable Vault cluster.

Deploying Vault on Red Hat OpenShift

Deploying Vault on OpenShift requires specialized configurations to accommodate strict platform rules, such as Security Context Constraints (SCCs), multi-replica high availability, and native routing. The standard deployment standard utilizes the official Helm chart customized with OpenShift-native parameters.

Vault Cluster Configuration

Below is the reference configuration blueprint using Raft integrated storage (eliminating the operational footprint of external storage backends like Consul):

# vault-values.yaml
global:
  enabled: true
  tlsDisable: false    # Forced TLS for all transit lines
  openshift: true      # Automatically tunes Helm chart to handle OpenShift compliance and SCCs

server:
  ha:
    enabled: true
    replicas: 3
    raft:
      enabled: true
      setNodeId: true
      config: |
        ui = true

        listener "tcp" {
          tls_disable     = 0
          address         = "[::]:8200"
          cluster_address = "[::]:8201"
        }

        storage "raft" {
          path    = "/vault/data"
          node_id = "HOSTNAME"

          # Auto-join configuration for Raft quorum consensus
          retry_join { leader_api_addr = "http://vault-0.vault-internal:8200" }
          retry_join { leader_api_addr = "http://vault-1.vault-internal:8200" }
          retry_join { leader_api_addr = "http://vault-2.vault-internal:8200" }
        }

Bootstrapping Identity-Based Authentication

Once the Vault instance is running on the cluster, we leverage Vault’s native Kubernetes Authentication Method. This allows applications to use their standard OpenShift ServiceAccount token as a secure, temporary proof of identity.

The configuration workflow establishes a secure cryptographic trust loop between Vault and the OpenShift Control Plane API:

Scripted Authentication Setup

The following reference script automates the creation of the auth backend, defines the system boundaries, and maps permissions directly to the target namespace service account:

#!/usr/bin/env bash
# 01-vault-k8s-auth.sh (Excerpt)
set -euo pipefail

echo "=== Configuring Vault Kubernetes Auth for Demo App ==="

# 1. Enable the Kubernetes auth engine
vault auth enable kubernetes || echo "Already enabled"

# 2. Point Vault to the host OpenShift API server for token verification
vault write auth/kubernetes/config \
    kubernetes_host="https://kubernetes.default.svc:443" \
    kubernetes_ca_cert=@/var/run/secrets/kubernetes.io/serviceaccount/ca.crt

# 3. Create a strict role matching the application's runtime platform identity
vault write auth/kubernetes/role/demo-role \
    bound_service_account_names="demo-sa" \
    bound_service_account_namespaces="demo" \
    policies="demo-policy" \
    ttl="1h" \
    max_ttl="24h"

Defining Fine-Grained ACL Rules

To adhere to the principle of least privilege, applications must never be handed a broad root token. Vault uses HashiCorp Configuration Language (HCL) policies to enforce strict path isolation. The demo-app is restricted entirely to its own data engine space:

# 00-vault-policy.hcl
# Allow the demo application read-only access to its dedicated database & infrastructure credentials
path "secret/data/demo" {
  capabilities = ["read"]
}

# Allow reading metadata details (e.g. tracking versions for rotation) without exposing values
path "secret/metadata/demo" {
  capabilities = ["list", "read"]
}

# Allow token renewal and self-inspection capabilities (crucial for automated sidecars)
path "auth/token/renew-self" {
  capabilities = ["update"]
}
path "auth/token/lookup-self" {
  capabilities = ["read"]
}

Injecting Secrets securely via the Key-Value Engine

With policies and authentication boundaries active, parameters can be loaded straight into the Vault Secure KV Store. Because this operates entirely through the API or CLI, configurations are maintained out-of-band — safely isolating them from human eyes and source repositories.

#!/usr/bin/env bash
# 02-vault-secrets.sh (Excerpt)
source .env # Local isolated parameters, ignored by Git version control

echo "=== Committing App Secrets to Vault Secure KV Store ==="

# Write all credentials atomically into Version 2 KV backend
vault kv put secret/demo \
    db_user="${APP_DB_USER}" \
    db_password="${APP_DB_PASSWORD}" \
    db_host="${APP_DB_HOST}" \
    db_port="${APP_DB_PORT}" \
    db_name="${APP_DB_NAME}" \
    ldap_url="${APP_LDAP_URL}"

echo "✅ Master credentials stored. These are completely absent from OpenShift's etcd."

In the next sections, we will explore how the Vault Agent Injector seamlessly introduces these paths directly to the pods as ephemeral files, allowing applications to stay zero-trust compliant without writing complex API clients or modification blocks.

Chapter 3: Implementation — Decoupled Vault Injection (Approach B)

Now that HashiCorp Vault is deployed and trusting OpenShift’s identities, we need to bridge the secrets into our application. The gold standard for this without rewriting application code is the Vault Agent Injector.

The Vault Agent Injector is a Mutating Admission Webhook. It monitors incoming pod specifications for specific annotations. When it spots them, it dynamically alters the Pod deployment manifest by embedding two specific platform sidecars into your application stack:

vault-agent-init (Init Container): Runs before your application container executes. It hits Vault using the pod's ServiceAccount identity, downloads the requested secrets, formats them, and writes them to a shared, high-speed ephemeral memory storage (tmpfs) volume mounted at /vault/secrets/.
vault-agent (Runtime Sidecar Container): Runs alongside your application indefinitely. It monitors Vault for updates and automatically rewrites the temporary workspace file on the fly if a secret value is rotated.

High-Level Architecture Flow

This setup decouples the application from Vault completely. The application code only needs to interact with a standard local filesystem file, remaining entirely unaware of Vault’s underlying REST APIs or authentication loops.

Manifest Configuration: Injecting Secrets via Annotations

To apply this pattern, your OpenShift deployment manifest doesn’t configure any native Kubernetes Secret values. Instead, you define annotations that guide the webhook engine.

# 04-deployment-with-vault.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: demo-app
  namespace: demo
  labels:
    app.kubernetes.io/name: demo-app
    secrets-approach: vault-injector
spec:
  replicas: 2
  selector:
    matchLabels:
      app: demo-app
  template:
    metadata:
      labels:
        app: demo-app
      annotations:
        # 1. Activate the injector webhook engine
        vault.hashicorp.com/agent-inject: "true"

        # 2. Specify the authentication role mapped inside Vault
        vault.hashicorp.com/role: "demo-role"

        # 3. Request secret paths and define the output filenames
        vault.hashicorp.com/agent-inject-secret-db.env: "secret/data/demo"
        vault.hashicorp.com/agent-inject-secret-ldap.env: "secret/data/demo"

        # 4. Define custom Go-Template structures to format the data cleanly for consumption
        vault.hashicorp.com/agent-inject-template-db.env: |
          {{- with secret "secret/data/demo" -}}
          export DB_USER="{{ .Data.data.db_user }}"
          export DB_PASSWORD="{{ .Data.data.db_password }}"
          export DB_HOST="{{ .Data.data.db_host }}"
          export DB_PORT="{{ .Data.data.db_port }}"
          export DB_NAME="{{ .Data.data.db_name }}"
          {{- end -}}

        vault.hashicorp.com/agent-inject-template-ldap.env: |
          {{- with secret "secret/data/demo" -}}
          export LDAP_URL="{{ .Data.data.ldap_url }}"
          {{- end -}}
    spec:
      serviceAccountName: demo-sa
      containers:
        - name: demo-app
          image: image-registry.openshift-image-registry.svc:5000/demo/demo-app:latest
          ports:
            - containerPort: 8080
          envFrom:
            - configMapRef:
                name: demo-config
          # Enforce runtime constraints
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            runAsNonRoot: true
            runAsUser: 1001
            capabilities:
              drop: ["ALL"]

Code Implementation: Reading secrets dynamically from Python

Because the sidecar drops standard configuration shell files into the ephemeral memory directory, reading them becomes a direct file parsing operation.

Below is how the application logic manages both initialization and dynamic monitoring loops to ingest updates without process restarts:

# app.py (Excerpt for Vault Approach B)
import os
import re
from pathlib import Path
from flask import Flask, jsonify

app = Flask(__name__)

# Defaults to the injector standard directory path
VAULT_SECRETS_PATH = os.environ.get("VAULT_SECRETS_PATH", "/vault/secrets")

class VaultConfig:
    SECRET_SOURCE = "Vault Agent Injector (Shared tmpfs Volume)"

    @staticmethod
    def parse_env_file(filename: str):
        """Parses a Go-templated export file into a dictionary format."""
        secrets = {}
        filepath = Path(VAULT_SECRETS_PATH) / filename

        if not filepath.exists():
            return secrets

        # Match pattern: export KEY="VALUE"
        pattern = re.compile(r'^\s*export\s+([A-Z0-9_]+)\s*=\s*"(.*)"\s*$')
        with open(filepath, "r") as f:
            for line in f:
                match = pattern.match(line)
                if match:
                    key, value = match.groups()
                    secrets[key] = value
        return secrets

    # Dynamic property mapping definitions
    @property
    def DB_PASSWORD(self):
        # Always read fresh from the filesystem to fetch hot rotations on the fly
        secrets = self.parse_env_file("db.env")
        return secrets.get("DB_PASSWORD", "not-found")

    @property
    def DB_USER(self):
        secrets = self.parse_env_file("db.env")
        return secrets.get("DB_USER", "not-found")

    @property
    def DB_HOST(self):
        secrets = self.parse_env_file("db.env")
        return secrets.get("DB_HOST", "localhost")

config_manager = VaultConfig()

@app.route("/config")
def config():
    return jsonify({
        "approach": "B — Vault Token Integration",
        "secret_source": VaultConfig.SECRET_SOURCE,
        "loaded_at": "On-demand filesystem reads (Hot-swappable dynamically)",
        "rotation_note": "Zero downtime or rollouts required during backend password shifts."
    })

@app.route("/db-status")
def db_status():
    return jsonify({
        "db_host": config_manager.DB_HOST,
        "db_user": config_manager.DB_USER,
        # Display masked password to prove it loaded successfully
        "db_password_used": f"{config_manager.DB_PASSWORD[:2]}******" if len(config_manager.DB_PASSWORD) > 2 else "short"
    })

The Lifecycle Revolution: Direct Structural Comparisons

| Security Lifecycle           | Approach A: Native OpenShift Secrets                         | Approach B: Centralized HashiCorp Vault                      |
| ---------------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ |
| **Storage Engine Location**  | Standard plain `etcd` Key-Value database.                    | Highly decoupled, encrypted Raft storage layer.              |
| **GitOps/CI-CD Handling**    | Requires third-party tools like `SealedSecrets` or encryption wrappers. | Manifests contain nothing but metadata annotations. Secure parameters are injected entirely out-of-band. |
| **Credential Rotation Loop** | Requires updating objects and performing a rolling deployment update. | The runtime sidecar modifies the local shared volume. Your application picks it up instantly without restarts. |
| **Audit Capabilities**       | Shared cluster-wide. Difficult to isolate individual container tracking parameters cleanly. | Fine-grained, immutable tracking policies. Records every read, renewal, and authentication event instantly. |

By implementing identity-driven secrets through HashiCorp Vault and utilizing the automated sidecar injector on Red Hat OpenShift, engineering teams achieve true zero-trust execution environments. You eliminate the leakage vectors inherent to version control repositories, decouple configurations from infrastructure databases, and enable seamless credential rotation — all without polluting your software assets with complex third-party SDK dependencies.

Conclusion: Elevating OpenShift Native Defense with Zero-Trust Automation

Red Hat OpenShift’s built-in secrets management sets an impressive enterprise baseline. By automating etcd encryption at rest out of the box and enforcing rigid boundaries through multi-tenant Projects and Security Context Constraints, it fixes the glaring security gaps found in vanilla Kubernetes.

However, enterprise governance demands more than just hardened storage; it requires dynamic agility. Transitioning to HashiCorp Vault elevates your security posture from a robust static platform to a proactive, identity-driven ecosystem. Even though OpenShift natively keeps data highly secure on disk, Vault solves the operational overhead of the application lifecycle. By eliminating the GitOps paradox of managing encrypted manifests in repositories, introducing granular, centralized audit logs for strict compliance tracking, and streaming live credential rotations to application containers without triggering disruptive rolling updates, Vault turns a complex security checklist into seamless, zero-trust automation.

| Feature / Security Lifecycle  | ☸️ Standard Kubernetes                                        | 🔴 Standard OpenShift                                         | 🔒 OpenShift + HashiCorp Vault                                |
| ----------------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ | ------------------------------------------------------------ |
| **Storage Engine & Security** | Stored in `etcd`. **Not encrypted at rest by default** (requires manual KMS provider configuration). base64 encoding only. | Stored in `etcd`. **Encrypted at rest by default** out of the box using AES-256-GCM cryptography. | Stored in highly decoupled, dedicated **Vault Raft storage**. Secrets never land in the cluster’s `etcd` layer. |
| **GitOps & Version Control**  | Plain YAML manifests contain secrets in clear text (base64). Risks repo leakage without 3rd party tools (e.g., `SealedSecrets`). | Same GitOps limitation as Kubernetes; manifests require client-side encryption before committing to git. | **Zero-secret manifests.** Deployment manifests contain only metadata annotations; actual values are handled out-of-band. |
| **Runtime Pod Delivery**      | Injected directly into the container as environment variables or mounted files via standard Kubelet API. | Injected similarly via environment variables or volume mounts, strictly governed by platform **Security Context Constraints (SCC)**. | Injected transparently into a shared, high-speed ephemeral memory storage (`tmpfs` volume) via a **Mutating Webhook Sidecar**. |
| **Credential Rotation Loop**  | Manual or custom scripting. Requires a rolling update deployment to restart pods and pick up new values. | Native rotation still requires object modification and a rolling deployment restart (`oc rollout restart`) to flush stale values. | **Dynamic & Hot-Swappable.** The `vault-agent` sidecar live-streams new values into the pod filesystem without restarting the container. |
| **Audit Capabilities**        | Basic cluster auditing. Difficult to trace exact API read requests to specific container parameters. | Strong platform-wide auditing via standard OpenShift cluster logging and cluster-level audit trails. | **Granular, cryptographic audit logs.** Tracks every authentication, lookup, token renewal, and precise key access. |
| **Least Privilege Isolation** | Scoped broadly by Namespace RBAC boundaries.                 | Scoped tightly by multi-tenant Projects and hardened Security Context Constraints (SCCs). | Scoped mathematically by fine-grained **Vault HCL Policies** mapped directly to individual `ServiceAccount` identities. |

Thanks for reading 🔏

Empowering the Team: How I Built a Lightweight Web UI for Bob (With Bob’s Help)

Alain Airom (Ayrom) — Fri, 17 Jul 2026 09:24:26 +0000

A lite UI to use Bob shell!

Introduction

As developers and tech professionals, we are incredibly spoiled by the modern Software Development Life Cycle (SDLC). Tools like IBM Bob have fundamentally changed how we work. Whether we are deep in our IDEs or navigating the terminal, having Bob as an AI co-pilot makes writing code, debugging, and system architecture feel like cheating.

But a funny thing happened recently.

My colleagues on the sales and business teams started noticing what Bob could do. They saw the power of automated file analysis, seamless document extraction, and instant business planning. Naturally, they wanted in.

The catch? I couldn’t exactly hand them a command-line interface or expect them to navigate an integrated development environment. Business planners and sales executives shouldn’t have to use a Bob IDE or Bob Shell just to run a document extraction query. They needed a simple, frictionless gateway.

So, I decided to build one: a lightweight, clean web application designed solely to expose Bob’s “Plan” and “Ask” features directly in a standard browser. And the best part? I had Bob build the entire application for me. Here is how the application works and how anyone on the team can use it.

The Web App at a Glance

The web implementation of the AskBob-LightSimulator is engineered specifically to eliminate the friction of technical interfaces (like IDE integrations or Terminal-based shells) for non-technical users. It accomplishes this by serving a responsive, client-side browser interface powered by a lightweight Node.js background router.

The web interface is designed to be completely distraction-free. No complex configurations, no terminal commands. Just two dedicated tabs representing Bob’s core business-facing powers:

Ask: A direct line to Bob’s cognitive engine for rapid file analysis, Q&A, and document breakdown.
Plan: A structured environment to generate business planners, step-by-step strategy maps, and project outlines.

Web Architecture Overview

Key Design Pillars of the Web Architecture

Decoupled Process Execution: Instead of packaging the application into a heavy, conflict-prone Electron wrapper (which can suffer from shell environment variable overrides like ELECTRON_RUN_AS_NODE), the system runs as a clean, local Node.js web server.
Real-time Server-Sent Events (SSE): To avoid the lag of waiting for a full CLI execution to complete, the server streams standard output directly back to the browser in real time.
Adaptive Workspace Targeting: The web interface permits non-technical stakeholders to swap directory contexts dynamically, altering where Plan-mode outputs are written.

Server-Side Route Implementation & Stream Spawning

The central brain of the web engine is server.js. It parses requests, constructs a sanitized shell environment to execute the bob CLI process, and continuously flushes stdout buffers as SSE chunks.

Code Excerpt: Dynamic Process Spawning & SSE Stream (server.js): Here is how the server handles the incoming chat request, isolates the process from Electron-related environmental pollution, and streams back chunk-by-chunk payload updates.

// From server.js
const { spawn } = require('child_process');

function handleChat(req, res) {
  // Set headers for Server-Sent Events (SSE) streaming
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
  });

  let body = '';
  req.on('data', chunk => { body += chunk; });
  req.on('end', () => {
    const { message, mode, workspace } = JSON.parse(body);

    // 1. Sanitize & build a clean shell environment
    const cleanEnv = { ...process.env };
    // Electron-related variables can cause the 'bob' node execution to crash
    delete cleanEnv.ELECTRON_RUN_AS_NODE; 

    // 2. Select CLI flags depending on the desired operation mode
    const modeFlag = mode === 'plan' ? '--chat-mode=code' : '--chat-mode=ask';
    const workspaceDir = workspace || process.env.BOB_WORKSPACE_DIR || __dirname;

    // 3. Spawn Bob as an active subprocess
    const bobProcess = spawn(
      BOB_CLI_PATH,
      [modeFlag, '-p', message],
      {
        cwd: workspaceDir,
        env: cleanEnv,
        shell: '/bin/zsh' // Target the system default shell safely
      }
    );

    // 4. Listen to standard output stream & emit events to the browser client
    bobProcess.stdout.on('data', (data) => {
      const textChunk = data.toString();
      // Render Markdown formatting & code blocks on-the-fly before shipping
      const htmlChunk = renderMarkdown(textChunk); 

      res.write(`data: ${JSON.stringify({ chunk: htmlChunk })}\n\n`);
    });

    bobProcess.on('close', (code) => {
      res.write(`data: [DONE]\n\n`);
      res.end();
    });
  });
}

Client-Side Interface & SSE Parser

On the browser side, renderer/app.js issues a standard fetch call but reads the response body progressively using a ReadableStream reader. This keeps the rendering snappy and ensures that standard text outputs and markdown code blocks are styled immediately as they stream in.

Code Excerpt: Client-Side Chunk Listener (renderer/app.js)

async function sendMessageToBob(promptText, activeMode) {
  const response = await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      message: promptText,
      mode: activeMode,
      workspace: currentWorkspacePath
    })
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let accumulatedMarkdown = '';

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;

    const chunkStr = decoder.decode(value);
    const lines = chunkStr.split('\n');

    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const payload = line.slice(6).trim();
        if (payload === '[DONE]') {
          finalizeChatUI();
          break;
        }

        try {
          const { chunk } = JSON.parse(payload);
          accumulatedMarkdown += chunk;
          updateChatBubble(accumulatedMarkdown);
        } catch (e) {
          // Keep parsing line by line in case of fragments
        }
      }
    }
  }
}

The Prompt Assistant: Simplifying File Access

To provide a contextual awareness and simplify file access and other precisions, commands and options helper is provided from the UI;

Relative paths (./workspace/input/brief.docx)
Absolute paths (/Users/admin/documents/...)
System mount points

How the File Access Helper Works

When users click the “📁 Reference File” helper button or use one of our pre-built Suggestion Chips, the UI automatically parses their selected workspace and constructs the perfect prompt structure for Bob.

Instead of requiring users to manually type out complex path strings, they are presented with a visual file picker or context-aware template chips that instantly inject the exact syntax Bob needs to resolve the file location.

UI Implementation: Suggestion Chips & Prompt Templates

We implemented a series of interactive, click-to-fill chips directly below the main prompt input. This gives users immediate, visual examples of how to query files correctly.

<div class="prompt-helper-container">
  <span class="helper-label">💡 Need help? Click a template to start:</span>
  <div class="suggestion-chips">
    <button class="chip" onclick="applyTemplate('ask', 'Analyze the file [file_name] and summarize the key action points.')">
      📄 Analyze a Document
    </button>
    <button class="chip" onclick="applyTemplate('plan', 'Read the project specs in [file_name] and build a skeleton structure in output/')">
      🏗️ Build Workspace Files
    </button>
    <button class="chip" onclick="applyTemplate('ask', 'Compare the data in [file_1] with [file_2] and highlight discrepancies.')">
      📊 Compare Data Sheets
    </button>
  </div>
</div>

Client-Side Parsing Engine

When a user selects a template, the client-side controller seamlessly swaps out generic placeholders ([file_name]) with actual file selections from the active workspace context.

Code Excerpt: Template Injection Engine (renderer/app.js)

/**
 * Injects quick-start templates with active workspace variables
 */
function applyTemplate(mode, templateText) {
  // Switch to the correct mode tab automatically (Ask or Plan)
  switchTab(mode);

  const inputField = document.getElementById('user-prompt-input');

  // If a workspace is active, we can pre-populate a realistic example path
  const targetWorkspace = currentWorkspacePath ? 'input/your-document.pdf' : 'document.pdf';

  // Cleanly replace path placeholders with easy-to-read instructions
  let finalizedPrompt = templateText
    .replace('[file_name]', targetWorkspace)
    .replace('[file_1]', 'input/sheet_A.csv')
    .replace('[file_2]', 'input/sheet_B.csv');

  inputField.value = finalizedPrompt;
  inputField.focus();

  // Visually highlight the parameter they need to customize
  const placeholderStart = finalizedPrompt.indexOf('input/');
  if (placeholderStart !== -1) {
    inputField.setSelectionRange(placeholderStart, finalizedPrompt.length);
  }
}

💡 Why this Matters
This small UX enhancement bridges the cognitive gap between business intent and CLI execution. A sales director doesn’t need to know that Bob is executing a targeted read command under the hood; they simply select “Analyze a Document”, replace the placeholder with their file’s name, and watch Bob stream the parsed results to their browser.

Last but not least, the commands helper is provided on the UI as well!

iOS Implementation (Work in Progress)

To take Bob’s planning capabilities on-the-go for sales leaders, a companion native iOS client is actively being developed under ios/AskBobIOS. It interfaces directly with the local Node.js backend using a SwiftUI native view and lightweight network handlers.

Mobile-Backend Integration Architecture

Key iOS Implementation Elements

SwiftUI Presentation Layer: Centered around a classic messenger conversational interface, allowing easy switching between the Ask and Plan execution schemes.
Flexible Target Endpoints: Because local simulators run on isolated host networks, the client implements a Server Connection Sheet (managed via ServerSheet.swift) that easily binds either to http://localhost:3747 (for same-machine simulators) or the local LAN IP address (for testing on real iPhones over Wi-Fi).

Core SwiftUI Network Client & View Model

The state of the iOS application, the active Server URL settings, and the streaming connections are managed by AppModel.swift. Below are the key architectural structures driving this mobile-to-web bridge.

Swift Code Excerpt: App Model State and Backend Configuration (AppModel.swift):

import Foundation
import Combine

class AppModel: ObservableObject {
    @Published var serverURL: String = "http://localhost:3747"
    @Published var isConnected: Bool = false
    @Published var messages: [ChatMessage] = []
    @Published var isThinking: Bool = false

    private var cancellables = Set<AnyCancellable>()

    /// Test the node web-server connection status
    func verifyConnection() {
        guard let url = URL(string: "\(serverURL)/api/ping") else { return }

        URLSession.shared.dataTaskPublisher(for: url)
            .map { response in
                // Expecting a successful HTTP status code from server.js
                guard let httpResponse = response.response as? HTTPURLResponse,
                      httpResponse.statusCode == 200 else { return false }
                return true
            }
            .replaceError(with: false)
            .receive(on: DispatchQueue.main)
            .assign(to: \.isConnected, on: self)
            .store(in: &cancellables)
    }

    /// Apply updated server URL configurations requested from the ServerSheet View
    func applyServerURL(_ newURL: String) {
        let sanitizedURL = newURL.trimmingCharacters(in: .whitespacesAndNewlines)
        self.serverURL = sanitizedURL
        verifyConnection()
    }
}

Streaming Chat Engine (Swift Implementation)

To stream characters into the SwiftUI interface, the mobile app connects to the /api/chat route. It translates HTTP chunk buffers into readable text, updating the active bubble state in real time.

Swift Code Excerpt: Stream Listener (ChatViewModel.swift)

import Foundation

class ChatViewModel: ObservableObject {
    @Published var activeChatBubbleText: String = ""

    func sendStreamingRequest(prompt: String, mode: String, serverURL: String) async {
        guard let url = URL(string: "\(serverURL)/api/chat") else { return }

        var request = URLRequest(url: url)
        request.httpMethod = "POST"
        request.setValue("application/json", forHTTPHeaderField: "Content-Type")

        let payload: [String: String] = [
            "message": prompt,
            "mode": mode
        ]

        request.httpBody = try? JSONSerialization.data(withJSONObject: payload)

        do {
            let (bytes, response) = try await URLSession.shared.bytes(for: request)
            guard let httpResponse = response as? HTTPURLResponse, httpResponse.statusCode == 200 else {
                return
            }

            // Read incoming line events progressively
            for try await line in bytes.lines {
                if line.hasPrefix("data: ") {
                    let jsonString = String(line.dropFirst(6))
                    if jsonString.contains("[DONE]") {
                        break
                    }

                    if let data = jsonString.data(using: .utf8),
                       let parsed = try? JSONDecoder().decode(ChatChunk.self, from: data) {
                        DispatchQueue.main.async {
                            self.activeChatBubbleText += parsed.chunk
                        }
                    }
                }
            }
        } catch {
            print("Streaming connection failed: \(error)")
        }
    }
}

struct ChatChunk: Codable {
    let chunk: String
}

Conclusion: Democratizing AI, One Interface at a Time

By wrapping the raw power of the IBM Bob SDLC into a frictionless, lightweight environment, we have successfully bridged the gap between engineering-grade capabilities and everyday business workflows. What started as an effort to save our sales and planning colleagues from terminal screens and IDE setups has materialized into a highly functional ecosystem. Crucially, the core philosophy behind this project is not to bypass, rewrite, or overwrite Bob’s robust native functionalities; rather, it is to provide a lightweight, accessible companion that acts as a helper to open up these powerful tools to a much broader public. Through this collaborative approach, we have delivered a responsive, real-time web portal powered by Server-Sent Events, integrated a dynamic Prompt Assistant to handle workspace pathing automatically, and laid the groundwork for an on-the-go SwiftUI mobile client. The ultimate irony remains our greatest success — using Bob himself to build the very tools that liberate others from the complexities of software development. As we finalize the iOS implementation, we aren’t just deploying an application; we are empowering our entire organization to extract insights, draft strategic plans, and build business documents at the speed of thought, proving that the best technology is the kind that belongs to everyone.

Thanks for reading 👍

Building a Secure, Local Outlook Mail Extractor with Electron, PKCE, and Microsoft Graph

Alain Airom (Ayrom) — Sun, 12 Jul 2026 09:33:06 +0000

Extracting your email(s) from MS Outlook client

Introduction

Extracting corporate data securely often comes with a major hurdle: credential management. When building tools to parse, back up, or process emails, handling user credentials safely is paramount.

In this post, we’ll break down how to build a desktop application using Electron, TypeScript, and the Microsoft Graph API that allows users to authenticate securely via OAuth 2.0 Authorization Code Flow with PKCE, discover their Outlook folder hierarchy, and safely extract message payloads locally without exposing sensitive access tokens to an external server.

⚠️ Disclaimer & Didactic Scope
Important Note: This application is intended solely for personal use and should be treated as a didactical demonstration. It is explicitly not designed or vetted for accessing corporate production email servers. In an enterprise environment, rigorous security controls — including robust credential encryption at rest, token validation lifecycles, advanced error boundary auditing, and corporate compliance guardrails — must be strictly enforced. Because this project prioritizes architectural clarity over enterprise hardening, it should not be used as a production-ready corporate solution.

The Architecture: Why PKCE in a Desktop App?

Standard OAuth 2.0 authorization code flows traditionally rely on a client_secret hosted on a secure backend server. Desktop applications are considered "public clients"—they run entirely on the user's machine, meaning a client secret cannot be safely hidden within the source code.

To overcome this, we implement Proof Key for Code Exchange (PKCE). Instead of a static secret, the application dynamically generates a cryptographic random string called a code_verifier, hashes it to create a code_challenge, and sends the challenge to the Microsoft Identity Platform. When exchanging the authorization code for tokens, the app sends the original verifier, proving it was the initiator of the login request.

The Lifecycle of the Extraction Flow

Here is exactly how the Electron Main process coordinates with the Renderer UI, a temporary local loopback server, and Microsoft’s APIs to pull the data:

Triggering Auth: The UI (Renderer) sends a message via IPC (IPC "connect") to the Main process.
Spinning Up a Local Server: The Main process boots a temporary, short-lived HTTP loopback server (localhost:8765) to await the authentication callback.
User Login: The app opens the system’s default browser targeting Microsoft’s /authorize endpoint, passing along the PKCE code challenge.
Token Exchange: Once the user consents, Microsoft redirects to localhost:8765 with an authorization code. The local server captures it, shuts down immediately, and the Main process POSTs this code alongside the plain text code_verifier to get the access_tokenand refresh_token.
Data Extraction: The tokens are securely cached locally, enabling recursive calls to Microsoft Graph (/me/mailFolders and /me/mailFolders/{id}/messages) to map folders and stream paginated emails down to disk.

Core Implementation Deep Dive

Let’s look at the actual mechanics making this happen under the hood, starting with how the UI triggers the action and how the Main process orchestrates the secure loopback.

Inter-Process Communication (IPC) Bridge

Electron keeps the UI isolated from the operating system for security. To talk to the filesystem or network libraries safely, we use a preload.ts script to expose a sanitized context bridge:

import { contextBridge, ipcRenderer } from 'electron';

contextBridge.exposeInMainWorld('electronAPI', {
  connect: () => ipcRenderer.send('connect'),
  listFolders: () => ipcRenderer.send('list-folders'),
  startExtraction: (folderIds: string[], exportParams: any) => 
    ipcRenderer.send('start-extraction', { folderIds, exportParams }),
  onProgress: (callback: (status: string) => void) =>
    ipcRenderer.on('progress', (_event, status) => callback(status)),
  onFoldersReturned: (callback: (folders: any[]) => void) =>
    ipcRenderer.on('return-folders', (_event, folders) => callback(folders)),
  onDone: (callback: (result: any) => void) =>
    ipcRenderer.on('done', (_event, result) => callback(result))
});

The Main Process: Managing the PKCE Challenge & Local Loopback

In main.ts, we handle the complex orchestration. When connect is invoked, we spin up a micro HTTP server via Node's native http module to intercept the incoming redirect from the browser.

Here is the structural framework of the token acquisition process:

import { app, BrowserWindow, ipcMain, shell } from 'electron';
import * as http from 'http';
import * as crypto from 'crypto';
import axios from 'axios';

let accessToken: string | null = null;
let codeVerifier: string = '';

// Helper to generate PKCE Code Verifier and Challenge
function generatePKCE() {
  codeVerifier = crypto.randomBytes(32).toString('base64url');
  const codeChallenge = crypto
    .createHash('sha256')
    .update(codeVerifier)
    .digest('base64url');
  return codeChallenge;
}

ipcMain.on('connect', async (event) => {
  const challenge = generatePKCE();

  // 1. Start temporary loopback server
  const server = http.createServer((req, res) => {
    const urlParams = new URL(req.url!, `http://${req.headers.host}`);
    const authCode = urlParams.searchParams.get('code');

    if (authCode) {
      res.writeHead(200, { 'Content-Type': 'text/html' });
      res.end('<h1>Authentication successful! You can close this window.</h1>');

      // Close the server immediately after capturing the code
      server.close();

      // 2. Exchange Code + Verifier for Tokens
      exchangeCodeForToken(authCode, event);
    } else {
      res.writeHead(400).end('Auth code missing');
    }
  });

  server.listen(8765, () => {
    // 3. Open system browser for explicit user authentication
    const authUrl = `https://login.microsoftonline.com/common/oauth2/v2.0/authorize?` +
      `client_id=YOUR_CLIENT_ID` +
      `&response_type=code` +
      `&redirect_uri=http://localhost:8765` +
      `&scope=Mail.Read offline_access` +
      `&code_challenge=${challenge}` +
      `&code_challenge_method=S256`;

    shell.openExternal(authUrl);
  });
});

async function exchangeCodeForToken(code: string, event: Electron.IpcMainEvent) {
  try {
    const response = await axios.post(
      'https://login.microsoftonline.com/common/oauth2/v2.0/token',
      new URLSearchParams({
        client_id: 'YOUR_CLIENT_ID',
        grant_type: 'authorization_code',
        code: code,
        redirect_uri: 'http://localhost:8765',
        code_verifier: codeVerifier, // Proves identity without client_secret
      }).toString(),
      { headers: { 'Content-Type': 'application/x-www-form-urlencoded' } }
    );

    accessToken = response.data.access_token;
    // Cache tokens locally if persistence across sessions is desired

    event.sender.send('progress', 'authenticated');
  } catch (error) {
    console.error('Token exchange failed', error);
  }
}

Fetching and Extracting Hierarchical Data

Once accessToken is set, the application uses it to target the Microsoft Graph API endpoints asynchronously. Because mailboxes can grow infinitely large, the extraction handles recursive folder mapping and cursor-based pagination using the @odata.nextLink property returned by Microsoft.

ipcMain.on('list-folders', async (event) => {
  try {
    // Graph call to list recursive mail folders
    const response = await axios.get('https://graph.microsoft.com/v1.0/me/mailFolders?$top=100', {
      headers: { Authorization: `Bearer ${accessToken}` }
    });

    event.sender.send('return-folders', response.data.value);
  } catch (error) {
    console.error('Failed to fetch mail folders', error);
  }
});

ipcMain.on('start-extraction', async (event, { folderIds, exportParams }) => {
  for (const id of folderIds) {
    let url = `https://graph.microsoft.com/v1.0/me/mailFolders/${id}/messages?$top=50`;

    while (url) {
      const res = await axios.get(url, { headers: { Authorization: `Bearer ${accessToken}` } });
      const messages = res.data.value;

      // Filter, transform, and stream raw email content to a local file path
      // e.g., outputting to JSON, CSV, or raw .eml strings

      url = res.data['@odata.nextLink']; // Keep pulling until pagination ends
    }
  }
  event.sender.send('done', { count: 150, format: 'JSON' });
});

Conclusion

By marrying Electron’s isolated execution architecture with OAuth 2.0 + PKCE, you eliminate the need to maintain an intermediate backend service just to hold secrets or route tokens. The credentials live where the data lives: completely on the client machine. This keeps the integration footprint small, highly performant, and perfectly aligned with privacy best practices.

Thanks for reading 😊

Elevating Document Parsing to Warp Speed: Introducing Docling-Rust and a Native Desktop GUI

Alain Airom (Ayrom) — Sat, 11 Jul 2026 13:01:53 +0000

Testing the brand new implementation of Docling in Rust

Introduction

In the enterprise Artificial Intelligence and Retrieval-Augmented Generation (RAG) landscape, the phrase “garbage in, garbage out” remains an absolute truth. Before an LLM can reason over corporate knowledge, documents must be ingested, cleaned, structural elements like tables and charts recognized, and the output exported into structured Markdown or JSON.

For a long time, Python has been the undisputed king of this pipeline, dominated by libraries like IBM’s powerful layout-parsing and open-sources engine, Docling. However, as operational scale increases, the deployment complexity, massive memory footprints, and raw latency of Python runtime environments present significant engineering bottlenecks.

Enter Docling-Rust (docling.rs), the official, high-performance Rust port of IBM's/ Linux Foundation’s open-source Docling document-conversion pipeline. In this post, Bob and I explore why Rust is changing the document processing landscape, dive deep into the newly unveiled docling.rs ecosystem, and unpack a production-ready, multi-threaded native desktop GUI application built entirely in Rust.

Introducing Docling-Rust (`docling.rs`)

Docling-Rust is the native port designed to address these deployment challenges directly. Maintained under the Docling project organization (github.com/docling-project/docling.rs), it brings the full multi-format parsing capabilities of IBM’s core document layout analysis algorithms into the Rust ecosystem.

With docling.rs, developers can process an incredible variety of enterprise formats out of the box, with absolutely no Python runtime required:

Document TypeSupported ExtensionsFixed Layout.pdfWord Processors.docx, .doc, .odt Presentations.pptx, .ppt Spreadsheets.xlsx, .xls, .csv Web & Structures .html, .htm, .mhtml, .mht, .md E-Books & Mail.epub, .eml Interchange .json (Docling Core Schema 1.10.0)

The engine extracts structural layout elements — hierarchical headings, ordered/unordered lists, complex mathematical equations, and nested data tables — and marshals them into a strongly typed DoclingDocument object model, ready to be serialized to strict spec-conformant Markdown or schema-validated JSON.

Deep Dive: Building a Multi-Threaded Desktop GUI App

To illustrate the capabilities of docling.rs, let’s analyze a production-grade, multi-threaded desktop GUI utility: Docling GUI. Built using eframe and egui (the leading immediate-mode Rust GUI framework), this application provides a sleek, drag-and-drop desktop workflow for batch document conversion.

Core Architectural Blueprint

The application follows an asynchronous worker architecture designed to keep the GUI thread rendering fluidly at 60 FPS, even when processing heavy, multi-page PDFs or multi-tab Excel files.

Technical Code Breakdown

Bootstrap and Environment Provisioning: Document parsers frequently require heavy underlying execution engines (such as ONNX Runtime weights for layout modeling and vision tasks, or dynamic libraries like libpdfium for PDF rasterization). In our Rust implementation, these dependencies are safely bootstrapped at application initialization inside src/app.rs:

/// Called once at startup. If `PDFIUM_DYNAMIC_LIB_PATH` is not already set,
/// look for `.pdfium/lib` relative to the current working directory and set
/// the required env vars so that `docling-pdf` can find them.
fn bootstrap_pdf_env() {
    let cwd = env::current_dir().unwrap_or_else(|_| PathBuf::from("."));

    // Provision pdfium dynamic library paths
    if env::var_os("PDFIUM_DYNAMIC_LIB_PATH").is_none() {
        let pdfium_lib_dir = cwd.join(".pdfium").join("lib");
        if pdfium_lib_dir.exists() {
            env::set_var("PDFIUM_DYNAMIC_LIB_PATH", &pdfium_lib_dir);
            log::info!("Set PDFIUM_DYNAMIC_LIB_PATH = {}", pdfium_lib_dir.display());
        }
    }

    let models_dir = cwd.join("models");

    macro_rules! set_if_missing {
        ($var:expr, $path:expr) => {
            if env::var_os($var).is_none() {
                let p = $path;
                if p.exists() {
                    env::set_var($var, &p);
                    log::info!("Set {} = {}", $var, p.display());
                }
            }
        };
    }

    // Map structural ONNX runtime models
    set_if_missing!("DOCLING_LAYOUT_ONNX",          models_dir.join("layout_heron.onnx"));
    set_if_missing!("DOCLING_OCR_REC_ONNX",          models_dir.join("ocr_rec.onnx"));
    set_if_missing!("DOCLING_OCR_DICT",               models_dir.join("ppocr_keys_v1.txt"));
    set_if_missing!("DOCLING_TABLEFORMER_ENCODER",    models_dir.join("tableformer").join("encoder.onnx"));
    set_if_missing!("DOCLING_TABLEFORMER_DECODER",    models_dir.join("tableformer").join("decoder.onnx"));
    set_if_missing!("DOCLING_TABLEFORMER_BBOX",       models_dir.join("tableformer").join("bbox.onnx"));
}

Fearless Background Conversion Worker: To ensure the immediate-mode rendering frame rate doesn’t stutter, the conversion processing is sent to a dedicated OS thread. Communication between the main GUI loop and the worker thread utilizes a thread-safe bounded crossbeam-channel. The snippet which follows, shows how the worker consumes the selected documents, configures the DocumentConverter, parses the data, writes the output files to disk, and sends progress updates back to the UI:

fn start_conversion(&mut self, ctx: egui::Context) {
    if self.picked_files.is_empty() { return; }

    // Clear output logs and switch UI view to Log tab
    self.reset_outputs_and_set_tab(ActiveTab::Log);

    // Instantiate communication channel: Worker -> UI Thread
    let (msg_tx, msg_rx) = crossbeam_channel::bounded::<WorkerMsg>(64);
    self.result_rx = Some(msg_rx);

    // Clone worker configurations
    let files = self.picked_files.clone();
    let out_dir = self.settings.output_dir.clone();
    let strict = self.settings.strict_mode;
    let export_md = self.settings.export_markdown;
    let export_json = self.settings.export_json;
    let job_state = Arc::clone(&self.job_state);

    *job_state.lock().unwrap() = JobState::Running;

    // Spawn native OS background thread
    thread::spawn(move || {
        if let Err(e) = std::fs::create_dir_all(&out_dir) {
            let _ = msg_tx.send(WorkerMsg::Error(format!("Cannot create output directory: {e}")));
            return;
        }

        let timestamp = Local::now().format("%Y%m%d_%H%M%S").to_string();
        let (mut combined_md, mut combined_json) = (String::new(), String::new());
        let (mut last_md_path, mut last_json_path) = (None, None);

        for file_path in &files {
            let file_name = file_path.file_name().map(|s| s.to_string_lossy().to_string()).unwrap_or_default();
            let _ = msg_tx.send(WorkerMsg::Progress(format!("Converting: {}", file_name)));

            // 1. Marshall the file into a Docling SourceDocument
            let source = match docling::SourceDocument::from_file(file_path) {
                Ok(s) => s,
                Err(e) => {
                    let _ = msg_tx.send(WorkerMsg::Progress(format!("  ✗ Failed to load {file_name}: {e}")));
                    continue;
                }
            };

            // 2. Configure converter settings (e.g., Strict Markdown Mode toggle)
            let converter = docling::DocumentConverter::new().strict(strict);

            // 3. Execute the core compilation/parsing engine
            let result = match converter.convert(source) {
                Ok(r) => r,
                Err(e) => {
                    let _ = msg_tx.send(WorkerMsg::Progress(format!("  ✗ Conversion error for {file_name}: {e}")));
                    continue;
                }
            };

            let stem = file_path.file_stem().map(|s| s.to_string_lossy().to_string()).unwrap_or_else(|| "document".into());

            // 4. Export Markdown AST structures
            if export_md {
                let md = result.document.export_to_markdown();
                combined_md.push_str(&format!("# {file_name}\n\n{md}\n\n---\n\n"));
                let md_path = out_dir.join(format!("{stem}_{timestamp}.md"));
                if std::fs::write(&md_path, &md).is_ok() {
                    let _ = msg_tx.send(WorkerMsg::Progress(format!("  ✓ Markdown saved: {}", md_path.display())));
                    last_md_path = Some(md_path);
                }
            }

            // 5. Export JSON Document Core Schema structures
            if export_json {
                let json = result.document.export_to_json();
                combined_json.push_str(&format!("{json}\n"));
                let json_path = out_dir.join(format!("{stem}_{timestamp}.json"));
                if std::fs::write(&json_path, &json).is_ok() {
                    let _ = msg_tx.send(WorkerMsg::Progress(format!("  ✓ JSON saved: {}", json_path.display())));
                    last_json_path = Some(json_path);
                }
            }
        }

        // Notify the UI that processing is complete
        let _ = msg_tx.send(WorkerMsg::Done {
            markdown: combined_md,
            json: combined_json,
            output_md_path: last_md_path,
            output_json_path: last_json_path,
        });

        // Request an immediate frame repaint on the main loop
        ctx.request_repaint();
    });
}

Polling and Non-blocking Main UI Integration: Every single frame, the immediate-mode UI calls poll_worker() inside its update loop. Because it utilizes try_recv(), the operation is completely non-blocking, checking for new messages and instantly updating internal state wrappers for rendering:

fn poll_worker(&mut self) {
    let rx = match &self.result_rx {
        Some(r) => r.clone(),
        None => return,
    };

    // Non-blocking drain of the worker channel
    while let Ok(msg) = rx.try_recv() {
        match msg {
            WorkerMsg::Progress(text) => {
                self.progress_msg = text.clone();
                self.log_lines.push(text);
            }
            WorkerMsg::Done { markdown, json, output_md_path, output_json_path } => {
                self.markdown_output = markdown;
                self.json_output = json;
                self.output_md_path = output_md_path;
                self.output_json_path = output_json_path;
                self.log_lines.push("✅ Conversion complete.".into());
                self.progress_msg.clear();
                *self.job_state.lock().unwrap() = JobState::Done;
                self.result_rx = None;

                // Automatically route focus to the optimal layout tab
                if !self.markdown_output.is_empty() {
                    self.active_tab = ActiveTab::Markdown;
                } else if !self.json_output.is_empty() {
                    self.active_tab = ActiveTab::Json;
                }
            }
            WorkerMsg::Error(err) => {
                self.log_lines.push(format!("❌ Error: {err}"));
                self.progress_msg = format!("Error: {err}");
                *self.job_state.lock().unwrap() = JobState::Failed(err);
                self.result_rx = None;
            }
        }
    }
}

Conclusion

The official launch of Docling-Rust represents a massive leap forward for production engineering teams building RAG pipelines, data ingestion workflows, or desktop tools. By replacing heavy Python environments with a lean, blazing-fast, compiled Rust crate, you can maximize multi-core hardware density, eliminate deployment friction, and guarantee predictability at scale.

Thanks for reading 🤗

From REST to ReAct: Transforming Legacy APIs into MCP Tools for AI Agents

Alain Airom (Ayrom) — Fri, 10 Jul 2026 09:46:15 +0000

A didactical guide to transform API calls into MCP Tools for agentic era!

Introduction

For more than two decades, Representational State Transfer (REST) APIs have served as the undisputed backbone of application interoperability. Designed for explicit execution, REST endpoints expect strict compliance, deterministic input types, and structured query definitions.

However, as Large Language Models (LLMs) rapidly pivot into autonomous AI Agents, traditional APIs become an architectural bottleneck. Rather than relying on hardcoded programmatic logic, agents rely on semantic discovery and reasoning loops. To bridge this structural divide, Anthropic open-sourced the Model Context Protocol (MCP) — a uniform standard that reshapes how AI interfaces with legacy data environments.

Why AI Agents Prefer MCP Over Traditional API Calls

When a traditional software client wants to read or write data, it calls an explicit endpoint with pre-compiled query variables. When an AI Agent tries to do the same over standard REST, it faces three main challenges:

Context Window Bloat: To understand an API, an agent needs its documentation. Injecting thousands of lines of an OpenAPI/Swagger JSON specification into an LLM’s system prompt consumes massive context tokens.
Brittle Input Mapping: LLMs are non-deterministic. If a REST API requires a strict format, slight deviations in the model’s output will trigger 400 Bad Request or 422 Unprocessable Entity errors, which break the application's runtime.
Lack of Native Discovery: Standard HTTP endpoints do not natively describe what they do to an external intelligence engine; they require pre-mapped code glue to translate text intent into actual network requests.

The Paradigm Shift

TRADITIONAL CLIENT WORKFLOW (Deterministic)
[ User UI Action ] ──► [ Hardcoded Client Proxy ] ──► [ Fixed HTTP Route ] ──► [ REST API Server ]

MODERN AI AGENT WORKFLOW (Autonomous via MCP)
[ User Text Query ] ──► [ LLM Reasoning Brain ] ──► [ Dynamic JSON-RPC Tool Call ] ──► [ MCP Server Bridge ] ──► [ Downstream REST API ]

| Architectural Dimension | Traditional REST API Integration                      | Model Context Protocol (MCP) Server                          |
| ----------------------- | ----------------------------------------------------- | ------------------------------------------------------------ |
| **Inference Handling**  | Expects strict parameters mapped at build-time.       | Exposes self-describing interfaces that LLMs bind to at runtime. |
| **Discovery Mechanism** | Manual mapping or external Swagger parsing.           | Native `tools/list` handshake schema built into the protocol. |
| **Error Handling**      | Returns raw structural status codes (e.g., 404, 500). | Converts exceptions into semantic text streams ideal for agent reflection. |

Architecture of the 4-App Reference Journey

To see how this works in practice, let’s explore a microservice-driven library system across four distinct applications. Together, they demonstrate how to transition an application from a raw backend database all the way to an autonomous agent system.

┌─────────────────────────────────────────────────────────────────┐
│                        User Browsers                            │
└─────┬──────────────┬──────────────┬──────────────┬──────────────┘
      │              │              │              │
  :8001          :8002          :8003          :8004
  App 1          App 2          App 3          App 4
  API Server     Client         MCP Server     Agent GUI
  (FastAPI)      (FastAPI)      (Node.js)      (FastAPI+WS)
      ▲              │              │              │
      │         HTTP Proxy          │         WebSocket
      │              ▼              │              │
      └──────────────┘         MCP Protocol       │
      ▲                             │              │
      └─────────────────────────────┘              │
           HTTP REST API calls               Ollama LLM
           (list_books, get_book, etc)    (granite local)

App 1 (Core API Server): The data backbone. A Python FastAPI service hosting our core book collection database on port 8001.
App 2 (API Client): The traditional layout. A standard Python frontend companion on port 8002 using hardcoded HTTP requests to display data.
App 3 (MCP Server Bridge): The architectural adapter. A Node.js/TypeScript instance on port 8003 that wraps App 1’s endpoints into 8 discoverable, schemas-validated MCP tools.
App 4 (Autonomous ReAct Agent): The final intelligence layer on port 8004. It communicates with App 3 over JSON-RPC, launching a Reason-Act loop via a local LLM to execute user tasks.

Structural Code Breakdown

Step 1: Exposing the Legacy Endpoints (App 1)

Our baseline application handles raw CRUD operations. The endpoint below shows how traditional backends use query parameters to narrow down records:

"""
App 1 - Library API Server
A sample REST API server exposing CRUD operations for a library book management system.
Runs on port 8001.
"""
import os
import uuid
from datetime import datetime
from typing import Optional
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from dotenv import load_dotenv

load_dotenv()

HOST = os.getenv("API_SERVER_HOST", "0.0.0.0")
PORT = int(os.getenv("API_SERVER_PORT", "8001"))

app = FastAPI(
    title="Library API Server",
    description="A sample REST API for managing a library's book collection",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

templates = Jinja2Templates(directory="templates")
app.mount("/static", StaticFiles(directory="static"), name="static")

# ─── Models ────────────────────────────────────────────────────────────────────

class BookCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200, description="Book title")
    author: str = Field(..., min_length=1, max_length=100, description="Author name")
    genre: str = Field(..., min_length=1, max_length=50, description="Book genre")
    year: int = Field(..., ge=1000, le=2100, description="Publication year")
    available: bool = Field(True, description="Whether the book is available for borrowing")
    description: Optional[str] = Field(None, max_length=500, description="Book description")

class BookUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    author: Optional[str] = Field(None, min_length=1, max_length=100)
    genre: Optional[str] = Field(None, min_length=1, max_length=50)
    year: Optional[int] = Field(None, ge=1000, le=2100)
    available: Optional[bool] = None
    description: Optional[str] = Field(None, max_length=500)

class Book(BaseModel):
    id: str
    title: str
    author: str
    genre: str
    year: int
    available: bool
    description: Optional[str] = None
    created_at: str
    updated_at: str

# ─── In-Memory Data Store ──────────────────────────────────────────────────────

def now_iso() -> str:
    return datetime.utcnow().isoformat() + "Z"

BOOKS: dict[str, dict] = {}

SEED_BOOKS = [
    {"title": "The Hitchhiker's Guide to the Galaxy", "author": "Douglas Adams", "genre": "Science Fiction", "year": 1979, "available": True, "description": "A comedic science fiction series following the adventures of Arthur Dent."},
    {"title": "Dune", "author": "Frank Herbert", "genre": "Science Fiction", "year": 1965, "available": True, "description": "An epic science fiction novel set in a distant future amidst a feudal interstellar society."},
    {"title": "The Name of the Wind", "author": "Patrick Rothfuss", "genre": "Fantasy", "year": 2007, "available": False, "description": "The story of Kvothe, a legendary figure in his world, telling his life story to a chronicler."},
    {"title": "Sapiens: A Brief History of Humankind", "author": "Yuval Noah Harari", "genre": "Non-Fiction", "year": 2011, "available": True, "description": "A narrative history from the Stone Age to the twenty-first century."},
    {"title": "The Pragmatic Programmer", "author": "David Thomas & Andrew Hunt", "genre": "Technology", "year": 1999, "available": True, "description": "Classic guide for software developers covering best practices and principles."},
    {"title": "1984", "author": "George Orwell", "genre": "Dystopian Fiction", "year": 1949, "available": False, "description": "A dystopian social science fiction novel and cautionary tale about the dangers of totalitarianism."},
    {"title": "Clean Code", "author": "Robert C. Martin", "genre": "Technology", "year": 2008, "available": True, "description": "A handbook of agile software craftsmanship."},
    {"title": "The Lord of the Rings", "author": "J.R.R. Tolkien", "genre": "Fantasy", "year": 1954, "available": True, "description": "An epic high-fantasy novel set in Middle-earth."},
]

for book_data in SEED_BOOKS:
    bid = str(uuid.uuid4())
    BOOKS[bid] = {
        "id": bid,
        **book_data,
        "created_at": now_iso(),
        "updated_at": now_iso(),
    }

# ─── Routes ───────────────────────────────────────────────────────────────────

@app.get("/", response_class=HTMLResponse, include_in_schema=False)
async def dashboard(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

@app.get("/health", tags=["System"])
async def health():
    return {"status": "ok", "service": "library-api-server", "book_count": len(BOOKS)}

@app.get("/books", response_model=list[Book], tags=["Books"])
async def list_books(genre: Optional[str] = None, available: Optional[bool] = None, search: Optional[str] = None):
    """List all books with optional filtering by genre, availability, or search term."""
    books = list(BOOKS.values())
    if genre:
        books = [b for b in books if b["genre"].lower() == genre.lower()]
    if available is not None:
        books = [b for b in books if b["available"] == available]
    if search:
        q = search.lower()
        books = [b for b in books if q in b["title"].lower() or q in b["author"].lower() or q in (b.get("description") or "").lower()]
    return books

@app.post("/books", response_model=Book, status_code=201, tags=["Books"])
async def create_book(book: BookCreate):
    """Create a new book in the library."""
    bid = str(uuid.uuid4())
    new_book = {
        "id": bid,
        **book.model_dump(),
        "created_at": now_iso(),
        "updated_at": now_iso(),
    }
    BOOKS[bid] = new_book
    return new_book

@app.get("/books/{book_id}", response_model=Book, tags=["Books"])
async def get_book(book_id: str):
    """Get a specific book by ID."""
    if book_id not in BOOKS:
        raise HTTPException(status_code=404, detail=f"Book '{book_id}' not found")
    return BOOKS[book_id]

@app.put("/books/{book_id}", response_model=Book, tags=["Books"])
async def update_book(book_id: str, updates: BookUpdate):
    """Update an existing book's details."""
    if book_id not in BOOKS:
        raise HTTPException(status_code=404, detail=f"Book '{book_id}' not found")
    book = BOOKS[book_id]
    patch = updates.model_dump(exclude_none=True)
    book.update(patch)
    book["updated_at"] = now_iso()
    return book

@app.delete("/books/{book_id}", tags=["Books"])
async def delete_book(book_id: str):
    """Delete a book from the library."""
    if book_id not in BOOKS:
        raise HTTPException(status_code=404, detail=f"Book '{book_id}' not found")
    del BOOKS[book_id]
    return {"message": f"Book '{book_id}' deleted successfully"}

@app.get("/genres", tags=["Books"])
async def list_genres():
    """Get all unique genres in the library."""
    genres = sorted(set(b["genre"] for b in BOOKS.values()))
    return {"genres": genres}

@app.post("/books/{book_id}/borrow", response_model=Book, tags=["Books"])
async def borrow_book(book_id: str):
    """Mark a book as borrowed (unavailable)."""
    if book_id not in BOOKS:
        raise HTTPException(status_code=404, detail=f"Book '{book_id}' not found")
    if not BOOKS[book_id]["available"]:
        raise HTTPException(status_code=409, detail="Book is already borrowed")
    BOOKS[book_id]["available"] = False
    BOOKS[book_id]["updated_at"] = now_iso()
    return BOOKS[book_id]

@app.post("/books/{book_id}/return", response_model=Book, tags=["Books"])
async def return_book(book_id: str):
    """Mark a book as returned (available)."""
    if book_id not in BOOKS:
        raise HTTPException(status_code=404, detail=f"Book '{book_id}' not found")
    if BOOKS[book_id]["available"]:
        raise HTTPException(status_code=409, detail="Book is already available")
    BOOKS[book_id]["available"] = True
    BOOKS[book_id]["updated_at"] = now_iso()
    return BOOKS[book_id]

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host=HOST, port=PORT, reload=True)

Step 2: The Traditional Integration Strategy (App 2)

In a traditional client architecture, every backend change requires updating a corresponding client method. If App 1 introduces new query parameters or pagination rules, App 2’s hardcoded proxy functions will fail or drop the data unless manually re-compiled and updated.

"""
App 2 - API Client Application
Consumes the Library API Server (App 1) and provides a rich dashboard GUI.
Runs on port 8002.
"""
import os
import httpx
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional
from dotenv import load_dotenv

load_dotenv()

HOST = os.getenv("API_CLIENT_HOST", "0.0.0.0")
PORT = int(os.getenv("API_CLIENT_PORT", "8002"))
API_SERVER_URL = os.getenv("API_SERVER_URL", "http://localhost:8001")

app = FastAPI(
    title="Library API Client",
    description="Client application that consumes the Library API Server",
    version="1.0.0",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

templates = Jinja2Templates(directory="templates")
app.mount("/static", StaticFiles(directory="static"), name="static")

# ─── Pydantic Models ──────────────────────────────────────────────────────────

class BookCreate(BaseModel):
    title: str
    author: str
    genre: str
    year: int
    available: bool = True
    description: Optional[str] = None

class BookUpdate(BaseModel):
    title: Optional[str] = None
    author: Optional[str] = None
    genre: Optional[str] = None
    year: Optional[int] = None
    available: Optional[bool] = None
    description: Optional[str] = None

# ─── Helper ───────────────────────────────────────────────────────────────────

async def api_request(method: str, path: str, **kwargs):
    """Proxy request to the upstream API server."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        url = f"{API_SERVER_URL}{path}"
        response = await client.request(method, url, **kwargs)
        return response

# ─── Dashboard Route ──────────────────────────────────────────────────────────

@app.get("/", response_class=HTMLResponse, include_in_schema=False)
async def dashboard(request: Request):
    return templates.TemplateResponse("index.html", {
        "request": request,
        "api_server_url": API_SERVER_URL
    })

@app.get("/config")
async def get_config():
    return {
        "api_server_url": API_SERVER_URL,
        "client_port": PORT,
    }

# ─── Proxy Routes ─────────────────────────────────────────────────────────────

@app.get("/api/health")
async def health_check():
    try:
        r = await api_request("GET", "/health")
        api_data = r.json() if r.status_code == 200 else {"status": "error"}
        return {"client_status": "ok", "api_server": api_data, "api_server_url": API_SERVER_URL}
    except Exception as e:
        return {"client_status": "ok", "api_server": {"status": "unreachable", "error": str(e)}, "api_server_url": API_SERVER_URL}

@app.get("/api/books")
async def list_books(genre: Optional[str] = None, available: Optional[bool] = None, search: Optional[str] = None):
    params = {}
    if genre: params["genre"] = genre
    if available is not None: params["available"] = str(available).lower()
    if search: params["search"] = search
    r = await api_request("GET", "/books", params=params)
    if not r.is_success:
        raise HTTPException(status_code=r.status_code, detail=r.text)
    return r.json()

@app.post("/api/books", status_code=201)
async def create_book(book: BookCreate):
    r = await api_request("POST", "/books", json=book.model_dump())
    if not r.is_success:
        raise HTTPException(status_code=r.status_code, detail=r.json().get("detail", r.text))
    return r.json()

@app.get("/api/books/{book_id}")
async def get_book(book_id: str):
    r = await api_request("GET", f"/books/{book_id}")
    if not r.is_success:
        raise HTTPException(status_code=r.status_code, detail=r.json().get("detail", r.text))
    return r.json()

@app.put("/api/books/{book_id}")
async def update_book(book_id: str, updates: BookUpdate):
    r = await api_request("PUT", f"/books/{book_id}", json=updates.model_dump(exclude_none=True))
    if not r.is_success:
        raise HTTPException(status_code=r.status_code, detail=r.json().get("detail", r.text))
    return r.json()

@app.delete("/api/books/{book_id}")
async def delete_book(book_id: str):
    r = await api_request("DELETE", f"/books/{book_id}")
    if not r.is_success:
        raise HTTPException(status_code=r.status_code, detail=r.json().get("detail", r.text))
    return r.json()

@app.post("/api/books/{book_id}/borrow")
async def borrow_book(book_id: str):
    r = await api_request("POST", f"/books/{book_id}/borrow")
    if not r.is_success:
        raise HTTPException(status_code=r.status_code, detail=r.json().get("detail", r.text))
    return r.json()

@app.post("/api/books/{book_id}/return")
async def return_book(book_id: str):
    r = await api_request("POST", f"/books/{book_id}/return")
    if not r.is_success:
        raise HTTPException(status_code=r.status_code, detail=r.json().get("detail", r.text))
    return r.json()

@app.get("/api/genres")
async def list_genres():
    r = await api_request("GET", "/genres")
    if not r.is_success:
        raise HTTPException(status_code=r.status_code, detail=r.text)
    return r.json()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host=HOST, port=PORT, reload=True)

Step 3: Mapping APIs to Self-Describing MCP Tools (App 3)

Instead of forcing our AI Agent to guess raw URL syntax, App 3 translates the FastAPI endpoints into standard tool definitions using the official @modelcontextprotocol/sdk. By integrating a Zod validation schema, the model learns the exact parameters it needs to provide.

// index-app3.ts
server.registerTool(
  "list_books",
  {
    description: "List all books in the library with optional filters",
    inputSchema: z.object({
      genre: z.string().optional().describe("Filter by genre (e.g., Fantasy, Technology, Sci-Fi)"),
      available: z.boolean().optional().describe("Filter by availability (true = available, false = borrowed)"),
      search: z.string().optional().describe("Search string matching titles, authors, or descriptions"),
    }),
  },
  async ({ genre, available, search }) => {
    const params: Record<string, string> = {};
    if (genre) params.genre = genre;
    if (available !== undefined) params.available = String(available);
    if (search) params.search = search;

    // Standard network call down to the core App 1 infrastructure
    const result = await apiCall("GET", "/books", undefined, params);
    const books = result as Array<Record<string, any>>;

    return {
      content: [{
        type: "text",
        text: `Found ${books.length} book(s):\n` + 
              books.map(b => `• "${b.title}" by ${b.author} [ID: ${b.id}] (${b.available ? 'Available' : 'Borrowed'})`).join("\n"),
      }],
    };
  }
);

Step 4: Orchestrating the Autonomous ReAct Loop (App 4)

When a user types an unstructured query (like “Find me a good tech book and check if it’s available”), App 4 boots up a Reason-Act (ReAct) loop.

The agent queries the tool list from App 3, thinks through its plan, triggers the list_books tool, observes the results, and formulates its final answer. The parser function below demonstrates how the loop reads the model's text reasoning steps to execute tools dynamically:

# main-app4.py
def parse_react_step(text: str) -> dict:
    """
    Parses the LLM's text output to extract either the next 
    structured tool invocation or the terminal 'Final Answer'.
    """
    pre_final, _, _ = text.partition("Final Answer:")

    action_match = re.search(r"Action:\s*(\w+)", pre_final, re.IGNORECASE)
    input_match = re.search(r"Action Input:\s*(\{.+?\})", pre_final, re.DOTALL | re.IGNORECASE)
    final_match = re.search(r"Final Answer:\s*(.+)", text, re.DOTALL | re.IGNORECASE)

    if action_match:
        try:
            return {
                "type": "action", 
                "tool": action_match.group(1).strip(), 
                "args": json.loads(input_match.group(1))
            }
        except Exception:
            # Fallback if the LLM output poorly-formatted JSON arguments
            return {"type": "action", "tool": action_match.group(1).strip(), "args": {}}

    elif final_match:
        return {"type": "final", "answer": final_match.group(1).strip()}

    return {"type": "fallback"}

Summary: The Modern Interface Matrix

Migrating legacy APIs to MCP tools shifts how we think about system integration. By introducing an adapter layer that translates raw endpoints into self-describing tool schemas, applications gain a new level of architectural flexibility.

When your core service changes, your agent doesn’t need to be rewritten. Instead, it reads the updated tool schemas at runtime, dynamically structuring its queries to meet the new backend requirements on the fly.

Thanks for reading 🫵

The ROI of SDLC (Part 5): Compiling Orchestration-Porting Our Multi-LLM Routing Engine to Pure Go for Low-Latency and Zero-Dependency Performance

Alain Airom (Ayrom) — Mon, 06 Jul 2026 12:17:00 +0000

Part 5 (bonus): Implementing Multi-LLM Routing in Go for performance

Introduction

Throughout this series, we have treated the LLM orchestration layer as an intelligent optimization problem. In my previous posts, the design of a mathematical utility model to dynamically match incoming prompts against a 4-tier pool of LLM “T-Shirt sizes” was showcased:

Where Q represents the expected quality score of a model given a task’s computed complexity, C represents normalized cost units, and the weights are bounded by β=1−α to represent a strict financial-versus-performance optimization tradeoff.

We (Bob and I) successfully implemented this routing layer in Python using FastAPI, leveraging the LiteLLM gateway to unify local Ollama nodes (granite4.1:3b, llama3.2, gemma3:4b, and mistral-small), and we integrated a live telemetry loop to update acceptable quality thresholds over a rolling execution window.

But as production workloads scale out, the performance characteristics of the orchestrator itself become critical. As traffic densities climb into hundreds of multi-task prompts per second, managing Python’s runtime environment, navigating single-threaded async event loop saturation under heavy CPU classification tasks, and dealing with significant baseline container footprints add unwanted operational friction.

In this fifth and final episode, let’s take the ultimate leap: compiling down to metal. We will walk through how the entire multi-LLM router/scheduler into a zero-dependency, hyper-performant engine written in pure Go (routing-v5) and how it was ported and translated. This post dissects and shows how replacing high-level abstractions with raw Go concurrency patterns, native memory mutexes, ring-buffer telemetry primitives, and strict circuit breaker state machines allows us to cut operational latency to the microsecond level while dramatically simplifying our production deployment topology.

Why Go? The Performance Trade-off of the AI Orchestrator

When evaluating an LLM gateway or router, engineers frequently assume that downstream model inference latency (the time an LLM spends inside Ollama generating tokens on CPUs and/or GPUs) entirely dwarfs the orchestrator’s overhead. In a trivial sequential chat setup, that is true. However, when a compound developer prompt is split into multiple sub-tasks that require simultaneous token classification, dynamic threshold lookups, concurrency synchronization, feedback auditing, and real-time streaming updates over WebSockets, the orchestrator’s language runtime becomes the primary bottleneck for system throughput.

Let’s look at the concrete operational paradigm shift achieved by moving from our Python FastAPI stack (v4) to our compiled pure Go engine (v5):

| Operational Metric            | Python Stack (v4)                                         | Go Engine (v5)                                               |
| ----------------------------- | --------------------------------------------------------- | ------------------------------------------------------------ |
| **Runtime Dependencies**      | Heavy `venv` (FastAPI, Uvicorn, LiteLLM, Pydantic, HTTPX) | **Zero** (Single self-contained native machine binary)       |
| **Baseline Memory Footprint** | ~75MB – 120MB idle baseline                               | **~12MB idle baseline**                                      |
| **Concurrency Architecture**  | `asyncio` loop (Single-threaded multiplexing)             | **Native Goroutines** (True multi-threaded OS thread multiplexing) |
| **State Thread-Safety**       | Asynchronous application primitives                       | **Native Mutexes (`sync.RWMutex`) & Memory Channels**        |
| **CPU-Bound Classification**  | Blocks or requires complex process pools                  | Fully non-blocking across available CPU cores                |

By adopting Go, the application eliminates interpreter latency. We replace layers of dependency wrappers with an optimized execution tree where token counting, regex-based task parsing, and mathematical utility optimizations execute in microseconds, leaving the system’s entire resource pool free to manage I/O throughput and concurrent client connections.

End-to-End System & Core Data Flow Blueprint

To understand the low-overhead architecture of the Go engine, we must trace how a compound prompt is ingested, processed, executed, and streamed. The system treats incoming text payloads as an optimization graph to be aggressively executed in parallel.

       ┌──────────────────────────────────────────────────────────────┐
       │                          main.go                             │
       │   Echo HTTP Server  ·  /api/chat  ·  WebSocket /ws/telemetry │
       └──────────────┬───────────────────────────────────────────────┘
                      │
              orchestrator.Handle()
                      │
   ┌──────────────────┼─────────────────────┬─────────────────────┐
   │                  │                     │                     │
   ▼                  ▼                     ▼                     ▼
classifier.         router.               llmclient.            feedback.
SplitAndClassify()  Dispatch()            CallModel()           Store.Record()
   │                  │                     │                     │
   │                  ├─────────────────────┤                     │
   │                  │                     │                     │
   ▼                  ▼                     ▼                     ▼
[Profiles]         failover.             telemetry.            [Append-Only]
                   Manager.Resolve()     Engine.Record()       JSONL Audit Logs

End-to-End Data Flow Matrix

Ingress Payload Consumption: A user client posts a structured text payload to the /api/chat Echo handler. This payload is handed directly to orchestrator.Handle(ctx, prompt).
Tokenization and Dissection: The string is analyzed by classifier.SplitAndClassify(). It matches text boundaries against explicit logical conjunction markers (and also, additionally, plus) via regular expressions, segmenting the prompt into discrete task clauses. Each clause is evaluated for complexity based on signal keywords, string length, and structural depth penalties, returning an array of TaskProfile structs.
Dynamic Utility Selection: The []TaskProfile array passes into router.Dispatch(). For each profile, the router fetches the latest running parameters from the thread-safe telemetry.Engine, cross-references the task type against the StaticTaskThresholds, and evaluates the model capability matrix. It then computes the highest utility option using the model spec's quality-curve functions: U=β⋅Q(complexity)−α⋅CostPer1k
Circuit-Breaker Interception: Before confirming the target tier, the candidate model is vetted by failover.Manager.Resolve(). If the model's circuit breaker state is OPEN due to repeated local timeout or connection failures at the Ollama layer, the request is intercepted and systematically rerouted to the next available provider in its configured FailoverChain.
Structured Concurrent Execution: The orchestrator instantiates a dedicated goroutine wrapper for each sub-task in the dispatch map. These routines run concurrently, multiplexing network calls over native OS threads.
Network Delivery & Telemetry Ingestion: Each worker goroutine triggers llmclient.CallModel(), which executes an HTTP POST to the local Ollama /v1/chat/completions OpenAI-compatible endpoint. Upon response completion, the worker records the transaction duration and token weight, passing this metadata into telemetry.Engine.Record() to recalculate the Exponential Moving Average (EMA) quality scores and latency percentiles (p50 and p95).
Asynchronous Streaming & Audit Logging: Concurrently, the orchestrator appends the finalized routing signatures to the thread-safe, append-only feedback.Store (writing row lines to output/feedback.jsonl). It also emits progress notifications over a centralized channel (eventCh), allowing the broadcastLoop to instantly push JSON frame payloads to all connected WebSocket visualization dashboards.
Aggregation Assembly: Once all concurrent routines join (sync.WaitGroup.Wait()), the system aggregates individual sections within merge(), compiling the outputs into unified Markdown pages returned to the caller inside an enriched HTTP JSON structure.

Component Deep-Dives & Production-Grade Code

Let’s dive into the core packages of the pure Go implementation (routing-v5) to analyze the code patterns that enforce thread safety, real-time metrics tracking, and low-latency execution.

Thread-Safe Live Telemetry Engine (`telemetry.go`)

In our Go implementation, we avoid relying on external analytical engines or global locks that create contention bottlenecks. Instead, telemetry.go maintains an explicit map of isolated fixed-size ring buffers, protected by a fine-grained sync.Mutex per model-and-task combination.

When a sub-task finishes, its execution signature updates an Exponential Moving Average (EMA) of quality scores and updates running latency tables to calculate precise p50 and p95 metrics. If percentiles slip past our targetLatencyMs, the engine increases the cost-penalty coefficient (α), causing the router to favor lighter, faster models until system pressure subsides.

// internal/telemetry/telemetry.go
package telemetry

import (
 "math"
 "os"
 "sort"
 "strconv"
 "sync"
)

var (
 windowSize      = envInt("TELEMETRY_WINDOW", 50)
 emaAlpha        = envFloat("EMA_ALPHA", 0.15)
 baseThreshold   = 0.72
 minThreshold    = 0.40
 maxThreshold    = 0.95
 baseCostWeight  = 0.40
 minCostWeight   = 0.10
 maxCostWeight   = 0.70
 targetLatencyMs = 3000.0
)

type Record struct {
 LatencyMs   int
 QualityEval float64
 Success     bool
}

type ringBuffer struct {
 data  []Record
 head  int
 size  int
 maxSz int
}

func newRingBuffer(max int) *ringBuffer {
 return &ringBuffer{data: make([]Record, max), maxSz: max}
}

func (r *ringBuffer) append(rec Record) {
 r.data[r.head] = rec
 r.head = (r.head + 1) % r.maxSz
 if r.size < r.maxSz {
  r.size++
 }
}

func (r *ringBuffer) slice() []Record {
 out := make([]Record, r.size)
 for i := 0; i < r.size; i++ {
  idx := (r.head - r.size + i + r.maxSz) % r.maxSz
  out[i] = r.data[idx]
 }
 return out
}

type Engine struct {
 mu      sync.Mutex
 buffers map[[2]string]*ringBuffer
}

func NewEngine() *Engine {
 return &Engine{buffers: make(map[[2]string]*ringBuffer)}
}

func (e *Engine) Record(modelLabel, taskType string, rec Record) map[string]float64 {
 e.mu.Lock()
 defer e.mu.Unlock()

 key := [2]string{modelLabel, taskType}
 if e.buffers[key] == nil {
  e.buffers[key] = newRingBuffer(windowSize)
 }
 e.buffers[key].append(rec)

 return e.tune(key)
}

func (e *Engine) tune(key [2]string) map[string]float64 {
 buf := e.buffers[key].slice()
 n := len(buf)
 if n == 0 {
  return map[string]float64{"quality_threshold": baseThreshold, "cost_weight": baseCostWeight}
 }

 qEMA := buf[0].QualityEval
 for _, r := range buf[1:] {
  qEMA = emaAlpha*r.QualityEval + (1-emaAlpha)*qEMA
 }

 lats := make([]int, n)
 successes := 0
 for i, r := range buf {
  lats[i] = r.LatencyMs
  if r.Success {
   successes++
  }
 }
 sort.Ints(lats)
 p95Idx := int(float64(n) * 0.95)
 if p95Idx >= n {
  p95Idx = n - 1
 }
 p95 := float64(lats[p95Idx])

 newThr := baseThreshold + 0.5*(qEMA-baseThreshold)
 newThr = math.Max(minThreshold, math.Min(maxThreshold, newThr))

 latPressure := p95 / targetLatencyMs
 costW := baseCostWeight * math.Exp(latPressure-1.0)
 costW = math.Max(minCostWeight, math.Min(maxCostWeight, costW))

 return map[string]float64{
  "quality_threshold": math.Round(newThr*1000) / 1000,
  "cost_weight":       math.Round(costW*1000) / 1000,
 }
}

func envInt(k string, d int) int {
 if v := os.Getenv(k); v != "" {
  if n, err := strconv.Atoi(v); err == nil { return n }
 }
 return d
}

func envFloat(k string, d float64) float64 {
 if v := os.Getenv(k); v != "" {
  if f, err := strconv.ParseFloat(v, 64); err == nil { return f }
 }
 return d
}

Circuit Breaker State Machine (`failover.go`)

Resilience at low latency requires keeping tabs on infrastructure node health without adding network overhead. The Go engine wraps every model endpoint inside an isolated state machine tracking three critical operational states: CLOSED (healthy processing), OPEN (failing back directly to alleviate latency bottlenecks), and HALF_OPEN (controlled probe windows testing for backend recovery).

          ┌──────────────────────────────────┐
          │                                  │
          ▼                                  │
     ┌──────────┐   Failure Threshold   ┌────────┐
  ──►│  CLOSED  ├──────────────────────►│  OPEN  │
     └────▲─────┘   Exceeded            └───┬────┘
          │                                 │
          │                                 │ Cooldown
          │ Probes Successful               │ Window
          │ (probeCount >= halfOpenProbes)  │ Elapsed
          │                                 │
     ┌────┴───────┐                         │
     │ HALF_OPEN  │◄────────────────────────┘
     └────┬───────┘
          │
          │ Probe Failure
          └─────────────────────────────────►

// internal/failover/failover.go
package failover

import (
 "sync"
 "time"

 "github.com/routing-v5/internal/registry"
)

type CircuitState string

const (
 Closed   CircuitState = "CLOSED"
 Open     CircuitState = "OPEN"
 HalfOpen CircuitState = "HALF_OPEN"
)

var (
 failureThreshold = 3
 cooldownSeconds  = 30
 halfOpenProbes   = 2
 healthEMAAlpha   = 0.20
)

type circuitBreakerState struct {
 mu               sync.RWMutex
 modelLabel       string
 state            CircuitState
 consecutiveFails int
 lastStateChange  time.Time
 probeCount       int
 healthEMA        float64
}

func newCircuitBreaker(label string) *circuitBreakerState {
 return &circuitBreakerState{
  modelLabel: label,
  state:      Closed,
  healthEMA:  1.0,
 }
}

func (cb *circuitBreakerState) IsCallable() bool {
 cb.mu.Lock()
 defer cb.mu.Unlock()

 if cb.state == Open {
  if time.Since(cb.lastStateChange) > time.Duration(cooldownSeconds)*time.Second {
   cb.state = HalfOpen
   cb.probeCount = 0
   cb.lastStateChange = time.Now()
   return true
  }
  return false
 }
 return true
}

func (cb *circuitBreakerState) RecordFailure() {
 cb.mu.Lock()
 defer cb.mu.Unlock()

 cb.consecutiveFails++
 cb.healthEMA = (1-healthEMAAlpha)*cb.healthEMA + healthEMAAlpha*0.0

 if (cb.state == Closed && cb.consecutiveFails >= failureThreshold) || cb.state == HalfOpen {
  cb.state = Open
  cb.lastStateChange = time.Now()
 }
}

func (cb *circuitBreakerState) RecordSuccess() {
 cb.mu.Lock()
 defer cb.mu.Unlock()

 cb.consecutiveFails = 0
 cb.healthEMA = (1-healthEMAAlpha)*cb.healthEMA + healthEMAAlpha*1.0

 if cb.state == HalfOpen {
  cb.probeCount++
  if cb.probeCount >= halfOpenProbes {
   cb.state = Closed
   cb.lastStateChange = time.Now()
  }
 }
}

func (cb *circuitBreakerState) HealthScore() float64 {
 cb.mu.RLock()
 defer cb.mu.RUnlock()
 return cb.healthEMA
}

type Manager struct {
 mu       sync.Mutex
 circuits map[string]*circuitBreakerState
 events   []FailoverEvent
}

func NewManager() *Manager {
 m := &Manager{circuits: make(map[string]*circuitBreakerState)}
 for _, mod := range registry.Models {
  m.circuits[mod.Label] = newCircuitBreaker(mod.Label)
 }
 return m
}

func (m *Manager) RecordFailure(label string) { m.circuits[label].RecordFailure() }
func (m *Manager) RecordSuccess(label string) { m.circuits[label].RecordSuccess() }

func (m *Manager) Resolve(primary *registry.ModelSpec, taskType string) (*registry.ModelSpec, *FailoverEvent) {
 if m.circuits[primary.Label].IsCallable() {
  return primary, nil
 }

 // Navigate fallback mapping
 for _, entry := range primary.FailoverChain {
  if m.circuits[entry.Label].IsCallable() {
   fb := registry.ByLabel(entry.Label)
   ev := &FailoverEvent{
    PrimaryLabel:  primary.Label,
    FallbackLabel: entry.Label,
    Reason:        entry.Reason,
    Timestamp:     time.Now(),
   }
   m.mu.Lock()
   m.events = append(m.events, *ev)
   m.mu.Unlock()
   return fb, ev
  }
 }

 // Last-resort backup: pick the node with the highest health EMA
 best := registry.Models[0]
 for _, mod := range registry.Models[1:] {
  if m.circuits[mod.Label].HealthScore() > m.circuits[best.Label].HealthScore() {
   best = mod
  }
 }
 ev := &FailoverEvent{
  PrimaryLabel:  primary.Label,
  FallbackLabel: best.Label,
  Reason:        "all_chains_exhausted",
  Timestamp:     time.Now(),
 }
 m.mu.Lock()
 m.events = append(m.events, *ev)
 m.mu.Unlock()
 return best, ev
}

The Orchestration Core & WebSocket Broadcasting Topology

To keep analytics and real-time dashboard events from blocking the primary API request loop, the Go implementation isolates client updates within a non-blocking buffered memory channel.

Instead of forcing the front-end to poll the database, the orchestrator broadcasts atomic execution snapshots over eventCh. The server's event loop fans these messages out to active browser clients, dropping slow readers instantly to protect core processing paths.

[Worker Goroutines]      [Central Routing backplane]      [Client Channels]
   (Sub-Task Workers)
  ┌───────────────────┐
  │  Sub-task Start   ├───────┐
  └───────────────────┘       │
                              ▼
  ┌───────────────────┐    ┌─────────┐                ┌──────────────────┐
  │ Sub-task Complete ├───►│ eventCh │───────────────►│ Client Message   │
  └───────────────────┘    │ (Buffer │                │ Ring-buffer loop │
                           │  Slots) │                └────────┬─────────┘
  ┌───────────────────┐       ▲                                │
  │   Routing Plans   ├───────┘                                ▼
  └───────────────────┘                                ┌───────────────┐
                                                       │ Browser Web   │
                                                       │ UI Dashboards │
                                                       └───────────────┘

// internal/orchestrator/orchestrator.go
package orchestrator

import (
 "context"
 "sync"
 "time"

 "github.com/routing-v5/internal/classifier"
 "github.com/routing-v5/internal/failover"
 "github.com/routing-v5/internal/feedback"
 "github.com/routing-v5/internal/llmclient"
 "github.com/routing-v5/internal/router"
 "github.com/routing-v5/internal/telemetry"
)

type Orchestrator struct {
 eng     *telemetry.Engine
 fm      *failover.Manager
 store   *feedback.Store
 eventCh chan map[string]interface{}
}

func NewOrchestrator(e *telemetry.Engine, f *failover.Manager, s *feedback.Store, ch chan map[string]interface{}) *Orchestrator {
 return &Orchestrator{eng: e, fm: f, store: s, eventCh: ch}
}

func (o *Orchestrator) Handle(ctx context.Context, prompt string) (*OrchestratorResult, error) {
 tIndex := time.Now()
 profiles := classifier.SplitAndClassify(prompt)
 routing := router.Dispatch(o.eng, o.fm, profiles)

 o.emit(map[string]interface{}{
  "type":         "routing_plan",
  "instructions": len(routing),
 })

 var wg sync.WaitGroup
 subResults := make([]SubTaskResult, len(routing))
 resMu := sync.Mutex{}

 for key, dispatchRes := range routing {
  wg.Add(1)
  go func(k string, dr router.DispatchResult) {
   defer wg.Done()
   o.emit(map[string]interface{}{"type": "subtask_start", "task_type": k})

   t0 := time.Now()
   cr := llmclient.CallModel(ctx, o.fm, dr.Selected.Label, dr.Profile.RawText, dr.Selected.MaxTokens)
   lat := int(time.Since(t0).Milliseconds())

   success := cr.Tokens > 0
   o.eng.Record(dr.Selected.Label, string(dr.Profile.TaskType), telemetry.Record{
    LatencyMs:   lat,
    QualityEval: dr.Selected.QualityAt(dr.Profile.Complexity),
    Success:     success,
   })

   cost := dr.Selected.CostPer1k * float64(cr.Tokens) / 1000.0
   o.store.Record(feedback.RoutingRecord{
    Timestamp:    time.Now().UTC().Format(time.RFC3339),
    TaskType:     k,
    Complexity:   dr.Profile.Complexity,
    Tier:         dr.Selected.Tier,
    ModelUsed:    dr.Selected.Label,
    LatencyMs:    lat,
    CostUnits:    cost,
    QualityEval:  dr.Selected.QualityAt(dr.Profile.Complexity),
    FailoverUsed: dr.FailoverEvent != nil,
   })

   resMu.Lock()
   subResults = append(subResults, SubTaskResult{
    Profile:      dr.Profile,
    Model:        dr.Selected,
    Response:     cr.Text,
    LatencyMs:    lat,
    OutTokens:    cr.Tokens,
    UtilityScore: dr.Utility,
    OllamaModel:  dr.Selected.Label,
    CostUnits:    cost,
   })
   resMu.Unlock()

   o.emit(map[string]interface{}{
    "type":       "subtask_complete",
    "task_type":  k,
    "model":      dr.Selected.Label,
    "latency_ms": lat,
   })
  }(key, dispatchRes)
 }

 wg.Wait()
 return &OrchestratorResult{
  Prompt:     prompt,
  SubResults: subResults,
  TotalMs:    int(time.Since(tIndex).Milliseconds()),
 }, nil
}

func (o *Orchestrator) emit(p map[string]interface{}) {
 if o.eventCh != nil {
  select {
  case o.eventCh <- p:
  default: // Non-blocking: skip slow dashboard consumers
  }
 }
}

To bridge this asynchronous pipeline with client browsers, main.go sets up a dedicated broadcasting loop. By isolating connection maps behind a mutex and pairing them with unblocked send frames, we ensure the core engine stays safe and responsive under load.

// snippet from main.go
var (
 wsClients  = make(map[*websocket.Conn]chan string)
 wsClientMu sync.Mutex
 eventCh    = make(chan map[string]interface{}, 500)
)

func broadcastLoop() {
 for payload := range eventCh {
  data, err := json.Marshal(payload)
  if err != nil {
   continue
  }
  msg := string(data)

  wsClientMu.Lock()
  dead := []*websocket.Conn{}
  for conn, ch := range wsClients {
   select {
   case ch <- msg:
   default:
    dead = append(dead, conn)
   }
  }
  for _, conn := range dead {
   close(wsClients[conn])
   delete(wsClients, conn)
  }
  wsClientMu.Unlock()
 }
}

Conclusion of these Series & Key Architectural Takeaways

By porting our multi-LLM routing pipeline from Python to compiled Go, we demonstrate that optimizing AI infrastructure goes beyond fine-tuning models — the efficiency of the orchestration layer itself plays a massive role.

Abstractions Have a Cost: Dropping external gateway frameworks in favor of native HTTP communication cuts orchestration overhead down to the microsecond level.
True System Concurrency: While not the main purpose, replacing Python’s single-threaded event loops with Go’s lightweight goroutines ensures that intensive text parsing and utility logic don’t block concurrent I/O streams.
Operational Independence: Compiling down to a single binary removes package dependency maintenance and simplifies deployment down to a simple executable drop.

This concludes our 5-part journey into building a production-grade multi-LLM router! By prioritizing live metrics tracking, smart task division, circuit breaker failovers, and low-latency code design, we can build workflows that run at peak efficiency — keeping costs minimal and performance high. The aim is to gain a practical, foundational understanding of how an intelligent SDLC engine manages multi-LLM orchestration and parallel routing. While this demonstration abstracts away proprietary or internal mechanics of a production agent, it illustrates the exact architectural paradigm required to industrialize software development: splitting compound engineering tasks, evaluating complexity curves, and dynamically balancing cost against quality thresholds.

Dynamic Task Deconstruction: Compound engineering prompts (e.g., “Write a Go helper function and document its edge cases”) are split into distinct execution branches rather than being sent as a single block to an expensive model.
The Utility Trade-Off: Decisions are mathematically routed using a utility function balancing expected quality (Q) against unit cost ( C ): U=β⋅Q−α⋅C.
T-Shirt Sized Tiering: Tasks are mapped to a model pool matching the required capabilities:

Tier 1 (Prose/Formatting): Lightweight, highly efficient models (e.g., IBM Granite).
Tier 2/3 (Structured Logic): Medium-tier generalists (e.g., LLaMA / Gemma) for standard logic.
Tier 4 (Complex Synthesis/Security): Heavyweight models (e.g., Mistral Small/Large) reserved for architecture reviews or security-critical implementations.

This structural framework effectively conveys that while the implementation details are a global approximation, the pattern itself is the standard for building predictable, cost-efficient, and secure AI-driven engineering tools.

🎯 That’s a wrap 💯 for these series and thanks for reading!

The ROI of SDLC: Proving the Financial Logic of Multi-Model Orchestration: An Advanced Dual-Layer Routing Implementation (Part 4)

Alain Airom (Ayrom) — Sun, 05 Jul 2026 14:02:35 +0000

Dealing with financial and failover side!

Introduction

Static routing rules are insufficient when building production-grade AI pipelines. If a local model container experiences a transient failure or a sudden spike in request volume, the entire application can stall. Furthermore, static quality thresholds fail to adapt to live performance conditions.

To solve this, we need a system that can handle infrastructure failures gracefully while dynamically adjusting its routing parameters based on real-time telemetry.

This post walks through a standalone, personal proof-of-concept — routing-v4 — built to demonstrate the immense financial and operational logic behind the multi-model orchestration paradigm. By layering a custom, telemetry-driven decision framework on top of an enterprise gateway (LiteLLM Router), we achieve a self-healing, self-optimizing AI routing pipeline.

The Dual-Layer Resilience Architecture

The system splits operational responsibilities into two distinct, isolated layers:

The Strategic Layer (router.py** + telemetry.py)**: Runs a continuous feedback loop. It analyzes incoming sub-tasks using an economic utility function (U=β⋅Q−α⋅C) where the cost weight (α) and quality threshold are continuously adjusted on the fly by a live telemetry engine.
The Tactical Layer (failover_manager.py** + llm_client.py)**: Protects application execution using per-model circuit breakers, explicit failover chains, and a unified connection layer.

          +-----------------------------------+
                  |   FastAPI Payload Intake Engine   |
                  +-----------------+-----------------+
                                    |
                                    v
                  +-----------------------------------+
                  |  Telemetry Engine & Param Tuner   |
                  |   (Live Ring Buffers & EMA)       |
                  +-----------------+-----------------+
                                    |
                        [Dynamic α and β Weights]
                                    |
                                    v
                  +-----------------------------------+
                  |   Utility-Based Task Router       |
                  |     U = β(t)·Q − α(t)·C           |
                  +-----------------+-----------------+
                                    |
                       [Primary Model Nominee]
                                    |
                                    v
                  +-----------------------------------+
                  |   Circuit-Breaker Failover Mgr    |
                  |   (CLOSED / OPEN / HALF_OPEN)     |
                  +-----------------+-----------------+
                                    |
                       [Resolved Execution Target]
                                    |
                                    v
                  +-----------------------------------+
                  |    LiteLLM Transport Gateway      |
                  +-----------------------------------+

Deep-Diving the Core Technical Core

State-Driven Circuit Breakers

To prevent requests from hanging on unresponsive endpoints, the FailoverManager tracks each model label inside an isolated state machine. Repeated failures trip the circuit into an OPEN state, instantly bypassing the model and routing requests to an explicit fallback chain.

# From src/failover_manager.py
class CircuitState(Enum):
    CLOSED = "CLOSED"        # Normal operations
    OPEN = "OPEN"            # Outage detected; block traffic
    HALF_OPEN = "HALF_OPEN"  # Testing recovery with probe calls

class CircuitBreaker:
    # ... tracking logic ...
    def record_failure(self) -> None:
        with self._lock:
            self._consecutive_failures += 1
            # Update health score using an Exponential Moving Average
            self.health_score = 0.7 * self.health_score + 0.3 * 0.0

            if self._state == CircuitState.CLOSED and self._consecutive_failures >= self._failure_threshold:
                self._state = CircuitState.OPEN
                self._open_since = time.time()

When a primary model nominee is selected, the manager intercepts the call. If the primary target’s circuit is broken, it traverses the task-specific failover prioritization matrix:

# From src/failover_manager.py
def resolve(self, primary: ModelSpec, task_type: str) -> tuple[ModelSpec, Optional[FailoverEvent]]:
    """Resolves the best available model, fallback chain, or healthy asset."""
    if self.is_callable(primary.label):
        return primary, None

    # Primary is blocked. Intercept and walk the fallback chain
    for fallback_label, reason in primary.failover_chain:
        if self.is_callable(fallback_label):
            fb_model = MODEL_BY_LABEL.get(fallback_label)
            if fb_model:
                event = FailoverEvent(
                    primary_label=primary.label,
                    fallback_label=fallback_label,
                    reason=reason,
                )
                return fb_model, event

    # Last resort: fallback to whichever model has the highest health score
    best = max(MODEL_REGISTRY, key=lambda m: self._circuits[m.label].health_score)
    return best, FailoverEvent(primary.label, best.label, "all_chains_exhausted")

Telemetry-Driven Parameter Tuning

The system does not use hardcoded thresholds. Instead, a TelemetryEngine maintains a fixed-size ring buffer for each model-task combination, computing p95 latencies and running quality scores.

If p95 latency spikes beyond a target SLA, the system calculates latency_pressure and automatically dials down the cost weight (α). This adjustments shifts preference toward faster, more expensive tiers to safeguard response speeds.

# From src/telemetry.py
def _tune_params(self, key: tuple[str, str], buf: Deque[TelemetryRecord]) -> DynamicParams:
    n = len(buf)
    q_ema = sum(r.quality_eval for r in buf) / n

    # Calculate Latency Percentiles
    latencies = sorted(r.latency_ms for r in buf)
    p95 = latencies[min(n - 1, int(n * 0.95))] if n else 0

    # Dynamic Threshold Calibration
    delta = q_ema - _BASE_THRESHOLD
    new_thr = _BASE_THRESHOLD + 0.5 * delta
    new_thr = max(_MIN_THRESHOLD, min(_MAX_THRESHOLD, new_thr))

    # Dynamic Cost Weight Modulation based on SLA Latency Pressure
    latency_pressure = p95 / _TARGET_LATENCY_MS
    cost_w = _BASE_COST_WEIGHT * (1.0 - min(0.5, latency_pressure * 0.3))
    cost_w = max(_MIN_COST_WEIGHT, min(_MAX_COST_WEIGHT, cost_w))

    return DynamicParams(
        model_label=key[0],
        task_type=key[1],
        quality_threshold=round(new_thr, 4),
        cost_weight=round(cost_w, 4),
        quality_ema=round(q_ema, 3),
        p95_latency_ms=p95
    )

Integrated Dynamic Execution Loop

The router.py module consumes these live-tuned parameters directly during the routing cycle. It evaluates Candidate models based on the current cost weights and quality thresholds before calling the failover resolver:

"""
router.py  —  routing-v4
=========================
Core routing logic enhanced with:
  · Dynamic quality thresholds from the telemetry engine
  · Failover-aware model resolution via the failover manager
  · Per-task utility function  U = β(t)·Q − α(t)·C
    where α and β are continuously adjusted per (model, task_type) pair
"""

from __future__ import annotations

from src.cost_registry import MODEL_REGISTRY, MODEL_BY_LABEL, TASK_FAILOVER_PRIORITY, ModelSpec
from src.task_classifier import TaskProfile
from src.failover_manager import failover_manager, FailoverEvent
from src.telemetry import telemetry_engine

# ── Static fallback parameters (used before telemetry has data) ───────────────
BASE_QUALITY_THRESHOLD = 0.72
BASE_COST_WEIGHT       = 0.40
BASE_QUALITY_WEIGHT    = 0.60

STATIC_TASK_THRESHOLDS: dict[str, float] = {
    "documentation":   0.60,
    "summarisation":   0.55,
    "qa_simple":       0.55,
    "code_generation": 0.72,
    "code_review":     0.85,
    "reasoning":       0.80,
}

SECURITY_OVERRIDES: set[str] = {"code_review"}


# ── Utility function ──────────────────────────────────────────────────────────

def _utility(model: ModelSpec, profile: TaskProfile) -> float:
    """
    U = β·Q − α·C   with dynamic α and β from the telemetry engine.
    """
    alpha = telemetry_engine.get_cost_weight(model.label, profile.task_type.value)
    beta  = 1.0 - alpha
    q     = model.quality_score(profile.complexity)
    c     = model.cost_per_1k
    return beta * q - alpha * c


def _effective_threshold(model_label: str, task_type: str) -> float:
    """
    Blend the static task-level threshold with the live telemetry threshold.
    The dynamic value wins once there are enough samples.
    """
    static_thr  = STATIC_TASK_THRESHOLDS.get(task_type, BASE_QUALITY_THRESHOLD)
    dynamic_thr = telemetry_engine.get_threshold(model_label, task_type)
    return round((static_thr + dynamic_thr) / 2.0, 4)


# ── Primary routing ───────────────────────────────────────────────────────────

def route(profile: TaskProfile) -> tuple[ModelSpec, FailoverEvent | None]:
    """
    Return (selected_model, failover_event_or_None) for the given TaskProfile.

    Selection rules:
      1. Security overrides → always tier-4 (then failover if circuit open)
      2. Filter by capability + context window
      3. Filter by blended quality threshold
      4. Best utility wins
      5. Safety net: highest quality model if no candidate passes threshold
      6. Resolve failover if the chosen model's circuit is not callable
    """
    ttype = profile.task_type.value

    if ttype in SECURITY_OVERRIDES:
        primary = next(m for m in MODEL_REGISTRY if m.tier == 4)
        return failover_manager.resolve(primary, ttype)

    threshold = _effective_threshold("", ttype)  # task-level threshold (no model prefix)

    candidates = [
        m for m in MODEL_REGISTRY
        if ttype in m.capable_types
        and m.max_tokens >= profile.token_estimate
        and m.quality_score(profile.complexity) >= threshold
    ]

    if not candidates:
        candidates = sorted(
            MODEL_REGISTRY,
            key=lambda m: m.quality_score(profile.complexity),
            reverse=True,
        )

    primary = max(candidates, key=lambda m: _utility(m, profile))
    return failover_manager.resolve(primary, ttype)


def dispatch(
    profiles: list[TaskProfile],
) -> dict[str, tuple[TaskProfile, ModelSpec, FailoverEvent | None]]:
    """
    Route each sub-task independently.
    Returns mapping of unique_key → (profile, selected_model, failover_event).
    """
    result: dict[str, tuple[TaskProfile, ModelSpec, FailoverEvent | None]] = {}
    type_counts: dict[str, int] = {}

    for profile in profiles:
        key   = profile.task_type.value
        count = type_counts.get(key, 0)
        type_counts[key] = count + 1
        unique_key = key if count == 0 else f"{key}_{count}"
        model, failover_event = route(profile)
        result[unique_key] = (profile, model, failover_event)

    return result

Verifying the Flow Execution with Examples

To test the end-to-end orchestration pipeline, copy .env.example to configuration state, start your local Ollama nodes, and run the terminal test scenario:

# Launch the pre-configured demonstration script
./scripts/demo.sh

This demo triggers a complex compound request: “Write the API reference documentation AND implement the OAuth2 middleware”.

The system fragments the string into its constituent clauses, routes the documentation task to a light, high-efficiency instance (e.g., granite4.1:3b), and dynamically assigns the complex security implementation to a heavy tier (e.g., mistral-small3.2).

Conclusion

By coupling LiteLLM’s multi-provider connection layer with a self-correcting Circuit Breaker and a live Telemetry Engine, this design proves that multi-model orchestration is highly viable for production.

The architecture guarantees high availability while continuously minimizing token costs, demonstrating how automated routing workflows can maximize efficiency in the software development lifecycle.

Up Next — Part 5: Breaking the Speed Barrier with Go 🏎️

While Python’s asynchronous event loop handles our v4 dual-layer routing beautifully at moderate scale, production systems tracking hundreds of live telemetry ring buffers and real-time streaming updates face a common bottleneck: runtime synchronization overhead.

In the next part of this series, we take this exact architectural blueprint and port it entirely to Go. We will explore how replacing single-threaded event loops with native goroutines and lock-free thread-safe primitives allows us to scale routing throughput, lower memory footprints, and achieve predictable, bare-metal efficiency. Stay tuned for the code drop!

🎯 That’s a wrap 💯 for part 4 and thanks for reading!

Stay tuned for the next episode… 📺

The ROI of SDLC: How Bob Orchestration of LLMs Makes Financially Sense (Part 3)-Showtime

Alain Airom (Ayrom) — Fri, 03 Jul 2026 09:50:35 +0000

Part 3: A theorical implementation of LLM selection and orchestration through a UI

Introduction-Bringing the Logic to Life by Integrating Our Custom Routing Engine

In the first installment of this series, we explored the core philosophy of intelligent, multi-LLM orchestration — unpacking the “T-Shirt size” optimization strategy that keeps cost-per-token at an absolute minimum without sacrificing performance. In Episode 2, we moved from theory to architecture, building a custom, plug-and-play routing engine from scratch using LiteLLM to establish a unified abstraction layer.

Now, it’s time to bring these pieces together in our own workspace.

In this third episode, we are diving into the practical integration phase. We will walk through how to plug this self-implemented routing layer directly into a custom orchestration workflow — simulating how a system like Bob evaluates live tasks, handles model failovers, and automatically routes every code generation or documentation request to its most cost-effective algorithmic match in real time.

> 🚨 Disclaimer: This article is for educational purposes only. It provides a conceptual, hands-on look at how LLM routing logic operates and should not be mistaken for the official enterprise implementation or architecture of the IBM Bob product.

Multi-LLM Routing — Part 3: (Almost) Final Implementation

Before diving into code, let’s look at how a raw, compound prompt travels through our custom system. Rather than passing the entire prompt blindly to a single heavy LLM, our integration pipeline enforces a strict four-stage flow:

┌─────────────────────────────────────────────────────────────────┐
│  Sidebar                │  Main                                  │
│  ─────────────────────  │  ────────────────────────────────────  │
│  Model Registry         │  Top bar (title + dry-run toggle)      │
│   T1 IBM Granite        │                                        │
│   T2 Meta LLaMA         │  Chat Thread                           │
│   T3 Google Gemma       │   ┌─ User bubble ──────────────────┐   │
│   T4 Mistral AI         │   │  "Write API docs AND OAuth2"   │   │
│                         │   └────────────────────────────────┘   │
│  Formula                │   ┌─ Routing Card ─────────────────┐   │
│  U = 0.60·Q − 0.40·C   │   │  🔀 2 sub-tasks detected       │    │
│                         │   │  ┌ documentation  T1 Granite ┐ │   │
│  Session Stats          │   │  │ complexity 26% Q 64% U+0.36│ │  │
│   Total calls           │   │  └────────────────────────────┘ │  │
│   Avg quality           │   │  ┌ code_generation T3 Gemma  ┐ │   │
│   Total cost            │   │  │ complexity 61% Q 85% U+0.37│ │  │
│                         │   │  └────────────────────────────┘ │  │
│                         │   └────────────────────────────────┘   │
│                         │   ┌─ Response Card ─────────────────┐  │
│                         │   │ [T1 documentation][T3 code][⊞]  │  │
│                         │   │  📡 IBM Granite  ⏱ 312ms  📝 214│ │
│                         │   │  <rendered markdown response>   │  │
│                         │   └────────────────────────────────┘   │
│                         │                                        │
│                         │  Input area (textarea + send button)   │
└─────────────────────────┴────────────────────────────────────────┘

Split & Classify (splitter.py + task_classifier.py): The prompt is tokenized and dissected into atomic sub-tasks, each stamped with a specific type and evaluated for complexity (0.0 to 1.0).
Dispatch & Route (router.py): For each sub-task, the router matches capabilities and applies the utility function U=β⋅Q−α⋅C to pick the optimal tier.
Parallel Execution & Abstraction (orchestrator.py + llm_client.py): Sub-tasks are dispatched simultaneously using asynchronous concurrency, communicating via LiteLLM to local Ollama nodes.
Aggregation & Feedback Storage (feedback.py): The final answers are merged into a unified markdown response, and execution metadata (latency, tokens, costs) is captured in an append-only analytics ledger.

Technical Synthesis: Key Code Implementations

Let’s look at the actual code mechanisms making this cross-model coordination possible.

The Evaluation Core: Splitting & Routing

When a compound request comes in (e.g., “Write a detailed API documentation section and code a secure OAuth2 decorator”), it must be handled independently. The Orchestrator triggers the classification pipeline to build a routing table before executing any model calls.

"""
orchestrator.py  —  enriched for the Chat UI
Ties together splitting, routing, parallel execution, and aggregation.
Returns full routing metadata (utility score, quality score, ollama model)
so the UI can display the decision rationale alongside each response.
"""

from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass, field

from src.splitter import split_and_classify
from src.router import dispatch, _utility, TASK_QUALITY_THRESHOLDS, QUALITY_THRESHOLD
from src.llm_client import call_model, TIER_TO_MODEL
from src.feedback import FeedbackStore
from src.task_classifier import TaskProfile
from src.cost_registry import ModelSpec


@dataclass
class SubTaskResult:
    profile:       TaskProfile
    model:         ModelSpec
    response:      str
    latency_ms:    int
    out_tokens:    int
    # ── routing metadata exposed to the UI ──────────────────────────────────
    utility_score: float        # U = β·Q − α·C
    quality_score: float        # model.quality_score(complexity)
    threshold:     float        # per-task quality threshold applied
    ollama_model:  str          # concrete ollama model name used
    cost_units:    float        # normalised cost for this call


@dataclass
class OrchestratorResult:
    prompt:       str
    sub_results:  list[SubTaskResult]
    merged:       str
    total_ms:     int

    def routing_table(self) -> list[dict]:
        return [
            {
                "task_type":    r.profile.task_type.value,
                "complexity":   r.profile.complexity,
                "tier":         r.profile.tier.name,
                "model":        r.model.label,
                "provider":     r.model.provider,
                "ollama_model": r.ollama_model,
                "utility":      round(r.utility_score, 4),
                "quality":      round(r.quality_score, 4),
                "threshold":    r.threshold,
                "latency_ms":   r.latency_ms,
                "out_tokens":   r.out_tokens,
                "cost_units":   r.cost_units,
                "response":     r.response,
            }
            for r in self.sub_results
        ]

    def to_ui_payload(self) -> dict:
        """Full payload for the chat UI — includes per-subtask routing details."""
        return {
            "prompt":      self.prompt,
            "total_ms":    self.total_ms,
            "merged":      self.merged,
            "sub_tasks":   self.routing_table(),
        }


class Orchestrator:
    def __init__(self, feedback_store: FeedbackStore | None = None) -> None:
        self._store = feedback_store or FeedbackStore()

    async def handle(self, prompt: str) -> OrchestratorResult:
        """
        Full pipeline:
          1. Split + classify
          2. Route via utility function U = β·Q − α·C
          3. Execute all sub-tasks in parallel via LiteLLM Router
          4. Aggregate + record
        """
        t_start = time.perf_counter()
        profiles = split_and_classify(prompt)
        routing  = dispatch(profiles)

        async def _exec(key: str, profile: TaskProfile, model: ModelSpec) -> SubTaskResult:
            response, latency, tokens = await call_model(
                model_label=model.label,
                prompt=profile.raw_text,
                max_tokens=min(2048, profile.token_estimate * 3),
            )
            lat_ms      = int(latency * 1000)
            u           = _utility(model, profile)
            q           = model.quality_score(profile.complexity)
            thr         = TASK_QUALITY_THRESHOLDS.get(profile.task_type.value, QUALITY_THRESHOLD)
            ollama_name = TIER_TO_MODEL.get(model.label, model.label)
            cost        = round(model.cost_per_1k * tokens / 1000, 6)

            self._store.record(
                task_type=profile.task_type.value,
                complexity=profile.complexity,
                tier=model.tier,
                model_used=model.label,
                latency_ms=lat_ms,
                output_tokens=tokens,
                cost_per_1k=model.cost_per_1k,
                prompt_len=len(profile.raw_text),
            )
            return SubTaskResult(
                profile=profile,
                model=model,
                response=response,
                latency_ms=lat_ms,
                out_tokens=tokens,
                utility_score=round(u, 4),
                quality_score=round(q, 4),
                threshold=thr,
                ollama_model=ollama_name,
                cost_units=cost,
            )

        sub_results = list(await asyncio.gather(
            *[_exec(k, p, m) for k, (p, m) in routing.items()]
        ))

        merged   = self._merge(sub_results)
        total_ms = int((time.perf_counter() - t_start) * 1000)

        return OrchestratorResult(
            prompt=prompt,
            sub_results=sub_results,
            merged=merged,
            total_ms=total_ms,
        )

    def handle_sync(self, prompt: str) -> OrchestratorResult:
        return asyncio.run(self.handle(prompt))

    @staticmethod
    def _merge(results: list[SubTaskResult]) -> str:
        sections = []
        for r in results:
            label  = r.profile.task_type.value.replace("_", " ").title()
            header = (
                f"## {label}  "
                f"_(via {r.model.label} [{r.model.provider}], {r.latency_ms} ms)_"
            )
            sections.append(f"{header}\n\n{r.response.strip()}")
        return "\n\n---\n\n".join(sections)

The dispatch routine in router.py picks the model using our utility metric. For example, a code task requires a higher threshold (0.72) than prose documentation (0.60), forcing a higher tier if complexity spikes:

"""
router.py
Core routing logic: selects the cheapest model that meets the quality threshold
for each TaskProfile, using a utility function U = β·Q − α·C.
"""

from __future__ import annotations
from src.cost_registry import MODEL_REGISTRY, ModelSpec
from src.task_classifier import TaskProfile

# ── Tunable parameters ────────────────────────────────────────────────────────
QUALITY_THRESHOLD = 0.72   # default minimum acceptable quality score
COST_WEIGHT       = 0.40   # α — penalise cost
QUALITY_WEIGHT    = 0.60   # β — reward quality

# Per-task-type quality thresholds.
# Documentation / summaries tolerate lower quality from a cheap model;
# code and reasoning tasks require a higher bar.
TASK_QUALITY_THRESHOLDS: dict[str, float] = {
    "documentation":   0.60,   # light model acceptable for prose
    "summarisation":   0.55,   # even more lenient for summaries
    "qa_simple":       0.55,
    "code_generation": 0.72,
    "code_review":     0.85,   # security: high bar
    "reasoning":       0.80,
}

# Task types that require the heaviest tier (Tier 4) regardless of complexity score.
# code_review always goes to Mistral (security-aware, largest context)
SECURITY_OVERRIDES: set[str] = {"code_review"}


def _utility(model: ModelSpec, profile: TaskProfile) -> float:
    """
    Utility  U = β·Q − α·C
      Q = model.quality_score(complexity)   ∈ [0, 1]
      C = model.cost_per_1k (normalised)    ∈ [0, 1]
    A higher U means the model is preferred.
    """
    q = model.quality_score(profile.complexity)
    c = model.cost_per_1k
    return QUALITY_WEIGHT * q - COST_WEIGHT * c


def route(profile: TaskProfile) -> ModelSpec:
    """
    Return the optimal ModelSpec for the given TaskProfile.

    Selection rules (in order):
      1. Model must support the task type.
      2. Model context window must fit the estimated token budget.
      3. Model quality_score must meet QUALITY_THRESHOLD
         (unless it's a security override — then tier 4 is forced).
      4. Among qualifying candidates, the one with the highest utility wins.
      5. If no candidate passes the threshold, fall back to the highest-quality model.
    """
    ttype = profile.task_type.value

    # Force tier-4 (heaviest) for security-sensitive task types
    if ttype in SECURITY_OVERRIDES:
        heavy = next(m for m in MODEL_REGISTRY if m.tier == 4)
        return heavy

    threshold = TASK_QUALITY_THRESHOLDS.get(ttype, QUALITY_THRESHOLD)

    candidates = [
        m for m in MODEL_REGISTRY
        if ttype in m.capable_types
        and m.max_tokens >= profile.token_estimate
        and m.quality_score(profile.complexity) >= threshold
    ]

    if not candidates:
        # Safety net: pick highest-quality model regardless of cost
        candidates = sorted(
            MODEL_REGISTRY,
            key=lambda m: m.quality_score(profile.complexity),
            reverse=True,
        )

    return max(candidates, key=lambda m: _utility(m, profile))


def dispatch(profiles: list[TaskProfile]) -> dict[str, tuple[TaskProfile, ModelSpec]]:
    """
    Route each sub-task independently.

    Returns a mapping of task_type → (profile, selected_model).
    When multiple sub-tasks share the same type, a suffix is appended.
    """
    result: dict[str, tuple[TaskProfile, ModelSpec]] = {}
    type_counts: dict[str, int] = {}

    for profile in profiles:
        key = profile.task_type.value
        count = type_counts.get(key, 0)
        type_counts[key] = count + 1
        unique_key = key if count == 0 else f"{key}_{count}"
        result[unique_key] = (profile, route(profile))

    return result

Concurrent Multi-LLM Execution via LiteLLM

Executing multiple sub-tasks sequentially kills the user experience. By wrapping our Router inside an asynchronous execution map using Python's asyncio.gather, we fire concurrent requests safely.

Here is how the asynchronous executor loops through the resolved routing map, hits llm_client.py (powered by litellm.Router), and tracks precise metrics:

# From src/orchestrator.py
async def handle(self, prompt: str) -> OrchestratorResult:
    t_start = time.perf_counter()
    profiles = split_and_classify(prompt)
    routing = dispatch(profiles)

    async def _exec(key: str, profile: TaskProfile, model: ModelSpec) -> SubTaskResult:
        # Resolve concrete Ollama mapping (e.g., model::light -> granite4.1:3b)
        ollama_name = TIER_TO_MODEL.get(model.label, model.label)
        thr = TASK_QUALITY_THRESHOLDS.get(profile.task_type.value, QUALITY_THRESHOLD)
        q = model.quality_score(profile.complexity)
        u = _utility(model, profile)

        # Execute call via LiteLLM gateway wrapper
        response, lat_sec, tokens = await call_model(
            model_label=model.label,
            prompt=profile.raw_text,
            max_tokens=model.max_tokens
        )

        lat_ms = int(lat_sec * 1000)
        cost = model.cost_per_1k * tokens / 1000.0

        # Persistent audit feedback trail
        self.feedback_store.record(
            task_type=profile.task_type.value,
            complexity=profile.complexity,
            tier=model.tier,
            model_used=ollama_name,
            latency_ms=lat_ms,
            cost_units=cost,
            output_tokens=tokens,
            prompt_len=len(profile.raw_text),
        )

        return SubTaskResult(
            profile=profile, model=model, response=response,
            latency_ms=lat_ms, out_tokens=tokens, utility_score=round(u, 4),
            quality_score=round(q, 4), threshold=thr, ollama_model=ollama_name, cost_units=cost
        )

    # Fire concurrent async execution workers across all models
    sub_results = list(await asyncio.gather(
        *[_exec(k, p, m) for k, (p, m) in routing.items()]
    ))

    merged = self._merge(sub_results)
    total_ms = int((time.perf_counter() - t_start) * 1000)

    return OrchestratorResult(prompt=prompt, sub_results=sub_results, merged=merged, total_ms=total_ms)

Persistent Telemetry and Analytics Tracking

To verify that our routing architecture actually saves money or routes accurately over time, every transaction leaves a signature trace in an append-only JSONL feedback vault.

"""
feedback.py
Records routing decisions and outcome signals.
Used to surface per-tier quality statistics and to guide future tuning.
"""

from __future__ import annotations
import json
import os
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path


@dataclass
class RoutingRecord:
    timestamp:    str    # ISO-8601
    task_type:    str
    complexity:   float
    tier:         int
    model_used:   str
    latency_ms:   int
    cost_units:   float   # cost_per_1k × output_tokens / 1000
    quality_eval: float   # auto-eval heuristic: 0.0 – 1.0
    prompt_len:   int     # character count of original prompt


class FeedbackStore:
    """Append-only JSONL store for routing records."""

    def __init__(self, path: str = "output/feedback.jsonl") -> None:
        self._path = Path(path)
        self._path.parent.mkdir(parents=True, exist_ok=True)
        self._records: list[RoutingRecord] = self._load()

    # ── Persistence ───────────────────────────────────────────────────────────

    def _load(self) -> list[RoutingRecord]:
        if not self._path.exists():
            return []
        records = []
        with self._path.open() as fh:
            for line in fh:
                line = line.strip()
                if line:
                    records.append(RoutingRecord(**json.loads(line)))
        return records

    def record(
        self,
        task_type:    str,
        complexity:   float,
        tier:         int,
        model_used:   str,
        latency_ms:   int,
        output_tokens: int,
        cost_per_1k:  float,
        prompt_len:   int,
    ) -> None:
        quality_eval = self._heuristic_quality(latency_ms, output_tokens)
        cost_units   = cost_per_1k * output_tokens / 1000.0

        rec = RoutingRecord(
            timestamp=datetime.now(timezone.utc).isoformat(),
            task_type=task_type,
            complexity=round(complexity, 3),
            tier=tier,
            model_used=model_used,
            latency_ms=latency_ms,
            cost_units=round(cost_units, 6),
            quality_eval=round(quality_eval, 3),
            prompt_len=prompt_len,
        )
        self._records.append(rec)
        with self._path.open("a") as fh:
            fh.write(json.dumps(asdict(rec)) + "\n")

    # ── Analytics ─────────────────────────────────────────────────────────────

    @staticmethod
    def _heuristic_quality(latency_ms: int, output_tokens: int) -> float:
        """
        Rough auto-eval: reward high token output at low latency.
        Real systems replace this with a judge-LLM or human rating.
        """
        if output_tokens == 0:
            return 0.0
        tokens_per_second = output_tokens / max(1, latency_ms / 1000)
        return min(1.0, 0.5 + tokens_per_second / 60)

    def summary(self) -> dict:
        if not self._records:
            return {"total_calls": 0}
        total_cost = sum(r.cost_units for r in self._records)
        avg_quality = sum(r.quality_eval for r in self._records) / len(self._records)
        by_tier: dict[int, int] = {}
        for r in self._records:
            by_tier[r.tier] = by_tier.get(r.tier, 0) + 1
        return {
            "total_calls":  len(self._records),
            "total_cost":   round(total_cost, 6),
            "avg_quality":  round(avg_quality, 3),
            "calls_by_tier": by_tier,
        }

Bringing It All Together: Exposing the Pipeline to an Interface

With our orchestration core wrapped up, exposing it requires an lightweight API layer. Using FastAPI, we can expose the entire orchestration run payload directly to a client web interface.

#!/usr/bin/env python3
"""
app.py  —  LLM Routing Chat UI
FastAPI backend that serves the single-page chat interface and exposes:
  GET  /              → chat UI (static/index.html)
  POST /api/chat      → route prompt, call LLMs, return enriched JSON
  GET  /api/models    → model registry (tiers, providers, ollama names)
  GET  /api/stats     → routing feedback statistics
  GET  /health        → liveness probe
"""
from __future__ import annotations
import os
import sys
from pathlib import Path

from dotenv import load_dotenv
load_dotenv()

import uvicorn
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

sys.path.insert(0, str(Path(__file__).parent))

from src.orchestrator import Orchestrator
from src.feedback import FeedbackStore
from src.cost_registry import MODEL_REGISTRY
from src.llm_client import TIER_TO_MODEL

# ── App ───────────────────────────────────────────────────────────────────────
app = FastAPI(
    title="LLM Routing Chat",
    description="Chat UI that demonstrates intelligent LLM routing via LiteLLM → Ollama",
    version="1.0.0",
    docs_url="/api/docs",
)

feedback_store = FeedbackStore(path="output/feedback.jsonl")
orchestrator   = Orchestrator(feedback_store=feedback_store)

# ── Serve the UI ──────────────────────────────────────────────────────────────
STATIC_DIR = Path(__file__).parent / "static"

@app.get("/", response_class=HTMLResponse, include_in_schema=False)
async def serve_ui():
    html = (STATIC_DIR / "index.html").read_text()
    return HTMLResponse(content=html)


# ── Request / Response models ─────────────────────────────────────────────────
class ChatRequest(BaseModel):
    prompt: str
    dry_run: bool = False


# ── API endpoints ─────────────────────────────────────────────────────────────

@app.post("/api/chat")
async def chat(req: ChatRequest):
    """
    Route a prompt through the LLM pipeline and return the full response
    with routing metadata for the UI.
    """
    if not req.prompt.strip():
        raise HTTPException(status_code=400, detail="Prompt must not be empty.")

    if req.dry_run:
        from src.splitter import split_and_classify
        from src.router import dispatch, _utility, TASK_QUALITY_THRESHOLDS, QUALITY_THRESHOLD
        profiles = split_and_classify(req.prompt)
        routing  = dispatch(profiles)
        sub_tasks = []
        for key, (p, m) in routing.items():
            sub_tasks.append({
                "task_type":    p.task_type.value,
                "complexity":   p.complexity,
                "tier":         p.tier.name,
                "model":        m.label,
                "provider":     m.provider,
                "ollama_model": TIER_TO_MODEL.get(m.label, m.label),
                "utility":      round(_utility(m, p), 4),
                "quality":      round(m.quality_score(p.complexity), 4),
                "threshold":    TASK_QUALITY_THRESHOLDS.get(p.task_type.value, QUALITY_THRESHOLD),
                "latency_ms":   None,
                "out_tokens":   None,
                "cost_units":   None,
                "response":     None,
            })
        return JSONResponse(content={
            "prompt":    req.prompt,
            "total_ms":  None,
            "merged":    None,
            "sub_tasks": sub_tasks,
            "dry_run":   True,
        })

    try:
        result = await orchestrator.handle(req.prompt)
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc

    payload = result.to_ui_payload()
    payload["dry_run"] = False
    return JSONResponse(content=payload)


@app.get("/api/models")
async def get_models():
    """Return the full model registry for the UI sidebar."""
    return JSONResponse(content=[
        {
            "label":        m.label,
            "tier":         m.tier,
            "provider":     m.provider,
            "ollama_model": TIER_TO_MODEL.get(m.label, "?"),
            "cost_per_1k":  m.cost_per_1k,
            "max_tokens":   m.max_tokens,
            "capable_types": list(m.capable_types),
            "quality_at_low":  round(m.quality_score(0.1), 3),
            "quality_at_high": round(m.quality_score(0.9), 3),
        }
        for m in MODEL_REGISTRY
    ])


@app.get("/api/stats")
async def get_stats():
    """Return aggregated routing statistics."""
    return JSONResponse(content=feedback_store.summary())


@app.get("/health")
async def health():
    return {"status": "ok", "gateway": "LiteLLM Router → Ollama"}


# ── Entry point ───────────────────────────────────────────────────────────────
if __name__ == "__main__":
    import argparse
    p = argparse.ArgumentParser(description="LLM Routing Chat UI")
    p.add_argument("--host", default="0.0.0.0")
    p.add_argument("--port", type=int, default=8080)
    args = p.parse_args()
    uvicorn.run("app:app", host=args.host, port=args.port, reload=False)

The resulting UI payload returns not only the merged markdown string text, but also the dynamic sub_tasks list showing the specific complexity scores, thresholds, and utility equations that calculated the outcome.

Conclusion & What’s Next

By anchoring our tasks within an automated routing structure, we achieve a system that behaves like an intelligent scheduler. High-complexity code reviews default automatically to secure, large-context models (Tier 4), while routine explanations map directly to fast, resource-light foundational parameters (Tier 1).

We have achieved real-time cost-per-token optimization without compromising code production viability.

Coming up next in Part 4: We will explore how to configure advanced model failovers, optimize dynamic thresholds based on live telemetry, and build a front-end interface that renders these multi-LLM decisions visually in real time!

🎯 That’s a wrap 💯 for part 3 and thanks for reading!

Stay tuned for the next episode… 📺

The ROI of SDLC: How Bob Orchestration of LLMs Makes Financially Sense (Part 2) Using LiteLLM

Alain Airom (Ayrom) — Thu, 02 Jul 2026 07:29:55 +0000

Part 2: A theorical implementation of LLM selection and orchestration using the open source LiteLLM

Introduction

In the first part of this series, we explored the theoretical foundation of intelligent orchestration (implemented by Bob :) ) — demonstrating how to dynamically route tasks to the most performant and cost-effective Large Language Model (LLM) based on context and complexity, and why this precise capability stands out as one of the definitive differentiators of IBM Bob. Now, it’s time to move from theory to production-grade architecture. In this second installment, we will implement this exact cost-routing logic using a powerful, production-ready open-source solution: LiteLLM with a unified abstraction layer, making a routing engine completely plug-and-play.

> 🚨 Disclaimer: This article is for educational purposes only. It provides a conceptual, hands-on look at how LLM routing logic (here with LiteLLM) operates and should not be mistaken for the official enterprise implementation or architecture of the IBM Bob product.

TL;DR-What is LiteLLM?

Image from LiteLLM Site

Excerpt of LiteLLM Github Repository
LiteLLM is an open source AI Gateway that gives you a single, unified interface to call 100+ LLM providers — OpenAI, Anthropic, Gemini, Bedrock, Azure, and more — using the OpenAI format.
Use it as a Python SDK for direct library integration, or deploy the AI Gateway (Proxy Server) as a centralized service for your team or organization.
Why LiteLLM
Managing LLM calls across providers gets complicated fast — different SDKs, auth patterns, request formats, and error types for every model. LiteLLM removes that friction:
Unified API — one interface for 100+ LLMs, no provider-specific SDK juggling
Drop-in OpenAI compatibility — swap providers without rewriting your code
Production-ready gateway — virtual keys, spend tracking, guardrails, load balancing, and an admin dashboard out of the box
8ms P95 latency at 1k RPS (benchmarks)

Multi-LLM Routing — Part 2: Implementation with LiteLLM

In the first part of this series, we saw the theoretical foundation of intelligent orchestration — demonstrating how to dynamically decompose a complex user prompt and route sub-tasks to the most performant and cost-effective model tier with an ad-hoc code.

In this post, we transition the concept of cost-routing pipeline into a scalable solution by integrating a powerful open-source solution: LiteLLM. By leveraging LiteLLM as a unified transport layer over our local Ollama instances, we decouple which tier is chosen from how the call is executed.

The Production-Grade Gateway Architecture

By integrating LiteLLM, the custom mathematical utility scoring while completely abstracting the underlying infrastructure transport remains. The orchestrator no longer addresses specific container endpoints directly; instead, it interfaces with a standardized, OpenAI-compatible abstraction layer.

Under this architecture, responsibility is split into two cleanly isolated layers:

The Custom Routing Pipeline (router.py): Handles the business intelligence. It scores task complexity and calculates which symbolic tier (model::light, model::medium, etc.) maximizes performance while minimizing token cost using our standard utility function:

The LiteLLM AI Gateway (llm_client.py): Handles transport resilience. It maps those symbolic tier aliases to concrete model weights and manages enterprise-grade networking features such as automatic retries, back-off parameters, fallback chains, and transient error cooling.

Let’s dig into that 🪏

Core Technical Implementation

Granular Task Decomposition

This part remains unchanged from the first version: to avoid processing unnecessary tokens through high-tier engines, the architecture relies on a specialized TaskSplitter. This component evaluates the raw user payload and breaks it down into discrete components so that documentation and code can travel along separate routing paths.

"""
splitter.py
Splits a compound prompt into independent clauses, then classifies each one.
"""

from __future__ import annotations
import re
from src.task_classifier import classify, TaskProfile


# Conjunctions that signal independent sub-tasks within a single prompt
_SPLIT_PATTERN = re.compile(
    r"(?i)\s*\b(and also|and then|additionally|plus|as well as|AND|also)\b\s*",
)


def split_and_classify(prompt: str) -> list[TaskProfile]:
    """
    1. Split the prompt on conjunction markers.
    2. Discard separator tokens.
    3. Classify each remaining clause.
    """
    raw_clauses = _SPLIT_PATTERN.split(prompt)

    # Keep only non-separator, non-empty parts
    clauses = [
        c.strip()
        for c in raw_clauses
        if c and not _SPLIT_PATTERN.match(c.strip()) and len(c.strip()) > 8
    ]

    if not clauses:
        # Fallback: treat the whole prompt as one task
        clauses = [prompt.strip()]

    return [classify(c) for c in clauses]

T-Shirt Sizing and Feature Classification

This logic still remains: once separated, each distinct sub-task passes through the TaskClassifier. The classifier looks for domain markers (e.g., security policies, coding constraints, or structural documentation formatting) and assigns a maximum required model complexity tier: SMALL, MEDIUM, or LARGE.

"""
task_classifier.py
Classifies a text prompt into a TaskProfile: type, complexity score, and token estimate.
"""

from __future__ import annotations
import math
import re
from dataclasses import dataclass, field
from enum import Enum


class TaskType(Enum):
    DOCUMENTATION   = "documentation"    # READMEs, API docs, inline comments
    CODE_GENERATION = "code_generation"  # new functions / modules / classes
    CODE_REVIEW     = "code_review"      # diff analysis, security audit
    REASONING       = "reasoning"        # architecture, design decisions
    QA_SIMPLE       = "qa_simple"        # factual lookups, definitions
    SUMMARISATION   = "summarisation"    # condense / shorten text


class ComplexityTier(Enum):
    LOW    = 1   # score 0.00 – 0.34
    MEDIUM = 2   # score 0.35 – 0.64
    HIGH   = 3   # score 0.65 – 1.00


@dataclass
class TaskProfile:
    raw_text:       str
    task_type:      TaskType
    complexity:     float          # 0.0 → 1.0
    tier:           ComplexityTier
    token_estimate: int
    sub_tasks:      list["TaskProfile"] = field(default_factory=list)

    def __repr__(self) -> str:
        return (
            f"TaskProfile(type={self.task_type.value!r}, "
            f"complexity={self.complexity}, tier={self.tier.name}, "
            f"tokens≈{self.token_estimate})"
        )


# ── Keyword signal table: (keywords, base_complexity_score) ───────────────────
_SIGNALS: dict[TaskType, tuple[list[str], float]] = {
    TaskType.DOCUMENTATION:   (
        ["write the api", "api reference", "write docs", "write documentation",
         "document", "readme", "docstring", "comment",
         "explain", "describe"], 0.20),
    TaskType.CODE_GENERATION: (
        ["implement", "create function", "write code", "build",
         "develop", "middleware", "endpoint", "class", "module"], 0.55),
    TaskType.CODE_REVIEW:     (
        ["review", "audit", "security", "vulnerability",
         "diff", "check", "analyse code", "analyze code"], 0.60),
    TaskType.REASONING:       (
        ["architecture", "design", "trade-off", "compare",
         "evaluate", "pros and cons", "should i", "best approach"], 0.80),
    TaskType.QA_SIMPLE:       (
        ["what is", "how does", "define", "tell me", "when was"], 0.10),
    TaskType.SUMMARISATION:   (
        ["summarise", "summarize", "tldr", "shorten",
         "brief", "condense", "overview"], 0.15),
}

# Multi-word phrases that force a specific TaskType regardless of keyword scoring.
_PHRASE_OVERRIDES: list[tuple[str, TaskType]] = [
    ("api reference",       TaskType.DOCUMENTATION),
    ("write the api",       TaskType.DOCUMENTATION),
    ("write docs",          TaskType.DOCUMENTATION),
    ("write documentation", TaskType.DOCUMENTATION),
    ("code review",         TaskType.CODE_REVIEW),
    ("security audit",      TaskType.CODE_REVIEW),
]


def _token_estimate(text: str) -> int:
    """Rough GPT-style tokenisation: ~4 chars per token."""
    return max(1, len(text) // 4)


def _depth_penalty(text: str) -> float:
    """
    Extra score for multi-step / conditional reasoning markers.
    Each hit adds 0.04, capped at 0.20.
    """
    markers = [
        "step 1", "first,", "then,", "finally,",
        "however,", "alternatively,", "trade-off",
        "on the other hand", "in addition", "furthermore",
    ]
    lower = text.lower()
    hits = sum(1 for m in markers if m in lower)
    return min(0.20, hits * 0.04)


def classify(text: str) -> TaskProfile:
    """Return a TaskProfile for the given text snippet."""
    best_type, base_score = TaskType.QA_SIMPLE, 0.10
    lower = text.lower()

    # Phase 1: phrase-level overrides (highest priority)
    for phrase, forced_type in _PHRASE_OVERRIDES:
        if phrase in lower:
            best_type = forced_type
            base_score = dict(_SIGNALS)[forced_type][1]
            break
    else:
        # Phase 2: single-keyword scoring (only if no phrase matched)
        for ttype, (keywords, score) in _SIGNALS.items():
            if any(k in lower for k in keywords):
                if score > base_score:
                    best_type, base_score = ttype, score

    tokens       = _token_estimate(text)
    token_factor = min(0.15, math.log10(max(1, tokens)) * 0.05)
    depth        = _depth_penalty(text)
    final_score  = min(1.0, base_score + token_factor + depth)

    tier = (ComplexityTier.LOW    if final_score < 0.35 else
            ComplexityTier.MEDIUM if final_score < 0.65 else
            ComplexityTier.HIGH)

    return TaskProfile(
        raw_text=text,
        task_type=best_type,
        complexity=round(final_score, 3),
        tier=tier,
        token_estimate=tokens,
    )

Unified Client Abstraction via `litellm.Router`

Rather than spawning raw, unmonitored HTTP connections to different endpoints, all asynchronous I/O travels through a centralized litellm.Router instance. The initializion of the router is done by passing a structured configuration list, assigning our custom symbolic tiers directly to our local Ollama models.

# From src/llm_client.py
async def call_model(
    model_label: str,
    prompt: str,
    max_tokens: int = 1024,
) -> tuple[str, float, int]:
    """
    Call the LiteLLM AI Gateway Router for the given symbolic model label.
    Normalizes response formats, handles retries, and tracks generation token metrics.
    """
    messages = [{"role": "user", "content": prompt}]

    t0 = time.perf_counter()
    try:
        response = await _router.acompletion(
            model=model_label,
            messages=messages,
            max_tokens=max_tokens,
            temperature=0.3,
        )
        latency = time.perf_counter() - t0

        text   = response.choices[0].message.content or ""
        tokens = (
            response.usage.completion_tokens
            if response.usage
            else max(1, len(text) // 4)
        )
        return text, latency, tokens

    except Exception as exc:
        latency = time.perf_counter() - t0
        real_model = _TIER_MODELS.get(model_label, model_label)
        return (
            f"[LiteLLM ERROR — {real_model}: {exc}]",
            latency,
            0,
        )

High-Resilience Execution Block

When a task is scheduled for execution, the pipeline invokes a uniform acompletion() call. LiteLLM dynamically normalizes response parsing across different endpoints, allowing us to safely extract generation data using a single, unified interface regardless of the model provider.

# From src/llm_client.py
async def call_model(
    model_label: str,
    prompt: str,
    max_tokens: int = 1024,
) -> tuple[str, float, int]:
    """
    Call the LiteLLM AI Gateway Router for the given symbolic model label.
    Normalizes response formats, handles retries, and tracks generation token metrics.
    """
    messages = [{"role": "user", "content": prompt}]

    t0 = time.perf_counter()
    try:
        response = await _router.acompletion(
            model=model_label,
            messages=messages,
            max_tokens=max_tokens,
            temperature=0.3,
        )
        latency = time.perf_counter() - t0

        text   = response.choices[0].message.content or ""
        tokens = (
            response.usage.completion_tokens
            if response.usage
            else max(1, len(text) // 4)
        )
        return text, latency, tokens

    except Exception as exc:
        latency = time.perf_counter() - t0
        real_model = _TIER_MODELS.get(model_label, model_label)
        return (
            f"[LiteLLM ERROR — {real_model}: {exc}]",
            latency,
            0,
        )

Parallel Workload Orchestration

To prevent independent model generations from bottlenecking each other, the pipeline schedules execution asynchronously. The orchestrator uses asyncio.gather to trigger all routed sub-tasks concurrently across the LiteLLM transport layer, making the overall latency bound only by the slowest single engine in the pool.

"""
orchestrator.py
Ties together splitting, routing, parallel execution, and aggregation.
Uses the LiteLLM Router (via llm_client.py) for all LLM calls.
"""

from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass

from src.splitter import split_and_classify
from src.router import dispatch
from src.llm_client import call_model
from src.feedback import FeedbackStore
from src.task_classifier import TaskProfile
from src.cost_registry import ModelSpec


@dataclass
class SubTaskResult:
    profile:     TaskProfile
    model:       ModelSpec
    response:    str
    latency_ms:  int
    out_tokens:  int


@dataclass
class OrchestratorResult:
    prompt:       str
    sub_results:  list[SubTaskResult]
    merged:       str
    total_ms:     int

    def routing_table(self) -> list[dict]:
        return [
            {
                "task_type":   r.profile.task_type.value,
                "complexity":  r.profile.complexity,
                "tier":        r.profile.tier.name,
                "model":       r.model.label,
                "latency_ms":  r.latency_ms,
                "out_tokens":  r.out_tokens,
                "cost_units":  round(r.model.cost_per_1k * r.out_tokens / 1000, 6),
            }
            for r in self.sub_results
        ]


class Orchestrator:
    def __init__(self, feedback_store: FeedbackStore | None = None) -> None:
        self._store = feedback_store or FeedbackStore()

    # ── Core ─────────────────────────────────────────────────────────────────

    async def handle(self, prompt: str) -> OrchestratorResult:
        """
        Full pipeline:
          1. Split prompt into independent sub-tasks.
          2. Classify each sub-task (type + complexity).
          3. Route each sub-task to the optimal model tier via utility function.
          4. Execute all sub-tasks in parallel via LiteLLM Router.
          5. Aggregate responses.
          6. Record to feedback store.
        """
        t_start = time.perf_counter()

        # Step 1 + 2: split & classify
        profiles = split_and_classify(prompt)

        # Step 3: route (utility-based selection)
        routing = dispatch(profiles)   # key → (profile, model)

        # Step 4: parallel execution through LiteLLM Router
        async def _exec(
            key: str, profile: TaskProfile, model: ModelSpec
        ) -> SubTaskResult:
            response, latency, tokens = await call_model(
                model_label=model.label,
                prompt=profile.raw_text,
                max_tokens=min(2048, profile.token_estimate * 3),
            )
            lat_ms = int(latency * 1000)
            self._store.record(
                task_type=profile.task_type.value,
                complexity=profile.complexity,
                tier=model.tier,
                model_used=model.label,
                latency_ms=lat_ms,
                output_tokens=tokens,
                cost_per_1k=model.cost_per_1k,
                prompt_len=len(profile.raw_text),
            )
            return SubTaskResult(profile, model, response, lat_ms, tokens)

        sub_results = await asyncio.gather(
            *[_exec(k, p, m) for k, (p, m) in routing.items()]
        )

        # Step 5: aggregate
        merged   = self._merge(list(sub_results))
        total_ms = int((time.perf_counter() - t_start) * 1000)

        return OrchestratorResult(
            prompt=prompt,
            sub_results=list(sub_results),
            merged=merged,
            total_ms=total_ms,
        )

    def handle_sync(self, prompt: str) -> OrchestratorResult:
        return asyncio.run(self.handle(prompt))

    # ── Merge ────────────────────────────────────────────────────────────────

    @staticmethod
    def _merge(results: list[SubTaskResult]) -> str:
        sections = []
        for r in results:
            label = r.profile.task_type.value.replace("_", " ").title()
            header = (
                f"## {label}  "
                f"_(via {r.model.label} [{r.model.provider}], {r.latency_ms} ms)_"
            )
            sections.append(f"{header}\n\n{r.response.strip()}")
        return "\n\n---\n\n".join(sections)

Enhancing Enterprise Resilience

Moving execution mechanics to LiteLLM provides several critical operational advantages out of the box:

Transparent Fallback Chains: If a local Ollama container gets saturated or crashes, LiteLLM can immediately catch the exception and redirect the payload to a backup cloud endpoint (such as IBM watsonx or an external API provider) without changing a line of your application code.
Built-in Error Handling: It encapsulates transient issues such as HTTP timeouts or context boundary overruns, implementing an exponential back-off strategy rather than failing the client request immediately.
Uniform Observability Footprint: Because every sub-task traverses a single router gateway, logging input/output token usage, collecting latency metrics, and auditing security telemetry becomes completely centralized inside our FeedbackStore.

Compared to the plain Ollama version:

Running and benchmarking the solution

To evaluate the routing engine in real time, simply execute the provided terminal scenario;

./multi-llm-router-LiteLLM/scripts/demo.sh
[demo] Verifying dependencies (litellm, fastapi, uvicorn)…



┌──────────────────────────────────────────────────────────────┐
│  Multi-LLM Task Router — LiteLLM Edition  demo               │
│  Gateway: LiteLLM AI Router → Ollama (localhost:11434)       │
│                                                              │
│  Tip: set DRY_RUN=1 to inspect routing without LLM calls     │
└──────────────────────────────────────────────────────────────┘

================================================================================
  Multi-LLM Task Router  [LiteLLM Edition]  —  Demonstration Scenario
================================================================================

Prompt:
  Write the API reference documentation for our REST /users endpoints AND implement the OAuth2 Bearer-token middleware in Python FastAPI.

LiteLLM Gateway → Ollama at: http://localhost:11434

Model tier mapping (LiteLLM alias → Ollama model):
  model::light          (IBM Granite   )  →  ollama/granite4.1:3b
  model::medium         (Meta LLaMA    )  →  ollama/llama3.2:latest
  model::balanced       (Google Gemma  )  →  ollama/gemma3:4b
  model::heavy          (Mistral AI    )  →  ollama/mistral-small3.2:latest

Detected 2 sub-task(s):
  [1] TaskProfile(type='documentation', complexity=0.26, tier=LOW, tokens≈16)
  [2] TaskProfile(type='code_generation', complexity=0.609, tier=MEDIUM, tokens≈15)

Routing decisions:
  Task Type              Complexity  Tier       Label                Provider              U
  ------------------------------------------------------------------------------
  documentation               0.260  LOW        model::light         IBM Granite     +0.3616
  code_generation             0.609  MEDIUM     model::balanced      Google Gemma    +0.3718

Executing sub-tasks via LiteLLM → Ollama…  (set DRY_RUN=1 to skip)


================================================================================
  ROUTING TABLE  (via LiteLLM AI Gateway)
================================================================================
  Task Type              Label                Provider         Latency   Tokens        Cost
  ------------------------------------------------------------------------------
  documentation          model::light         IBM Granite       2720 ms       48    0.002400
  code_generation        model::balanced      Google Gemma      5595 ms       45    0.015750

  Total wall-clock: 5601 ms

======================================================================
  MERGED RESPONSE
======================================================================
## Documentation  _(via model::light [IBM Granite], 2720 ms)_

# API Reference: `/users` Endpoints

Our RESTful API provides a set of endpoints under the `/users` resource to manage user-related operations. Below is the detailed documentation for each endpoint, including request methods, expected input parameters,

---

## Code Generation  _(via model::balanced [Google Gemma], 5595 ms)_

`python
from fastapi import Depends, HTTPException, status
from fastapi.security import CredentialsTokenAuthentication
import jwt  # You'll need to install this: pip install PyJWT


async def authenticate(token: str

======================================================================
  FEEDBACK SUMMARY
======================================================================
{
  "total_calls": 2,
  "total_cost": 0.01815,
  "avg_quality": 0.714,
  "calls_by_tier": {
    "1": 1,
    "3": 1
  }
}

Conclusion

By wrapping our budget-conscious routing logic in LiteLLM, we have constructed an AI orchestration layer that balances economic efficiency with enterprise reliability. We no longer write custom code to handle network hiccups or provider differences; instead, we let our core pipeline focus entirely on making optimal routing decisions.

🎯 Et voilà, that’s a wrap 💯 for part 2 and thanks for reading!

Stay tuned for the next episode 📺

The ROI of SDLC: How Bob Orchestration of LLMs Makes Financially Sense (Part 1)

Alain Airom (Ayrom) — Wed, 01 Jul 2026 11:37:10 +0000

Part 1: A theorical implementation of LLM selection and orchestration

Introduction

One of the greatest strengths of Bob Orchestrator lies in its intelligent multi-LLM routing logic. Instead of relying on a single, costly model for everything, Bob dynamically selects the best tool for the specific task at hand. Need comprehensive documentation? Bob routes it to a model optimized for text. Need complex code generation? Bob leverages the exact LLM architecture best suited for that language or framework. By matching the task to the right LLM “T-Shirt size,” Bob automatically optimizes your cost-per-token, delivering maximum performance at the lowest possible price point.

In this opening installment of the series, I will implement and demonstrate how an orchestration scenario scaffolds a sample application built specifically to implement this multi-LLM routing logic.

To keep this walkthrough straightforward, reproducible, and cost-effective, I am running the orchestration entirely locally via Ollama. Below is the specific roster of local models — ranging from lightweight utility models to heavy-duty reasoning engines — that we will use to test an intelligent application’s dynamic routing decisions:

| Model Name                     | Image ID       | Size   |
| ------------------------------ | -------------- | ------ |
| **mistral-small3.2:latest**    | `5a408ab55df5` | 15 GB  |
| **ibm/granite3.3-guardian:8b** | `90a8aabc98eb` | 6.7 GB |
| **granite3.3:8b**              | `fd429f23b909` | 4.9 GB |
| **gemma3:4b**                  | `a2af6cc3eb7f` | 3.3 GB |
| **ibm/granite3.2-guardian:3b** | `4f7cdff62b7e` | 2.7 GB |
| **granite3-guardian:latest**   | `ba81a177bd23` | 2.7 GB |
| **granite4.1:3b**              | `6fd349357287` | 2.1 GB |
| **ibm/granite4:3b**            | `89962fcc7523` | 2.1 GB |
| **llama3.2:latest**            | `a80c4f17acd5` | 2.0 GB |
| **ibm/granite4:350m**          | `5eee845b49c4` | 708 MB |
| **granite-embedding:30m**      | `eb4c533ba6f7` | 62 MB  |

Architecture Overview: The Intelligent Cost-Routing Pipeline

At the heart of orchestrator’s cost-efficiency engine is a modular pipeline designed to intercept user intent, break down complex requirements, and match each sub-task to the most cost-effective LLM tier. Instead of blindly passing a massive multi-part prompt to a premium, high-cost reasoning model, an orchestrator acts as a smart traffic controller.

                  +------------------------+
                  |  User Complex Request  |
                  +-----------+------------+
                              |
                              v
                  +------------------------+
                  |    Task Splitter       |
                  +-----------+------------+
                              |
                    [Splits into Sub-Tasks]
                              |
                              v
                  +------------------------+
                  |    Task Classifier     |
                  +-----------+------------+
                              |
                    [Identifies Domain & Size]
                              |
                              v
                  +------------------------+
                  |   Dynamic LLM Router   |
                  +-----------+------------+
                              |
            [Queries Cost Registry & Selects Best Fit]
                              |
         +--------------------+--------------------+
         |                    |                    |
         v                    v                    v
+-----------------+  +-----------------+  +-----------------+
| Mistral (Small) |  | Granite (Guard) |  |  Llama (Utility)|
+-----------------+  +-----------------+  +-----------------+

The pipeline operates in four distinct stages:

Task Splitting: Parsing complex, compound requests into a sequence of isolated, clean sub-tasks.
Intent & Domain Classification: Scoring each sub-task to determine its exact domain (e.g., code, documentation, text transformation) and complexity requirement (“T-Shirt sizing”).
Registry-Driven Routing: Cross-referencing the classification payload against a centralized model tier and cost ledger to pick the absolute optimal local Ollama target.
Isolated Execution & Reassembly: Invoking the targeted models and stitching the final context back together flawlessly.

Technical Deep Dive & Core Implementations

Granular Task Decomposition

To avoid processing unnecessary tokens through high-tier engines, the architecture relies on a specialized TaskSplitter. This component evaluates the raw user payload and breaks it down into discrete components so that documentation and code can travel along separate routing paths.

"""
splitter.py
Splits a compound prompt into independent clauses, then classifies each one.
"""

from __future__ import annotations
import re
from src.task_classifier import classify, TaskProfile


# Conjunctions that signal independent sub-tasks within a single prompt
_SPLIT_PATTERN = re.compile(
    r"(?i)\s*\b(and also|and then|additionally|plus|as well as|AND|also)\b\s*",
)


def split_and_classify(prompt: str) -> list[TaskProfile]:
    """
    1. Split the prompt on conjunction markers.
    2. Discard separator tokens.
    3. Classify each remaining clause.
    """
    raw_clauses = _SPLIT_PATTERN.split(prompt)

    # Keep only non-separator, non-empty parts
    clauses = [
        c.strip()
        for c in raw_clauses
        if c and not _SPLIT_PATTERN.match(c.strip()) and len(c.strip()) > 8
    ]

    if not clauses:
        # Fallback: treat the whole prompt as one task
        clauses = [prompt.strip()]

    return [classify(c) for c in clauses]

T-Shirt Sizing and Feature Classification

Once separated, each distinct sub-task passes through the TaskClassifier. The classifier looks for domain markers (e.g., security policies, coding constraints, or structural documentation formatting) and assigns a maximum required model complexity tier: SMALL, MEDIUM, or LARGE.

"""
task_classifier.py
Classifies a text prompt into a TaskProfile: type, complexity score, and token estimate.
"""

from __future__ import annotations
import math
import re
from dataclasses import dataclass, field
from enum import Enum


class TaskType(Enum):
    DOCUMENTATION   = "documentation"    # READMEs, API docs, inline comments
    CODE_GENERATION = "code_generation"  # new functions / modules / classes
    CODE_REVIEW     = "code_review"      # diff analysis, security audit
    REASONING       = "reasoning"        # architecture, design decisions
    QA_SIMPLE       = "qa_simple"        # factual lookups, definitions
    SUMMARISATION   = "summarisation"    # condense / shorten text


class ComplexityTier(Enum):
    LOW    = 1   # score 0.00 – 0.34
    MEDIUM = 2   # score 0.35 – 0.64
    HIGH   = 3   # score 0.65 – 1.00


@dataclass
class TaskProfile:
    raw_text:       str
    task_type:      TaskType
    complexity:     float          # 0.0 → 1.0
    tier:           ComplexityTier
    token_estimate: int
    sub_tasks:      list["TaskProfile"] = field(default_factory=list)

    def __repr__(self) -> str:
        return (
            f"TaskProfile(type={self.task_type.value!r}, "
            f"complexity={self.complexity}, tier={self.tier.name}, "
            f"tokens≈{self.token_estimate})"
        )


# ── Keyword signal table: (keywords, base_complexity_score) ───────────────────
_SIGNALS: dict[TaskType, tuple[list[str], float]] = {
    TaskType.DOCUMENTATION:   (
        ["write the api", "api reference", "write docs", "write documentation",
         "document", "readme", "docstring", "comment",
         "explain", "describe"], 0.20),
    TaskType.CODE_GENERATION: (
        ["implement", "create function", "write code", "build",
         "develop", "middleware", "endpoint", "class", "module"], 0.55),
    TaskType.CODE_REVIEW:     (
        ["review", "audit", "security", "vulnerability",
         "diff", "check", "analyse code", "analyze code"], 0.60),
    TaskType.REASONING:       (
        ["architecture", "design", "trade-off", "compare",
         "evaluate", "pros and cons", "should i", "best approach"], 0.80),
    TaskType.QA_SIMPLE:       (
        ["what is", "how does", "define", "tell me", "when was"], 0.10),
    TaskType.SUMMARISATION:   (
        ["summarise", "summarize", "tldr", "shorten",
         "brief", "condense", "overview"], 0.15),
}

# Multi-word phrases that force a specific TaskType regardless of keyword scoring.
_PHRASE_OVERRIDES: list[tuple[str, TaskType]] = [
    ("api reference",       TaskType.DOCUMENTATION),
    ("write the api",       TaskType.DOCUMENTATION),
    ("write docs",          TaskType.DOCUMENTATION),
    ("write documentation", TaskType.DOCUMENTATION),
    ("code review",         TaskType.CODE_REVIEW),
    ("security audit",      TaskType.CODE_REVIEW),
]


def _token_estimate(text: str) -> int:
    """Rough GPT-style tokenisation: ~4 chars per token."""
    return max(1, len(text) // 4)


def _depth_penalty(text: str) -> float:
    """
    Extra score for multi-step / conditional reasoning markers.
    Each hit adds 0.04, capped at 0.20.
    """
    markers = [
        "step 1", "first,", "then,", "finally,",
        "however,", "alternatively,", "trade-off",
        "on the other hand", "in addition", "furthermore",
    ]
    lower = text.lower()
    hits = sum(1 for m in markers if m in lower)
    return min(0.20, hits * 0.04)


def classify(text: str) -> TaskProfile:
    """Return a TaskProfile for the given text snippet."""
    best_type, base_score = TaskType.QA_SIMPLE, 0.10
    lower = text.lower()

    # Phase 1: phrase-level overrides (highest priority)
    for phrase, forced_type in _PHRASE_OVERRIDES:
        if phrase in lower:
            best_type = forced_type
            base_score = dict(_SIGNALS)[forced_type][1]
            break
    else:
        # Phase 2: single-keyword scoring (only if no phrase matched)
        for ttype, (keywords, score) in _SIGNALS.items():
            if any(k in lower for k in keywords):
                if score > base_score:
                    best_type, base_score = ttype, score

    tokens       = _token_estimate(text)
    token_factor = min(0.15, math.log10(max(1, tokens)) * 0.05)
    depth        = _depth_penalty(text)
    final_score  = min(1.0, base_score + token_factor + depth)

    tier = (ComplexityTier.LOW    if final_score < 0.35 else
            ComplexityTier.MEDIUM if final_score < 0.65 else
            ComplexityTier.HIGH)

    return TaskProfile(
        raw_text=text,
        task_type=best_type,
        complexity=round(final_score, 3),
        tier=tier,
        token_estimate=tokens,
    )

Dynamic Cost Routing Mechanics

The LLMRouter pairs the classification metadata with a static CostRegistry. By calculating cost thresholds based on local resource sizes or token-equivalent costs, it isolates which Ollama container is running the model best suited for the job.

"""
router.py
Core routing logic: selects the cheapest model that meets the quality threshold
for each TaskProfile, using a utility function U = β·Q − α·C.
"""

from __future__ import annotations
from src.cost_registry import MODEL_REGISTRY, ModelSpec
from src.task_classifier import TaskProfile

# ── Tunable parameters ────────────────────────────────────────────────────────
QUALITY_THRESHOLD = 0.72   # default minimum acceptable quality score
COST_WEIGHT       = 0.40   # α — penalise cost
QUALITY_WEIGHT    = 0.60   # β — reward quality

# Per-task-type quality thresholds.
# Documentation / summaries tolerate lower quality from a cheap model;
# code and reasoning tasks require a higher bar.
TASK_QUALITY_THRESHOLDS: dict[str, float] = {
    "documentation":   0.60,   # light model acceptable for prose
    "summarisation":   0.55,   # even more lenient for summaries
    "qa_simple":       0.55,
    "code_generation": 0.72,
    "code_review":     0.85,   # security: high bar
    "reasoning":       0.80,
}

# Task types that require the heaviest tier (Tier 4) regardless of complexity score.
# code_review always goes to Mistral (security-aware, largest context)
SECURITY_OVERRIDES: set[str] = {"code_review"}


def _utility(model: ModelSpec, profile: TaskProfile) -> float:
    """
    Utility  U = β·Q − α·C
      Q = model.quality_score(complexity)   ∈ [0, 1]
      C = model.cost_per_1k (normalised)    ∈ [0, 1]
    A higher U means the model is preferred.
    """
    q = model.quality_score(profile.complexity)
    c = model.cost_per_1k
    return QUALITY_WEIGHT * q - COST_WEIGHT * c


def route(profile: TaskProfile) -> ModelSpec:
    """
    Return the optimal ModelSpec for the given TaskProfile.

    Selection rules (in order):
      1. Model must support the task type.
      2. Model context window must fit the estimated token budget.
      3. Model quality_score must meet QUALITY_THRESHOLD
         (unless it's a security override — then tier 3 is forced).
      4. Among qualifying candidates, the one with the highest utility wins.
      5. If no candidate passes the threshold, fall back to the highest-quality model.
    """
    ttype = profile.task_type.value

    # Force tier-4 (heaviest) for security-sensitive task types
    if ttype in SECURITY_OVERRIDES:
        heavy = next(m for m in MODEL_REGISTRY if m.tier == 4)
        return heavy

    threshold = TASK_QUALITY_THRESHOLDS.get(ttype, QUALITY_THRESHOLD)

    candidates = [
        m for m in MODEL_REGISTRY
        if ttype in m.capable_types
        and m.max_tokens >= profile.token_estimate
        and m.quality_score(profile.complexity) >= threshold
    ]

    if not candidates:
        # Safety net: pick highest-quality model regardless of cost
        candidates = sorted(
            MODEL_REGISTRY,
            key=lambda m: m.quality_score(profile.complexity),
            reverse=True,
        )

    return max(candidates, key=lambda m: _utility(m, profile))


def dispatch(profiles: list[TaskProfile]) -> dict[str, tuple[TaskProfile, ModelSpec]]:
    """
    Route each sub-task independently.

    Returns a mapping of task_type → (profile, selected_model).
    When multiple sub-tasks share the same type, a suffix is appended.
    """
    result: dict[str, tuple[TaskProfile, ModelSpec]] = {}
    type_counts: dict[str, int] = {}

    for profile in profiles:
        key = profile.task_type.value
        count = type_counts.get(key, 0)
        type_counts[key] = count + 1
        unique_key = key if count == 0 else f"{key}_{count}"
        result[unique_key] = (profile, route(profile))

    return result

Concrete Cost-Efficiency Proof

By decoupling tasks, the application completely redefines the cost equation for local and hybrid enterprise architectures.

Consider a compound prompt asking to: “Generate a secure Python CRUD interface and write 3 paragraphs of Markdown documentation for it.”

| Action Strategy            | Traditional Monolithic Approach | Orchestrator Pipeline Approach                  | Financial / Resource Impact          |
| -------------------------- | ------------------------------- | --------------------------------------------------- | ------------------------------------ |
| **Code Generation Step**   | Runs on `mistral-small` (15 GB) | Runs on `granite3.3:8b` (4.9 GB) or `mistral-small` | **Targeted Resource Allocation**     |
| **Documentation Step**     | Runs on `mistral-small` (15 GB) | Runs on `llama3.2:latest` (2.0 GB)                  | **~86% VRAM / Compute reduction**    |
| **Guardrail/Safety Check** | Ignored or rerun on flagship    | Runs on `granite3.2-guardian:3b` (2.7 GB)           | **Dedicated, low-latency alignment** |

The Token-Size Formula

When processing large workloads, the total resource overhead or token expense can be represented by the formula:

Where a traditional monolithic setup applies the maximum rate (RateMax) to all tokens (TokensTotal), the application drastically optimizes (in this case all is free as we use Ollama locally) the total cost footprint by ensuring that only high-complexity sub-tasks scale up to premium tiers, while documentation and utility steps run at significantly reduced rates.

To test the application we can just run the “demo.sh” script;

./multi-llm-router/scripts/demo.sh

Running the documentation-vs-code routing scenario…
(Set DRY_RUN=1 to skip LLM calls and inspect routing only)

================================================================================
  Multi-LLM Task Router  —  Demonstration Scenario  (4 providers)
================================================================================

Prompt:
  Write the API reference documentation for our REST /users endpoints AND implement the OAuth2 Bearer-token middleware in Python FastAPI.

Model tier mapping:
  model::light          IBM Granite     →  granite4.1:3b
  model::medium         Meta LLaMA      →  llama3.2:latest
  model::balanced       Google Gemma    →  gemma3:4b
  model::heavy          Mistral AI      →  mistral-small3.2:latest

Detected 2 sub-task(s):
  [1] TaskProfile(type='documentation', complexity=0.26, tier=LOW, tokens≈16)
  [2] TaskProfile(type='code_generation', complexity=0.609, tier=MEDIUM, tokens≈15)

Routing decisions:
  Task Type              Complexity  Tier       Label                Provider              U
  ------------------------------------------------------------------------------
  documentation               0.260  LOW        model::light         IBM Granite     +0.3616
  code_generation             0.609  MEDIUM     model::balanced      Google Gemma    +0.3718

Executing sub-tasks via Ollama…  (set DRY_RUN=1 to skip LLM calls)


================================================================================
  ROUTING TABLE
================================================================================
  Task Type              Label                Provider         Latency   Tokens        Cost
  ------------------------------------------------------------------------------
  documentation          model::light         IBM Granite       2839 ms       48    0.002400
  code_generation        model::balanced      Google Gemma     15279 ms       45    0.015750

  Total wall-clock: 15327 ms

======================================================================
  MERGED RESPONSE
======================================================================
## Documentation  _(via model::light, 2839 ms)_

# REST API Reference: `/users` Endpoints

Our RESTful API provides a set of endpoints under the `/users` resource to manage user-related operations. Below is a detailed description of each endpoint, including supported HTTP methods, request/response

---

## Code Generation  _(via model::balanced, 15279 ms)_

`python
from fastapi import Depends, HTTPException, status
from fastapi.security import CredentialsTokenAuthentication
import jwt

# Replace with your secret key (keep this secure!)
JWT_SECRET = "your-secret-

======================================================================
  FEEDBACK SUMMARY
======================================================================
{
  "total_calls": 2,
  "total_cost": 0.01815,
  "avg_quality": 0.665,
  "calls_by_tier": {
    "1": 1,
    "3": 1
  }
}

Conclusion

By treating LLMs as modular, specialized workers rather than monolithic black boxes, this simple orchestrator demonstrates that the most high-performing AI strategy is also the most cost-effective one. Through automated splitting, precise T-shirt sizing, and registry-driven routing, you can systematically lower your token footprint without sacrificing a single drop of output quality.

🎯 Et voilà, that’s a wrap 💯 for part 1 and thanks for reading! Stay tuned for the next episode 📺

DEV Community: Alain Airom (Ayrom)

Book Review: Build Applications with Local AI Models on a Mac

Introduction

Part 1: First Steps — Entering the World of the Terminal

Chapter 1: Getting Started with Local AI

Chapter 2: Installing and Using Homebrew

Chapter 3: Installing and Setting Up Ollama

Part 2: Core Skills — Building Your First AI Chatbot

Chapter 4: Downloading an LLM and Your First Conversation

Chapter 5: Setting Up VS Code and Your Python Development Environment

Chapter 6: Creating and Managing Python Virtual Environments

Chapter 7: Controlling LLMs via the Ollama Python API

Chapter 8: Building a Web UI with Streamlit

Chapter 9: Recording Audio and Transcribing Speech with MLX Whisper

Chapter 10: Building a Voice-Enabled AI Chat Application

Chapter 11: Managing Session State and Chat History

Part 3: Advanced Topics — Deeper Understanding and Customization

Chapter 12: Comparing and Selecting LLM Models

Chapter 13: System Prompts and Parameter Tuning

Chapter 14: Offline LLMs and Their Benefits

Chapter 15: Common Errors and Troubleshooting

Chapter 16: Advancing to RAG, Web UI, and Fine-Tuning

Chapter 17: The Open Model Revolution of 2026

Chapter 18: Where to Go from Here

Conclusion & Personal Opinion

Personal Opinion

Links

Building an AI-Native Document Platform with Doclang

🚀 Introduction & Overview

What is Doclang?

Implementation

🏗️ System Architecture & Workflow Schemas

High-Level System Architecture

Document Processing & Ingestion Pipeline

🛠️ Key Implementation Highlights

Solving the macOS Metal / PyTorch MPS Threading Crash

Hand-Rolled Markdown to DocLang XML Parser

Local LLM Q&A Integration with Ollama

📊 Token Efficiency Benchmark

🧪 Automated Testing Suite

💡 Conclusion

Links

Vaulting Beyond Defaults: The Evolution of Secrets Management on Red Hat OpenShift

The Baseline: Secrets Management in Kubernetes

The Upgraded Standard: How OpenShift Secures Native Secrets

Chapter 1: The Classical Approach — OpenShift Native Secrets (Approach A)

High-Level Architecture

Step-by-Step Reference Implementation

Declaring the Hardened Project and RBAC Boundaries

Defining the Platform Secrets

Injecting Secrets into the Application Deployment

How the Application Consumes Native Secrets?

The Operational Reality of Native Secrets

The Rotation Problem

The GitOps Paradox

Chapter 2: The Modern Alternative — Identity-Driven Secrets with HashiCorp Vault (Approach B)

Deploying Vault on Red Hat OpenShift

Vault Cluster Configuration

Bootstrapping Identity-Based Authentication

Scripted Authentication Setup

Defining Fine-Grained ACL Rules

Injecting Secrets securely via the Key-Value Engine

Chapter 3: Implementation — Decoupled Vault Injection (Approach B)

High-Level Architecture Flow

Manifest Configuration: Injecting Secrets via Annotations

Code Implementation: Reading secrets dynamically from Python

The Lifecycle Revolution: Direct Structural Comparisons

Conclusion: Elevating OpenShift Native Defense with Zero-Trust Automation

Links

Empowering the Team: How I Built a Lightweight Web UI for Bob (With Bob’s Help)

Introduction

The Web App at a Glance

Web Architecture Overview

Key Design Pillars of the Web Architecture

Server-Side Route Implementation & Stream Spawning

Client-Side Interface & SSE Parser

The Prompt Assistant: Simplifying File Access

How the File Access Helper Works

UI Implementation: Suggestion Chips & Prompt Templates

Client-Side Parsing Engine

Introducing Docling-Rust (`docling.rs`)

Thread-Safe Live Telemetry Engine (`telemetry.go`)

Circuit Breaker State Machine (`failover.go`)