Build an Autonomous DevOps Copilot with Gemma 3 & Python (Lab 3: OpsAgent)

Watch Full Video on YouTube: https://www.youtube.com/watch?v=0p4uzniuMKw

Download notebook to follow along: https://github.com/MysteryBytes-Labs/agentic_ai/tree/main/local-agentic-ai-gemma3/

The Terminal Access Dilemma: Why Raw Bash Is Fatal for LLMs

Connecting an AI agent to a terminal shell is one of the most powerful capabilities in software engineering—and simultaneously one of the most hazardous.

When developers grant an autonomous LLM unconstrained access to subprocess.run(shell=True) or raw bash execution, catastrophic failure is only one hallucination away:

1. Destructive Command Hallucination: A model attempting to clear temporary caches might emit rm -rf /tmp /var/* or alter critical /etc/ configurations.
2. Flag and Syntax Mismatches: A subtle typo in flags (e.g., kill -9 targeting an unverified PID) can terminate essential database or system daemons.
3. Prompt Injection Risks: Hostile data embedded inside server logs or web requests can hijack the agent's reasoning loop and execute arbitrary shell commands.
4. Data Sovereignty & Privacy Leaks: Sending proprietary host telemetry, system architecture details, and internal network IP addresses to commercial cloud LLMs exposes sensitive infrastructure.

The Solution: We must replace raw shell access with strictly typed, read-only Pydantic data contracts. The LLM never touches bash; instead, it selects discrete, sandboxed inspection tools implemented in Python using native psutil APIs.

Agent Capability Comparison

3-Phase Closed-Loop Execution Architecture

OpsAgent operates via a deterministic, three-phase execution cycle:

┌────────────────────────────────────────────────────────┐
│             User Infrastructure Inquiry                │
│     "What is my current CPU and memory usage?"         │
└──────────────────────────┬─────────────────────────────┘
                           │
                           ▼
┌────────────────────────────────────────────────────────┐
│       Phase 1: Structured Intent Analysis (Ollama)     │
│  Gemma 3 maps natural language to OpsAgentDecision    │
│  with mandatory technical reasoning deduction          │
└──────────────────────────┬─────────────────────────────┘
                           │
                           ▼
┌────────────────────────────────────────────────────────┐
│       Phase 2: Sandboxed Python Tool Dispatch          │
│  psutil collects hardware metrics (CPU, RAM, Disks)    │
│  using safe, strictly bounded parameters               │
└──────────────────────────┬─────────────────────────────┘
                           │
                           ▼
┌────────────────────────────────────────────────────────┐
│       Phase 3: Technical Synthesis with Gemma 3        │
│  Gemma 3 interprets telemetry and formats an           │
│  actionable, publication-ready Markdown report         │
└────────────────────────────────────────────────────────┘

Step 1: Environment Setup & Host Platform Verification

We initialize the local Ollama client and inspect host telemetry to confirm that Python can query the system kernel:

pip install -q ollama pydantic psutil ipython

from pydantic import BaseModel, Field
from typing import Optional, Literal, Dict, Any, List
import json
import os
import psutil
import platform
import shutil
from datetime import datetime
import ollama
from IPython.display import Markdown, display

# Initialize Ollama client connected to localhost
client = ollama.Client(host="http://localhost:11434")
MODEL_NAME = "gemma3:4b"

# Host telemetry verification
cpu_count = psutil.cpu_count(logical=True)
total_ram_gb = round(psutil.virtual_memory().total / (1024 ** 3), 1)

print(f"✅ Connected to Ollama at {client._client.base_url}")
print(f"  Target Model: {MODEL_NAME}")
print(f"  Host System : {platform.system()} {platform.machine()} ({cpu_count} CPU cores, {total_ram_gb} GB RAM)")

Step 2: Defining Strongly Typed DevOps Tool Schemas

To prevent the LLM from inventing parameters, we define three discrete Pydantic models:

1. Hardware Telemetry Schema (SystemTelemetryParams)

Restricts hardware inspections to supported subsystems:

class SystemTelemetryParams(BaseModel):
    """Inspect current hardware utilization metrics."""
    resource: Literal["cpu", "memory", "disk", "all"] = Field(
        "all", description="Subsystem to inspect: 'cpu', 'memory', 'disk', or 'all'"
    )

2. Process Inspector Schema (ProcessInspectorParams)

Sorts active processes by CPU or memory with bounded limits ($1 \le \text{top_n} \le 10$):

class ProcessInspectorParams(BaseModel):
    """Identify top resource-consuming processes running on the machine."""
    sort_by: Literal["cpu", "memory"] = Field(
        "cpu", description="Metric to sort processes by: 'cpu' or 'memory'"
    )
    top_n: int = Field(3, ge=1, le=10, description="Number of top processes to return (1-10)")

3. Diagnostic Health Report Schema (HealthReportParams)

Triggers automated threshold evaluations:

class HealthReportParams(BaseModel):
    """Evaluate overall system health against a resource threshold."""
    alert_threshold_pct: int = Field(80, ge=50, le=95, description="Utilization threshold triggering alerts")
    save_markdown: bool = Field(False, description="Whether to save a markdown summary file locally")

Step 3: Compiling the Unified OpsAgent Router Contract

We encapsulate all capabilities into the unified OpsAgentDecision schema. Notice that reasoning: str is mandatory—forcing Gemma 3 to justify its choice before executing:

class OpsAgentDecision(BaseModel):
    """Unified routing schema for OpsAgent autonomous decisions."""
    tool: Literal["system_telemetry", "inspect_processes", "generate_health_report", "none"] = Field(
        ..., description="Selected tool name, or 'none' if query can be answered directly"
    )
    reasoning: str = Field(..., description="Technical explanation for selecting this action")
    telemetry_params: Optional[SystemTelemetryParams] = Field(None, description="Parameters if system_telemetry selected")
    process_params: Optional[ProcessInspectorParams] = Field(None, description="Parameters if inspect_processes selected")
    report_params: Optional[HealthReportParams] = Field(None, description="Parameters if generate_health_report selected")

ops_agent_schema = OpsAgentDecision.model_json_schema()
print("✓ Unified OpsAgent Decision Schema compiled for Ollama constrained decoding.")

Step 4: System Tool Implementations & Central Registry

All tools are implemented in Python as pure, read-only inspections:

# 1. Hardware Telemetry Inspection Tool
def get_system_telemetry(resource: str = "all") -> Dict[str, Any]:
    data: Dict[str, Any] = {}
    if resource in ("cpu", "all"):
        data["cpu_percent"] = psutil.cpu_percent(interval=0.2)
        data["cpu_cores_logical"] = psutil.cpu_count(logical=True)
    if resource in ("memory", "all"):
        vm = psutil.virtual_memory()
        data["memory_used_gb"] = round((vm.total - vm.available) / (1024 ** 3), 2)
        data["memory_total_gb"] = round(vm.total / (1024 ** 3), 2)
        data["memory_percent"] = vm.percent
    if resource in ("disk", "all"):
        du = shutil.disk_usage("/")
        data["disk_used_gb"] = round(du.used / (1024 ** 3), 1)
        data["disk_total_gb"] = round(du.total / (1024 ** 3), 1)
        data["disk_percent"] = round((du.used / du.total) * 100, 1)
    return data

# 2. Top Process Inspection Tool
def inspect_top_processes(sort_by: str = "cpu", top_n: int = 3) -> List[Dict[str, Any]]:
    procs = []
    for p in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
        try:
            info = p.info
            if info['name'] and info['name'] != 'kernel_task':
                procs.append(info)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    sort_key = 'cpu_percent' if sort_by == 'cpu' else 'memory_percent'
    sorted_procs = sorted(procs, key=lambda x: x.get(sort_key) or 0, reverse=True)[:top_n]
    return [
        {
            "pid": p["pid"],
            "name": p["name"],
            "cpu_pct": f"{p['cpu_percent']}%",
            "mem_pct": f"{round(p['memory_percent'] or 0, 1)}%"
        }
        for p in sorted_procs
    ]

# 3. System Diagnostic Health Report Tool
def generate_system_health_report(alert_threshold_pct: int = 80, save_markdown: bool = False) -> Dict[str, Any]:
    telemetry = get_system_telemetry("all")
    alerts = []
    if telemetry.get("cpu_percent", 0) > alert_threshold_pct:
        alerts.append(f"High CPU load detected: {telemetry['cpu_percent']}%")
    if telemetry.get("memory_percent", 0) > alert_threshold_pct:
        alerts.append(f"High Memory usage detected: {telemetry['memory_percent']}%")
    if telemetry.get("disk_percent", 0) > alert_threshold_pct:
        alerts.append(f"High Disk usage detected: {telemetry['disk_percent']}%")
    
    status = "HEALTHY" if not alerts else "WARNING"
    report = {
        "status": status,
        "alert_count": len(alerts),
        "alerts": alerts,
        "telemetry": telemetry,
        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    }
    return report

# Central Dispatcher Registry
TOOL_REGISTRY = {
    "system_telemetry": lambda p: get_system_telemetry(p.resource if p else "all"),
    "inspect_processes": lambda p: inspect_top_processes(p.sort_by if p else "cpu", p.top_n if p else 3),
    "generate_health_report": lambda p: generate_system_health_report(p.alert_threshold_pct if p else 80, p.save_markdown if p else False)
}

Step 5: Compiling the Autonomous OpsAgent Orchestrator

The run_ops_agent() orchestrator manages the full lifecycle:

def run_ops_agent(user_query: str) -> str:
    print(f"\n💬 User Query: \"{user_query}\"")
    
    # Phase 1: Structured Intent Analysis & Routing with Pydantic Schema
    response = client.chat(
        model=MODEL_NAME,
        messages=[
            {
                "role": "system",
                "content": (
                    "You are OpsAgent, an autonomous DevOps system copilot with access to live machine tools. "
                    "Follow these tool selection rules strictly:\n"
                    "1. If the user asks about live CPU, memory, or disk usage, select 'system_telemetry'.\n"
                    "2. If the user asks to identify, rank, or inspect running processes, select 'inspect_processes'.\n"
                    "3. If the user requests a health audit or diagnostic report, select 'generate_health_report'.\n"
                    "4. Only select 'none' for conceptual, educational, or theoretical questions that do not require live data."
                )
            },
            {"role": "user", "content": user_query}
        ],
        format=ops_agent_schema  # Forces output to conform to OpsAgentDecision
    )
    
    # Parse and validate with Pydantic
    decision = OpsAgentDecision.model_validate_json(response["message"]["content"])
    print(f"🤖 Agent Reasoning: {decision.reasoning}")
    print(f"🔧 Tool Selected  : {decision.tool}")
    
    if decision.tool == "none":
        return decision.reasoning
        
    # Phase 2: Execute matched system tool
    if decision.tool == "system_telemetry":
        tool_output = TOOL_REGISTRY["system_telemetry"](decision.telemetry_params)
    elif decision.tool == "inspect_processes":
        tool_output = TOOL_REGISTRY["inspect_processes"](decision.process_params)
    elif decision.tool == "generate_health_report":
        tool_output = TOOL_REGISTRY["generate_health_report"](decision.report_params)
    else:
        tool_output = {"error": "Unknown tool"}
        
    print(f"⚙️ Telemetry Output: {tool_output}")
    
    # Phase 3: Technical Synthesis with Gemma 3
    synth_response = client.chat(
        model=MODEL_NAME,
        messages=[
            {
                "role": "system",
                "content": "You are an expert DevOps engineer. Formulate a concise, beautifully formatted Markdown summary of the system telemetry findings."
            },
            {"role": "user", "content": user_query},
            {"role": "assistant", "content": f"System inspection output: {json.dumps(tool_output)}"},
            {"role": "user", "content": "Summarize these system findings clearly for an engineer."}
        ]
    )
    
    return synth_response["message"]["content"]

Step 6: Live Multi-Scenario DevOps Benchmark Evaluation

We put OpsAgent through three realistic operational scenarios:

Scenario 1: Live Hardware Telemetry Lookup

ans1 = run_ops_agent("What is my current CPU and memory utilization, and do I have plenty of RAM free?")
display(Markdown(ans1))

• Agent Reasoning: The user is asking for live CPU and memory metrics and available RAM. Selects system_telemetry.
• Telemetry Collected: CPU: 5.4%, Memory: 82.4% (13.18 GB used / 16.0 GB total), Disk: 93.4%.
• Gemma 3 Synthesis:
  • CPU Load: 5.4% (Low — no bottleneck).
  • Memory Pressure: 82.4% (Approaching threshold — recommends monitoring for potential memory pressure).
  • Storage Alert: 93.4% disk utilization flagged as an immediate warning.

Scenario 2: Active Process Resource Triage

ans2 = run_ops_agent("Which top 3 processes are consuming the most memory right now?")
display(Markdown(ans2))

• Agent Reasoning: User requested a real-time ranking of running processes by memory. Selects inspect_processes with sort_by="memory", top_n=3.
• Telemetry Collected: Accurately extracts top processes (launchd, logd, UserEventAgent) along with PIDs and memory usage percentages.

Scenario 3: Conceptual System Diagnosis (Swap vs Physical RAM)

ans3 = run_ops_agent("Explain the difference between swap memory and physical RAM, and when a machine starts swapping.")
display(Markdown(ans3))

• Agent Reasoning: Factual inquiry on memory architecture and kernel paging thresholds.
• Synthesis: Explains physical RAM vs SSD/HDD swap space, kernel page-out thresholds, and latency implications without triggering redundant tool calls.

Key Takeaways

1. Security by Design: Never allow an LLM to generate raw bash commands. Enforcing strict read-only Pydantic contracts completely eliminates command injection and destructive hallucination risks.
2. Deterministic Observability: OpsAgent transforms messy machine metrics into structured, auditable JSON logs before formulating clear engineering recommendations.
3. True Local Edge Computing: OpsAgent operates entirely on-device with Google Gemma 3 (4B) and Ollama—delivering sub-second latency, zero cloud costs, and complete privacy for sensitive server telemetry.

Masterclass Series Roadmap

• Lab 1: High-Performance Local LLMs with Gemma 3 & Token Streaming
• Lab 2: Pydantic Schemas & Constrained Tool Calling
• Lab 3 (This Guide): OpsAgent — Autonomous DevOps & Hardware Telemetry Copilot
• Lab 4 (Series Finale): StockAgent — Autonomous Financial Market Screener (Live Web Scraping + Institutional Memos)

Download the companion notebook lab3_devops-agent-running_local_using_gemma.ipynb and follow along with the complete hands-on tutorial.