DEV Community: MUHAMMAD ABIODUN SULAIMAN

Engineering a Multi-Agent AI Platform — Part 1: Adaptive Depth

MUHAMMAD ABIODUN SULAIMAN — Thu, 13 Aug 2026 19:32:27 +0000

This is Part 1 of a series on engineering AI systems that learn in production. This part covers the classification and strategy dispatch layer.

The problem nobody talks about

Here's a question that keeps coming up in production AI systems: should your model think before it answers?

Not philosophically. Mechanically. When a user asks "What's my account balance?", firing up a chain-of-thought reasoning engine with 16,000 thinking tokens is like hiring a PhD mathematician to count change. You burn compute, you add latency, and the answer isn't any better for it. But when someone asks "Compare the ROI of three deployment architectures under peak load, factoring in failure modes and regional latency differences," a zero-reasoning direct call produces garbage.

Most platforms pick one approach and apply it everywhere. The expensive ones over-reason on simple queries, bleeding money on thinking tokens that produce nothing useful. The cheap ones under-reason on hard problems, producing responses that sound confident but miss the point entirely.

Consider a platform where dozens of autonomous agents serve thousands of tenants across different industries. A scheduling agent handles "move my 3pm to Thursday" alongside an analytics agent wrestling with "why did churn spike in Q3 among enterprise accounts." These requests hit the same infrastructure, but they need radically different cognitive investment.

The system we're building doesn't pick a single depth. It scores each request and routes it to the appropriate reasoning tier. And then, as we'll see in Parts 2 and 3, it learns from outcomes and human feedback to get better at that scoring over time.

Three tiers of thinking

Think of reasoning depth as a spectrum with three practical zones.

Tier 1: No intermediate reasoning (NONE)

The model receives the prompt and generates a response directly. No thinking tokens, no structured intermediate steps. Input goes in, output comes out. This is your workhorse for factual lookups, simple status checks, greetings, and FAQ-style questions. Latency sits around 200ms. Cost is minimal.

When does this fail? When the question requires the model to hold multiple constraints in working memory simultaneously, or when the answer depends on comparing several options against each other. The model will produce something plausible-sounding, but it hasn't actually reasoned through the tradeoffs.

Tier 2: Lightweight structured reasoning (DRAFT)

Here, the system injects a structured instruction into the prompt: "Before answering, write your reasoning inside <draft> tags, limited to 512 tokens. Then provide your final answer inside <answer> tags."

The model gets a small scratchpad. It can organize its thoughts, check a few constraints, maybe outline a comparison. But the budget is tight. 512 tokens of reasoning is enough to decompose a moderately complex question into sub-parts and address each one, but not enough to explore rabbit holes or consider five alternative framings.

The system parses the response, extracts what's between the <answer> tags, discards the draft. The user never sees the intermediate reasoning unless the operator wants to inspect it for debugging.

Tier 3: Full chain-of-thought (FULL_COT)

This tier hands the problem to a model with native thinking-token support. Models like o3 or DeepSeek-R1 have a separate "thinking" channel where they can reason at length before producing a visible response. The system sets a reasoning_effort parameter and allocates a thinking budget (more on budgets in a moment).

The model can now spend thousands of tokens working through a problem. It can consider alternatives, backtrack when it hits a dead end, verify its own logic. For genuinely hard analytical questions, this produces qualitatively different answers than the other two tiers.

The tradeoff is obvious: latency climbs to 2-5 seconds and token costs multiply. You don't want this running on "What time do you close?"

Scoring complexity

So how does the system decide which tier to use? It runs the incoming request through a multi-signal classifier that produces a single score between 0 and 1.

Five signals feed the score:

Task type (weight: 0.30). Is this a factual lookup? A comparison? An analysis? A debugging request? A creative task? The classifier maintains a taxonomy of task types with associated difficulty priors. A status check scores low. A multi-criteria comparison scores high.

Ambiguity (weight: 0.20). How many ways could this request be interpreted? "Tell me about performance" is ambiguous (performance of what? the system? the team? a specific metric?). "What was the p95 latency for the /checkout endpoint last Tuesday" is not. Ambiguity markers include vague pronouns, missing context, and underspecified scope.

Domain vocabulary density (weight: 0.20). Requests loaded with specialized terminology signal that the answer needs domain expertise and careful reasoning. A message full of financial modeling terms or medical terminology likely needs deeper thinking than a conversational greeting.

Lexical complexity (weight: 0.15). Sentence length, subordinate clause depth, vocabulary sophistication. A proxy for how much cognitive load the request itself carries. Not a perfect signal (a short request can demand complex reasoning), but a useful one in combination with the others.

Context dependency (weight: 0.15). How much does the correct answer depend on prior conversation history, tenant-specific data, or cross-referencing multiple information sources? A standalone question scores lower than one that requires synthesizing three previous messages and two documents.

The composite score:

C(x) = 0.30 · S_task(x) + 0.20 · S_amb(x) + 0.20 · S_dom(x) + 0.15 · S_lex(x) + 0.15 · S_ctx(x)

Each signal function S(x) outputs a value in [0, 1]. The weighted sum produces the final complexity score C(x), also in [0, 1].

The routing thresholds:

C(x) < 0.30 → NONE (direct call, no reasoning)
0.30 ≤ C(x) < 0.65 → DRAFT (lightweight scratchpad, 512-token cap)
C(x) ≥ 0.65 → FULL_COT (native thinking tokens with budget)

A concrete example makes this tangible. "What is my account balance?" scores roughly: task_type = 0.1 (factual lookup), ambiguity = 0.1 (clear intent), domain = 0.05 (no specialized vocabulary), lexical = 0.1 (short, simple), context = 0.2 (needs account data but straightforward). Composite: 0.30(0.1) + 0.20(0.1) + 0.20(0.05) + 0.15(0.1) + 0.15(0.2) = 0.03 + 0.02 + 0.01 + 0.015 + 0.03 = 0.105. NONE tier. Just answer it.

Now try "Compare the ROI of three deployment architectures considering latency, cost, and failure modes under peak load." Task_type = 0.9 (multi-criteria analysis), ambiguity = 0.4 (what counts as "ROI" needs interpretation), domain = 0.8 (infrastructure terminology), lexical = 0.7 (complex sentence, multiple constraints), context = 0.6 (needs architecture details from prior context). Composite: 0.30(0.9) + 0.20(0.4) + 0.20(0.8) + 0.15(0.7) + 0.15(0.6) = 0.27 + 0.08 + 0.16 + 0.105 + 0.09 = 0.705. FULL_COT. This one needs to think.

Why this matters in multi-agent systems

A single-agent chatbot can get away with a fixed reasoning depth. But multi-agent platforms have a topology that makes static approaches break down.

Picture a supervisor agent that receives every inbound message. It classifies intent and routes to specialist agents: a scheduling agent, a billing agent, a technical support agent, an analytics agent. Each specialist handles a different slice of the problem space.

Here's what's interesting: the same specialist agent might need different reasoning depths depending on the specific request. The billing agent handles "what's my next payment date?" (NONE) alongside "explain why my invoice increased 40% this month considering the three pricing tiers, the mid-cycle plan change, and the prorated adjustments" (FULL_COT).

The reasoning strategy is selected per-request, not per-agent. The billing agent doesn't have a fixed reasoning depth. It gets classified fresh on every interaction based on what the user actually asked.

It gets more layered when you have agent teams. A lead agent might decompose a complex request into subtasks, delegating each to a different specialist. One subtask might be simple data retrieval (NONE), another might require cross-referencing multiple sources (DRAFT), and a third might involve genuine analytical reasoning (FULL_COT). The system scores each subtask independently.

This per-request granularity is what makes the whole thing tractable economically. If you set every agent to FULL_COT because some of their tasks are hard, you're paying ten times more for the easy ones. If you set them all to NONE because most tasks are easy, the hard ones produce bad answers and you lose user trust. Scoring at the request level means you pay for reasoning exactly where it creates value.

The thinking budget

Even within FULL_COT, not every hard problem deserves the maximum thinking allocation. A moderately complex question might need 4,096 thinking tokens to work through. A genuinely difficult analytical task might need 16,384. Allocating 32,000 thinking tokens to a question that only needs 4,000 wastes compute and sometimes degrades quality (the model can over-think, going in circles or second-guessing correct intermediate steps).

The system maps complexity sub-ranges to budget tiers:

Simple (below threshold): 0 tokens (NONE/DRAFT handle it)
Moderate (0.65-0.80): 4,096 tokens
Complex (0.80-1.0): 16,384 tokens

Budget enforcement works as a state machine. When the model starts thinking, the state is ACTIVE. If the model reaches the budget limit, the state transitions to EXHAUSTED, and the system signals the model to wrap up its reasoning. A grace window of 512 tokens allows the model to reach a natural stopping point rather than being cut off mid-thought. After the grace window, the state transitions to TERMINATED and the model must produce its final answer from whatever reasoning it completed.

This prevents runaway thinking. Without budget enforcement, a model given native thinking tokens can sometimes spend 20,000+ tokens reasoning in circles on a problem that doesn't benefit from it. The budget creates a ceiling without preventing deep reasoning when it's needed.

The diagram

Here's the full request lifecycle through strategy selection:

A request enters the complexity classifier, which evaluates five signal dimensions and produces a weighted score. The score crosses one of two thresholds (0.30 and 0.65), routing the request into one of three strategy lanes. Each lane uses a different mechanical approach to produce the final response: a simple pass-through call, a structured draft envelope with tag parsing, or a full thinking-token API with budget enforcement. All three converge on a response delivered to the user.

What's missing

The classifier works. It routes most requests to reasonable tiers. But it's a heuristic, and heuristics don't learn.

If the system consistently under-reasons on a particular type of request for a specific agent, the classifier keeps making the same mistake. If tenant patterns shift (maybe a business changes industries and their support queries become more technical), the static weights don't adapt.

What we actually want is a system that observes outcomes. Did the DRAFT strategy produce a good result for this context, or did it fall short? Was the FULL_COT response worth the extra latency and cost, or would DRAFT have been equally good?

That's the reinforcement learning router, and it's the subject of Part 2. The core idea: maintain a probabilistic belief about how well each (model, strategy) pair performs in each context, sample from that belief to select an action, observe the outcome, and update the belief. A contextual bandit that learns the terrain.

The classifier gives us a reasonable starting position. The bandit learns to do better.

Next in the series: Part 2 explores how Thompson Sampling learns optimal model+strategy routing from production outcomes.

You can follow me on LinkedIn and Twitter for more updates.

Engineering a Multi-Agent AI Platform — Part 2: The Bandit

MUHAMMAD ABIODUN SULAIMAN — Thu, 13 Aug 2026 19:24:44 +0000

This is Part 2 of a series on engineering AI systems that learn in production. Part 1 introduced the complexity classifier and three reasoning tiers. This part covers the reinforcement learning layer that replaces static thresholds with a contextual bandit that learns optimal (model, strategy) pairs from production outcomes.

The map is not the terrain

Part 1 gave us a complexity classifier. It scores each request across five dimensions, maps that score to a reasoning tier, and dispatches accordingly. For a static system, it works reasonably well.

But here's the thing about heuristic classifiers — they encode your assumptions at design time and then never update them. Those weights (0.30 for task type, 0.20 for ambiguity, and so on) were chosen based on intuition and early testing. They don't adapt when the system encounters patterns that violate those assumptions.

Consider a technical support agent handling database troubleshooting. The classifier might score most requests as moderate complexity (DRAFT tier). But maybe for this specific agent dealing with this specific domain, DRAFT consistently produces responses that require operator correction. The classifier doesn't know that. It keeps routing to DRAFT because the input signals look moderate. The problem isn't the classification of the input. The problem is that input classification doesn't predict which strategy will produce a good outcome for this particular context.

Think about the difference between a map and a navigator. The classifier is the map: static, useful for orientation, gives you a reasonable default route. What we need is a navigator that drives the actual roads, learns where the potholes are, and adjusts the route based on experience.

That navigator is a contextual bandit.

Thompson Sampling in plain language

Before the math, the intuition.

You moved to a new city and you're trying to find the best lunch spot near your office. Nine restaurants to choose from (three cuisines times three price points, say). You don't know which one you'll like best. How do you figure it out?

One approach: always go to whatever place seems best based on what you know so far. Problem is you'll settle on the first decent restaurant and never discover the amazing one two blocks further. You exploit what you know without exploring what you don't.

Another approach: go to a random restaurant every day. Problem is you'll eat bad meals indefinitely because you never concentrate on the places that have proven good.

Thompson Sampling does something more natural. You maintain a mental model of how good you think each restaurant is, but with uncertainty. For the place you've been to ten times and loved, you're pretty confident it's great. For the place you tried once and it was mediocre — you think it's probably mediocre but you're not sure. Maybe the chef was having a bad day.

Each lunchtime, you mentally "sample" from your uncertainty about each restaurant. Sometimes that sample for the once-visited place comes out surprisingly high (your uncertainty is wide, so extreme values are possible). When that happens, you go try it again. Most of the time, your confident favorites produce the highest samples, so you go there. But occasionally, your uncertainty about less-explored options produces a high enough sample to trigger exploration.

Over time, you visit everything enough to form confident beliefs. Restaurants you keep going back to are genuinely the best ones. You explored efficiently because you explored in proportion to your uncertainty, not randomly.

That's Thompson Sampling. Now replace "restaurants" with "(model, strategy) pairs" and "how much you liked the meal" with "how good the AI response was, accounting for cost and latency."

The action space

The bandit selects from the Cartesian product of available models and reasoning strategies.

Say the system has access to three models (gpt-4o, o3, deepseek-r1) and three strategies (NONE, DRAFT, FULL_COT). That's nine possible arms:

Arm	Model	Strategy
1	gpt-4o	NONE
2	gpt-4o	DRAFT
3	gpt-4o	FULL_COT
4	o3	NONE
5	o3	DRAFT
6	o3	FULL_COT
7	deepseek-r1	NONE
8	deepseek-r1	DRAFT
9	deepseek-r1	FULL_COT

Not all combinations make equal sense. You wouldn't typically run FULL_COT on a model without native thinking-token support. But the bandit can learn that. If arm 4 (o3 with no reasoning) consistently underperforms arm 6 (o3 with full reasoning), the posterior for arm 4 will shrink and it'll rarely get selected. You don't need to manually prune bad combinations — the learning handles it.

Per-agent policies define which models and strategies are available. An agent optimized for cost might only have gpt-4o in its allowlist. An agent handling high-stakes analytical work might have all three models available. The bandit operates within whatever action space the policy permits.

Context: the bandit isn't blind

A plain multi-armed bandit treats every request the same. Pull the arm with the best track record, regardless of what the request looks like.

That's too coarse. The optimal (model, strategy) pair for "summarize this email" is probably different from the optimal pair for "diagnose why our deployment is failing under load." We need the bandit to condition its selection on the request context.

The system encodes each request into a context bucket — a composite key built from:

Task complexity (from the classifier): low / medium / high
Intent label: the detected user intent (scheduling, billing, analysis, debugging, etc.)
Conversation length bucket: short (1-3 turns), medium (4-10), long (11+)
Tenant plan tier: free, professional, enterprise
Time bucket: business hours, off-hours, weekend

The full context key looks something like: high | debugging | medium_conv | enterprise | business_hours

Each unique context bucket gets its own set of Beta distributions, one per arm. So the system maintains separate beliefs about "how good is (gpt-4o, DRAFT) for high-complexity debugging requests from enterprise tenants during business hours" versus "how good is (gpt-4o, DRAFT) for low-complexity scheduling requests from free-tier tenants on weekends."

Why buckets instead of continuous features? Three reasons.

First, manageability: maintaining a Beta distribution per arm per context bucket works. Second, interpretability: an operator can look at the bucket high | analysis | long_conv | enterprise | business_hours and understand exactly what context it represents. Third, cold-start sharing: a new agent handling similar request types can inherit priors from existing agents with the same context patterns, instead of starting from zero.

The selection algorithm

Here's the core of Thompson Sampling with the Beta-Bernoulli model.

Each arm a_i in context bucket b has parameters (α_i, β_i) representing the system's belief about that arm's reward probability. Both start at (1, 1), which is a uniform prior. ("I know nothing. This arm could be anywhere from terrible to excellent.")

When a request arrives with context bucket b:

For each arm a_i in the agent's allowlist:
    Sample θ_i ~ Beta(α_i^b, β_i^b)

Select a* = argmax_i(θ_i)
Execute the request using (model, strategy) = a*

That's it. Sample from your current beliefs, pick whichever sample came out highest, use that arm. Arms you're uncertain about (low α + β, meaning wide distributions) will occasionally produce high samples and get explored. Arms you're confident about (high α + β, narrow distributions) will consistently produce samples near their true mean, so good arms get exploited.

The warm-up period. When the system hasn't seen many requests for a given context, every arm has wide distributions and selection is nearly random. We define a warm-up threshold: if the selected arm has fewer than 50 observations in this context, the selection gets flagged as low-confidence. During warm-up (fewer than 50 total observations per arm in the context), exploration rate starts at 30%. With exploration active, the system bypasses Thompson Sampling entirely and selects a random arm uniformly. This guarantees that even arms the sampling might overlook get a fair number of trials early on.

Exploration decay. That 30% exploration rate decays by 1% per week, settling at a 2% floor. Even in steady state, the system occasionally tries a random arm. This handles non-stationarity: if a model provider improves their API quality, or if token costs change, the system can discover that a previously inferior arm has become competitive.

The reward signal

After every request, the system observes three outcomes:

Quality: how good was the response? (Evaluated by automated quality metrics, sentence coherence scoring, factual grounding checks, or downstream task completion.)
Cost: how many tokens were consumed? (Translated to monetary cost using the model's pricing.)
Latency: how long did the user wait?

These three measurements get normalized and combined into a single scalar reward:

R = w_q · Q_norm + w_c · (1 - C_norm) + w_l · (1 - L_norm)

The default weights: w_q = 0.5, w_c = 0.3, w_l = 0.2. Quality matters most, cost matters second, latency matters third. These weights are configurable per agent, so cost-sensitive deployments can raise w_c and latency-sensitive ones can raise w_l.

Normalization is relative. Q_norm, C_norm, and L_norm are each scaled to [0, 1] based on the observed min/max across all arms in the model pool over a trailing 7-day window. If the cheapest model costs $0.001 per request and the most expensive costs $0.05, then a request costing $0.025 gets C_norm = 0.49. Cost and latency are inverted (lower is better), which is why they appear as (1 - C_norm) and (1 - L_norm).

Edge case: when min equals max (no variance observed, maybe the system just started), the normalized value defaults to 0.5. System assumes middling performance until it has enough data to differentiate.

A concrete example. An enterprise tenant asks a debugging question. The bandit selects (o3, FULL_COT). Response takes 3.2 seconds, costs $0.038, and scores 0.92 on quality evaluation. Over the trailing 7 days, quality ranges [0.6, 0.95], cost ranges [$0.001, $0.05], latency ranges [0.2s, 5.0s].

Q_norm = (0.92 - 0.6) / (0.95 - 0.6) = 0.914
C_norm = (0.038 - 0.001) / (0.05 - 0.001) = 0.755
L_norm = (3.2 - 0.2) / (5.0 - 0.2) = 0.625

R = 0.5(0.914) + 0.3(1 - 0.755) + 0.2(1 - 0.625)
  = 0.457 + 0.074 + 0.075
  = 0.606

A reward of 0.606. Decent but not outstanding. Quality was high, but cost and latency pulled it down. If a cheaper arm could have produced similar quality, the bandit will learn to prefer it.

Updating beliefs

After computing the reward, the system updates the Beta distribution for the arm that was selected, in the context bucket where it was selected:

α_new = α_old + R
β_new = β_old + (1 - R)

With our example reward of 0.606: if the arm previously had α = 12.4, β = 8.2, it becomes α = 13.006, β = 8.594. Distribution shifts slightly toward higher expected reward. Over hundreds of observations, the distribution narrows and the expected value converges on the arm's true performance in that context.

Why does this work? Beta distribution's mean is α / (α + β). Adding R to α and (1 - R) to β pushes the mean toward R while gradually reducing variance (as α + β grows, the distribution gets tighter). High rewards increase α faster, shifting the mean upward. Low rewards increase β faster, shifting the mean downward. Update is proportional to the reward magnitude, so a reward of 0.9 shifts the distribution more than a reward of 0.55.

Batch processing. Rewards don't update the posterior immediately after each request. They accumulate in a Redis queue and get processed every 5 minutes. This is a pragmatic choice: it reduces database write pressure, allows the system to handle bursts without locking contention, and makes the update process idempotent (if a batch fails, it retries the same set of rewards).

Persistence. Every 15 minutes, current state of all Beta distributions gets persisted to PostgreSQL with a SHA-256 checksum. On startup, the system loads the latest valid checkpoint. If the checksum doesn't match (corrupted state from a partial write or a weird failure mode), the system falls back to Beta(1, 1) priors (uniform, "I know nothing") and rebuilds from the last 7 days of routing decisions stored in the audit log.

Worst-case data loss is 7 days of learning, not a catastrophic reset to zero knowledge. And because the system maintains the raw routing decisions alongside the computed posteriors, it can always reconstruct its beliefs from history.

What makes this work in practice

A few design decisions that matter for production but don't show up in the math:

Per-agent isolation. Each agent has its own set of posteriors. The billing agent's beliefs about (gpt-4o, DRAFT) are completely separate from the analytics agent's beliefs about the same arm. What works for billing queries doesn't necessarily transfer to analytical queries, even in the same context bucket.

Policy constraints. Bandit can only select from the agent's configured allowlist. An enterprise-tier agent might have all nine arms available. A cost-constrained free-tier agent might only have three (gpt-4o with each strategy). The bandit optimizes within whatever boundaries the policy sets.

The classifier as a prior. The complexity classifier from Part 1 doesn't disappear. It provides the initial routing decision for new contexts where the bandit has insufficient data (the warm-up period). As the bandit accumulates observations, its learned posteriors gradually override the classifier's heuristic. In contexts where the bandit has 200+ observations per arm, the classifier's initial score barely influences the final selection.

Non-stationarity handling. Model providers change their APIs. Pricing shifts. Models get updated. The 2% exploration floor ensures the system can detect these changes, but slowly. For abrupt changes (a model provider announces a price cut), operators can manually reset the posteriors for affected arms, triggering a new warm-up period that quickly re-learns the new performance landscape.

The diagram

Here's the full bandit decision loop:

A request arrives with context features (complexity, intent, conversation depth, tenant tier, time). Context encoder maps these to a bucket string, which determines which set of Beta distributions to load, one per arm. System checks the exploration rate; if exploring, it picks a random arm. If exploiting, it samples from each arm's Beta distribution and picks the arm whose sample is highest. Selected arm executes, system observes quality/cost/latency, computes a normalized reward, enqueues it to Redis. Every 5 minutes a batch process applies the Bayesian update to the posterior. Loop closes. Next time this context appears, the distributions reflect what was learned.

What's still missing

The reward signal so far is fully automated. Quality is measured by an eval function. Cost and latency are measured by instrumentation. System learns from these numbers.

But numbers miss something. An automated quality scorer might rate a response highly because it's coherent, grammatically correct, and addresses the topic. But the human reading it might find it unhelpful, off-target, or missing the point in a way that's hard for an automated metric to capture.

What happens when an operator reads the response and corrects it? What happens when a customer gives a thumbs-down? What happens when a human takes over the conversation entirely because the agent's response was so inadequate that no correction could salvage it?

These are human preference signals, and they carry information that automated metrics can't replicate. Part 3 covers how to collect them, how to convert them into reward adjustments, and how to feed them back into the bandit so that it learns not just from numbers but from human judgment.

We'll also cover what happens when things go wrong at the infrastructure level: model timeouts, rate limits, budget exhaustion. System needs to degrade gracefully, and that graceful degradation is itself a learned behavior.

Next in the series: Part 3 explores how human preference signals enrich the learning loop, and how the system survives production failure modes.

System Design for Agentic AI Projects: What to Do, What Not to Do, and When

MUHAMMAD ABIODUN SULAIMAN — Sat, 04 Oct 2025 17:43:58 +0000

Agentic AI, where independent AI agents make complex decisions or orchestrate workflows, is revolutionizing business processes. Yet architecting such systems brings fresh design challenges and high stakes. Here are actionable principles, informed by real-world builds and strategic lessons, for designing robust, scalable Agentic AI solutions, complete with Python pseudocode examples to illustrate implementation patterns.

What to Do: Core System Design Principles Anchor to Business Value

Start by defining explicit, measurable business outcomes for the AI system. Every agent’s function should map directly to a high-impact organizational goal. Avoid speculative builds; only add autonomy where it demonstrably amplifies value and innovation.

class BusinessAlignedAgent:
    def __init__(self, name, business_kpi, success_threshold):
        self.name = name
        self.business_kpi = business_kpi  # e.g., "customer_satisfaction_score"
        self.success_threshold = success_threshold  # e.g., 0.85
        self.metrics_tracker = MetricsTracker()

    def execute_task(self, task_input):
        result = self.process_task(task_input)
        # Always measure business impact
        kpi_value = self.metrics_tracker.measure_kpi(self.business_kpi, result)

        if kpi_value < self.success_threshold:
            self.escalate_to_human(result, kpi_value)

        return result

Use Clear Orchestration and Task Decomposition

Architect systems with a two-tier model: a primary (“supervisor”) agent that owns orchestration and specialized subagents that execute well-bounded, stateless tasks. Decompose complex operations either vertically (sequential dependency) or horizontally (parallel independence), based on the workflow.

class AgentOrchestrator:
    def __init__(self):
        self.subagents = {
            'data_processor': DataProcessorAgent(),
            'validator': ValidatorAgent(),
            'reporter': ReporterAgent()
        }

    def execute_workflow(self, task):
        # Sequential decomposition example
        if task.type == "data_analysis":
            raw_data = self.subagents['data_processor'].process(task.input)
            validated_data = self.subagents['validator'].validate(raw_data)
            report = self.subagents['reporter'].generate_report(validated_data)
            return report

        # Parallel decomposition example
        elif task.type == "multi_source_research":
            parallel_tasks = task.split_into_parallel()
            results = []

            for subtask in parallel_tasks:
                agent = self.select_agent_for_task(subtask)
                results.append(agent.execute_async(subtask))

            return self.merge_results(results)

Favor Statelessness and Modularity

Design subagents as pure functions, with no internal state and no reliance on conversation history. This ensures predictable, parallel execution, easy troubleshooting, reproducible outputs, and streamlined caching. Isolate and test each module rigorously before integration.

class StatelessAgent:
    def __init__(self, tools, config):
        self.tools = tools
        self.config = config
        # No conversation state stored here

    def process(self, task_input, context=None):
        """Pure function - same input always produces same output"""
        # Extract only necessary context
        relevant_context = self.extract_relevant_context(context, task_input)

        # Process without side effects
        result = self.execute_logic(task_input, relevant_context)

        # Return structured output with metadata
        return {
            'result': result,
            'confidence': self.calculate_confidence(result),
            'reasoning': self.explain_reasoning(),
            'metadata': {
                'processing_time': self.get_processing_time(),
                'tools_used': self.get_tools_used()
            }
        }

Explicit Protocols and Structured Data

All agent-to-agent communication should use strict, structured formats: specify objective, input data, expected output, constraints, and success criteria. Responses must report status, reason codes, results, and relevant metadata.

from dataclasses import dataclass
from typing import Dict, Any, List
from enum import Enum

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    ESCALATED = "escalated"

@dataclass
class AgentTask:
    objective: str
    input_data: Dict[str, Any]
    constraints: List[str]
    expected_output_format: str
    priority: int = 1
    timeout_seconds: int = 300

@dataclass
class AgentResponse:
    status: TaskStatus
    result: Dict[str, Any]
    confidence_score: float
    reasoning: str
    metadata: Dict[str, Any]
    error_details: str = None
    next_actions: List[str] = None

class CommunicationProtocol:
    def send_task(self, agent_id: str, task: AgentTask) -> str:
        """Send task to agent, return task_id"""
        pass

    def get_response(self, task_id: str) -> AgentResponse:
        """Get agent response for task"""
        pass

Integrate Monitoring, Observability, and Human Oversight Early

From the outset, design dashboards and trace logs to capture every action, error, and decision path. Incorporate humans-in-the-loop to review high-stakes outcomes, intervene on edge cases, and absorb critical feedback for continual refinement.

import logging
from opentelemetry import trace
from datetime import datetime

class ObservableAgent:
    def __init__(self, name):
        self.name = name
        self.tracer = trace.get_tracer(__name__)
        self.logger = logging.getLogger(f"agent.{name}")
        self.metrics_collector = MetricsCollector()

    def execute_with_observability(self, task):
        # Create distributed trace
        with self.tracer.start_as_current_span(f"agent_{self.name}_execute") as span:
            span.set_attribute("agent.name", self.name)
            span.set_attribute("task.type", task.type)

            start_time = datetime.now()

            try:
                # Log task start
                self.logger.info(f"Starting task: {task.objective}")

                # Execute task with monitoring
                result = self.monitored_execute(task)

                # Record success metrics
                self.metrics_collector.record_success(
                    agent_name=self.name,
                    task_type=task.type,
                    duration=(datetime.now() - start_time).total_seconds(),
                    confidence=result.confidence_score
                )

                # Check if human oversight needed
                if self.requires_human_review(result):
                    return self.escalate_to_human(task, result)

                return result

            except Exception as e:
                # Record failure metrics
                self.metrics_collector.record_failure(
                    agent_name=self.name,
                    task_type=task.type,
                    error=str(e)
                )

                span.set_attribute("error", True)
                span.set_attribute("error.message", str(e))

                self.logger.error(f"Task failed: {e}")
                raise

    def requires_human_review(self, result):
        """Determine if result needs human oversight"""
        return (result.confidence_score < 0.7 or 
                "high_stakes" in result.metadata or
                result.status == TaskStatus.ESCALATED)

What Not to Do: Preventable Pitfalls Do Not Over-Engineer Hierarchies

More than two levels of agent delegation often result in convoluted debugging, unclear accountability, and diminished returns. Resist deep trees of subagents; complexity compounds risk.

class OverEngineeredSystem:  # DON'T DO THIS
    def __init__(self):
        # Too many levels - hard to debug and maintain
        self.level1_orchestrator = MainOrchestrator([
            SubOrchestrator([
                SpecializedAgent([
                    MicroAgent(), MicroAgent()
                ]),
                SpecializedAgent([
                    MicroAgent(), MicroAgent()
                ])
            ])
        ])

# PREFER - Simple two-level hierarchy
class SimpleSystem:  # DO THIS INSTEAD
    def __init__(self):
        self.orchestrator = MainOrchestrator()
        self.agents = {
            'processor': ProcessorAgent(),
            'validator': ValidatorAgent(),
            'reporter': ReporterAgent()
        }

Avoid Context Creep and Unbounded State

Do not simply pass all conversation history or business context to every agent. Instead, use the minimal, strictly relevant context needed for the immediate task.

class ContextManager:
    def __init__(self):
        self.context_store = {}
        self.context_filters = {
            'data_processor': ['data_schema', 'processing_rules'],
            'validator': ['validation_rules', 'error_thresholds'],
            'reporter': ['report_template', 'audience_type']
        }

    def get_filtered_context(self, agent_type, task):
        """Return only relevant context for specific agent"""
        full_context = self.context_store.get(task.session_id, {})

        # Filter to only relevant keys for this agent
        relevant_keys = self.context_filters.get(agent_type, [])
        filtered_context = {k: v for k, v in full_context.items() 
                          if k in relevant_keys}

        # Add task-specific context
        filtered_context.update(task.get_immediate_context())

        return filtered_context

When to Do What: Phased Execution Discovery & Design

Interview stakeholders, clarify business KPIs, and identify the smallest, highest-value opportunity for autonomy. Map out explicit agent roles and approval boundaries.

class ProjectDiscovery:
    def __init__(self):
        self.stakeholders = []
        self.business_requirements = {}
        self.agent_specifications = {}

    def conduct_stakeholder_interviews(self):
        for stakeholder in self.stakeholders:
            requirements = self.interview(stakeholder)
            self.business_requirements[stakeholder.role] = requirements

    def identify_automation_opportunities(self):
        opportunities = []

        for process in self.business_requirements['processes']:
            if (process.is_repetitive() and 
                process.has_clear_success_criteria() and
                process.risk_level < 'high'):
                opportunities.append(process)

        # Prioritize by value and feasibility
        return sorted(opportunities, 
                     key=lambda x: x.business_value / x.complexity_score)

Prototype & Evaluate

Build a minimal orchestration layer and 1–3 specialized subagents. Rigorously test each in isolation and as an integrated chain. Validate statelessness and boundary clarity with sample workflows.

class PrototypeValidator:
    def __init__(self, agents, test_cases):
        self.agents = agents
        self.test_cases = test_cases
        self.test_results = {}

    def run_isolation_tests(self):
        """Test each agent individually"""
        for agent_name, agent in self.agents.items():
            for test_case in self.test_cases[agent_name]:
                result = self.test_agent_isolation(agent, test_case)
                self.test_results[f"{agent_name}_isolation"] = result

    def run_integration_tests(self):
        """Test agent interactions and workflows"""
        for workflow in self.test_cases['workflows']:
            result = self.test_workflow_integration(workflow)
            self.test_results[f"workflow_{workflow.name}"] = result

    def validate_statelessness(self, agent):
        """Ensure same input produces same output"""
        test_input = self.generate_test_input()

        # Run same input multiple times
        results = [agent.process(test_input) for _ in range(3)]

        # All results should be identical (stateless)
        return all(r == results[0] for r in results)

Incremental Extension

Expand agent count, memory, and privileges only after robust monitoring and human-in-the-loop review demonstrate sustained, safe performance. Each new extension must undergo regression and error-handling tests.

class SafeExpansionManager:
    def __init__(self, current_system):
        self.current_system = current_system
        self.safety_metrics = SafetyMetrics()
        self.approval_gate = ApprovalGate()

    def propose_expansion(self, new_capability):
        # Check current system health
        if not self.safety_metrics.is_system_healthy():
            return False, "Current system showing issues"

        # Run safety analysis
        risk_assessment = self.analyze_expansion_risk(new_capability)

        if risk_assessment.risk_level > 'medium':
            return False, f"Risk too high: {risk_assessment.details}"

        # Require human approval for expansion
        approval = self.approval_gate.request_approval(
            expansion=new_capability,
            risk_assessment=risk_assessment,
            current_performance=self.safety_metrics.get_performance_summary()
        )

        return approval, "Ready for controlled rollout"

    def controlled_rollout(self, new_capability):
        """Gradual deployment with monitoring"""
        # Start with 5% of traffic
        self.deploy_to_percentage(new_capability, 5)

        # Monitor for 24 hours
        if self.monitor_for_duration(hours=24):
            # Increase to 25%
            self.deploy_to_percentage(new_capability, 25)
            # Continue gradual expansion...

What to Do When: Handling Change and Failure On System Drift or Error

Instantly trigger rollback, escalate to human review, and only proceed to retrain agents with the learnings post-mortem. Never ignore recurring anomalies, investigate, audit, and patch.

class ErrorRecoverySystem:
    def __init__(self):
        self.error_detector = ErrorDetector()
        self.rollback_manager = RollbackManager()
        self.human_escalation = HumanEscalationSystem()
        self.retry_policy = RetryPolicy()

    def handle_system_error(self, error, context):
        # Immediate containment
        if error.severity == 'critical':
            self.rollback_manager.immediate_rollback()
            self.human_escalation.emergency_alert(error, context)

        # Retry with exponential backoff for transient errors
        elif error.is_transient():
            return self.retry_with_backoff(error.failed_operation, context)

        # Pattern analysis for recurring errors
        elif self.error_detector.is_recurring_pattern(error):
            self.escalate_for_investigation(error)

        # Standard error recovery
        else:
            return self.standard_recovery(error, context)

    def retry_with_backoff(self, operation, context, max_retries=3):
        """Implement exponential backoff retry pattern"""
        for attempt in range(max_retries):
            try:
                return operation.execute(context)
            except Exception as e:
                if attempt == max_retries - 1:
                    # Final attempt failed
                    self.human_escalation.notify(e, context)
                    raise

                # Exponential backoff: 2^attempt seconds
                wait_time = 2 ** attempt
                time.sleep(wait_time)

    def post_mortem_analysis(self, error_incident):
        """Analyze failures and improve system"""
        analysis = {
            'root_cause': self.analyze_root_cause(error_incident),
            'prevention_measures': self.identify_prevention_measures(error_incident),
            'system_improvements': self.suggest_improvements(error_incident)
        }

        # Update agent training with lessons learned
        self.update_agent_training(analysis)
        return analysis

Amidst New Constraints

When regulations or requirements shift, update agent privileges, audit trails, and test compliance before resuming autonomous actions.

class ComplianceManager:
    def __init__(self):
        self.regulation_tracker = RegulationTracker()
        self.audit_system = AuditSystem()
        self.constraint_engine = ConstraintEngine()

    def handle_regulatory_change(self, new_regulation):
        # Immediate system pause for critical changes
        if new_regulation.impact_level == 'critical':
            self.pause_autonomous_operations()

        # Update constraint engine
        new_constraints = self.regulation_tracker.translate_to_constraints(new_regulation)
        self.constraint_engine.update_constraints(new_constraints)

        # Re-validate all agents against new constraints
        validation_results = self.validate_agents_compliance()

        if all(result.compliant for result in validation_results):
            self.resume_operations_with_new_constraints()
        else:
            self.remediate_non_compliant_agents(validation_results)

    def continuous_compliance_monitoring(self):
        """Ongoing compliance checking"""
        while self.system_is_running():
            for agent in self.active_agents():
                compliance_check = self.audit_system.check_compliance(
                    agent, self.constraint_engine.current_constraints
                )

                if not compliance_check.passed:
                    self.handle_compliance_violation(agent, compliance_check)

            time.sleep(self.compliance_check_interval)

For Coverage or Performance Gaps

Analyze dashboard metrics and user feedback, then upgrade agent skills or add specialized nodes incrementally, not en masse. Each new addition should prove its value through KPIs before full deployment.

class PerformanceOptimizer:
    def __init__(self):
        self.metrics_analyzer = MetricsAnalyzer()
        self.feedback_processor = FeedbackProcessor()
        self.improvement_engine = ImprovementEngine()

    def identify_performance_gaps(self):
        # Analyze quantitative metrics
        performance_data = self.metrics_analyzer.get_performance_summary()
        gaps = []

        for metric, value in performance_data.items():
            if value < self.get_threshold(metric):
                gaps.append(PerformanceGap(metric, value, self.get_threshold(metric)))

        # Analyze qualitative feedback
        user_feedback = self.feedback_processor.analyze_recent_feedback()
        gaps.extend(self.feedback_processor.identify_capability_gaps(user_feedback))

        return sorted(gaps, key=lambda x: x.business_impact, reverse=True)

    def implement_targeted_improvements(self, performance_gaps):
        for gap in performance_gaps[:3]:  # Focus on top 3 gaps
            improvement = self.improvement_engine.design_improvement(gap)

            # A/B test the improvement
            test_result = self.run_ab_test(improvement)

            if test_result.shows_improvement():
                self.deploy_improvement(improvement)
                self.monitor_improvement_impact(improvement)
            else:
                self.log_failed_improvement_attempt(improvement, test_result)

Conclusion

Well-designed Agentic AI solutions are anchored by clear intent, strong modular boundaries, real-time observability, and phased expansion under active human guidance. The code examples above illustrate how these principles translate into practical implementation patterns that you can adapt for your specific use cases. Adhering to these principles avoids costly missteps and speeds the path from breakthrough to sustainable business impact.

You can follow me on LinkedIn and Twitter for more updates.

🎯 ML Done Right: Versioning Datasets and Models with DVC & MLflow

MUHAMMAD ABIODUN SULAIMAN — Sat, 01 Feb 2025 20:32:06 +0000

Introduction

Data versioning is a crucial aspect of Machine Learning (ML) workflows. It ensures that datasets are reproducible, traceable, and manageable throughout the ML lifecycle. Unlike traditional software development, where Git efficiently tracks code changes, ML workflows require specialized tools to version datasets, models, and metadata.

Two powerful tools for data versioning in ML pipelines are:
DVC (Data Version Control): Designed to manage large datasets efficiently and integrates seamlessly with Git.
MLflow: Focused on experiment tracking, model versioning, and lifecycle management. In this article, we will explore how to set up DVC and MLflow for data versioning in a machine learning workflow using Python.

Why Do We Need Data Versioning?

Before going into implementation, let’s first understand why data versioning is essential:

Reproducibility → Ensures that ML models can be recreated using the exact dataset used during training.
Collaboration → Enables teams to share and work on different versions of datasets.
Traceability → Keeps track of dataset changes, helping to identify issues in models.
Rollback & Experimentation → Facilitates easy rollback to previous dataset versions for comparison.
Storage Efficiency → Avoids redundancy by tracking only changes in datasets instead of duplicating entire files.

1️⃣ Setting Up DVC for Data Versioning

DVC is an open-source tool designed to handle large datasets efficiently while integrating seamlessly with Git.

Installation

To install DVC, run:

pip install dvc

If you’re using cloud storage (e.g., AWS S3, Google Drive, or Azure), install the appropriate extension:

pip install 'dvc[s3]'
pip install 'dvc[gdrive]'

Initialize DVC in a Project

Inside a Git-tracked ML project, initialize DVC:

git init
dvc init
git commit -m "Initialize DVC"

Adding and Versioning Data

Assume we have a dataset stored in data/:

dvc add data/

This creates:

data.dvc → A metadata file that tracks the dataset version.
Updates .gitignore to prevent large files from being committed to Git.

Now, commit these changes to Git:

git add data.dvc .gitignore
git commit -m "Track dataset with DVC"

Remote Storage Configuration

To store the dataset in cloud storage:

dvc remote add myremote s3://mybucket/path/
dvc push

This uploads the dataset to S3. Other options like Google Drive, Azure, and SSH are also supported.

Restoring Previous Versions

To retrieve a previous dataset version:

git checkout <commit_id>
dvc pull

This ensures that data and models remain synchronized with the desired version.

2️⃣ Using MLflow for Data Tracking

MLflow helps track datasets, experiments, and models during development.

Installation

pip install mlflow

Initializing MLflow Tracking

Start the MLflow tracking server:

mlflow server --backend-store-uri sqlite:///mlflow.db --default-artifact-root ./mlruns

This creates a local database (mlflow.db) to store ML experiments.

Logging Dataset Versions with MLflow

Modify your Python script to track dataset versions:

import mlflow
import mlflow.artifacts
import os

# Log dataset version
def log_dataset_version(dataset_path):
    with mlflow.start_run():
        mlflow.log_artifact(dataset_path, artifact_path="datasets")
        print("Dataset version logged in MLflow")

log_dataset_version("data/")

Tracking Experiments

When training an ML model, log key parameters and metrics:

import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

# Load dataset
data = load_iris()
X_train, X_test, y_train, y_test = train_test_split(data.data, data.target, test_size=0.2, random_state=42)

# Train model
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

# Evaluate model
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)

# Log model and metrics
with mlflow.start_run():
    mlflow.log_param("n_estimators", 100)
    mlflow.log_metric("accuracy", accuracy)
    mlflow.sklearn.log_model(model, "random_forest_model")

    print(f"Model logged with accuracy: {accuracy}")

Retrieving Dataset Versions

To list previous dataset versions in MLflow:

mlflow artifacts list <run_id>

3️⃣ Combining DVC and MLflow for a Complete Workflow

A best practice is to integrate DVC for dataset versioning and MLflow for experiment tracking.

🔄 Workflow Summary

Store dataset using DVC → dvc add data/
Push dataset to remote storage → dvc push
Track dataset version in MLflow → mlflow.log_artifact("data/")
Train ML models and log parameters in MLflow
Retrieve dataset versions via DVC and experiments via MLflow.

🛠️ Example: End-to-End Pipeline

import dvc.api
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

# Fetch dataset version from DVC
dataset_path = "data/"
dvc.api.get_url(dataset_path)

# Log dataset version in MLflow
mlflow.log_artifact(dataset_path, artifact_path="datasets")

# Load dataset
data = load_iris()
X_train, X_test, y_train, y_test = train_test_split(data.data, data.target, test_size=0.2, random_state=42)

# Train model
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

# Evaluate model
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)

# Log results in MLflow
with mlflow.start_run():
    mlflow.log_param("n_estimators", 100)
    mlflow.log_metric("accuracy", accuracy)
    mlflow.sklearn.log_model(model, "random_forest_model")

    print(f"Model logged with accuracy: {accuracy}")

Data versioning is critical for:
✅ Reproducibility → Ensure consistent ML results.
✅ Collaboration → Manage dataset changes efficiently.
✅ Traceability → Keep track of dataset & model versions.

By integrating DVC and MLflow, you can create a scalable, reproducible, and traceable ML pipeline.

You can connect with me via email or LinkedIn or medium

STREAMLINE YOUR CI/CD PIPELINE WITH GITHUB ACTIONS

MUHAMMAD ABIODUN SULAIMAN — Fri, 12 Apr 2024 14:23:18 +0000

In the current dynamic software development landscape, security and efficiency are critical. Companies and developers work hard to swiftly roll out apps while making sure they are safe from security flaws. This is where the industry-leading containerization technology Docker comes into play. By wrapping up applications in containers, Docker makes the deployment process simpler. However, if done manually, managing these containers, creating them, checking for vulnerabilities, and pushing them safely can be difficult and time-consuming. For this reason, it's critical to automate these procedures via a Continuous Integration/Continuous Deployment (CI/CD) pipeline.

The Role of CI/CD in Modern Software Development

Software release process automation is the goal of Continuous Integration (CI) and Continuous Deployment (CD). Building software that is reliable, safe, and quickly deployable is the aim.

Continuous Integration:

CI involves merging all developers' working copies to a shared mainline several times a day. The main objectives of CI include:

Reducing bugs: Automated testing in CI helps detect and fix bugs quickly, improves software quality, and reduces the time it takes to validate and release new software updates.
Improving software quality: Continuous integration leads to significantly reduced assumptions as integration issues are detected and fixed continuously.

Continuous Deployment:

CD extends CI by automatically deploying all code changes to a testing and/or production environment after the build stage. This means that on top of the automated testing, automated release processes further streamline the development lifecycle. Benefits of CD include:

Faster time to market: Accelerated release cycles ensure features reach production faster.
Higher release rates: Frequent releases promote smaller, more manageable changes and less deployment risk.
Improved customer satisfaction: Continuous delivery of features addresses user feedback more promptly and enhances the user experience.

Implementing GitHub Actions for Docker Management

From the first code change to the last production deployment, GitHub Actions is a CI/CD tool that streamlines the software process. Building processes that automatically create, test, and launch Docker containers is a requirement of using GitHub Actions for Docker administration.

GitHub Actions Workflow: The "Complete Docker Workflow"

An extensive dissection of the "Complete Docker Workflow,” a system for managing Docker deployments with GitHub Actions, is given in this section. There are multiple stages in this workflow, all designed to improve security and expedite procedures across the container management lifecycle.

Workflow Activation Triggers

Scheduled Runs: Set to trigger daily at 16:21 UTC, this ensures regular updates and checks, keeping the application up to date with the latest base image vulnerabilities addressed.

    - cron: '21 16 * * *'

Push Events: Activates on pushes to specific branches or tags (specified branch and semantic versioned tags). This ensures that all changes undergo rigorous testing and security checks before deployment.

push:
    branches:
      - branch-name

Pull Requests: Targets pull requests to the dev-2 branch, allowing for automated reviews and tests and ensuring that new code integrations meet quality standards before merging.

pull_request:
    branches:
      - branch-name

Environment Configuration

Using environment variables and secrets for configurations like Docker registry credentials secures sensitive information and streamlines the setup process across multiple environments or projects. docker.io is specified as our REGISTRY in this instance.

env:
  REGISTRY: ${{ secrets.REGISTRY }}
  IMAGE_NAME: ${{ secrets.DOCKER_USERNAME }}/${{ secrets.IMAGE_NAME }}

Initial Setup and Configurations

Checkout Repository

Uses GitHub's actions/checkout@v3.5.2 to clone the repository, with a minimal fetch depth of 1 to speed up the checkout process.

- name: Checkout repository
        uses: actions/checkout@v3.5.2
        with:
          fetch-depth: 1

Install Cosign

Implements sigstore/cosign-installer@v3.4.0 to install Cosign, which is used later to sign Docker images.

- name: Install Cosign
        uses: sigstore/cosign-installer@v3.4.0

Set up QEMU

Employs docker/setup-qemu-action@v2.1.0 to configure QEMU, facilitating the emulation of different architectures which is essential for cross-platform Docker builds.

- name: Set up QEMU
        uses: docker/setup-qemu-action@v2.1.0

Set up Docker Buildx

Engages docker/setup-buildx-action@v2.5.0 to set up Docker Buildx, enhancing the ability to perform multi-platform builds directly from a single command.

- name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v2.5.0

Log in to Docker Registry

Utilizes docker/login-action@v2.1.0 for logging into the Docker registry using credentials stored in GitHub secrets.

- name: Log in to Docker Registry
        uses: docker/login-action@v2.1.0
        with:
          registry: ${{ env.REGISTRY }}
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}

Extract Docker Metadata

Activates docker/metadata-action@v4.4.0 to generate and format Docker image metadata, such as tags, from the environment variables.

- name: Extract Docker metadata
        id: meta
        uses: docker/metadata-action@v4.4.0
        with:
          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}

Building and Pushing Images

The procedure ensures interoperability across various hardware settings by supporting the development of images for many architectures using Docker Buildx and QEMU. Developers' work is made easier by automating the push to registries, freeing them up to concentrate on essential features rather than operational setups. Here, linux/amd64 is the chosen OS. Please be aware that you can use multiple operating systems.

- name: Build and Push container images
        uses: docker/build-push-action@v4.0.0
        with:
          platforms: linux/amd64
          push: true
          tags: ${{ steps.meta.outputs.tags }}

Security Scanning with Trivy

Preventing potential security risks before they arise in production requires integrating Trivy scans to evaluate image vulnerabilities. This scan is essential for keeping a secure deployment as it looks for vulnerabilities at the OS and library levels. If there are multiple tags for the image, you can duplicate the below step and specify each of the tags.

- name: Scan Docker image with Trivy (specifying a tag)
        uses: aquasecurity/trivy-action@master
        with:
          image-ref: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:tag
          format: 'table'
          severity: 'CRITICAL,HIGH'
          vuln-type: 'os,library'

Digital Signing of Images with Cosign

By confirming the source of the images and signing them using Cosign, GitHub's OIDC integration adds an extra degree of security to ensure that only validated images are released.

- name: Sign the images with GitHub OIDC Token (Non-interactive)
        run: |
          IFS=',' read -ra ADDR <<< "${{ steps.meta.outputs.tags }}"
          for tag in "${ADDR[@]}"; do
            echo "Signing $tag"
            cosign sign --oidc-issuer=https://token.actions.githubusercontent.com --yes "$tag"
          done
        env:
          COSIGN_EXPERIMENTAL: "true"

Complete CI/CD Pipeline

name: Complete Docker Workflow

on:
  schedule:
    - cron: '21 16 * * *'
  push:
    branches:
      - branch-name
    tags:
      - 'v*.*.*'
  pull_request:
    branches:
      - branch-name

env:
  REGISTRY: ${{ secrets.REGISTRY }}
  IMAGE_NAME: ${{ secrets.DOCKER_USERNAME }}/${{ secrets.IMAGE_NAME }}

jobs:
  build:
    runs-on: ubuntu-latest
    permissions:
      contents: read
      packages: write
      id-token: write # needed for signing the images with GitHub OIDC Token
    steps:
      - name: Checkout repository
        uses: actions/checkout@v3.5.2
        with:
          fetch-depth: 1

      - name: Install Cosign
        uses: sigstore/cosign-installer@v3.4.0

      - name: Set up QEMU
        uses: docker/setup-qemu-action@v2.1.0

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v2.5.0

      - name: Log in to Docker Registry
        uses: docker/login-action@v2.1.0
        with:
          registry: ${{ env.REGISTRY }}
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}

      - name: Extract Docker metadata
        id: meta
        uses: docker/metadata-action@v4.4.0
        with:
          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}

      - name: Build and Push container images
        uses: docker/build-push-action@v4.0.0
        with:
          platforms: linux/amd64
          push: true
          tags: ${{ steps.meta.outputs.tags }}

      - name: Scan Docker image with Trivy (tag1)
        uses: aquasecurity/trivy-action@master
        with:
          image-ref: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:tag1
          format: 'table'
          severity: 'CRITICAL,HIGH'
          vuln-type: 'os,library'

      - name: Scan Docker image with Trivy (tag2)
        uses: aquasecurity/trivy-action@master
        with:
          image-ref: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:tag2
          format: 'table'
          severity: 'CRITICAL,HIGH'
          vuln-type: 'os,library'

      - name: Sign the images with GitHub OIDC Token (Non-interactive)
        run: |
          IFS=',' read -ra ADDR <<< "${{ steps.meta.outputs.tags }}"
          for tag in "${ADDR[@]}"; do
            echo "Signing $tag"
            cosign sign --oidc-issuer=https://token.actions.githubusercontent.com --yes "$tag"
          done
        env:
          COSIGN_EXPERIMENTAL: "true"

Advantages over Traditional Methods

There are several benefits to automating these procedures over more conventional manual ones.

Reduced Human Error: Automating repetitive tasks lowers the possibility of human error, including deployment script misconfiguration or omission of stages.
Consistency and Reliability: Automation guarantees consistency and repeatability by ensuring that each step is carried out in the same way. This makes the process of developing, testing, and deploying software dependable and predictable.
Security: By methodically identifying and addressing security concerns, automated vulnerability scans considerably lower the risk of releasing vulnerable code.

Conclusion

This pipeline comprehensively covers all aspects of Docker image management, from build to security checks to deployment, ensuring high standards of automation and security using GitHub Actions. This setup not only automates the build and deployment process but also incorporates critical security practices like scanning and signing images, pivotal for maintaining the integrity and trustworthiness of software in a CI/CD environment.

Guide to Resizing EC2 Instance Volume without Deleting the Instance

MUHAMMAD ABIODUN SULAIMAN — Tue, 05 Dec 2023 13:51:24 +0000

As engineers or developers who use AWS Cloud infrastructure, when we provision an EC2 instance, we often face the challenge of the instance storage getting exhausted while our deployment is midway.

While starting out my career with AWS Cloud a couple of years ago, whenever I'm faced with this challenge, I find myself deleting the instance and provisioning a new instance with a higher volume size.

Over the years, I found that I wasn't implementing the best practice. Rather, I was only meant to create a shell script that will increase the volume size to a size of my choice while restarting the instance to ensure that the partition takes up all the space it can.

The script explained in part:

A. Specify the desired volume size in GiB as a command line argument. If not specified, default to 20 GiB.

SIZE=${1:-20}

B. Get the ID of the environment host Amazon EC2 instance.

INSTANCEID=$(curl http://169.254.169.254/latest/meta-data/instance-id)
REGION=$(curl -s http://169.254.169.254/latest/meta-data/placement/availability-zone | sed 's/\(.*\)[a-z]/\1/')

C. Get the ID of the Amazon EBS volume associated with the instance.

VOLUMEID=$(aws ec2 describe-instances \
  --instance-id $INSTANCEID \
  --query "Reservations[0].Instances[0].BlockDeviceMappings[0].Ebs.VolumeId" \
  --output text \
  --region $REGION)

D. Resize the EBS volume.

aws ec2 modify-volume --volume-id $VOLUMEID --size $SIZE

E. Wait for the resize to finish.

while [ \
  "$(aws ec2 describe-volumes-modifications \
    --volume-id $VOLUMEID \
    --filters Name=modification-state,Values="optimizing","completed" \
    --query "length(VolumesModifications)"\
    --output text)" != "1" ]; do
sleep 1
done

F. Check if we're on an NVMe filesystem

if [[ -e "/dev/xvda" && $(readlink -f /dev/xvda) = "/dev/xvda" ]]
then
  # Rewrite the partition table to take up all the space it can.
  sudo growpart /dev/xvda 1

  # Expand the size of the file system.
  # Check if we're on AL2
  STR=$(cat /etc/os-release)
  SUB="VERSION_ID=\"2\""
  if [[ "$STR" == *"$SUB"* ]]
  then
    sudo xfs_growfs -d /
  else
    sudo resize2fs /dev/xvda1
  fi

else
  # Rewrite the partition table to take up all the space it can.
  sudo growpart /dev/nvme0n1 1

  # Expand the size of the file system.
  # Check if we're on AL2
  STR=$(cat /etc/os-release)
  SUB="VERSION_ID=\"2\""
  if [[ "$STR" == *"$SUB"* ]]
  then
    sudo xfs_growfs -d /
  else
    sudo resize2fs /dev/nvme0n1p1
  fi
fi

The complete script is below:

#!/bin/bash

# Specify the desired volume size in GiB as a command line argument. If not specified, default to 20 GiB.
SIZE=${1:-20}

# Get the ID of the environment host Amazon EC2 instance.
INSTANCEID=$(curl http://169.254.169.254/latest/meta-data/instance-id)
REGION=$(curl -s http://169.254.169.254/latest/meta-data/placement/availability-zone | sed 's/\(.*\)[a-z]/\1/')

# Get the ID of the Amazon EBS volume associated with the instance.
VOLUMEID=$(aws ec2 describe-instances \
  --instance-id $INSTANCEID \
  --query "Reservations[0].Instances[0].BlockDeviceMappings[0].Ebs.VolumeId" \
  --output text \
  --region $REGION)

# Resize the EBS volume.
aws ec2 modify-volume --volume-id $VOLUMEID --size $SIZE

# Wait for the resize to finish.
while [ \
  "$(aws ec2 describe-volumes-modifications \
    --volume-id $VOLUMEID \
    --filters Name=modification-state,Values="optimizing","completed" \
    --query "length(VolumesModifications)"\
    --output text)" != "1" ]; do
sleep 1
done

#Check if we're on an NVMe filesystem
if [[ -e "/dev/xvda" && $(readlink -f /dev/xvda) = "/dev/xvda" ]]
then
  # Rewrite the partition table to take up all the space it can.
  sudo growpart /dev/xvda 1

  # Expand the size of the file system.
  # Check if we're on AL2
  STR=$(cat /etc/os-release)
  SUB="VERSION_ID=\"2\""
  if [[ "$STR" == *"$SUB"* ]]
  then
    sudo xfs_growfs -d /
  else
    sudo resize2fs /dev/xvda1
  fi

else
  # Rewrite the partition table to take up all the space it can.
  sudo growpart /dev/nvme0n1 1

  # Expand the size of the file system.
  # Check if we're on AL2
  STR=$(cat /etc/os-release)
  SUB="VERSION_ID=\"2\""
  if [[ "$STR" == *"$SUB"* ]]
  then
    sudo xfs_growfs -d /
  else
    sudo resize2fs /dev/nvme0n1p1
  fi
fi

You can connect with me via Email or LinkedIn or medium

Passing Credentials to Deployed Streamlit Apps using Streamlit Secrets

MUHAMMAD ABIODUN SULAIMAN — Sat, 05 Nov 2022 16:32:52 +0000

Often as programmers, we get to deal with credentials, for instance, when we need to connect to a database or ingest data from sources besides our local computers. Hence, it becomes imperative to find a way to securely pass the credentials such that they are not exposed in our code scripts.

In this short article, I will explain how I used #StreamlitSecrets to resolve a problem with passing credentials to a #MachineLearning system I deployed to the web using #Streamlit. Before fixing the problem, the deployed app worked fine on my local computer while testing it because the credentials were passed to the ingestion pipeline using an environment variable file (.env); however, upon deploying to the web, the system failed to work due to unavailability of the credentials. Some steps in solving this problem include removing the environment variable file from gitignoreand creating a Python (.py) file; since the programming language I was working with was Python. Upon taking these steps, I could not push the local repository to the remote repository as pre-commit hook detected that credentials had been exposed and did not allow the push. Hence, I finally had to use the #StreamlitSecrets file (secrets.toml) to pass the credentials while updating the credentials on the app settings section of the deployed streamlit app. The steps taken to achieve this are listed below:

Create the directory .streamlit/
Create the secret file secrets.toml
In the created secret file, type in the credentials in the below format:

[db_credentials]
user = 'someuser'
password = 'somepassword'
host = 'somehost'
database = 'somedatabase'

PS: You can include other credentials as needed.

In the python file, pass in the credentials with the following lines of code:

host=st.secrets.db_credentials.host,
user=st.secrets.db_credentials.user,
password=st.secrets.db_credentials.password,
db=st.secrets.db_credentials.database

Navigate to the url link of the deployed app, open settings, and type in the provided credentials in .streamlit/secrets.toml

The above steps ensure that your streamlit app works seamlessly on your local computer and the Streamlit deployed version.

If you find the article helpful, kindly click the like button and comment.

You can follow me on LinkedIn and Twitter for more updates.

Automating AWS Infrastructure with Terraform

MUHAMMAD ABIODUN SULAIMAN — Tue, 30 Aug 2022 16:30:41 +0000

What is Cloud Automation and Why is it Important in IT?

Cloud automation broadly refers to the processes and tools used to provision and manage cloud computing workloads and infrastructure. These processes and tools aim to reduce or eliminate manual processes, saving costs and resources.

With the increasing demand of serving a wider group of customers/clients and expanding the customer base, organizations need to consider the adoption of Cloud Automation as it affords them the ability to scale efficiently. Several IT companies have adopted cloud automation to optimize their business resources while also staying on top of their games.
Some of the reasons why companies will want to consider the adoption of cloud adoption are listed below

It saves an IT team time and money;
It is faster, more secure and more scalable than manually performing tasks;
It leads to fewer errors, as organizations can construct more predictable and reliable workflows; and
It contributes directly to better IT and corporate governance.

Next, we shall explore cloud automation tools which are categorized as:
Cloud Providers (private and public): These include:-

i. AWS: AWS Config, AWS CloudFormation, AWS EC2 Systems Manager;
ii. Azure: Microsoft Azure Resource Manager, Azure Automation;
iii. Google: Google Cloud Composer, Cloud Deployment Manager; and
iv. IBM: IBM Cloud Orchestrator.

Configuration Management tools: Most of these tools allows for Infrastructure-as-a-service (IaaS) setup, and has the below as some of its examples:

i. Red Hat Ansible,
ii. Puppet Enterprise,
iii. Chef Automate,
iv. Salt/SaltStack, and
v. HashiCorp Terraform

Many multi-cloud management vendors incorporate automation capabilities into their tools. Some prominent ones are:

i. VMware,
ii. CloudBolt,
iii. CloudSphere (Hypergrid),
iv. Snow (Embotics),
v. Morpheus Data,
vi. Scalr, and
vii. Flexera (RightScale).

To read more about cloud automation, click here.

Having introduced us to the basics of cloud automation, its importance, and tools used for cloud automation. We shall employ some of the earlier mentioned cloud automation tools to build a project. We will use AWS, HashiCorp Terraform, and Microsoft Visual Studio Code (VSC) as our IDE.

This project introduces beginners to Cloud Infrastructure Automation, with AWS as the cloud provider. A detailed explanation of the processes and resources created in this small project can be found here. Simply follow the guides here, and you should be fine. Feel free to fork this repo, raise a pull request to contribute to this project, and raise an issue if you encounter any challenges

This project created nine resources/processes in our AWS instance right from terraform. These resources are:

Virtual Private Cloud (VPC): A virtual private cloud (VPC) is a secure, isolated private cloud hosted within a public cloud. VPCs combines the scalability and convenience of public cloud computing with private cloud computing data isolation. Click here to read more about VPC.
Internet Gateway: An Internet gateway is a network "node" connecting two networks that use different protocols (rules) to communicate. In the most basic terms, an Internet gateway is where data stops on its way to or from other networks. Click here to read more about internet gateway.
Custom Route Table: A route table contains a set of rules, called routes, that are used to determine where network traffic from your subnet or gateway is directed. In a more explicit term, a route table tells network packets which way they need to go to get to their destination. Click here to read more about custom route table.
Subnet: A subnet, or subnetwork, is a segmented piece of a more extensive network. More specifically, subnets are a logical partition of an IP network into multiple, smaller network segments. Click here to read more about subnets.
We associated the subnet created in step 4 with the route table created in step 3.
We created a Security Group to allow ports 22, 80, 443.
Create a network interface with an IP in the subnet created in step 4.
Assigned an elastic IP to the network interface created in step 7
Created an Ubuntu server and installed/enabled apache2

Remember to drop a comment and drop a like if you have benefitted from this article.

You can connect with me via:
Email: abiodun.msulaiman@gmail.com or
LinkedIn: Muhammad Abiodun Sulaiman