Building a CI/CD Automation Agent with llama.cpp: Complete Guide

Building a CI/CD Automation Agent with llama.cpp: Complete Guide

Continuous Integration and Continuous Deployment (CI/CD) pipelines have become the backbone of modern software development. As teams scale and infrastructure grows more complex, the need for intelligent automation within these pipelines has never been greater. By combining the power of local large language models through llama.cpp with traditional CI/CD tooling, developers can build autonomous agents capable of analyzing build failures, suggesting fixes, generating test cases, and even orchestrating deployment decisions. This guide walks you through the entire process of building such an agent from scratch.

What Is a CI/CD Automation Agent?

A CI/CD automation agent is an AI-powered system that integrates into your existing pipeline to perform tasks that traditionally require human intervention. Unlike simple scripted automation, an AI agent can reason about context, interpret ambiguous error messages, and make informed decisions based on historical patterns and project-specific knowledge.

When powered by llama.cpp, the inference engine for running LLaMA and compatible models locally, the agent gains several unique advantages over cloud-based alternatives. The entire inference happens on your own infrastructure, meaning no source code or logs ever leave your environment. This is particularly critical in enterprise settings where compliance and data sovereignty are non-negotiable requirements.

The agent typically operates by receiving structured inputs from pipeline events, such as build failures, test results, or deployment triggers. It processes these inputs through a locally hosted language model, generates actionable outputs, and then executes or recommends specific actions within the pipeline.

Why It Matters

Traditional CI/CD pipelines are deterministic by design. They follow predefined rules and fail loudly when something unexpected happens. While this reliability is essential, it also means that every failure requires human attention. A developer must read logs, identify root causes, implement fixes, and re-trigger the pipeline. This cycle can consume hours of productive time, especially in large teams managing microservice architectures.

• Reduced Mean Time to Recovery (MTTR): The agent can instantly analyze failure logs and suggest or apply fixes, dramatically cutting down the time between a failed build and a successful one.
• Consistent Code Quality: By reviewing pull requests automatically, the agent ensures that coding standards are enforced uniformly without relying on human reviewers being available.
• Cost Efficiency: Running inference locally with llama.cpp eliminates per-token API costs, making it economically viable to run the agent on every single commit.
• Data Privacy: Source code, internal architecture details, and proprietary logs never traverse external networks.
• Offline Capability: The agent functions in air-gapped environments where internet access is restricted or prohibited.

Prerequisites and Setup

Before building the agent, you need to set up your environment. This includes compiling llama.cpp, downloading a suitable model, and preparing the Python dependencies that will serve as the glue between your CI/CD system and the language model.

Compiling llama.cpp

Start by cloning the llama.cpp repository and compiling it for your platform. The build process is straightforward but varies slightly depending on whether you want GPU acceleration.

# Clone the repository
git clone https://github.com/ggerganov/llama.cpp.git
cd llama.cpp

# Build with CPU support
make

# Alternatively, build with CUDA support for NVIDIA GPUs
make LLAMA_CUDA=1

# Verify the build
./llama-cli --version

Once compiled, you will have access to the llama-cli binary, which provides a command-line interface for running inference. You will also have llama-server, which exposes a REST API compatible with the OpenAI API format. For building an agent, the server mode is generally more practical because it allows your Python code to communicate with the model over HTTP.

Choosing and Downloading a Model

The choice of model significantly impacts the agent's capabilities. For CI/CD tasks, you need a model that excels at code understanding, log analysis, and structured output generation. Models in the 7B to 13B parameter range typically offer a good balance between performance and resource requirements.

# Download a code-focused model (example: CodeLlama 7B Instruct)
# Using huggingface-cli for convenience
pip install huggingface-hub
huggingface-cli download TheBloke/CodeLlama-7B-Instruct-GGUF \
    codellama-7b-instruct.Q4_K_M.gguf \
    --local-dir ./models

# Verify the model file exists
ls -lh ./models/codellama-7b-instruct.Q4_K_M.gguf

The Q4_K_M quantization provides a good trade-off between model quality and memory usage. On a machine with 16GB of RAM, a 7B model with this quantization will run comfortably while leaving resources for the rest of your pipeline tooling.

Starting the Inference Server

Launch the llama-server binary to expose the model via an HTTP API. This server will be the backbone of your agent's reasoning capabilities.

./llama-server \
    -m ./models/codellama-7b-instruct.Q4_K_M.gguf \
    --port 8080 \
    --ctx-size 4096 \
    --threads 4 \
    --cont-batching

The --ctx-size parameter determines how many tokens the model can process in a single request. For CI/CD tasks, 4096 is usually sufficient since you will be sending log snippets and code diffs rather than entire repositories. The --cont-batching flag enables continuous batching, which improves throughput when multiple requests arrive concurrently.

Building the Agent Core

With the inference server running, you can now build the Python application that will serve as the agent's core logic. The agent needs to handle several responsibilities: receiving pipeline events, formatting prompts, calling the model, parsing responses, and executing actions.

Project Structure

ci-cd-agent/
├── agent/
│   ├── __init__.py
│   ├── core.py
│   ├── prompts.py
│   ├── actions.py
│   └── pipeline.py
├── config.yaml
├── requirements.txt
└── main.py

Dependencies

Create a requirements.txt file with the necessary Python packages.

requests>=2.31.0
pyyaml>=6.0
jinja2>=3.1.0
python-gitlab>=4.0.0
subprocess-run>=0.0.1

Configuration Management

The agent should be configurable without code changes. A YAML configuration file allows you to specify model endpoints, pipeline settings, and action permissions.

# config.yaml
server:
  url: "http://localhost:8080"
  model: "codellama-7b-instruct"
  temperature: 0.2
  max_tokens: 1024

pipeline:
  workspace: "/var/lib/ci-agent/workspace"
  max_retries: 3
  auto_fix_enabled: true
  auto_fix_risk_threshold: "low"

actions:
  allowed:
    - "suggest_fix"
    - "create_issue"
    - "notify_slack"
    - "run_tests"
  blocked:
    - "deploy_production"
    - "delete_resources"

logging:
  level: "INFO"
  file: "/var/log/ci-agent/agent.log"

The Core Agent Class

The core class manages communication with the llama.cpp server and orchestrates the agent's decision-making process.

# agent/core.py
import requests
import json
import logging
from typing import Dict, Any, Optional

logger = logging.getLogger(__name__)

class CICDAgent:
    def __init__(self, config: Dict[str, Any]):
        self.server_url = config["server"]["url"]
        self.model = config["server"]["model"]
        self.temperature = config["server"]["temperature"]
        self.max_tokens = config["server"]["max_tokens"]
        self.config = config

    def _call_model(self, system_prompt: str, user_prompt: str) -> str:
        """Send a completion request to the llama.cpp server."""
        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            "temperature": self.temperature,
            "max_tokens": self.max_tokens,
            "stream": False
        }

        try:
            response = requests.post(
                f"{self.server_url}/v1/chat/completions",
                json=payload,
                timeout=120
            )
            response.raise_for_status()
            result = response.json()
            return result["choices"][0]["message"]["content"]
        except requests.exceptions.RequestException as e:
            logger.error(f"Model call failed: {e}")
            raise

    def analyze_build_failure(self, build_log: str, 
                                repo_context: str) -> Dict[str, Any]:
        """Analyze a build failure and return structured recommendations."""
        from agent.prompts import BUILD_FAILURE_SYSTEM, BUILD_FAILURE_USER

        user_prompt = BUILD_FAILURE_USER.format(
            repo_context=repo_context,
            build_log=build_log[:3000]  # Truncate to fit context window
        )

        raw_response = self._call_model(BUILD_FAILURE_SYSTEM, user_prompt)

        try:
            # Attempt to parse JSON response
            parsed = json.loads(raw_response)
            return parsed
        except json.JSONDecodeError:
            # Fallback: wrap in a standard structure
            return {
                "analysis": raw_response,
                "suggested_actions": [],
                "confidence": "low",
                "auto_fixable": False
            }

    def review_pull_request(self, diff: str, 
                             file_context: str) -> Dict[str, Any]:
        """Review a pull request diff and provide feedback."""
        from agent.prompts import PR_REVIEW_SYSTEM, PR_REVIEW_USER

        user_prompt = PR_REVIEW_USER.format(
            file_context=file_context,
            diff=diff[:3000]
        )

        raw_response = self._call_model(PR_REVIEW_SYSTEM, user_prompt)

        try:
            return json.loads(raw_response)
        except json.JSONDecodeError:
            return {
                "summary": raw_response,
                "issues": [],
                "approved": True
            }

Prompt Engineering for CI/CD Tasks

The quality of your agent's output depends heavily on prompt design. CI/CD tasks require structured, actionable responses rather than conversational text. Using Jinja2-style templates with explicit JSON output instructions produces the most reliable results.

# agent/prompts.py

BUILD_FAILURE_SYSTEM = """You are a CI/CD automation agent integrated into 
a software development pipeline. Your job is to analyze build failures and 
provide structured recommendations.

You MUST respond with valid JSON only. No markdown, no explanations outside 
the JSON structure. The JSON must follow this schema:

{
  "root_cause": "Brief description of the identified root cause",
  "category": "one of: dependency|syntax|test|config|infrastructure|unknown",
  "confidence": "one of: high|medium|low",
  "suggested_actions": [
    {
      "action": "one of: suggest_fix|create_issue|notify_slack|run_tests",
      "description": "What should be done",
      "risk_level": "one of: low|medium|high",
      "auto_fixable": true or false,
      "fix_code": "Code snippet if auto_fixable, otherwise empty string"
    }
  ],
  "estimated_fix_time": "Brief estimate like '5 minutes' or '2 hours'"
}

Be conservative. If you are not confident about a fix, set auto_fixable to 
false and suggest human review instead."""

BUILD_FAILURE_USER = """Analyze the following build failure.

Repository Context:
{repo_context}

Build Log (truncated):
{build_log}
Provide your analysis as valid JSON."""

PR_REVIEW_SYSTEM = """You are a CI/CD automation agent performing automated 
code review on pull requests. Focus on:
1. Potential bugs and logic errors
2. Security vulnerabilities
3. Performance concerns
4. Adherence to common coding best practices

Respond with valid JSON only:
{{
  "summary": "One paragraph summary of the changes",
  "issues": [
    {{
      "severity": "one of: critical|warning|info",
      "file": "File path",
      "line": "Approximate line number or range",
      "description": "Description of the issue",
      "suggestion": "Suggested fix"
    }}
  ],
  "approved": true or false,
  "approval_notes": "Explanation if not approved"
}}

Do not block PRs for stylistic preferences. Only flag issues that could 
cause bugs, security problems, or significant performance degradation."""

PR_REVIEW_USER = """Review the following pull request diff.

File Context:
{file_context}

Diff:
diff
{diff}
Provide your review as valid JSON."""

Action Execution Layer

The agent's recommendations are only useful if they can be translated into concrete actions. The action layer acts as a safety boundary, ensuring that only permitted actions are executed and that risky operations require human approval.

# agent/actions.py
import subprocess
import json
import logging
from typing import Dict, Any, List

logger = logging.getLogger(__name__)

class ActionExecutor:
    def __init__(self, config: Dict[str, Any]):
        self.allowed = set(config["actions"]["allowed"])
        self.blocked = set(config["actions"]["blocked"])
        self.auto_fix_enabled = config["pipeline"]["auto_fix_enabled"]
        self.risk_threshold = config["pipeline"]["auto_fix_risk_threshold"]

    def execute(self, action: Dict[str, Any]) -> Dict[str, Any]:
        """Execute a single action if permitted."""
        action_type = action.get("action")
        risk_level = action.get("risk_level", "high")

        if action_type in self.blocked:
            logger.warning(f"Blocked action attempted: {action_type}")
            return {
                "status": "blocked",
                "reason": f"Action '{action_type}' is in blocked list"
            }

        if action_type not in self.allowed:
            logger.warning(f"Disallowed action: {action_type}")
            return {
                "status": "rejected",
                "reason": f"Action '{action_type}' not in allowed list"
            }

        # Check risk level for auto-fixes
        risk_order = {"low": 0, "medium": 1, "high": 2}
        if (action.get("auto_fixable") and 
            risk_order.get(risk_level, 2) > risk_order.get(self.risk_threshold, 0)):
            return {
                "status": "deferred",
                "reason": f"Risk level '{risk_level}' exceeds threshold"
            }

        # Dispatch to appropriate handler
        handlers = {
            "suggest_fix": self._handle_suggest_fix,
            "create_issue": self._handle_create_issue,
            "notify_slack": self._handle_notify_slack,
            "run_tests": self._handle_run_tests
        }

        handler = handlers.get(action_type)
        if handler:
            return handler(action)
        return {"status": "error", "reason": "No handler found"}

    def _handle_suggest_fix(self, action: Dict[str, Any]) -> Dict[str, Any]:
        """Output a suggested fix for human review."""
        fix_code = action.get("fix_code", "")
        description = action.get("description", "")
        logger.info(f"Suggested fix: {description}")
        return {
            "status": "suggested",
            "description": description,
            "fix_code": fix_code
        }

    def _handle_create_issue(self, action: Dict[str, Any]) -> Dict[str, Any]:
        """Create an issue in the issue tracker."""
        # Integration with GitLab, GitHub, or Jira would go here
        logger.info(f"Creating issue: {action.get('description')}")
        return {
            "status": "created",
            "issue_id": "PLACEHOLDER-001",
            "description": action.get("description")
        }

    def _handle_notify_slack(self, action: Dict[str, Any]) -> Dict[str, Any]:
        """Send a notification to a Slack channel."""
        # Slack webhook integration would go here
        logger.info(f"Slack notification: {action.get('description')}")
        return {"status": "notified"}

    def _handle_run_tests(self, action: Dict[str, Any]) -> Dict[str, Any]:
        """Trigger a test run."""
        try:
            result = subprocess.run(
                ["pytest", "--tb=short", "-q"],
                capture_output=True,
                text=True,
                timeout=300
            )
            return {
                "status": "completed",
                "exit_code": result.returncode,
                "output": result.stdout[-2000:]
            }
        except subprocess.TimeoutExpired:
            return {"status": "timeout"}
        except Exception as e:
            return {"status": "error", "reason": str(e)}

    def execute_batch(self, actions: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Execute multiple actions in sequence."""
        results = []
        for action in actions:
            result = self.execute(action)
            results.append({
                "action": action.get("action"),
                "result": result
            })
            # Stop if an action fails critically
            if result.get("status") == "error":
                logger.error("Stopping batch due to error")
                break
        return results

Pipeline Integration

The pipeline integration module connects the agent to your CI/CD system. This example shows integration with a generic pipeline that could be adapted for GitLab CI, GitHub Actions, or Jenkins.

# agent/pipeline.py
import os
import json
import logging
from typing import Dict, Any
from agent.core import CICDAgent
from agent.actions import ActionExecutor

logger = logging.getLogger(__name__)

class PipelineIntegration:
    def __init__(self, agent: CICDAgent, executor: ActionExecutor):
        self.agent = agent
        self.executor = executor

    def handle_build_failure(self, event: Dict[str, Any]) -> Dict[str, Any]:
        """Process a build failure event from the CI/CD pipeline."""
        build_log = event.get("build_log", "")
        repo_context = self._build_repo_context(event)
        
        logger.info("Analyzing build failure...")
        analysis = self.agent.analyze_build_failure(build_log, repo_context)
        
        logger.info(f"Root cause: {analysis.get('root_cause')}")
        logger.info(f"Confidence: {analysis.get('confidence')}")

        # Execute suggested actions
        actions = analysis.get("suggested_actions", [])
        results = self.executor.execute_batch(actions)

        return {
            "analysis": analysis,
            "action_results": results
        }

    def handle_pull_request(self, event: Dict[str, Any]) -> Dict[str, Any]:
        """Process a pull request event."""
        diff = event.get("diff", "")
        file_context = event.get("file_context", "")

        logger.info("Reviewing pull request...")
        review = self.agent.review_pull_request(diff, file_context)

        if not review.get("approved", True):
            logger.warning("PR not approved by agent")
            # Could post comment on the PR here

        return review

    def _build_repo_context(self, event: Dict[str, Any]) -> str:
        """Build context string about the repository."""
        return (
            f"Repository: {event.get('repo', 'unknown')}\n"
            f"Branch: {event.get('branch', 'unknown')}\n"
            f"Commit: {event.get('commit_sha', 'unknown')}\n"
            f"Language: {event.get('language', 'unknown')}\n"
            f"Build System: {event.get('build_system', 'unknown')}"
        )

Entry Point

The main entry point ties everything together and provides a simple interface for triggering the agent from pipeline scripts.

# main.py
import yaml
import json
import sys
import logging
from agent.core import CICDAgent
from agent.actions import ActionExecutor
from agent.pipeline import PipelineIntegration

def setup_logging(config):
    log_config = config.get("logging", {})
    logging.basicConfig(
        level=getattr(logging, log_config.get("level", "INFO")),
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        handlers=[
            logging.StreamHandler(),
            logging.FileHandler(log_config.get("file", "agent.log"))
        ]
    )

def load_config(path: str = "config.yaml") -> dict:
    with open(path, "r") as f:
        return yaml.safe_load(f)

def main():
    config = load_config()
    setup_logging(config)

    agent = CICDAgent(config)
    executor = ActionExecutor(config)
    pipeline = PipelineIntegration(agent, executor)

    # Read event from stdin or file
    if len(sys.argv) > 1:
        with open(sys.argv[1], "r") as f:
            event = json.load(f)
    else:
        event = json.load(sys.stdin)

    event_type = event.get("type")

    if event_type == "build_failure":
        result = pipeline.handle_build_failure(event)
    elif event_type == "pull_request":
        result = pipeline.handle_pull_request(event)
    else:
        print(json.dumps({"error": f"Unknown event type: {event_type}"}))
        sys.exit(1)

    print(json.dumps(result, indent=2))

if __name__ == "__main__":
    main()

Integrating with CI/CD Systems

Once the agent is built, you need to wire it into your actual CI/CD pipeline. The integration approach varies depending on your platform, but the pattern is consistent: capture relevant data on failure, send it to the agent, and act on the response.

GitLab CI Integration Example

The following GitLab CI configuration shows how to invoke the agent when a build fails. The agent runs on a dedicated runner with sufficient resources to host the model.

# .gitlab-ci.yml
stages:
  - build
  - test
  - analyze
  - deploy

variables:
  AGENT_SERVER_URL: "http://ci-agent-server:8080"

build:
  stage: build
  script:
    - make build
  artifacts:
    paths:
      - build/
    when: always
  after_script:
    - |
      if [ $? -ne 0 ]; then
        echo "Build failed, triggering agent analysis..."
        cat build.log > /tmp/build_log.txt
        python3 /opt/ci-agent/main.py < /tmp/pr_diff.txt
      python3 /opt/ci-agent/main.py <

# .github/actions/agent-analysis/action.yml
name: 'CI/CD Agent Analysis'
description: 'Run llama.cpp-powered agent analysis on failures'
inputs:
  event-type:
    description: 'Type of event (build_failure or pull_request)'
    required: true
  log-content:
    description: 'Build log content for failure analysis'
    required: false
  diff-content:
    description: 'PR diff for review'
    required: false

runs:
  using: 'composite'
  steps:
    - name: Install dependencies
      run: |
        pip install pyyaml requests
      shell: bash

    - name: Run agent analysis
      run: |
        python3 /opt/ci-agent/main.py <<'EVENT_EOF'
        {
          "type": "${{ inputs.event-type }}",
          "build_log": "${{ inputs.log-content }}",
          "diff": "${{ inputs.diff-content }}",
          "repo": "${{ github.repository }}",
          "branch": "${{ github.ref_name }}",
          "commit_sha": "${{ github.sha }}"
        }
        EVENT_EOF
      shell: bash

# agent/memory.py
import sqlite3
import json
from datetime import datetime
from typing import Dict, Any, Optional, List

class AgentMemory:
    def __init__(self, db_path: str = "agent_memory.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS analyses (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT,
                event_type TEXT,
                repo TEXT,
                branch TEXT,
                root_cause TEXT,
                category TEXT,
                analysis_json TEXT,
                outcome TEXT
            )
        """)
        self.conn.commit()

    def store(self, event_type: str, repo: str, branch: str,
              analysis: Dict[str, Any], outcome: str = "pending"):
        self.conn.execute(
            """INSERT INTO analyses 
               (timestamp, event_type, repo, branch, root_cause, 
                category, analysis_json, outcome)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
            (datetime.now().isoformat(), event_type, repo, branch,
             analysis.get("root_cause", ""), analysis.get("category", ""),
             json.dumps(analysis), outcome)
        )
        self.conn.commit()

    def find_similar(self, repo: str, category: str, 
                     limit: int = 5) -> List[Dict[str, Any]]:
        cursor = self.conn.execute(
            """SELECT analysis_json, outcome FROM analyses 
               WHERE repo = ? AND category = ?
               ORDER BY timestamp DESC LIMIT ?""",
            (repo, category, limit)
        )
        results = []
        for row in cursor:
            results.append({
                "analysis": json.loads(row[0]),
                "outcome": row[1]
            })
        return results

    def update_outcome(self, analysis_id: int, outcome: str):
        self.conn.execute(
            "UPDATE analyses SET outcome = ? WHERE id = ?",
            (outcome, analysis_id)
        )
        self.conn.commit()

    def close(self):
        self.conn.close()

# Multi-model configuration
triage_server:
  url: "http://localhost:8080"
  model: "llama-3.2-3b-instruct"
  temperature: 0.1
  max_tokens: 256

analysis_server:
  url: "http://localhost:8081"
  model: "codellama-13b-instruct"
  temperature: 0.2
  max_tokens: 2048