Chapter 11: Practical Project 2: Automated R&D and Operations Agent

In software development, a large amount of repetitive work consumes engineers' time: code reviews, bug localization, log analysis, and troubleshooting. If an Agent can take on these tasks, engineers can focus on more creative work. This chapter will build an automated R&D and operations Agent covering code review, log analysis, CI/CD integration, and knowledge accumulation.

11.1 Scenario Definition: Automated Code Review and Bug Fix Assistant

Project Goals

Capability	Input	Output
Code Review	Git Diff / PR Description	Review Comments + Improvement Suggestions
Bug Localization	Error Logs + Code Repository	Possible Causes + Fix Plan
Log Analysis	Application Log Files	Anomaly Patterns + Root Cause Speculation
Fault Response	Alert Information	Diagnostic Steps + Repair Actions

Architecture Design

Developer Submits PR / Alert Triggered
       |
  [Event Listener] ──> GitHub Webhook / Monitoring System
       |
  [Context Building] ──> Read Code/Logs/History
       |
  [Agent Analysis] ──> Review/Diagnose/Fix
       |
  [Result Output] ──> PR Comment / Ticket / Notification
       |
  [Knowledge Accumulation] ──> Write to Knowledge Base

Technology Selection

Component	Selection	Reason
LLM	GPT-4o	Code understanding requires strong reasoning
Code Hosting	GitHub	Mature Webhook ecosystem
CI/CD	GitHub Actions	Seamless integration with GitHub
Log Storage	ELK / Loki	Structured log retrieval
Knowledge Base	Chroma	Vector storage for incident cases

11.2 Context Building: Parsing Code Repositories and Dependencies

Git Repository Operations

from langchain_core.tools import tool
import subprocess

@tool
def get_git_diff(repo_path: str, base_branch: str = "main") -> str:
    """Get the code changes of the current branch relative to the base branch in a Git repository.

    Args:
        repo_path: Local path of the code repository
        base_branch: Base branch name, default is main
    """
    try:
        result = subprocess.run(
            ["git", "diff", base_branch, "--stat"],
            capture_output=True, text=True, cwd=repo_path
        )
        stat = result.stdout

        result = subprocess.run(
            ["git", "diff", base_branch],
            capture_output=True, text=True, cwd=repo_path
        )
        diff = result.stdout

        # Limit diff size to avoid exceeding context window
        if len(diff) > 10000:
            diff = diff[:10000] + "\n... (diff truncated, total {} characters)".format(len(diff))

        return f"Change Statistics:\n{stat}\n\nDetailed Diff:\n{diff}"
    except Exception as e:
        return f"Failed to get diff: {e}"

@tool
def read_file_content(file_path: str, repo_path: str = "", start_line: int = 1, end_line: int = -1) -> str:
    """Read the content of a code file.

    Args:
        file_path: Relative file path
        repo_path: Repository root directory
        start_line: Starting line number (starting from 1)
        end_line: Ending line number (-1 means to the end of the file)
    """
    full_path = os.path.join(repo_path, file_path)
    try:
        with open(full_path, 'r', encoding='utf-8') as f:
            lines = f.readlines()
        if end_line == -1:
            end_line = len(lines)
        selected = lines[start_line - 1:end_line]
        return "".join(selected)
    except Exception as e:
        return f"Failed to read file: {e}"

@tool
def list_directory(dir_path: str, repo_path: str = "") -> str:
    """List directory contents.

    Args:
        dir_path: Relative directory path
        repo_path: Repository root directory
    """
    full_path = os.path.join(repo_path, dir_path)
    try:
        entries = os.listdir(full_path)
        dirs = [e for e in entries if os.path.isdir(os.path.join(full_path, e))]
        files = [e for e in entries if os.path.isfile(os.path.join(full_path, e))]
        return f"Directory: {dir_path}\nSubdirectories: {dirs}\nFiles: {files}"
    except Exception as e:
        return f"Failed to list directory: {e}"

Dependency Analysis

@tool
def analyze_python_imports(file_path: str, repo_path: str = "") -> str:
    """Analyze the import dependencies of a Python file.

    Args:
        file_path: Relative path of the Python file
        repo_path: Repository root directory
    """
    import ast

    full_path = os.path.join(repo_path, file_path)
    try:
        with open(full_path, 'r') as f:
            tree = ast.parse(f.read())

        imports = []
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                for alias in node.names:
                    imports.append(alias.name)
            elif isinstance(node, ast.ImportFrom):
                imports.append(f"{node.module}.*" if node.module else "relative_import")

        return f"Dependencies of file {file_path}:\n" + "\n".join(f"  - {imp}" for imp in imports)
    except Exception as e:
        return f"Failed to analyze dependencies: {e}"

11.3 Interactive Debugging: How the Agent Reads Logs and Restarts Services

Log Analysis Tools

@tool
def search_logs(pattern: str, log_path: str = "/var/log/app.log", 
                context_lines: int = 3, max_results: int = 20) -> str:
    """Search for matching lines in a log file.

    Args:
        pattern: Search pattern (supports regular expressions)
        log_path: Log file path
        context_lines: Number of context lines
        max_results: Maximum number of matches to return
    """
    try:
        result = subprocess.run(
            ["grep", "-n", "-E", f"-C{context_lines}", pattern, log_path],
            capture_output=True, text=True, timeout=10
        )
        lines = result.stdout.split("\n")
        if len(lines) > max_results:
            lines = lines[:max_results] + [f"... (total {len(lines)} lines, truncated display)"]
        return "\n".join(lines) if lines else "No matching logs found"
    except Exception as e:
        return f"Failed to search logs: {e}"

@tool
def analyze_error_patterns(log_path: str = "/var/log/app.log", 
                           hours: int = 24) -> str:
    """Analyze error patterns in logs.

    Args:
        log_path: Log file path
        hours: Analyze logs from the last how many hours
    """
    try:
        # Extract ERROR level logs
        result = subprocess.run(
            ["grep", "-c", "ERROR", log_path],
            capture_output=True, text=True
        )
        error_count = result.stdout.strip()

        # Extract the most common error types
        result = subprocess.run(
            ["grep", "ERROR", log_path, "|", "awk", "{print $NF}", "|",
             "sort", "|", "uniq", "-c", "|", "sort", "-rn", "|", "head", "-10"],
            capture_output=True, text=True, shell=True
        )
        top_errors = result.stdout

        return f"Error statistics for the past {hours} hours:\nTotal errors: {error_count}\nTop error types:\n{top_errors}"
    except Exception as e:
        return f"Failed to analyze error patterns: {e}"

Service Management Tools

@tool
def restart_service(service_name: str) -> str:
    """Restart a system service (requires confirmation).

    Args:
        service_name: Service name
    """
    # In production environments, this operation requires secondary confirmation
    try:
        result = subprocess.run(
            ["sudo", "systemctl", "restart", service_name],
            capture_output=True, text=True, timeout=30
        )
        if result.returncode == 0:
            return f"Service {service_name} restarted successfully"
        return f"Restart failed: {result.stderr}"
    except Exception as e:
        return f"Service restart exception: {e}"

@tool
def check_service_status(service_name: str) -> str:
    """Check the running status of a service.

    Args:
        service_name: Service name
    """
    try:
        result = subprocess.run(
            ["systemctl", "status", service_name, "--no-pager"],
            capture_output=True, text=True
        )
        return result.stdout[:2000]
    except Exception as e:
        return f"Failed to check status: {e}"

Building the Operations Agent

system_prompt = """You are a senior DevOps engineer assistant.

## Responsibilities
- Analyze code changes and perform automatic code reviews
- Analyze logs and locate root causes of failures
- Restart services when necessary (requires confirmation first)
- Accumulate incident cases into the knowledge base

## Code Review Standards
- Security vulnerabilities (SQL injection, XSS, hardcoded keys)
- Performance issues (N+1 queries, memory leaks)
- Code conventions (naming, comments, error handling)
- Best practices (SOLID principles, DRY principle)

## Log Analysis Method
1. First, count error frequencies to find high-frequency errors
2. Analyze error context to find trigger conditions
3. Trace the call chain to locate the root cause
4. Search historical cases to see if it's a known issue

## Security Rules
- Must confirm with the user before restarting services
- Do not execute any irreversible operations (such as deleting data)
- Sensitive information (keys, passwords) must be masked
"""

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

tools = [
    get_git_diff, read_file_content, list_directory, 
    analyze_python_imports, search_logs, analyze_error_patterns,
    check_service_status, restart_service,
]

llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_openai_tools_agent(llm, tools, prompt)
devops_agent = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=10)

11.4 CI/CD Pipeline Integration: Combining GitHub Actions with the Agent

GitHub Webhook Listener

from fastapi import FastAPI, Request
import hmac, hashlib

app = FastAPI()

WEBHOOK_SECRET = os.getenv("GITHUB_WEBHOOK_SECRET")

def verify_signature(payload: bytes, signature: str) -> bool:
    """Verify GitHub Webhook signature"""
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode(), payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.post("/webhook/github")
async def handle_github_webhook(request: Request):
    payload = await request.body()
    signature = request.headers.get("X-Hub-Signature-256", "")

    if not verify_signature(payload, signature):
        return {"status": "unauthorized"}

    data = json.loads(payload)
    event = request.headers.get("X-GitHub-Event")

    if event == "pull_request" and data["action"] in ["opened", "synchronize"]:
        # Trigger code review when a PR is created or updated
        pr_info = {
            "repo": data["repository"]["full_name"],
            "pr_number": data["number"],
            "diff_url": data["pull_request"]["diff_url"],
        }
        # Asynchronously trigger Agent review
        review_result = devops_agent.invoke({
            "input": f"Review code changes for PR #{pr_info['pr_number']}: {pr_info['diff_url']}"
        })
        # Post review results as a comment on the PR
        post_pr_comment(pr_info, review_result["output"])

    return {"status": "ok"}

GitHub Actions Integration

# .github/workflows/ai-code-review.yml
name: AI Code Review

on:
  pull_request:
    types: [opened, synchronize]

jobs:
  ai-review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - name: Get Diff
        id: diff
        run: |
          DIFF=$(git diff origin/main...HEAD)
          echo "diff<<EOF" >> $GITHUB_OUTPUT
          echo "$DIFF" >> $GITHUB_OUTPUT
          echo "EOF" >> $GITHUB_OUTPUT

      - name: AI Review
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python3 scripts/ai_review.py --diff "${{ steps.diff.outputs.diff }}" --pr ${{ github.event.pull_request.number }}

      - name: Post Review Comment
        env:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        run: |
          python3 scripts/post_comment.py --pr ${{ github.event.pull_request.number }}

Auto-commenting Review Results

import requests

def post_pr_comment(pr_info: dict, review_content: str):
    """Post review results as a comment on the PR"""
    GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
    repo = pr_info["repo"]
    pr_number = pr_info["pr_number"]

    url = f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments"
    headers = {
        "Authorization": f"Bearer {GITHUB_TOKEN}",
        "Accept": "application/vnd.github.v3+json",
    }

    comment_body = f"## AI Code Review\n\n{review_content}"
    requests.post(url, json={"body": comment_body}, headers=headers)

Reference Documentation: GitHub Webhooks Documentation

11.5 Operations Knowledge Accumulation: Incident Postmortems and Automatic Knowledge Base Updates

Structuring Incident Cases

After each incident is handled, the Agent automatically structures the case and writes it to the knowledge base:

from pydantic import BaseModel
from datetime import datetime

class IncidentCase(BaseModel):
    """Incident Case"""
    incident_id: str
    title: str
    symptoms: list[str]           # Incident symptoms
    root_cause: str               # Root cause
    resolution: str               # Solution
    timeline: list[dict]          # Timeline
    affected_services: list[str]  # Affected services
    severity: str                 # Severity level
    tags: list[str]               # Tags (for retrieval)
    created_at: str

class IncidentKnowledgeBase:
    """Incident Knowledge Base"""

    def __init__(self, vectorstore):
        self.vectorstore = vectorstore
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)

    def add_incident(self, case: IncidentCase):
        """Add an incident case to the knowledge base"""
        # Convert the case into retrievable text
        doc_text = f"""
Incident Title: {case.title}
Symptoms: {"; ".join(case.symptoms)}
Root Cause: {case.root_cause}
Solution: {case.resolution}
Severity Level: {case.severity}
Tags: {", ".join(case.tags)}
"""
        self.vectorstore.add_texts(
            texts=[doc_text],
            metadatas=[{
                "incident_id": case.incident_id,
                "severity": case.severity,
                "services": ",".join(case.affected_services),
                "created_at": case.created_at,
            }]
        )

    def search_similar_incidents(self, symptoms: str, k: int = 3) -> list[str]:
        """Search for similar incident cases"""
        retriever = self.vectorstore.as_retriever(search_kwargs={"k": k})
        docs = retriever.invoke(symptoms)
        return [doc.page_content for doc in docs]

    def auto_create_incident(self, diagnosis: str, resolution: str) -> IncidentCase:
        """Automatically create an incident case from diagnosis and solution"""
        extract_prompt = f"""
Please extract a structured incident case from the following diagnostic information:

Diagnosis Process: {diagnosis}
Solution: {resolution}

Please output in JSON format, including fields:
title, symptoms(array), root_cause, resolution, affected_services(array), severity(high/medium/low), tags(array)
"""
        response = self.llm.invoke(extract_prompt)
        data = json.loads(response.content)

        return IncidentCase(
            incident_id=f"INC-{datetime.now().strftime('%Y%m%d%H%M%S')}",
            title=data["title"],
            symptoms=data["symptoms"],
            root_cause=data["root_cause"],
            resolution=data["resolution"],
            timeline=[],
            affected_services=data["affected_services"],
            severity=data["severity"],
            tags=data["tags"],
            created_at=datetime.now().isoformat(),
        )

Automated Incident Postmortems

class PostMortemGenerator:
    """Automatically generate incident postmortem reports"""

    def __init__(self, llm):
        self.llm = llm

    def generate(self, incident: IncidentCase) -> str:
        template = f"""# Incident Postmortem Report: {incident.title}

## Basic Information
- Severity Level: {incident.severity}
- Affected Services: {", ".join(incident.affected_services)}
- Occurrence Time: {incident.created_at}

## Incident Symptoms
{chr(10).join(f"- {s}" for s in incident.symptoms)}

## Root Cause Analysis
{incident.root_cause}

## Solution
{incident.resolution}

## Improvement Measures
Based on the above information, please propose 3-5 improvement measures to prevent similar issues from happening again.
"""
        # Use LLM to supplement improvement measures
        full_report = self.llm.invoke(template).content
        return full_report

Complete Operations Agent Workflow

class DevOpsWorkflow:
    """Complete DevOps Agent Workflow"""

    def __init__(self, agent_executor, knowledge_base):
        self.agent = agent_executor
        self.kb = knowledge_base

    def handle_incident(self, alert_info: str) -> dict:
        """Complete process for handling an alert event"""
        # 1. First, search historical cases
        similar_cases = self.kb.search_similar_incidents(alert_info)
        context = f"Similar Historical Cases:\n{chr(10).join(similar_cases)}" if similar_cases else "No similar historical cases"

        # 2. Agent diagnosis
        diagnosis = self.agent.invoke({
            "input": f"Alert Information: {alert_info}\n\n{context}\n\nPlease diagnose and provide a solution."
        })

        # 3. Automatically create an incident case
        incident = self.kb.auto_create_incident(
            diagnosis["output"],
            "To be confirmed"  # Solution needs manual confirmation before updating
        )

        # 4. Store in knowledge base
        self.kb.add_incident(incident)

        return {
            "incident_id": incident.incident_id,
            "diagnosis": diagnosis["output"],
            "similar_past_incidents": len(similar_cases),
        }

Chapter Summary

Module	Core Implementation	Key Points
Scenario Definition	Code Review + Log Analysis + Fault Response	Clearly define Agent responsibility boundaries
Context Building	Git Operations + File Reading + Dependency Analysis	Agent needs sufficient code context
Interactive Debugging	Log Search + Error Pattern Analysis + Service Management	Dangerous operations require secondary confirmation
CI/CD Integration	Webhook + GitHub Actions	Automatic triggering is more reliable than manual triggering
Knowledge Accumulation	Structured Cases + Vector Retrieval	Every incident is knowledge accumulation

In the next chapter, we will build the third practical project—a personalized education tutoring Agent.