AI Infrastructure Operations Copilot

Build a multi-agent RAG pipeline with LangGraph for incident analysis.

Introduction

30 Second Summary

When a production server crashes at 2 AM, engineers scramble through runbooks and past incidents searching for the right fix. What if an AI system could instantly pull relevant context and walk you through the resolution?

In this project, you will build a multi-agent AI pipeline that analyzes infrastructure incidents, retrieves relevant runbook context, and returns structured remediation steps with confidence scores.

What You'll Build

A local FastAPI server that orchestrates 5 specialized AI agents through a LangGraph state graph to analyze infrastructure incidents and produce actionable responses.

By the end of this project, you'll have:

  • A LangGraph StateGraph pipeline with 5 agent nodes that process incidents through query analysis, retrieval, root cause identification, resolution generation, and validation.
  • A ChromaDB vector store ingested with sample infrastructure runbooks that the pipeline queries for relevant context.
  • A FastAPI POST endpoint that accepts incident descriptions and returns structured JSON with confidence scores, root cause analysis, and source-referenced remediation steps.
  • Secret Mission: Add a conditional validation loop where the pipeline re-analyzes incidents when confidence is too low, showcasing LangGraph's conditional routing.

Are there any prerequisites?

You should be comfortable with Python basics (functions, dictionaries, imports) and have some familiarity with REST APIs. Experience with FastAPI is helpful but not required.

Before We Start

Before We Start...

Set Up Your Environment

Your AI operations copilot will use several specialized libraries working together. LangGraph orchestrates the multi-agent pipeline, ChromaDB stores and retrieves runbook documents, and FastAPI exposes everything as a REST endpoint.

Before any of that can come together, you need a clean project environment with the right packages installed and a working API key. In this step, you will create the project folder, install dependencies, and verify your connection to Gemini.

In this step, get ready to:
  • Create your project directory and a Python virtual environment.
  • Install all packages needed for the multi-agent pipeline.
  • Get a Gemini API key and verify it works.
Create your project directory and virtual environment

A virtual environment gives your project its own isolated set of Python packages. This prevents version conflicts with other projects on your machine.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir ops-copilot
cd ops-copilot
  • Confirm the folder exists by running:
ls

You should see ops-copilot listed in the output.

  • Create a Python virtual environment by running:
python3 -m venv venv

What does this command do?

python3 -m venv venv tells Python to create a virtual environment in a folder called venv inside your project directory.

This isolated environment means every package you install stays local to this project.

  • Activate the virtual environment by running:

macOS

source venv/bin/activate

Windows

venv\Scripts\Activate.ps1

Seeing an execution policy error?

Run this command first, then try the activate command again:

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

Make sure you ran the activate command from inside the ops-copilot folder. If you're on macOS, confirm you used source before the path.

help me activate my Python virtual environment

  • Open the ops-copilot folder in Cursor by running:
cursor .

What does cursor . do?

The cursor . command opens the current directory in Cursor. The dot (.) means "this folder."

Command not recognized?

If cursor is not recognized, open Cursor manually. Press Cmd+Space (macOS) or the Windows key (Windows), type Cursor, and press Enter. Then click File > Open Folder and select the ops-copilot folder on your Desktop.

Install all dependencies

Your project uses six packages that handle different responsibilities across the pipeline. You will install them all in a single command.

  • In Cursor, open a new terminal by pressing Ctrl+` (backtick key, below Escape).
  • Make sure (venv) still appears in your terminal prompt. If it does not, re-run the activate command from the previous substep.
  • Install the project dependencies by running:
pip install langgraph langchain-google-genai chromadb "fastapi[standard]" langchain-core python-dotenv

What are these packages?

  • langgraph builds and runs the multi-agent state graph.
  • langchain-google-genai connects to the Gemini LLM.
  • chromadb provides the local vector database for runbook storage.
  • fastapi[standard] is the web framework that exposes the API endpoint.
  • langchain-core provides base abstractions used by LangGraph and the Gemini integration.
  • python-dotenv loads environment variables from a .env file.

The installation may take a minute or two as it downloads the packages and their dependencies.

  • Verify the packages installed correctly by running:
python -c "import langgraph; import chromadb; import fastapi; print('All packages installed')"

You should see All packages installed printed in your terminal.

Seeing an import error?

Make sure (venv) appears in your prompt. If you accidentally installed packages outside the virtual environment, activate it first, then re-run the pip install command.

help me fix a Python package import error after pip install

Get a Gemini API key and verify it works

The pipeline uses Google's Gemini model for all LLM calls. You will get a free API key from Google AI Studio and confirm it connects successfully.

  • Open Google AI Studio API Keys in your browser.
  • Sign in with your Google account.
  • Click Create API Key.
  • Copy the generated API key and keep it somewhere safe. You will only see it once.

Is this free?

Yes. Google AI Studio offers a free tier for the Gemini API with no credit card required. The usage in this project stays well within the free allowance.

  • In Cursor's file explorer on the left, right-click and select New File.
  • Name the file .env.
  • Add your API key to the file by pasting the following, replacing your-api-key-here with the key you just copied:
GOOGLE_API_KEY=your-api-key-here

Why a .env file?

Storing your API key in a .env file keeps it out of your source code. The python-dotenv package loads it into your environment at runtime so your code never hardcodes the key.

  • Save .env.

Now verify the key connects to Gemini successfully.

  • Create a new file called test_key.py and paste the following code:
import os
from dotenv import load_dotenv
from langchain_google_genai import ChatGoogleGenerativeAI

# Load the API key from .env
load_dotenv()

# Create the Gemini LLM instance
llm = ChatGoogleGenerativeAI(
    model="gemini-3.5-flash",
    google_api_key=os.getenv("GOOGLE_API_KEY"),
)

# Send a simple test prompt
response = llm.invoke("Say hello in one sentence.")
print(response.content)

What does this script do?

  • load_dotenv() reads .env and loads GOOGLE_API_KEY into the environment.
  • ChatGoogleGenerativeAI creates an LLM instance pointed at the gemini-3.5-flash model.
  • llm.invoke() sends a prompt and returns the model's response.
  • Save test_key.py and run it in your terminal:
python test_key.py

You should see a short greeting printed in your terminal, confirming the Gemini API key is working.

Seeing an authentication error?

Double-check that your .env file has no extra spaces around the = sign. The format must be exactly GOOGLE_API_KEY=your-key with no quotes around the key value.

Also confirm you copied the full key from Google AI Studio. If the key was truncated, generate a new one.

help me fix my Gemini API key authentication error

  • Once the test succeeds, delete test_key.py since it was only needed for verification. Right-click the file in Cursor's file explorer and select Delete.

✔️ Awesome, I've got everything!

Great. Double check you have saved your .env file.

ⓧ I'd like to double check the full code

Your .env file should look like this (with your real key in place of the placeholder):

GOOGLE_API_KEY=your-api-key-here

Your environment is ready to go. Next up, you will define the data models and load infrastructure runbooks into your vector database.

Define Data Models and Ingest Runbooks

Your environment is ready, and Gemini is responding to prompts. Now you need the building blocks that every agent in your pipeline will share.

Before writing any agent logic, you need two things: structured data shapes that define what flows through the pipeline, and a knowledge base the agents can query. In this step, you will create Pydantic models for the API layer, a TypedDict for the pipeline state, and a ChromaDB vector store loaded with infrastructure runbooks.

In this step, get ready to:
  • Define Pydantic response schemas and a typed pipeline state in models.py.
  • Create an ingestion script that loads 6 infrastructure runbooks into ChromaDB.
  • Run the script and verify the vector store contains all 6 documents.
Create the API and pipeline data models

Every agent in the pipeline reads from and writes to a shared state object. You also need Pydantic models that define what the FastAPI endpoint accepts and returns.

  • In Cursor, create a new file called models.py in your ops-copilot folder.
  • Add the imports and Pydantic request/response models by pasting this code:
from typing import TypedDict
from pydantic import BaseModel, Field


class IncidentRequest(BaseModel):
    description: str = Field(..., description="Description of the infrastructure incident")


class IncidentAnalysis(BaseModel):
    root_cause: str = Field(..., description="Identified root cause of the incident")
    confidence_score: float = Field(..., ge=0, le=1, description="Confidence in the analysis")
    remediation_steps: list[str] = Field(..., description="Ordered steps to resolve the incident")
    sources: list[str] = Field(..., description="Runbook sources referenced")
    severity: str = Field(..., description="Incident severity: low, medium, high, critical")

What does this code do?

IncidentRequest is the shape of incoming API requests. It has a single description field where the user describes the infrastructure incident.

IncidentAnalysis is the structured response. It includes a confidence_score constrained between 0 and 1, a list of remediation_steps, and sources referencing which runbooks informed the analysis.

  • Now add the PipelineState class below IncidentAnalysis. This defines the shared state that flows between all agents in the LangGraph pipeline:
class PipelineState(TypedDict):
    query: str
    keywords: list[str]
    context_docs: list[str]
    sources: list[str]
    root_cause: str
    resolution: list[str]
    confidence: float
    severity: str

What is PipelineState?

This TypedDict is the contract between all 5 agents in your pipeline. Each agent reads the fields it needs, performs its work, and writes its output back to the shared state.

LangGraph uses this type to track what data is available at each node. Each field starts empty and gets populated as the state moves through the graph.

  • Save models.py.
  • Verify it has no import errors by running this in your terminal:
python -c "from models import IncidentAnalysis, PipelineState; print('Models OK')"

You should see Models OK printed with no errors.

Seeing an ImportError?

Make sure your virtual environment is still active. You should see (venv) or similar at the start of your terminal prompt. If not, activate it again.

Also confirm you saved models.py in the ops-copilot folder (not a subfolder).

help me fix an ImportError when importing my models

✔️ Awesome, I've got everything!

Great. Double check you have saved models.py before moving on.

ⓧ I'd like to double check the full code

from typing import TypedDict
from pydantic import BaseModel, Field


class IncidentRequest(BaseModel):
    description: str = Field(..., description="Description of the infrastructure incident")


class IncidentAnalysis(BaseModel):
    root_cause: str = Field(..., description="Identified root cause of the incident")
    confidence_score: float = Field(..., ge=0, le=1, description="Confidence in the analysis")
    remediation_steps: list[str] = Field(..., description="Ordered steps to resolve the incident")
    sources: list[str] = Field(..., description="Runbook sources referenced")
    severity: str = Field(..., description="Incident severity: low, medium, high, critical")


class PipelineState(TypedDict):
    query: str
    keywords: list[str]
    context_docs: list[str]
    sources: list[str]
    root_cause: str
    resolution: list[str]
    confidence: float
    severity: str
Create the runbook ingestion script

Your agents need a knowledge base to search. You will create a script that loads 6 sample infrastructure runbooks into a ChromaDB vector store.

ChromaDB handles embedding automatically using its built-in default embedding function. You just provide the raw text and it converts everything into searchable vectors.

  • Create a new file called ingest.py in your ops-copilot folder.
  • Add the import and the RUNBOOKS data list by pasting this code:
import chromadb

RUNBOOKS = [
    {
        "id": "kb-001",
        "content": (
            "Runbook: Kubernetes Pod CrashLoopBackOff\n"
            "Symptoms: Pod repeatedly crashes and restarts. Status shows CrashLoopBackOff.\n"
            "Root Causes: OOMKilled (memory limit exceeded), missing environment variables, "
            "failed health checks, application startup errors, missing dependencies.\n"
            "Resolution Steps:\n"
            "1. Check pod logs: kubectl logs <pod-name> --previous\n"
            "2. Describe pod for events: kubectl describe pod <pod-name>\n"
            "3. Check resource limits in deployment manifest\n"
            "4. Verify all ConfigMaps and Secrets are mounted\n"
            "5. Test container image locally with same env vars\n"
            "Severity: High"
        ),
        "metadata": {"category": "kubernetes", "severity": "high"},
    },
    {
        "id": "kb-002",
        "content": (
            "Runbook: High CPU Usage on Application Servers\n"
            "Symptoms: CPU utilization above 90% sustained for more than 5 minutes.\n"
            "Root Causes: Infinite loops in application code, unoptimized database queries, "
            "thread pool exhaustion, garbage collection storms, cryptomining malware.\n"
            "Resolution Steps:\n"
            "1. Identify top processes: top -c or htop\n"
            "2. Check application thread dumps\n"
            "3. Review recent deployments for code changes\n"
            "4. Scale horizontally if load-related\n"
            "5. Profile application with async-profiler or perf\n"
            "Severity: Medium"
        ),
        "metadata": {"category": "compute", "severity": "medium"},
    },
    {
        "id": "kb-003",
        "content": (
            "Runbook: Database Connection Pool Exhaustion\n"
            "Symptoms: Application throws connection timeout errors. Database shows max "
            "connections reached.\n"
            "Root Causes: Connection leaks in application code, long-running queries holding "
            "connections, misconfigured pool size, sudden traffic spike.\n"
            "Resolution Steps:\n"
            "1. Check active connections: SELECT count(*) FROM pg_stat_activity\n"
            "2. Identify long-running queries and terminate if needed\n"
            "3. Review connection pool settings (min, max, idle timeout)\n"
            "4. Check for connection leaks in application code\n"
            "5. Increase max_connections if hardware supports it\n"
            "Severity: High"
        ),
        "metadata": {"category": "database", "severity": "high"},
    },
    {
        "id": "kb-004",
        "content": (
            "Runbook: Disk Space Alert - Filesystem Above 90%\n"
            "Symptoms: Monitoring alerts on disk usage. Application may fail to write logs.\n"
            "Root Causes: Log files not rotated, large temp files, database WAL accumulation, "
            "container image layer buildup, core dumps filling disk.\n"
            "Resolution Steps:\n"
            "1. Identify large files: du -sh /* | sort -rh | head -20\n"
            "2. Check and clean log files older than 7 days\n"
            "3. Clear temp directories and old container images\n"
            "4. Implement log rotation with logrotate\n"
            "5. Set up automated disk cleanup cron jobs\n"
            "Severity: Medium"
        ),
        "metadata": {"category": "storage", "severity": "medium"},
    },
    {
        "id": "kb-005",
        "content": (
            "Runbook: Memory Leak Detection and Resolution\n"
            "Symptoms: Application memory usage grows continuously without releasing. "
            "Eventually leads to OOM kills.\n"
            "Root Causes: Unclosed resources (files, connections), growing caches without "
            "eviction, event listener accumulation, circular references.\n"
            "Resolution Steps:\n"
            "1. Monitor memory over time with Grafana/Prometheus\n"
            "2. Take heap dumps at intervals: jmap -dump:format=b,file=heap.bin <pid>\n"
            "3. Analyze with Eclipse MAT or VisualVM\n"
            "4. Check for unclosed resources in code review\n"
            "5. Implement memory limits and restart policies\n"
            "Severity: High"
        ),
        "metadata": {"category": "application", "severity": "high"},
    },
    {
        "id": "kb-006",
        "content": (
            "Runbook: Network Timeout Between Services\n"
            "Symptoms: Requests between microservices fail with timeout errors. "
            "Intermittent 504 Gateway Timeout.\n"
            "Root Causes: DNS resolution failures, network policy misconfiguration, "
            "service mesh sidecar issues, upstream service overloaded, TCP connection limits.\n"
            "Resolution Steps:\n"
            "1. Test connectivity: curl -v http://service-name:port/health\n"
            "2. Check DNS resolution: nslookup service-name\n"
            "3. Review network policies and security groups\n"
            "4. Check service mesh proxy logs (Envoy/Istio)\n"
            "5. Verify upstream service health and resource usage\n"
            "Severity: High"
        ),
        "metadata": {"category": "networking", "severity": "high"},
    },
]

What are these runbooks?

Each runbook is a structured document that an operations team would reference when an incident occurs. They cover symptoms, root causes, and resolution steps for common infrastructure problems.

The 6 documents cover Kubernetes pod crashes, CPU spikes, database connection exhaustion, disk space alerts, memory leaks, and network timeouts. Each has a metadata tag for category and severity that ChromaDB stores alongside the document text.

  • Now add the ingestion function and main block below the RUNBOOKS list:
def ingest_runbooks():
    client = chromadb.PersistentClient(path="./chroma_data")
    collection = client.get_or_create_collection(name="ops_runbooks")

    collection.add(
        ids=[doc["id"] for doc in RUNBOOKS],
        documents=[doc["content"] for doc in RUNBOOKS],
        metadatas=[doc["metadata"] for doc in RUNBOOKS],
    )

    print(f"Ingested {collection.count()} documents into 'ops_runbooks' collection.")


if __name__ == "__main__":
    ingest_runbooks()

What does this function do?

PersistentClient creates a ChromaDB instance that saves data to disk in a chroma_data folder. This means your runbooks survive between script runs.

get_or_create_collection creates the ops_runbooks collection if it does not exist, or returns the existing one. The collection.add call inserts all 6 documents with their IDs, text content, and metadata in one batch.

  • Save ingest.py.

Seeing a red underline on chromadb?

If Cursor shows a warning on the import chromadb line, make sure your Cursor Python interpreter is set to the virtual environment in your ops-copilot folder. The package is installed in the venv, not globally.

help me fix Cursor not recognizing my chromadb import

✔️ Awesome, I've got everything!

Great. Double check you have saved ingest.py before moving on.

ⓧ I'd like to double check the full code

import chromadb

RUNBOOKS = [
    {
        "id": "kb-001",
        "content": (
            "Runbook: Kubernetes Pod CrashLoopBackOff\n"
            "Symptoms: Pod repeatedly crashes and restarts. Status shows CrashLoopBackOff.\n"
            "Root Causes: OOMKilled (memory limit exceeded), missing environment variables, "
            "failed health checks, application startup errors, missing dependencies.\n"
            "Resolution Steps:\n"
            "1. Check pod logs: kubectl logs <pod-name> --previous\n"
            "2. Describe pod for events: kubectl describe pod <pod-name>\n"
            "3. Check resource limits in deployment manifest\n"
            "4. Verify all ConfigMaps and Secrets are mounted\n"
            "5. Test container image locally with same env vars\n"
            "Severity: High"
        ),
        "metadata": {"category": "kubernetes", "severity": "high"},
    },
    {
        "id": "kb-002",
        "content": (
            "Runbook: High CPU Usage on Application Servers\n"
            "Symptoms: CPU utilization above 90% sustained for more than 5 minutes.\n"
            "Root Causes: Infinite loops in application code, unoptimized database queries, "
            "thread pool exhaustion, garbage collection storms, cryptomining malware.\n"
            "Resolution Steps:\n"
            "1. Identify top processes: top -c or htop\n"
            "2. Check application thread dumps\n"
            "3. Review recent deployments for code changes\n"
            "4. Scale horizontally if load-related\n"
            "5. Profile application with async-profiler or perf\n"
            "Severity: Medium"
        ),
        "metadata": {"category": "compute", "severity": "medium"},
    },
    {
        "id": "kb-003",
        "content": (
            "Runbook: Database Connection Pool Exhaustion\n"
            "Symptoms: Application throws connection timeout errors. Database shows max "
            "connections reached.\n"
            "Root Causes: Connection leaks in application code, long-running queries holding "
            "connections, misconfigured pool size, sudden traffic spike.\n"
            "Resolution Steps:\n"
            "1. Check active connections: SELECT count(*) FROM pg_stat_activity\n"
            "2. Identify long-running queries and terminate if needed\n"
            "3. Review connection pool settings (min, max, idle timeout)\n"
            "4. Check for connection leaks in application code\n"
            "5. Increase max_connections if hardware supports it\n"
            "Severity: High"
        ),
        "metadata": {"category": "database", "severity": "high"},
    },
    {
        "id": "kb-004",
        "content": (
            "Runbook: Disk Space Alert - Filesystem Above 90%\n"
            "Symptoms: Monitoring alerts on disk usage. Application may fail to write logs.\n"
            "Root Causes: Log files not rotated, large temp files, database WAL accumulation, "
            "container image layer buildup, core dumps filling disk.\n"
            "Resolution Steps:\n"
            "1. Identify large files: du -sh /* | sort -rh | head -20\n"
            "2. Check and clean log files older than 7 days\n"
            "3. Clear temp directories and old container images\n"
            "4. Implement log rotation with logrotate\n"
            "5. Set up automated disk cleanup cron jobs\n"
            "Severity: Medium"
        ),
        "metadata": {"category": "storage", "severity": "medium"},
    },
    {
        "id": "kb-005",
        "content": (
            "Runbook: Memory Leak Detection and Resolution\n"
            "Symptoms: Application memory usage grows continuously without releasing. "
            "Eventually leads to OOM kills.\n"
            "Root Causes: Unclosed resources (files, connections), growing caches without "
            "eviction, event listener accumulation, circular references.\n"
            "Resolution Steps:\n"
            "1. Monitor memory over time with Grafana/Prometheus\n"
            "2. Take heap dumps at intervals: jmap -dump:format=b,file=heap.bin <pid>\n"
            "3. Analyze with Eclipse MAT or VisualVM\n"
            "4. Check for unclosed resources in code review\n"
            "5. Implement memory limits and restart policies\n"
            "Severity: High"
        ),
        "metadata": {"category": "application", "severity": "high"},
    },
    {
        "id": "kb-006",
        "content": (
            "Runbook: Network Timeout Between Services\n"
            "Symptoms: Requests between microservices fail with timeout errors. "
            "Intermittent 504 Gateway Timeout.\n"
            "Root Causes: DNS resolution failures, network policy misconfiguration, "
            "service mesh sidecar issues, upstream service overloaded, TCP connection limits.\n"
            "Resolution Steps:\n"
            "1. Test connectivity: curl -v http://service-name:port/health\n"
            "2. Check DNS resolution: nslookup service-name\n"
            "3. Review network policies and security groups\n"
            "4. Check service mesh proxy logs (Envoy/Istio)\n"
            "5. Verify upstream service health and resource usage\n"
            "Severity: High"
        ),
        "metadata": {"category": "networking", "severity": "high"},
    },
]


def ingest_runbooks():
    client = chromadb.PersistentClient(path="./chroma_data")
    collection = client.get_or_create_collection(name="ops_runbooks")

    collection.add(
        ids=[doc["id"] for doc in RUNBOOKS],
        documents=[doc["content"] for doc in RUNBOOKS],
        metadatas=[doc["metadata"] for doc in RUNBOOKS],
    )

    print(f"Ingested {collection.count()} documents into 'ops_runbooks' collection.")


if __name__ == "__main__":
    ingest_runbooks()
Run the ingestion script and verify the vector store

With the script ready, you will now run it to populate ChromaDB. This creates the chroma_data directory and stores all 6 runbook documents as searchable vectors.

  • Run the ingestion script by executing this command in your terminal:
python ingest.py

You should see this output:

Ingested 6 documents into 'ops_runbooks' collection.
  • Verify the collection document count by running:
python -c "import chromadb; client = chromadb.PersistentClient(path='./chroma_data'); coll = client.get_or_create_collection('ops_runbooks'); print(f'Documents: {coll.count()}')"

You should see Documents: 6 confirming all runbooks are stored.

What is a vector store?

A vector database converts text into numerical representations (embeddings) that capture meaning. When your Retriever agent searches for "pod keeps crashing," ChromaDB finds the Kubernetes CrashLoopBackOff runbook because their embeddings are mathematically similar.

This is the "R" in RAG (Retrieval-Augmented Generation). Instead of relying solely on the LLM's training data, your agents retrieve relevant documents first and use them as context.

  • Confirm the chroma_data directory was created by running:
ls chroma_data

You should see files and folders that ChromaDB uses for persistent storage.

Seeing 0 documents or a DuplicateIDError?

If you ran the script more than once, ChromaDB may throw a duplicate ID error because the documents already exist. Delete the chroma_data folder and re-run python ingest.py to start fresh.

If you see 0 documents, make sure you saved ingest.py before running it.

help me troubleshoot my ChromaDB ingestion script

Your data layer is ready. You have typed models for the API and pipeline, plus a ChromaDB collection loaded with 6 runbooks. Next up, you will build the 5 agent nodes that read from this state and query this vector store.

Build the Multi-Agent LangGraph Pipeline

Your data models are defined and your vector store is loaded with runbook knowledge. Now it's time to build the core intelligence of your copilot.

In this step, you'll create five specialized agent functions that each transform a shared state object. Then you'll wire them together using a LangGraph StateGraph so each agent runs in sequence, passing its output to the next.

In this step, get ready to:
  • Create agent node functions that extract keywords, retrieve context, analyze root causes, generate resolutions, and validate confidence.
  • Wire agents into a LangGraph StateGraph with linear edges.
  • Test the pipeline end-to-end with a sample incident.
Create the agent node functions

Each agent in this pipeline is a plain Python function. It receives the full pipeline state (a TypedDict), does its job, and returns a dictionary with only the keys it changed. LangGraph merges that partial dict back into the shared state before calling the next node.

  • Install one additional dependency needed for loading environment variables from your .env file by running:
pip install python-dotenv
  • Create a new file called agents.py in your ops-copilot folder.
  • Add the module setup that initializes the LLM and connects to your ChromaDB collection by copying the following code:
import os
from dotenv import load_dotenv
import chromadb
from langchain_google_genai import ChatGoogleGenerativeAI
from models import PipelineState

load_dotenv()

llm = ChatGoogleGenerativeAI(
    model="gemini-3.5-flash",
    google_api_key=os.getenv("GOOGLE_API_KEY"),
    temperature=0.2,
)

chroma_client = chromadb.PersistentClient(path="./chroma_data")
collection = chroma_client.get_or_create_collection(name="ops_runbooks")

What does this code do?

  • load_dotenv() reads your .env file and loads GOOGLE_API_KEY into the environment so the LLM can authenticate.
  • ChatGoogleGenerativeAI creates a Gemini LLM client with low temperature (0.2) for consistent, factual responses.
  • The PersistentClient reconnects to the same chroma_data directory you populated in the previous step, so the agents can query your runbooks.

You won't see a change yet. The next chunk adds the agent functions that use this setup.

  • Add the first two agent functions below the collection line in agents.py:

def query_analyzer(state: PipelineState) -> dict:
    query = state["query"]
    response = llm.invoke(
        f"Extract the key technical keywords from this infrastructure incident description. "
        f"Return only a comma-separated list of keywords, nothing else.\n\n"
        f"Incident: {query}"
    )
    keywords = [k.strip() for k in response.content.split(",")]
    return {"keywords": keywords}


def retriever(state: PipelineState) -> dict:
    query = state["query"]
    keywords = state.get("keywords", [])
    search_query = f"{query} {' '.join(keywords)}"

    results = collection.query(
        query_texts=[search_query],
        n_results=3,
    )

    docs = results["documents"][0] if results["documents"] else []
    ids = results["ids"][0] if results["ids"] else []

    return {"context_docs": docs, "sources": ids}

What do these agents do?

  • query_analyzer asks Gemini to extract technical keywords from the incident description. These keywords improve the vector search in the next step.
  • retriever combines the original query with extracted keywords, then queries ChromaDB for the 3 most relevant runbook documents. It returns both the document text and the source IDs.

You won't see a change yet. The next chunk adds the analysis and validation agents that complete the pipeline.

  • Add the remaining three agent functions below retriever in agents.py:

def root_cause_analyzer(state: PipelineState) -> dict:
    query = state["query"]
    context = "\n\n---\n\n".join(state.get("context_docs", []))

    response = llm.invoke(
        f"You are an infrastructure operations expert. Based on the following runbook "
        f"context, identify the most probable root cause for this incident.\n\n"
        f"Incident: {query}\n\n"
        f"Runbook Context:\n{context}\n\n"
        f"Provide a concise root cause analysis (2-3 sentences)."
    )
    return {"root_cause": response.content}


def resolution_generator(state: PipelineState) -> dict:
    query = state["query"]
    root_cause = state.get("root_cause", "")
    context = "\n\n---\n\n".join(state.get("context_docs", []))

    response = llm.invoke(
        f"You are an infrastructure operations expert. Given the incident and root cause, "
        f"provide specific remediation steps.\n\n"
        f"Incident: {query}\n"
        f"Root Cause: {root_cause}\n\n"
        f"Runbook Context:\n{context}\n\n"
        f"Return ONLY a numbered list of actionable remediation steps (3-5 steps)."
    )

    steps = [
        line.strip().lstrip("0123456789.)")
        .strip()
        for line in response.content.strip().split("\n")
        if line.strip() and line.strip()[0].isdigit()
    ]
    if not steps:
        steps = [response.content.strip()]

    return {"resolution": steps}


def validator(state: PipelineState) -> dict:
    context_docs = state.get("context_docs", [])
    root_cause = state.get("root_cause", "")
    resolution = state.get("resolution", [])

    score = 0.0
    if context_docs:
        score += 0.4
    if root_cause and len(root_cause) > 20:
        score += 0.3
    if resolution and len(resolution) >= 2:
        score += 0.3

    severity = "low"
    context_text = " ".join(context_docs).lower()
    if "critical" in context_text:
        severity = "critical"
    elif "high" in context_text:
        severity = "high"
    elif "medium" in context_text:
        severity = "medium"

    return {"confidence": round(score, 2), "severity": severity}

What do these agents do?

  • root_cause_analyzer sends the incident description plus retrieved runbook context to Gemini, asking it to pinpoint the most likely root cause.
  • resolution_generator takes the identified root cause and runbook context, then asks Gemini for numbered remediation steps. It parses the numbered lines into a clean list.
  • validator scores confidence (0.0 to 1.0) based on whether context documents were found, the root cause is substantive, and enough resolution steps were generated. It also determines severity by scanning the runbook text for keywords.
  • Save agents.py.
  • Verify all agents load correctly by running:
python -c "from agents import query_analyzer, retriever, root_cause_analyzer, resolution_generator, validator; print('All agents loaded successfully')"

You should see All agents loaded successfully printed in your terminal.

Seeing an import error?

  • If you see ModuleNotFoundError: No module named 'dotenv', confirm you ran pip install python-dotenv inside your virtual environment.
  • If you see an error about GOOGLE_API_KEY, make sure your .env file is in the ops-copilot folder (not a subfolder).
  • If ChromaDB throws a path error, confirm the chroma_data directory exists from running ingest.py in the previous step.

Help me debug my agents.py import error.

✔️ Awesome, I've got everything!

Great. Double check you've saved agents.py before moving on.

ⓧ I'd like to double check the full code

import os
from dotenv import load_dotenv
import chromadb
from langchain_google_genai import ChatGoogleGenerativeAI
from models import PipelineState

load_dotenv()

llm = ChatGoogleGenerativeAI(
    model="gemini-3.5-flash",
    google_api_key=os.getenv("GOOGLE_API_KEY"),
    temperature=0.2,
)

chroma_client = chromadb.PersistentClient(path="./chroma_data")
collection = chroma_client.get_or_create_collection(name="ops_runbooks")


def query_analyzer(state: PipelineState) -> dict:
    query = state["query"]
    response = llm.invoke(
        f"Extract the key technical keywords from this infrastructure incident description. "
        f"Return only a comma-separated list of keywords, nothing else.\n\n"
        f"Incident: {query}"
    )
    keywords = [k.strip() for k in response.content.split(",")]
    return {"keywords": keywords}


def retriever(state: PipelineState) -> dict:
    query = state["query"]
    keywords = state.get("keywords", [])
    search_query = f"{query} {' '.join(keywords)}"

    results = collection.query(
        query_texts=[search_query],
        n_results=3,
    )

    docs = results["documents"][0] if results["documents"] else []
    ids = results["ids"][0] if results["ids"] else []

    return {"context_docs": docs, "sources": ids}


def root_cause_analyzer(state: PipelineState) -> dict:
    query = state["query"]
    context = "\n\n---\n\n".join(state.get("context_docs", []))

    response = llm.invoke(
        f"You are an infrastructure operations expert. Based on the following runbook "
        f"context, identify the most probable root cause for this incident.\n\n"
        f"Incident: {query}\n\n"
        f"Runbook Context:\n{context}\n\n"
        f"Provide a concise root cause analysis (2-3 sentences)."
    )
    return {"root_cause": response.content}


def resolution_generator(state: PipelineState) -> dict:
    query = state["query"]
    root_cause = state.get("root_cause", "")
    context = "\n\n---\n\n".join(state.get("context_docs", []))

    response = llm.invoke(
        f"You are an infrastructure operations expert. Given the incident and root cause, "
        f"provide specific remediation steps.\n\n"
        f"Incident: {query}\n"
        f"Root Cause: {root_cause}\n\n"
        f"Runbook Context:\n{context}\n\n"
        f"Return ONLY a numbered list of actionable remediation steps (3-5 steps)."
    )

    steps = [
        line.strip().lstrip("0123456789.)")
        .strip()
        for line in response.content.strip().split("\n")
        if line.strip() and line.strip()[0].isdigit()
    ]
    if not steps:
        steps = [response.content.strip()]

    return {"resolution": steps}


def validator(state: PipelineState) -> dict:
    context_docs = state.get("context_docs", [])
    root_cause = state.get("root_cause", "")
    resolution = state.get("resolution", [])

    score = 0.0
    if context_docs:
        score += 0.4
    if root_cause and len(root_cause) > 20:
        score += 0.3
    if resolution and len(resolution) >= 2:
        score += 0.3

    severity = "low"
    context_text = " ".join(context_docs).lower()
    if "critical" in context_text:
        severity = "critical"
    elif "high" in context_text:
        severity = "high"
    elif "medium" in context_text:
        severity = "medium"

    return {"confidence": round(score, 2), "severity": severity}
Wire the agents into a StateGraph

A StateGraph is LangGraph's way of defining which agents run and in what order. You register each agent as a named node, then draw edges between them to create the execution flow.

  • Create a new file called graph.py in your ops-copilot folder.
  • Define the graph by copying the following code:
from langgraph.graph import StateGraph, START, END
from models import PipelineState
from agents import (
    query_analyzer,
    retriever,
    root_cause_analyzer,
    resolution_generator,
    validator,
)

workflow = StateGraph(PipelineState)

workflow.add_node("query_analyzer", query_analyzer)
workflow.add_node("retriever", retriever)
workflow.add_node("root_cause_analyzer", root_cause_analyzer)
workflow.add_node("resolution_generator", resolution_generator)
workflow.add_node("validator", validator)

workflow.add_edge(START, "query_analyzer")
workflow.add_edge("query_analyzer", "retriever")
workflow.add_edge("retriever", "root_cause_analyzer")
workflow.add_edge("root_cause_analyzer", "resolution_generator")
workflow.add_edge("resolution_generator", "validator")
workflow.add_edge("validator", END)

graph = workflow.compile()

What does this code do?

  • StateGraph(PipelineState) creates a graph whose shared state matches the PipelineState TypedDict you defined in models.py.
  • add_node registers each agent function under a string name.
  • add_edge draws a connection from one node to the next, creating the linear pipeline: START to query_analyzer to retriever to root_cause_analyzer to resolution_generator to validator to END.
  • workflow.compile() finalizes the graph into an executable object you can invoke with an initial state dict.
  • Save graph.py.

You won't see a change yet. The next substep adds a test block and runs the full pipeline.

Test the pipeline end-to-end

Time to see your multi-agent pipeline in action. You'll add a test block to graph.py that invokes the compiled graph with a sample Kubernetes incident and prints the results.

  • Add the following test block at the bottom of graph.py, below the graph = workflow.compile() line:

if __name__ == "__main__":
    result = graph.invoke({
        "query": "Kubernetes pod CrashLoopBackOff on payment-service",
        "keywords": [],
        "context_docs": [],
        "sources": [],
        "root_cause": "",
        "resolution": [],
        "confidence": 0.0,
        "severity": "",
    })
    print(f"Root Cause: {result['root_cause']}")
    print(f"Resolution: {result['resolution']}")
    print(f"Confidence: {result['confidence']}")
    print(f"Sources: {result['sources']}")
    print(f"Severity: {result['severity']}")

What does this test do?

The if __name__ == "__main__" block only runs when you execute graph.py directly. It passes an initial state with the incident query and empty fields, then prints the final state after all 5 agents have processed it.

  • Save graph.py.
  • Run the pipeline test by executing:
python graph.py

You should see output with all five fields populated. The root cause will reference memory limits or missing environment variables. Resolution will be a list of remediation steps. Confidence should be 1.0 (all three scoring criteria met). Sources will list runbook IDs like kb-001.

Pipeline returning empty results or errors?

  • If you see a google.api_core authentication error, double check that your GOOGLE_API_KEY in .env is correct and has no extra spaces.
  • If confidence is 0.0 and all fields are empty, make sure you ran ingest.py in the previous step so the ops_runbooks collection has documents.
  • If you see KeyError on a state field, ensure the initial state dict in the test block includes all keys from PipelineState.

Help me debug my LangGraph pipeline test output.

✔️ Awesome, I've got everything!

Great. Double check you've saved graph.py before moving on.

ⓧ I'd like to double check the full code

from langgraph.graph import StateGraph, START, END
from models import PipelineState
from agents import (
    query_analyzer,
    retriever,
    root_cause_analyzer,
    resolution_generator,
    validator,
)

workflow = StateGraph(PipelineState)

workflow.add_node("query_analyzer", query_analyzer)
workflow.add_node("retriever", retriever)
workflow.add_node("root_cause_analyzer", root_cause_analyzer)
workflow.add_node("resolution_generator", resolution_generator)
workflow.add_node("validator", validator)

workflow.add_edge(START, "query_analyzer")
workflow.add_edge("query_analyzer", "retriever")
workflow.add_edge("retriever", "root_cause_analyzer")
workflow.add_edge("root_cause_analyzer", "resolution_generator")
workflow.add_edge("resolution_generator", "validator")
workflow.add_edge("validator", END)

graph = workflow.compile()

if __name__ == "__main__":
    result = graph.invoke({
        "query": "Kubernetes pod CrashLoopBackOff on payment-service",
        "keywords": [],
        "context_docs": [],
        "sources": [],
        "root_cause": "",
        "resolution": [],
        "confidence": 0.0,
        "severity": "",
    })
    print(f"Root Cause: {result['root_cause']}")
    print(f"Resolution: {result['resolution']}")
    print(f"Confidence: {result['confidence']}")
    print(f"Sources: {result['sources']}")
    print(f"Severity: {result['severity']}")

Your multi-agent pipeline is working end-to-end. Next up, you'll expose this pipeline as a FastAPI endpoint so you can send incident descriptions via HTTP and get structured JSON responses back.

Expose the Pipeline via FastAPI

Your multi-agent pipeline processes incidents correctly when invoked from a script. But a Python script isn't how real systems communicate.

In this step, you'll wrap the pipeline in a FastAPI endpoint so any service can submit an incident description over HTTP and receive structured JSON back. This turns your local experiment into something that could slot into a real monitoring stack.

In this step, get ready to:
  • Create a FastAPI app with a POST endpoint that invokes your LangGraph pipeline.
  • Start the dev server and send a test incident via the interactive docs.
  • Verify that different incidents retrieve different runbook context and produce relevant analysis.
Create the FastAPI app file

The endpoint needs to accept an incident description, initialize blank pipeline state, invoke the graph, and map the result into the response model you already defined in models.py.

  • Create a new file called main.py in your ops-copilot folder.
  • Add the imports and app setup by pasting this code:
from fastapi import FastAPI
from models import IncidentRequest, IncidentAnalysis, PipelineState
from graph import graph

# Create the FastAPI application instance
app = FastAPI(title="AI Infrastructure Operations Copilot")

What does this code do?

You import the Pydantic models (IncidentRequest and IncidentAnalysis) that define what the endpoint accepts and returns. PipelineState provides the typed structure for the initial graph input.

The compiled graph object from graph.py is what actually runs your 5-agent pipeline. FastAPI handles the HTTP layer so you can focus on the pipeline logic.

  • Below the app setup, add the endpoint function:
@app.post("/analyze", response_model=IncidentAnalysis)
async def analyze_incident(request: IncidentRequest):
    # Build the initial state with empty fields for each pipeline node to fill
    initial_state: PipelineState = {
        "query": request.description,
        "keywords": [],
        "context_docs": [],
        "sources": [],
        "root_cause": "",
        "resolution": [],
        "confidence": 0.0,
        "severity": "",
    }

    # Run the full 5-agent pipeline
    result = graph.invoke(initial_state)

    # Map pipeline state keys to the Pydantic response fields
    return IncidentAnalysis(
        root_cause=result["root_cause"],
        confidence_score=result["confidence"],
        remediation_steps=result["resolution"],
        sources=result["sources"],
        severity=result["severity"],
    )

What does this endpoint do?

The @app.post("/analyze") decorator registers this function as a POST handler at the /analyze path. FastAPI automatically validates the incoming JSON body against IncidentRequest.

The function creates blank pipeline state, slots in the user's description, then calls graph.invoke() to run all five agents in sequence. The result dict is mapped into an IncidentAnalysis response, which FastAPI serializes to JSON.

  • Save main.py.

Seeing an import error?

Make sure main.py is in the same folder as models.py and graph.py. If your terminal shows a ModuleNotFoundError, confirm that your virtual environment is still active (you should see (venv) or similar in your prompt).

help me debug an import error in main.py

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py before continuing.

ⓧ I'd like to double check the full code

from fastapi import FastAPI
from models import IncidentRequest, IncidentAnalysis, PipelineState
from graph import graph

app = FastAPI(title="AI Infrastructure Operations Copilot")


@app.post("/analyze", response_model=IncidentAnalysis)
async def analyze_incident(request: IncidentRequest):
    initial_state: PipelineState = {
        "query": request.description,
        "keywords": [],
        "context_docs": [],
        "sources": [],
        "root_cause": "",
        "resolution": [],
        "confidence": 0.0,
        "severity": "",
    }

    result = graph.invoke(initial_state)

    return IncidentAnalysis(
        root_cause=result["root_cause"],
        confidence_score=result["confidence"],
        remediation_steps=result["resolution"],
        sources=result["sources"],
        severity=result["severity"],
    )
Start the dev server and test your endpoint

FastAPI ships with a built-in development server that auto-reloads when you save changes. It also generates interactive API docs you can use to test without writing any client code.

  • Start the FastAPI development server by running this command:
fastapi dev main.py
  • You should see output indicating the server is running at http://127.0.0.1:8000.
  • Open a browser and go to http://127.0.0.1:8000/docs to see the interactive Swagger UI.

What is the /docs page?

FastAPI auto-generates interactive documentation from your Pydantic models. The /docs page lets you fill in a request body and click **Execute** to send a real request to your running server. No curl or Postman needed.

  • In the Swagger UI, click on the POST /analyze section to expand it.
  • Click Try it out.
  • Replace the example request body with this JSON:
{
  "description": "Kubernetes pod CrashLoopBackOff on payment-service"
}
  • Click Execute.

The response takes a few seconds while the pipeline calls Gemini twice (once for keyword extraction, once for root cause analysis, once for resolution). You should see a JSON response with root_cause, confidence_score, remediation_steps, sources, and severity fields populated.

Server won't start or returns a 500 error?

If fastapi dev main.py fails to start, check that your virtual environment is active and all packages are installed. If the server starts but returns a 500 error when you send a request, check the terminal running the server for the full traceback.

Common causes: your .env file is missing or GOOGLE_API_KEY is not set correctly. The chroma_data directory must also exist (run ingest.py again if you deleted it).

help me debug my FastAPI server error

Test with different incident types

A good RAG system retrieves different context for different queries. Sending varied incidents confirms your pipeline doesn't just return the same runbook every time.

  • In the Swagger UI, click Try it out on the POST /analyze endpoint again.
  • Send this request body for a database incident:
{
  "description": "Database connection timeout errors across all application pods"
}
  • Click Execute and inspect the response. The sources field should reference kb-003 (the database connection pool runbook).
  • Now test with a disk space incident:
{
  "description": "Disk space alert - production filesystem at 95% capacity"
}
  • Click Execute. The sources should include kb-004 (the disk space runbook).
  • Finally, test with a memory issue:
{
  "description": "High memory usage on checkout-service pods, approaching OOM threshold"
}
  • Click Execute. The sources should include kb-005 (the memory leak runbook).

Each response should show different sources and remediation steps tailored to the specific incident type. The confidence_score should be between 0 and 1 for each response.

Why do different incidents retrieve different runbooks?

ChromaDB uses semantic similarity to match your query against the 6 ingested documents. When you describe a database timeout, the embedding is closest to the database connection pool runbook. A disk space query lands nearest the disk space runbook. This is RAG in action.

All responses return the same sources?

If every query returns the same runbook IDs, your chroma_data directory may have been ingested multiple times with duplicate data. Delete the chroma_data folder and run python ingest.py once more. Verify it reports exactly 6 documents.

help me debug why ChromaDB returns the same results for every query

Secret mission

Add a Conditional Validation Loop

Your pipeline currently runs in a straight line from start to finish, regardless of how confident the final analysis is. What if the Validator could send weak results back for deeper analysis? In this mission, you'll add a conditional edge that loops back to Root Cause Analysis when confidence is too low, showcasing LangGraph's most powerful feature: runtime branching.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • The ops-copilot project folder on your Desktop (contains all Python files, the virtual environment, and the chroma_data/ ChromaDB storage).
  • Google AI Studio API key (created at aistudio.google.com/apikey).

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with the ops copilot.

  • Your project folder stays on your Desktop ready to use.
  • Your Google AI Studio API key remains active for future Gemini API calls.
  • To restart the server later, activate your virtual environment and run fastapi dev main.py from inside the ops-copilot folder.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files and API key stay intact for when you return.

  • Stop the FastAPI dev server by pressing Ctrl+C in the terminal where it is running.
  • Deactivate your virtual environment by running:
deactivate

Everything stays on disk. When you come back, navigate to the folder, reactivate the virtual environment, and start the server again.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

  • Stop the FastAPI dev server by pressing Ctrl+C in the terminal if it is still running.
  • Deactivate your virtual environment by running:
deactivate
  • Delete the entire project folder. On macOS/Linux, run:
cd ~/Desktop
rm -rf ops-copilot
  • On Windows, run:
cd ~/Desktop
rmdir /s /q ops-copilot
  • Delete your Google AI Studio API key. Go to aistudio.google.com/apikey, find the key you created for this project, and click the trash icon next to it to delete it.

This removes all project files, the virtual environment, installed packages, ChromaDB data, and revokes API access.

Nice Work!

Nice Work!

You just built a fully functional AI-powered infrastructure operations copilot. Your system takes raw incident descriptions, runs them through a multi-agent pipeline, and returns structured remediation plans backed by real runbook context.

You've learned how to:

  • Build a multi-agent RAG pipeline with 5 specialized nodes orchestrated by LangGraph, where each agent transforms shared typed state and has a single focused responsibility.
  • Ingest infrastructure runbooks into ChromaDB and retrieve relevant vector-matched context per incident query, giving your LLM grounded source material instead of hallucinated answers.
  • Expose the pipeline as a FastAPI POST endpoint returning structured Pydantic responses with confidence scores, severity ratings, and source citations.
  • Secret Mission: Implemented conditional validation looping with add_conditional_edges, demonstrating LangGraph's branching capability to re-analyze low-confidence results automatically.

Ready to quiz yourself?