Build an Incident Triage Agent

Build an evaluated AI workflow that routes incidents to the right runbook.

Introduction

30 Second Summary

Operations alerts rarely use the same words as the instructions written for responders. A weak match can send someone toward the wrong fix when every minute matters.

In this project, you will build an incident triage agent in Google Colab that matches everyday alert language to operational runbooks. A labelled evaluation proves when its automatic decisions deserve trust.

What You'll Build

Typing Checkout requests are taking too long into the notebook reveals a confidence-scored payment runbook marked AUTO_APPROVED.

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

  • A side-by-side retrieval test that exposes the paraphrase gap in keyword retrieval. Semantic retrieval then returns Payment API latency for Checkout requests are taking too long.
  • A LangGraph workflow trace that lets you follow retrieve, draft, assess, plus the final approval or escalation branch.
  • A four-case evaluation report that shows retrieval precision plus retrieval recall for each labelled incident.
  • Secret Mission: An optional challenge to push your skills further.

Are there any prerequisites?

You need a Google account to save the notebook in Google Drive.

No local Python installation, API key, GPU, or Java setup is required.

Before We Start

Before any hands-on work, define the operational decision your agent must make. It matches natural-language incidents to runbooks, recommends concrete response actions, and hands weak retrieval matches to a human for a safer decision.

Set Up Your Colab Environment

Your incident workflow needs a stable foundation before it can match runbooks safely. Different local Python setups can make later results drift.

Google Colab gives you a hosted notebook without a local Python installation. Its pinned runtime keeps every later result on the same Python plus NumPy foundation.

In this step, get ready to:
  • Create a Google Colab notebook under your Google account.
  • Connect the notebook to the pinned runtime without hardware acceleration.
  • Install the project packages and verify their versions.
Create and pin your notebook

A Colab runtime is the hosted computer that executes your notebook cells. Selecting a documented runtime removes Windows installation differences from this project.

  • Open Google Colab in your browser.
  • Create a Google account from the sign-in page if you do not already have one.
  • Sign in with your Google account.
  • Create a new notebook from the Colab welcome page.

You should now see a blank notebook that is available to save in Google Drive.

Why Use a Hosted Notebook?

Colab runs your notebook inside a hosted virtual machine. Your Windows laptop only needs a browser.

Your notebook can remain in Google Drive after the temporary runtime is recycled. This keeps your project cells available between sessions.

  • Open the Runtime menu at the top of the notebook.
  • Select Change runtime type.
  • Select 2026.07 from the Runtime Version dropdown.
  • Choose no hardware accelerator.
  • Confirm the runtime change in the dialog.

The notebook reconnects to the pinned runtime. Its cells now execute inside the same hosted environment used throughout this project.

Install the pinned packages

The notebook needs a fixed set of workflow and retrieval libraries. Exact package pins keep later behavior reproducible.

  • Install the four project packages by running this command in the first code cell:
%pip install langgraph==1.2.11 sentence-transformers==6.0.1 faiss-cpu==1.15.0 numpy==2.0.2

What Does This Install?

  • The %pip command current notebook kernel mein packages install karta hai.
  • LangGraph organizes the agent as a stateful workflow.
  • Sentence Transformers converts incident text into semantic embeddings.
  • FAISS searches those embeddings for similar runbooks.
  • NumPy provides the numerical arrays used by the retrieval pipeline.
  • For a Java or Spring developer, this cell plays the same role as resolving a project's pinned dependencies.
  • Let the installation cell finish running.
  • Confirm that the cell completes without an installation error.

Good work. The notebook now has every library the incident triage workflow needs.

Package Installation Stopped Early?

  • Reconnect the notebook runtime from the connection control.
  • Run the same installation cell again.

Ask for help with the failed installation.

Verify the runtime and packages

A successful installation needs a concrete version check. This cell reads the active Python runtime plus the installed package metadata.

Before you run the check, decide whether you expect every value to match the pinned environment.

  • Verify the runtime and installed distributions by running this cell:
import platform
from importlib.metadata import version

print("Python", platform.python_version())
for package in ["langgraph", "sentence-transformers", "faiss-cpu", "numpy"]:
    print(package, version(package))

What Should I See?

Yeh cell active runtime se Python version read karta hai. For a Java or Spring developer, this is a runtime plus dependency report.

  • The first line shows Python 3.12.13.
  • The langgraph line shows 1.2.11.
  • The sentence-transformers line shows 6.0.1.
  • The faiss-cpu line shows 1.15.0.
  • The numpy line shows 2.0.2.

✔️ I see all pinned versions

Everything matches. Your notebook now has the exact runtime and package foundation used throughout this project.

ⓧ I see a different version

A different version means the runtime selection or installation cell needs resetting.

  • Return to the Runtime menu.
  • Select Change runtime type.
  • Select 2026.07 from the Runtime Version dropdown.
  • Choose no hardware accelerator.
  • Confirm the runtime change in the dialog.
  • Rerun the installation cell from the previous substep.
  • Rerun the version-check cell.

ⓧ The cell does not run

A disconnected runtime prevents notebook cells from producing output.

  • Reconnect the notebook through the connection control.
  • Rerun the installation cell.
  • Rerun the version-check cell.

Still Unable to Run Cells?

Ask for help reconnecting the notebook runtime.

Your pinned Colab environment is ready for repeatable experiments. Next, you'll build a keyword baseline and watch a paraphrased incident expose its weakness.

Build a Keyword Baseline

Your Google Colab environment is ready. You can now turn that reproducible setup into the first working slice of your incident triage agent.

A keyword retriever ranks runbooks using shared words. Operational alerts often use different phrasing from the runbooks they describe, so this baseline gives you a concrete way to test the limits of literal overlap.

In this step, get ready to:
  • Create a temporary Python file containing operational runbooks.
  • Rank those runbooks using shared keywords.
  • Compare an exact-word incident with a paraphrased incident.
Create the baseline file

A Python script keeps the runbooks plus retrieval logic in one reusable file. You will start with one runbook so the first test loop stays small.

  • Select the folder-shaped control in the left sidebar of the Colab notebook from earlier.
  • Select the new-file control at the top of the runtime file pane.
  • Name the file incident_triage_agent.py.
  • Double-click incident_triage_agent.py to open it in the editor.
  • Paste the initial runbook data into incident_triage_agent.py using the code below:
from __future__ import annotations
from typing import Any, Literal, TypedDict
RUNBOOKS: list[dict[str, Any]] = [
    {
        "id": "payment_latency",
        "title": "Payment API latency",
        "symptoms": "Payment service responses are slow and timeout rate increases.",
        "actions": [
            "Check payment service latency and timeout logs.",
            "Compare the current error rate with the normal baseline.",
            "Escalate if an external payment dependency is degraded.",
        ],
    },
]
STOPWORDS = {
    "a", "an", "and", "are", "is", "of", "on", "the", "to", "too", "taking"
}
def tokenize(text: str) -> set[str]:
    words = {
        word.strip(".,:;!?()[]{}\"").lower()
        for word in text.split()
    }
    return {word for word in words if word and word not in STOPWORDS}

What does this code do?

  • Think of RUNBOOKS as a list of small DTO records in Java. Each dictionary holds one incident type plus the actions an operator can take.
  • The STOPWORDS set common filler words hataata hai. This keeps the comparison focused on useful incident terms.
  • The tokenize() function text ko lowercase word set mein badalta hai. A set makes shared-word comparisons direct.
  • Save incident_triage_agent.py using the editor's save control.
  • Return to the runtime file pane to confirm the file exists.

You should see incident_triage_agent.py listed in the connected Colab runtime.

File missing from the runtime?

Check that the filename ends with .py. Make sure the file was created in the connected runtime file pane.

Ask for help if the file still does not appear: help me create incident_triage_agent.py in my connected Google Colab runtime.

The data needs a scoring path before it can answer an incident. This next chunk builds one searchable string per runbook before ranking shared words.

  • Return to the editor for incident_triage_agent.py.
  • Paste the retrieval functions plus the baseline entry point at the end of the file:
def runbook_text(runbook: dict[str, Any]) -> str:
    actions = " ".join(runbook["actions"])
    return f"{runbook['title']}. {runbook['symptoms']} {actions}"
def keyword_retrieve(query: str, top_k: int = 1) -> list[dict[str, Any]]:
    query_words = tokenize(query)
    ranked: list[tuple[int, dict[str, Any]]] = []
    for runbook in RUNBOOKS:
        score = len(query_words & tokenize(runbook_text(runbook)))
        ranked.append((score, runbook))
    ranked.sort(key=lambda item: item[0], reverse=True)
    return [
        {**runbook, "score": float(score)}
        for score, runbook in ranked
        if score > 0
    ][:top_k]
def demonstrate_keyword_failure() -> None:
    for query in [
        "payment latency spike",
        "Checkout requests are taking too long",
    ]:
        hits = keyword_retrieve(query)
        title = hits[0]["title"] if hits else "NO MATCH"
        print(f"Keyword query: {query!r} -> {title}")
def main() -> None:
    print("\n1. KEYWORD BASELINE")
    demonstrate_keyword_failure()
if __name__ == "__main__":
    main()

How does the baseline rank runbooks?

  • The runbook_text() function combines the title plus symptoms plus response actions into one searchable string.
  • The keyword_retrieve() function measures set intersection. Java mein yeh do word sets ke common members count karne jaisa hai.
  • A zero score is removed from the results. This produces NO MATCH when no useful words overlap.
  • The main() function gives %run a predictable entry point for the two test incidents.
  • Save incident_triage_agent.py using the editor's save control.

Before you run the baseline, do you think both incident phrases will reach the same payment runbook?

  • Run the baseline from a new Colab code cell using:
%run incident_triage_agent.py

Your first query reaches Payment API latency. The paraphrased checkout query returns NO MATCH.

Why does the paraphrase disappear?

The keyword score only counts words shared by the query plus runbook text. The checkout sentence uses different terms for the same operational symptom.

This shortfall is intentional. You now have visible evidence that literal word overlap cannot cover every natural incident description.

Baseline cell failed?

Check that incident_triage_agent.py is still listed in the connected runtime. Check the indentation inside each function against the snippets above.

Ask for targeted help here: help me find the syntax or indentation problem in my keyword baseline.

Add the remaining runbooks

One runbook proves the scoring path. A broader incident catalogue now tests that the same retriever can rank several operational choices.

  • In RUNBOOKS, find the dictionary whose ID line is "id": "payment_latency",.
  • Paste the database dictionary directly above the payment dictionary:
    {
        "id": "db_cpu",
        "title": "Database CPU saturation",
        "symptoms": "Database processor usage is high and queries become slow.",
        "actions": [
            "Inspect the most expensive queries.",
            "Check whether traffic changed suddenly.",
            "Escalate before changing database capacity.",
        ],
    },

What does this runbook cover?

The db_cpu record describes high database processor usage plus slow queries. Iska action list diagnosis ko capacity changes se pehle rakhta hai.

  • Save incident_triage_agent.py using the editor's save control.
  • Check that the expanded file still runs by using:
%run incident_triage_agent.py

You should see the same two baseline lines. This confirms the database dictionary was added without breaking the script.

Database runbook broke the file?

Make sure the database dictionary sits inside RUNBOOKS. Keep the comma after its closing brace.

Ask for help with the insertion: help me place the db_cpu dictionary correctly inside RUNBOOKS.

  • Find the final ] that closes RUNBOOKS.
  • Paste the authentication dictionary directly above that closing bracket:
    {
        "id": "auth_failures",
        "title": "Authentication failure spike",
        "symptoms": "Login attempts fail and invalid credential errors increase.",
        "actions": [
            "Check authentication error counts.",
            "Compare failures by client and region.",
            "Escalate before blocking users or rotating credentials.",
        ],
    },

What does this runbook add?

The auth_failures record adds failed sign-in symptoms. Java service catalogue ki tarah each record keeps its own diagnosis path plus escalation boundary.

  • Save incident_triage_agent.py using the editor's save control.
  • Check the updated file by running:
%run incident_triage_agent.py

You should see both baseline lines again. The authentication runbook now parses as part of the same list.

Authentication runbook broke the file?

Check the commas between dictionaries in RUNBOOKS. Keep every action inside the actions list.

Ask for help with the list structure: help me correct the auth_failures dictionary inside RUNBOOKS.

  • Find the final ] that closes RUNBOOKS.
  • Paste the storage dictionary directly above that closing bracket:
    {
        "id": "storage_full",
        "title": "Storage capacity exhausted",
        "symptoms": "Disk usage is full and file writes fail.",
        "actions": [
            "Identify the filesystem with the highest usage.",
            "Find recent growth in logs or temporary files.",
            "Escalate before deleting production data.",
        ],
    },

What does the fourth runbook add?

The storage_full record covers failed writes caused by exhausted storage. Iska escalation step protects production data from an unsafe automatic deletion.

  • Save incident_triage_agent.py using the editor's save control.
  • Check that all four runbooks load by running:
%run incident_triage_agent.py

The two baseline lines should still print. Your retriever now loads db_cpu, payment_latency, auth_failures, plus storage_full without a parsing failure.

Storage runbook broke the file?

Confirm the storage dictionary appears before the final bracket of RUNBOOKS. Check that its closing brace keeps the trailing comma.

Ask for help with the final dictionary: help me fix the storage_full insertion without changing the other runbooks.

✔️ Awesome, I've got everything!

Great. Double-check that incident_triage_agent.py is saved in the connected Colab runtime.

ⓧ I'd like to double check the full code

Compare your complete baseline with this version. The runbook order plus every identifier should match.

from __future__ import annotations

from typing import Any, Literal, TypedDict

RUNBOOKS: list[dict[str, Any]] = [
    {
        "id": "db_cpu",
        "title": "Database CPU saturation",
        "symptoms": "Database processor usage is high and queries become slow.",
        "actions": [
            "Inspect the most expensive queries.",
            "Check whether traffic changed suddenly.",
            "Escalate before changing database capacity.",
        ],
    },
    {
        "id": "payment_latency",
        "title": "Payment API latency",
        "symptoms": "Payment service responses are slow and timeout rate increases.",
        "actions": [
            "Check payment service latency and timeout logs.",
            "Compare the current error rate with the normal baseline.",
            "Escalate if an external payment dependency is degraded.",
        ],
    },
    {
        "id": "auth_failures",
        "title": "Authentication failure spike",
        "symptoms": "Login attempts fail and invalid credential errors increase.",
        "actions": [
            "Check authentication error counts.",
            "Compare failures by client and region.",
            "Escalate before blocking users or rotating credentials.",
        ],
    },
    {
        "id": "storage_full",
        "title": "Storage capacity exhausted",
        "symptoms": "Disk usage is full and file writes fail.",
        "actions": [
            "Identify the filesystem with the highest usage.",
            "Find recent growth in logs or temporary files.",
            "Escalate before deleting production data.",
        ],
    },
]

STOPWORDS = {
    "a", "an", "and", "are", "is", "of", "on", "the", "to", "too", "taking"
}


def tokenize(text: str) -> set[str]:
    words = {
        word.strip(".,:;!?()[]{}\"").lower()
        for word in text.split()
    }
    return {word for word in words if word and word not in STOPWORDS}


def runbook_text(runbook: dict[str, Any]) -> str:
    actions = " ".join(runbook["actions"])
    return f"{runbook['title']}. {runbook['symptoms']} {actions}"


def keyword_retrieve(query: str, top_k: int = 1) -> list[dict[str, Any]]:
    query_words = tokenize(query)
    ranked: list[tuple[int, dict[str, Any]]] = []
    for runbook in RUNBOOKS:
        score = len(query_words & tokenize(runbook_text(runbook)))
        ranked.append((score, runbook))
    ranked.sort(key=lambda item: item[0], reverse=True)
    return [
        {**runbook, "score": float(score)}
        for score, runbook in ranked
        if score > 0
    ][:top_k]


def demonstrate_keyword_failure() -> None:
    for query in [
        "payment latency spike",
        "Checkout requests are taking too long",
    ]:
        hits = keyword_retrieve(query)
        title = hits[0]["title"] if hits else "NO MATCH"
        print(f"Keyword query: {query!r} -> {title}")


def main() -> None:
    print("\n1. KEYWORD BASELINE")
    demonstrate_keyword_failure()


if __name__ == "__main__":
    main()
Verify the designed failure

The final check proves two things with one run. Literal overlap can find a clearly worded incident, while natural paraphrasing exposes the baseline's blind spot.

Before you run the final check, which query do you expect to produce a runbook title?

  • Run the completed keyword baseline in a Colab code cell using:
%run incident_triage_agent.py

You should see Payment API latency for payment latency spike. You should then see NO MATCH for Checkout requests are taking too long.

You have a working baseline plus clear evidence of its wording gap. Next, you will replace literal matching with embeddings in a FAISS index.

Add Semantic FAISS Retrieval

Your keyword retrieval baseline catches literal overlap. Its NO MATCH result for the checkout paraphrase exposed the gap between exact wording and human language.

This step adds embeddings that place related language close together. FAISS then ranks the runbooks by vector similarity.

In this step, get ready to:
  • Load a Sentence Transformer model for semantic search.
  • Build a FAISS index from normalized runbook embeddings.
  • Retrieve the payment runbook from a paraphrased checkout incident.

Why Sentence Transformers With FAISS?

Keyword matching only sees shared words. Sentence Transformers converts similar meanings into nearby vectors.

FAISS searches those vectors directly. This makes the semantic retrieval process visible inside a small deterministic project.

Prepare the semantic retrieval pieces

The baseline only uses standard Python data structures. Semantic retrieval needs a model loader plus a vector index library.

  • Return to the file editor for incident_triage_agent.py from the previous step.
  • Find the line containing from typing import Any.
  • Add the semantic retrieval imports plus the model identifier immediately below that line by pasting this code:
from sentence_transformers import SentenceTransformer
import faiss

MODEL_ID = "sentence-transformers/all-MiniLM-L6-v2"

What Does This Code Do?

  • SentenceTransformer text ko embeddings mein badalne wala model loader hai.
  • faiss in vectors ko in-memory index mein search karta hai.
  • MODEL_ID public model ka fixed naam rakhta hai. Java mein yeh reusable configuration constant jaisa hai.
  • Save incident_triage_agent.py.
  • Return to the notebook from earlier.
  • Check that the imports preserve the baseline by running this cell:
%run incident_triage_agent.py

What Does This Check Prove?

%run current file ko notebook program ke roop mein chalata hai. The existing keyword results prove that the new imports did not break the baseline.

You should still see Payment API latency for the exact-word query. You should still see NO MATCH for the paraphrased query.

Seeing an Import Failure?

Confirm that the notebook is still connected to the runtime prepared earlier. Rerun the package installation cell from the setup step if the runtime restarted.

Help me fix the semantic retrieval imports in Colab.

Build the normalized FAISS index

Each runbook needs a numeric representation before FAISS can search it. Normalizing every vector lets inner product act as cosine similarity.

  • Return to incident_triage_agent.py in the notebook file editor.
  • Locate the line def demonstrate_keyword_failure() -> None:.
  • Insert the model and index-building code immediately above that line by pasting this code:
MODEL = SentenceTransformer(MODEL_ID)


def build_index(runbooks: list[dict[str, Any]]):
    texts = [runbook_text(runbook) for runbook in runbooks]
    vectors = MODEL.encode(
        texts,
        precision="float32",
        convert_to_numpy=True,
        normalize_embeddings=True,
    )
    index = faiss.IndexFlatIP(vectors.shape[1])
    index.add(vectors)
    return index


FAISS_INDEX = build_index(RUNBOOKS)

How Is the Index Built?

  • MODEL model identifier se public embedding model load karta hai.
  • texts har runbook ka title plus operational detail rakhta hai.
  • MODEL.encode runbook text ko normalized float32 vectors mein convert karta hai.
  • faiss.IndexFlatIP exact inner-product index banata hai. Unit-length vectors ke saath yeh cosine similarity ranking deta hai.
  • FAISS_INDEX application ke liye ready search structure rakhta hai. Java mein build_index startup service method jaisa hai.
  • Save incident_triage_agent.py.

The first model load can feel slow because Google Colab downloads the model into the runtime cache. Later runs reuse that cached model.

  • Build the index while running the existing baseline by executing this cell:
%run incident_triage_agent.py

What Just Happened?

Python loads the model before calling main(). It then converts all four runbooks into vectors inside FAISS_INDEX.

When the familiar keyword results print, the in-memory index is ready. The downloaded model also remains in the current runtime cache.

Model Download Stuck?

Confirm that the Colab runtime is connected to the internet. Stop the cell only if the runtime disconnects before the baseline output appears.

Help me diagnose a stalled Sentence Transformer model download.

Search by meaning and verify the fix

The index now contains runbook vectors. A query must pass through the same model before FAISS can compare its meaning with those vectors.

  • Return to incident_triage_agent.py.
  • Find the line FAISS_INDEX = build_index(RUNBOOKS).
  • Add the semantic retrieval function immediately below that line by pasting this code:
def semantic_retrieve(query: str, top_k: int = 2) -> list[dict[str, Any]]:
    query_vector = MODEL.encode(
        [query],
        precision="float32",
        convert_to_numpy=True,
        normalize_embeddings=True,
    )
    result_count = min(top_k, len(RUNBOOKS))
    scores, indices = FAISS_INDEX.search(query_vector, result_count)
    hits: list[dict[str, Any]] = []
    for score, index_position in zip(scores[0], indices[0]):
        runbook = RUNBOOKS[int(index_position)]
        hits.append({**runbook, "score": float(score)})
    return hits

How Does Semantic Search Work?

  • query_vector incident query ko runbooks ke same normalized vector format mein rakhta hai.
  • result_count requested results ko available runbook count tak limit karta hai.
  • FAISS_INDEX.search closest vectors ke scores plus positions return karta hai.
  • hits matched runbooks ke saath unka similarity score store karta hai.
  • semantic_retrieve Java repository method jaisa contract deta hai. Query andar jaati hai aur ranked domain records bahar aate hain.
  • Scroll to the function beginning with def main() -> None:.
  • Replace that function with this expanded demonstration:
def main() -> None:
    print("\n1. KEYWORD BASELINE")
    demonstrate_keyword_failure()
    print("\n2. SEMANTIC RETRIEVAL")
    semantic_hit = semantic_retrieve(
        "Checkout requests are taking too long", top_k=1
    )[0]
    print(semantic_hit["title"], round(semantic_hit["score"], 3))

What Does the Demonstration Compare?

The first section preserves the failed keyword result. The second section sends the same paraphrase through semantic_retrieve.

Keeping both results visible makes the improvement measurable. The final line prints the top runbook plus its similarity score.

  • Save incident_triage_agent.py.

Before you run it, predict the result that the semantic retriever gives the checkout paraphrase.

  • Run the complete semantic retrieval demonstration with this cell:
%run incident_triage_agent.py

What Should You See?

The keyword section still prints NO MATCH for the paraphrase. The semantic section now prints Payment API latency followed by a similarity score.

That result proves the retriever can connect checkout delays with payment latency despite the different wording. Your first semantic retrieval loop now works.

Seeing the Wrong Runbook?

Check that both runbook vectors plus the query vector use normalize_embeddings=True. Confirm that FAISS_INDEX is created after all four entries in RUNBOOKS.

Help me debug an unexpected semantic retrieval result.

✔️ Awesome, I've got everything!

Great work. Your saved script now retrieves the payment runbook from the checkout paraphrase and prints its similarity score.

ⓧ I'd like to double check the full code

Compare your complete incident_triage_agent.py with this cumulative version.

from __future__ import annotations

from typing import Any

from sentence_transformers import SentenceTransformer
import faiss

MODEL_ID = "sentence-transformers/all-MiniLM-L6-v2"

RUNBOOKS: list[dict[str, Any]] = [
    {
        "id": "db_cpu",
        "title": "Database CPU saturation",
        "symptoms": "Database processor usage is high and queries become slow.",
        "actions": [
            "Inspect the most expensive queries.",
            "Check whether traffic changed suddenly.",
            "Escalate before changing database capacity.",
        ],
    },
    {
        "id": "payment_latency",
        "title": "Payment API latency",
        "symptoms": "Payment service responses are slow and timeout rate increases.",
        "actions": [
            "Check payment service latency and timeout logs.",
            "Compare the current error rate with the normal baseline.",
            "Escalate if an external payment dependency is degraded.",
        ],
    },
    {
        "id": "auth_failures",
        "title": "Authentication failure spike",
        "symptoms": "Login attempts fail and invalid credential errors increase.",
        "actions": [
            "Check authentication error counts.",
            "Compare failures by client and region.",
            "Escalate before blocking users or rotating credentials.",
        ],
    },
    {
        "id": "storage_full",
        "title": "Storage capacity exhausted",
        "symptoms": "Disk usage is full and file writes fail.",
        "actions": [
            "Identify the filesystem with the highest usage.",
            "Find recent growth in logs or temporary files.",
            "Escalate before deleting production data.",
        ],
    },
]

STOPWORDS = {
    "a", "an", "and", "are", "is", "of", "on", "the", "to", "too", "taking"
}


def tokenize(text: str) -> set[str]:
    words = {
        word.strip(".,:;!?()[]{}\"").lower()
        for word in text.split()
    }
    return {word for word in words if word and word not in STOPWORDS}


def runbook_text(runbook: dict[str, Any]) -> str:
    actions = " ".join(runbook["actions"])
    return f"{runbook['title']}. {runbook['symptoms']} {actions}"


def keyword_retrieve(query: str, top_k: int = 1) -> list[dict[str, Any]]:
    query_words = tokenize(query)
    ranked: list[tuple[int, dict[str, Any]]] = []
    for runbook in RUNBOOKS:
        score = len(query_words & tokenize(runbook_text(runbook)))
        ranked.append((score, runbook))
    ranked.sort(key=lambda item: item[0], reverse=True)
    return [
        {**runbook, "score": float(score)}
        for score, runbook in ranked
        if score > 0
    ][:top_k]


MODEL = SentenceTransformer(MODEL_ID)


def build_index(runbooks: list[dict[str, Any]]):
    texts = [runbook_text(runbook) for runbook in runbooks]
    vectors = MODEL.encode(
        texts,
        precision="float32",
        convert_to_numpy=True,
        normalize_embeddings=True,
    )
    index = faiss.IndexFlatIP(vectors.shape[1])
    index.add(vectors)
    return index


FAISS_INDEX = build_index(RUNBOOKS)


def semantic_retrieve(query: str, top_k: int = 2) -> list[dict[str, Any]]:
    query_vector = MODEL.encode(
        [query],
        precision="float32",
        convert_to_numpy=True,
        normalize_embeddings=True,
    )
    result_count = min(top_k, len(RUNBOOKS))
    scores, indices = FAISS_INDEX.search(query_vector, result_count)
    hits: list[dict[str, Any]] = []
    for score, index_position in zip(scores[0], indices[0]):
        runbook = RUNBOOKS[int(index_position)]
        hits.append({**runbook, "score": float(score)})
    return hits


def demonstrate_keyword_failure() -> None:
    for query in [
        "payment latency spike",
        "Checkout requests are taking too long",
    ]:
        hits = keyword_retrieve(query)
        title = hits[0]["title"] if hits else "NO MATCH"
        print(f"Keyword query: {query!r} -> {title}")


def main() -> None:
    print("\n1. KEYWORD BASELINE")
    demonstrate_keyword_failure()
    print("\n2. SEMANTIC RETRIEVAL")
    semantic_hit = semantic_retrieve(
        "Checkout requests are taking too long", top_k=1
    )[0]
    print(semantic_hit["title"], round(semantic_hit["score"], 3))


if __name__ == "__main__":
    main()

Your retriever now understands paraphrased incidents through vector similarity. Next, you'll turn that match into a decision workflow with a safe human review path.

Build the LangGraph Safety Workflow

Your semantic retrieval layer can find related runbooks. However, it always returns its closest option. That becomes risky when the incident has no useful match.

In this step, LangGraph turns retrieval into a stateful graph. The graph approves strong matches. It sends weak matches to a human.

In this step, get ready to:
  • Define the shared state that moves through the workflow.
  • Build retrieval, drafting, assessment, approval, and escalation nodes.
  • Invoke both safety branches and inspect their traces.
Define shared workflow state

Every graph node needs a shared contract for reading inputs and returning updates. A confidence threshold gives the final routing rule one consistent cutoff.

  • In incident_triage_agent.py, locate the import group at the top of the file.
  • Replace that import group with the following code:
from typing import Any, Literal, TypedDict

from sentence_transformers import SentenceTransformer
import faiss
from langgraph.graph import END, START, StateGraph

What do these imports add?

  • Literal limits the routing function to its two named destinations.
  • TypedDict describes the fields that can travel through the graph.
  • START and END mark the workflow boundaries.
  • StateGraph provides the graph builder.
  • Find the existing MODEL_ID declaration.
  • Add the confidence threshold directly below it using this line:
MIN_CONFIDENCE = 0.42

What does the threshold control?

MIN_CONFIDENCE stores the cutoff used by this small demo corpus. Scores at or above 0.42 follow the approval branch.

Lower scores follow the human-review branch. Production systems tune this value against representative labelled incidents.

  • Save incident_triage_agent.py.
  • Return to the Colab notebook from earlier.
  • Confirm the new imports load by running this cell:
%run incident_triage_agent.py

What does this check prove?

Your existing semantic retrieval output should still show Payment API latency. A completed cell with no traceback proves the graph imports and threshold are valid.

  • Return to incident_triage_agent.py.
  • Find the end of semantic_retrieve().
  • Paste the response-drafting function directly below it:
def draft_response(query: str, hits: list[dict[str, Any]]) -> str:
    if not hits:
        return "No supported runbook was found. Send this incident to a human."
    best = hits[0]
    actions = "\n".join(f"- {action}" for action in best["actions"])
    return (
        f"Incident: {query}\n"
        f"Matched runbook: {best['title']}\n"
        f"Recommended actions:\n{actions}"
    )

What does this function do?

  • draft_response() receives the original incident plus the retrieved runbooks.
  • best holds the highest-scoring runbook.
  • actions formats that runbook's response steps as readable bullets.
  • The fallback sends an empty retrieval result to a human.
  • Paste the shared state definition directly below draft_response() using this code:
class IncidentState(TypedDict, total=False):
    query: str
    retrieved: list[dict[str, Any]]
    response: str
    confidence: float
    status: str
    trace: list[str]

How does this map to Java?

IncidentState ek workflow-state DTO jaisa hai. Har node is shared object se values padhta hai.

Each node returns only the fields it updates. LangGraph merges those updates into the current state.

  • Save incident_triage_agent.py.
  • Return to the Colab notebook.
  • Confirm the new definitions load by running the script again:
%run incident_triage_agent.py

What should still work?

You should see the existing keyword and semantic retrieval demonstration complete. This confirms the new response function and state schema load without disrupting retrieval.

Does the file stop while loading?

Check that Literal and TypedDict appear in the typing import. Check that the LangGraph import sits outside every function.

Compare the indentation inside draft_response() and IncidentState.

Help me debug these workflow definitions.

Build the routing graph

A graph node is a small function that reads the current state and returns an update. The first three nodes retrieve context, draft the response, and record that assessment happened.

  • Paste the retrieval, drafting, and assessment nodes directly below IncidentState using this code:
def retrieve_node(state: IncidentState) -> IncidentState:
    hits = semantic_retrieve(state["query"], top_k=2)
    confidence = hits[0]["score"] if hits else 0.0
    return {
        "retrieved": hits,
        "confidence": confidence,
        "trace": state.get("trace", []) + ["retrieve"],
    }


def draft_node(state: IncidentState) -> IncidentState:
    return {
        "response": draft_response(state["query"], state["retrieved"]),
        "trace": state.get("trace", []) + ["draft"],
    }


def assess_node(state: IncidentState) -> IncidentState:
    return {"trace": state.get("trace", []) + ["assess"]}

How do these nodes work?

  • retrieve_node() searches for two runbooks. It records the best score as confidence.
  • draft_node() turns the retrieved runbook into response steps.
  • assess_node() adds its name to the trace before routing begins.
  • Each node behaves like a focused Spring service method that returns a workflow update.
  • Paste the routing function and final-route nodes directly below assess_node() using this code:
def route_after_assessment(state: IncidentState) -> Literal["approve", "escalate"]:
    return "approve" if state.get("confidence", 0.0) >= MIN_CONFIDENCE else "escalate"


def approve_node(state: IncidentState) -> IncidentState:
    return {
        "status": "AUTO_APPROVED",
        "trace": state.get("trace", []) + ["approve"],
    }


def escalate_node(state: IncidentState) -> IncidentState:
    response = state.get("response", "")
    warning = "Human review required because retrieval confidence is low."
    return {
        "status": "HUMAN_REVIEW",
        "response": f"{response}\n\n{warning}".strip(),
        "trace": state.get("trace", []) + ["escalate"],
    }

Where is the safety decision?

route_after_assessment() is the business routing rule. It chooses approve or escalate from the stored confidence.

approve_node() records automatic approval. escalate_node() records human review and appends a warning.

  • Save incident_triage_agent.py.
  • Return to the Colab notebook.
  • Check that all five node functions load by running the script:
%run incident_triage_agent.py

What does this run confirm?

The cell should complete with the earlier retrieval output and no traceback. This confirms every node can be defined before the graph is assembled.

  • Return to incident_triage_agent.py.
  • Paste the graph builder directly below escalate_node() using this code:
def build_graph():
    builder = StateGraph(IncidentState)
    builder.add_node("retrieve", retrieve_node)
    builder.add_node("draft", draft_node)
    builder.add_node("assess", assess_node)
    builder.add_node("approve", approve_node)
    builder.add_node("escalate", escalate_node)
    builder.add_edge(START, "retrieve")
    builder.add_edge("retrieve", "draft")
    builder.add_edge("draft", "assess")
    builder.add_conditional_edges(
        "assess", route_after_assessment, ["approve", "escalate"]
    )
    builder.add_edge("approve", END)
    builder.add_edge("escalate", END)
    return builder.compile()


TRIAGE_GRAPH = build_graph()

How is the graph wired?

  • add_node() registers each function under a graph node name.
  • add_edge() creates the fixed sequence from retrieval to assessment.
  • add_conditional_edges() applies the confidence-based routing rule.
  • compile() checks the graph structure and creates the runnable workflow.

Java mein fixed edges orchestration sequence jaise hain. The conditional edge works like a business-rule branch in a service layer.

  • Paste the invocation and printing helpers directly below TRIAGE_GRAPH using this code:
def run_incident(query: str) -> IncidentState:
    return TRIAGE_GRAPH.invoke({"query": query, "trace": []})


def print_graph_result(result: IncidentState) -> None:
    print("Status:", result["status"])
    print("Confidence:", round(result["confidence"], 3))
    print("Trace:", " -> ".join(result["trace"]))
    print(result["response"])

What do these helpers expose?

run_incident() creates the initial state and invokes the compiled graph. The empty trace lets each visited node record its name.

print_graph_result() exposes the status, confidence, trace, and drafted response. Those values make the routing decision inspectable.

  • Save incident_triage_agent.py.
  • Return to the Colab notebook.
  • Confirm the graph compiles by running the script:
%run incident_triage_agent.py

What does successful compilation prove?

The cell should finish without a graph-structure traceback. TRIAGE_GRAPH is now a compiled workflow that can accept incident state.

Does the graph fail to compile?

Check that every node name in build_graph() matches its registered name. Check that both conditional destinations are registered before compilation.

Confirm that TRIAGE_GRAPH = build_graph() appears after the complete build_graph() function.

Help me debug my LangGraph structure.

Run both safety branches

A clear payment incident should follow the approval route. Unrelated text should receive a weak similarity score and follow the human-review route.

  • Find the semantic retrieval output inside main().
  • Paste the workflow demonstration directly below that output using this code:
    print("\n3. LANGGRAPH WORKFLOW")
    clear_result = run_incident(
        "Payment service response latency is high and timeouts increased"
    )
    print_graph_result(clear_result)
    print()
    uncertain_result = run_incident("Purple banana quantum orchestra")
    print_graph_result(uncertain_result)

What will this demonstration test?

The first invocation describes symptoms covered by the payment runbook. The second invocation has no meaningful relationship to the four runbooks.

Running both through the same graph reveals whether the confidence route separates supported incidents from weak matches.

  • Save incident_triage_agent.py.

Before you run the workflow, which incident do you expect the graph to escalate?

  • Return to the Colab notebook.
  • Run the completed workflow with this cell:
%run incident_triage_agent.py

What should you see?

The clear payment incident should show AUTO_APPROVED. Its trace should end with approve.

Purple banana quantum orchestra should show HUMAN_REVIEW. Its trace should end with escalate.

Both traces should pass through retrieve, draft, and assess first. That visible path proves the safety decision belongs to the workflow.

Are both incidents taking the same route?

Check that retrieve_node() saves the first hit's score under confidence. Check that route_after_assessment() compares that field with MIN_CONFIDENCE.

Confirm the unrelated query is spelled exactly as shown in the demonstration. A different query can produce a different similarity score.

Help me diagnose the graph routing result.

✔️ Awesome, I've got everything!

Great work. Your graph now approves supported incidents and escalates weak matches with a visible trace.

ⓧ I'd like to double check the full code

Compare your complete incident_triage_agent.py with this version.

from __future__ import annotations

from typing import Any, Literal, TypedDict

from sentence_transformers import SentenceTransformer
import faiss
from langgraph.graph import END, START, StateGraph

MODEL_ID = "sentence-transformers/all-MiniLM-L6-v2"
MIN_CONFIDENCE = 0.42

RUNBOOKS: list[dict[str, Any]] = [
    {
        "id": "db_cpu",
        "title": "Database CPU saturation",
        "symptoms": "Database processor usage is high and queries become slow.",
        "actions": [
            "Inspect the most expensive queries.",
            "Check whether traffic changed suddenly.",
            "Escalate before changing database capacity.",
        ],
    },
    {
        "id": "payment_latency",
        "title": "Payment API latency",
        "symptoms": "Payment service responses are slow and timeout rate increases.",
        "actions": [
            "Check payment service latency and timeout logs.",
            "Compare the current error rate with the normal baseline.",
            "Escalate if an external payment dependency is degraded.",
        ],
    },
    {
        "id": "auth_failures",
        "title": "Authentication failure spike",
        "symptoms": "Login attempts fail and invalid credential errors increase.",
        "actions": [
            "Check authentication error counts.",
            "Compare failures by client and region.",
            "Escalate before blocking users or rotating credentials.",
        ],
    },
    {
        "id": "storage_full",
        "title": "Storage capacity exhausted",
        "symptoms": "Disk usage is full and file writes fail.",
        "actions": [
            "Identify the filesystem with the highest usage.",
            "Find recent growth in logs or temporary files.",
            "Escalate before deleting production data.",
        ],
    },
]

STOPWORDS = {
    "a", "an", "and", "are", "is", "of", "on", "the", "to", "too", "taking"
}


def tokenize(text: str) -> set[str]:
    words = {
        word.strip(".,:;!?()[]{}\"").lower()
        for word in text.split()
    }
    return {word for word in words if word and word not in STOPWORDS}


def runbook_text(runbook: dict[str, Any]) -> str:
    actions = " ".join(runbook["actions"])
    return f"{runbook['title']}. {runbook['symptoms']} {actions}"


def keyword_retrieve(query: str, top_k: int = 1) -> list[dict[str, Any]]:
    query_words = tokenize(query)
    ranked: list[tuple[int, dict[str, Any]]] = []
    for runbook in RUNBOOKS:
        score = len(query_words & tokenize(runbook_text(runbook)))
        ranked.append((score, runbook))
    ranked.sort(key=lambda item: item[0], reverse=True)
    return [
        {**runbook, "score": float(score)}
        for score, runbook in ranked
        if score > 0
    ][:top_k]


MODEL = SentenceTransformer(MODEL_ID)


def build_index(runbooks: list[dict[str, Any]]):
    texts = [runbook_text(runbook) for runbook in runbooks]
    vectors = MODEL.encode(
        texts,
        precision="float32",
        convert_to_numpy=True,
        normalize_embeddings=True,
    )
    index = faiss.IndexFlatIP(vectors.shape[1])
    index.add(vectors)
    return index


FAISS_INDEX = build_index(RUNBOOKS)


def semantic_retrieve(query: str, top_k: int = 2) -> list[dict[str, Any]]:
    query_vector = MODEL.encode(
        [query],
        precision="float32",
        convert_to_numpy=True,
        normalize_embeddings=True,
    )
    result_count = min(top_k, len(RUNBOOKS))
    scores, indices = FAISS_INDEX.search(query_vector, result_count)
    hits: list[dict[str, Any]] = []
    for score, index_position in zip(scores[0], indices[0]):
        runbook = RUNBOOKS[int(index_position)]
        hits.append({**runbook, "score": float(score)})
    return hits


def draft_response(query: str, hits: list[dict[str, Any]]) -> str:
    if not hits:
        return "No supported runbook was found. Send this incident to a human."
    best = hits[0]
    actions = "\n".join(f"- {action}" for action in best["actions"])
    return (
        f"Incident: {query}\n"
        f"Matched runbook: {best['title']}\n"
        f"Recommended actions:\n{actions}"
    )


class IncidentState(TypedDict, total=False):
    query: str
    retrieved: list[dict[str, Any]]
    response: str
    confidence: float
    status: str
    trace: list[str]


def retrieve_node(state: IncidentState) -> IncidentState:
    hits = semantic_retrieve(state["query"], top_k=2)
    confidence = hits[0]["score"] if hits else 0.0
    return {
        "retrieved": hits,
        "confidence": confidence,
        "trace": state.get("trace", []) + ["retrieve"],
    }


def draft_node(state: IncidentState) -> IncidentState:
    return {
        "response": draft_response(state["query"], state["retrieved"]),
        "trace": state.get("trace", []) + ["draft"],
    }


def assess_node(state: IncidentState) -> IncidentState:
    return {"trace": state.get("trace", []) + ["assess"]}


def route_after_assessment(state: IncidentState) -> Literal["approve", "escalate"]:
    return "approve" if state.get("confidence", 0.0) >= MIN_CONFIDENCE else "escalate"


def approve_node(state: IncidentState) -> IncidentState:
    return {
        "status": "AUTO_APPROVED",
        "trace": state.get("trace", []) + ["approve"],
    }


def escalate_node(state: IncidentState) -> IncidentState:
    response = state.get("response", "")
    warning = "Human review required because retrieval confidence is low."
    return {
        "status": "HUMAN_REVIEW",
        "response": f"{response}\n\n{warning}".strip(),
        "trace": state.get("trace", []) + ["escalate"],
    }


def build_graph():
    builder = StateGraph(IncidentState)
    builder.add_node("retrieve", retrieve_node)
    builder.add_node("draft", draft_node)
    builder.add_node("assess", assess_node)
    builder.add_node("approve", approve_node)
    builder.add_node("escalate", escalate_node)
    builder.add_edge(START, "retrieve")
    builder.add_edge("retrieve", "draft")
    builder.add_edge("draft", "assess")
    builder.add_conditional_edges(
        "assess", route_after_assessment, ["approve", "escalate"]
    )
    builder.add_edge("approve", END)
    builder.add_edge("escalate", END)
    return builder.compile()


TRIAGE_GRAPH = build_graph()


def run_incident(query: str) -> IncidentState:
    return TRIAGE_GRAPH.invoke({"query": query, "trace": []})


def print_graph_result(result: IncidentState) -> None:
    print("Status:", result["status"])
    print("Confidence:", round(result["confidence"], 3))
    print("Trace:", " -> ".join(result["trace"]))
    print(result["response"])


def demonstrate_keyword_failure() -> None:
    for query in [
        "payment latency spike",
        "Checkout requests are taking too long",
    ]:
        hits = keyword_retrieve(query)
        title = hits[0]["title"] if hits else "NO MATCH"
        print(f"Keyword query: {query!r} -> {title}")


def main() -> None:
    print("\n1. KEYWORD BASELINE")
    demonstrate_keyword_failure()
    print("\n2. SEMANTIC RETRIEVAL")
    semantic_hit = semantic_retrieve(
        "Checkout requests are taking too long", top_k=1
    )[0]
    print(semantic_hit["title"], round(semantic_hit["score"], 3))
    print("\n3. LANGGRAPH WORKFLOW")
    clear_result = run_incident(
        "Payment service response latency is high and timeouts increased"
    )
    print_graph_result(clear_result)
    print()
    uncertain_result = run_incident("Purple banana quantum orchestra")
    print_graph_result(uncertain_result)


if __name__ == "__main__":
    main()

How should you compare the file?

Check the order of the state, node, graph, and helper definitions. Confirm that main() invokes the graph only after TRIAGE_GRAPH is compiled.

Your incident triage workflow now makes its safety decision through an inspectable graph. Next, you'll measure retrieval precision and recall across labelled test cases.

Evaluate Retrieval Quality

Your LangGraph workflow now approves strong matches. It also escalates weak matches for human review.

A convincing demonstration only proves that two selected examples worked. Labelled test cases expose retrieval quality across several incidents through retrieval precision plus retrieval recall.

In this step, get ready to:
  • Add four labelled incidents with their expected runbook IDs.
  • Calculate precision plus recall with transparent set-based formulas.
  • Run a repeatable evaluation report across every labelled case.
Add labelled test cases

A labelled test case pairs an incident query with the runbook ID that counts as relevant. These labels give the evaluation a reference answer for every retrieval.

  • Return to incident_triage_agent.py in Colab.
  • Find the closing line of print_graph_result.
  • Add the labelled cases below that function by pasting this code:
TEST_CASES = [
    {
        "query": "Database processor usage is pinned and reads are sluggish",
        "reference_ids": {"db_cpu"},
    },
    {
        "query": "Checkout calls are taking too long",
        "reference_ids": {"payment_latency"},
    },
    {
        "query": "Users cannot sign in after repeated credential errors",
        "reference_ids": {"auth_failures"},
    },
    {
        "query": "The service cannot write files because the disk is full",
        "reference_ids": {"storage_full"},
    },
]

What do these cases represent?

  • Each query uses natural incident language that differs from the matching runbook text.
  • Each reference_ids set records the result that counts as relevant.
  • Yeh list Java mein a parameterized test fixture jaisi hai. Each dictionary supplies one input with its expected outcome.
  • Save incident_triage_agent.py.
  • Load the updated script into the notebook by running this cell:
%run incident_triage_agent.py
TEST_CASES

What does this check do?

The %run incident_triage_agent.py line loads the saved script into the notebook kernel. The final TEST_CASES expression displays the labelled cases.

You should see four dictionaries. Their reference IDs should be db_cpu, payment_latency, auth_failures, plus storage_full.

Do the test cases fail to load?

Check that TEST_CASES sits below the complete print_graph_result function. A missing closing brace or comma can prevent the file from loading.

Compare each dictionary with the code block above. Pay close attention to the braces around every reference_ids value.

help me debug my TEST_CASES definition

Calculate retrieval metrics

Precision measures how much of the retrieved set is relevant. Recall measures how much of the reference set was successfully retrieved.

  • Add the metric function below TEST_CASES by pasting this code:
def retrieval_metrics(
    retrieved_ids: list[str], reference_ids: set[str]
) -> tuple[float, float]:
    retrieved_set = set(retrieved_ids)
    relevant_retrieved = retrieved_set & reference_ids
    precision = len(relevant_retrieved) / len(retrieved_set) if retrieved_set else 0.0
    recall = len(relevant_retrieved) / len(reference_ids) if reference_ids else 0.0
    return precision, recall

How are the metrics calculated?

  • The set conversion removes duplicate retrieved IDs before scoring.
  • The & operation keeps IDs that appear in both sets.
  • Precision divides relevant retrieved IDs by every retrieved ID.
  • Recall divides relevant retrieved IDs by every reference ID.

Yeh function Java mein a small stateless service method jaisa hai. It accepts two collections before returning two calculated values.

Why handle empty sets?

An empty denominator would cause a division error. The conditional expressions return 0.0 when either scored set is empty.

  • Save incident_triage_agent.py.
  • Confirm the function ends with return precision, recall.

Seeing a syntax warning in the metric function?

Check that the return type uses tuple[float, float]. Confirm that both calculation lines remain indented inside retrieval_metrics.

help me fix my retrieval_metrics function

Run the evaluation report

The evaluator applies the same retrieval process to every labelled incident. It prints each retrieved ID so the final averages stay auditable.

  • Add the evaluator below retrieval_metrics by pasting this code:
def evaluate_retrieval(top_k: int = 2) -> None:
    precision_total = 0.0
    recall_total = 0.0
    for case in TEST_CASES:
        hits = semantic_retrieve(case["query"], top_k=top_k)
        retrieved_ids = [hit["id"] for hit in hits]
        precision, recall = retrieval_metrics(retrieved_ids, case["reference_ids"])
        precision_total += precision
        recall_total += recall
        print(
            f"Query={case['query']!r}\n"
            f"Retrieved={retrieved_ids} "
            f"Precision={precision:.2f} Recall={recall:.2f}\n"
        )
    case_count = len(TEST_CASES)
    print(
        "AVERAGE",
        f"Precision={precision_total / case_count:.2f}",
        f"Recall={recall_total / case_count:.2f}",
    )

How does the evaluator work?

  • The loop sends each labelled query through semantic_retrieve with the requested top_k value.
  • The list comprehension extracts runbook IDs from the retrieved hits.
  • The metric totals accumulate one score from each case.
  • The final division produces the average precision plus average recall.

Yeh loop Java mein a test suite runner jaisa hai. It executes the same assertion logic against every fixture.

  • Save incident_triage_agent.py.

Before you run the evaluation, do you expect retrieving two runbooks per case to produce perfect precision?

  • Load the saved file before calling the evaluator by running this cell:
%run incident_triage_agent.py
evaluate_retrieval(top_k=2)

What does this evaluation run?

The first line reloads your saved definitions. The second line evaluates all four cases with up to two retrieved runbooks per query.

You should see four labelled rows. You should also see an AVERAGE row containing Precision= plus Recall=.

Why can precision be lower than recall?

A case can retrieve its expected runbook plus one irrelevant runbook. The expected match keeps recall high.

The extra result increases the retrieved set without increasing the relevant set. That lowers precision.

Missing rows or metric values?

Confirm that evaluate_retrieval loops over TEST_CASES. Check that the call uses top_k=2.

If the function name is unavailable, save the file before rerunning the notebook cell.

help me debug my retrieval evaluation output

The individual evaluator works. The final script now needs one entry point that replays the complete project demonstration.

  • Add the keyword demonstration below evaluate_retrieval by pasting this code:
def demonstrate_keyword_failure() -> None:
    for query in [
        "payment latency spike",
        "Checkout requests are taking too long",
    ]:
        hits = keyword_retrieve(query)
        title = hits[0]["title"] if hits else "NO MATCH"
        print(f"Keyword query: {query!r} -> {title}")

Why keep the failed baseline?

The function preserves the designed obstacle from your keyword baseline. It makes the semantic improvement visible whenever the full script runs.

Yeh regression evidence hai. It records the exact weakness that the later retriever solves.

  • Save incident_triage_agent.py.
  • Confirm demonstrate_keyword_failure includes both incident queries.

Does the demonstration function look incomplete?

Confirm that both strings remain inside the query list. Check that the final print line stays inside the loop.

help me check my keyword demonstration function

  • Add the final entry point below demonstrate_keyword_failure by pasting this code:
def main() -> None:
    print("\n1. KEYWORD BASELINE")
    demonstrate_keyword_failure()
    print("\n2. SEMANTIC RETRIEVAL")
    semantic_hit = semantic_retrieve(
        "Checkout requests are taking too long", top_k=1
    )[0]
    print(semantic_hit["title"], round(semantic_hit["score"], 3))
    print("\n3. LANGGRAPH WORKFLOW")
    clear_result = run_incident(
        "Payment service response latency is high and timeouts increased"
    )
    print_graph_result(clear_result)
    print()
    uncertain_result = run_incident("Purple banana quantum orchestra")
    print_graph_result(uncertain_result)
    print("\n4. RETRIEVAL EVALUATION")
    evaluate_retrieval(top_k=2)


if __name__ == "__main__":
    main()

What does the entry point prove?

  • The first section reproduces the brittle keyword match.
  • The second section shows the semantic retriever handling the paraphrase.
  • The third section exercises both LangGraph safety branches.
  • The fourth section runs the labelled quality report.

The main function maps to an application entry point in Java. It coordinates existing components without hiding their individual responsibilities.

  • Save incident_triage_agent.py.

Before you run the complete script, which four sections do you expect to appear in order?

  • Run the finished incident triage agent with this Colab cell:
%run incident_triage_agent.py

What does the finished run show?

The notebook executes main because the file runs as a program. Every project result appears in one repeatable report.

You should see Payment API latency followed by NO MATCH in the keyword section. The semantic section should retrieve Payment API latency for the paraphrased checkout incident.

The workflow section should contain AUTO_APPROVED plus HUMAN_REVIEW. The evaluation section should print four query rows followed by an AVERAGE row with both metrics.

Does the full script stop early?

Use the traceback line number to find the first incomplete function. A missing indentation level near main can prevent later sections from running.

Confirm the final two lines remain outside the main function. They should call main() only when the file runs as a program.

help me debug the complete incident triage script

✔️ Awesome, I've got everything!

Your script now turns retrieval quality into a repeatable report. Save incident_triage_agent.py so the complete base project remains available in the current Colab runtime.

ⓧ I'd like to double check the full code

Compare your saved incident_triage_agent.py with this complete base project script.

from __future__ import annotations

from typing import Any, Literal, TypedDict

from sentence_transformers import SentenceTransformer
import faiss
from langgraph.graph import END, START, StateGraph

MODEL_ID = "sentence-transformers/all-MiniLM-L6-v2"
MIN_CONFIDENCE = 0.42

RUNBOOKS: list[dict[str, Any]] = [
    {
        "id": "db_cpu",
        "title": "Database CPU saturation",
        "symptoms": "Database processor usage is high and queries become slow.",
        "actions": [
            "Inspect the most expensive queries.",
            "Check whether traffic changed suddenly.",
            "Escalate before changing database capacity.",
        ],
    },
    {
        "id": "payment_latency",
        "title": "Payment API latency",
        "symptoms": "Payment service responses are slow and timeout rate increases.",
        "actions": [
            "Check payment service latency and timeout logs.",
            "Compare the current error rate with the normal baseline.",
            "Escalate if an external payment dependency is degraded.",
        ],
    },
    {
        "id": "auth_failures",
        "title": "Authentication failure spike",
        "symptoms": "Login attempts fail and invalid credential errors increase.",
        "actions": [
            "Check authentication error counts.",
            "Compare failures by client and region.",
            "Escalate before blocking users or rotating credentials.",
        ],
    },
    {
        "id": "storage_full",
        "title": "Storage capacity exhausted",
        "symptoms": "Disk usage is full and file writes fail.",
        "actions": [
            "Identify the filesystem with the highest usage.",
            "Find recent growth in logs or temporary files.",
            "Escalate before deleting production data.",
        ],
    },
]

STOPWORDS = {
    "a", "an", "and", "are", "is", "of", "on", "the", "to", "too", "taking"
}


def tokenize(text: str) -> set[str]:
    words = {
        word.strip(".,:;!?()[]{}\"").lower()
        for word in text.split()
    }
    return {word for word in words if word and word not in STOPWORDS}


def runbook_text(runbook: dict[str, Any]) -> str:
    actions = " ".join(runbook["actions"])
    return f"{runbook['title']}. {runbook['symptoms']} {actions}"


def keyword_retrieve(query: str, top_k: int = 1) -> list[dict[str, Any]]:
    query_words = tokenize(query)
    ranked: list[tuple[int, dict[str, Any]]] = []
    for runbook in RUNBOOKS:
        score = len(query_words & tokenize(runbook_text(runbook)))
        ranked.append((score, runbook))
    ranked.sort(key=lambda item: item[0], reverse=True)
    return [
        {**runbook, "score": float(score)}
        for score, runbook in ranked
        if score > 0
    ][:top_k]


MODEL = SentenceTransformer(MODEL_ID)


def build_index(runbooks: list[dict[str, Any]]):
    texts = [runbook_text(runbook) for runbook in runbooks]
    vectors = MODEL.encode(
        texts,
        precision="float32",
        convert_to_numpy=True,
        normalize_embeddings=True,
    )
    index = faiss.IndexFlatIP(vectors.shape[1])
    index.add(vectors)
    return index


FAISS_INDEX = build_index(RUNBOOKS)


def semantic_retrieve(query: str, top_k: int = 2) -> list[dict[str, Any]]:
    query_vector = MODEL.encode(
        [query],
        precision="float32",
        convert_to_numpy=True,
        normalize_embeddings=True,
    )
    result_count = min(top_k, len(RUNBOOKS))
    scores, indices = FAISS_INDEX.search(query_vector, result_count)
    hits: list[dict[str, Any]] = []
    for score, index_position in zip(scores[0], indices[0]):
        runbook = RUNBOOKS[int(index_position)]
        hits.append({**runbook, "score": float(score)})
    return hits


def draft_response(query: str, hits: list[dict[str, Any]]) -> str:
    if not hits:
        return "No supported runbook was found. Send this incident to a human."
    best = hits[0]
    actions = "\n".join(f"- {action}" for action in best["actions"])
    return (
        f"Incident: {query}\n"
        f"Matched runbook: {best['title']}\n"
        f"Recommended actions:\n{actions}"
    )


class IncidentState(TypedDict, total=False):
    query: str
    retrieved: list[dict[str, Any]]
    response: str
    confidence: float
    status: str
    trace: list[str]


def retrieve_node(state: IncidentState) -> IncidentState:
    hits = semantic_retrieve(state["query"], top_k=2)
    confidence = hits[0]["score"] if hits else 0.0
    return {
        "retrieved": hits,
        "confidence": confidence,
        "trace": state.get("trace", []) + ["retrieve"],
    }


def draft_node(state: IncidentState) -> IncidentState:
    return {
        "response": draft_response(state["query"], state["retrieved"]),
        "trace": state.get("trace", []) + ["draft"],
    }


def assess_node(state: IncidentState) -> IncidentState:
    return {"trace": state.get("trace", []) + ["assess"]}


def route_after_assessment(state: IncidentState) -> Literal["approve", "escalate"]:
    return "approve" if state.get("confidence", 0.0) >= MIN_CONFIDENCE else "escalate"


def approve_node(state: IncidentState) -> IncidentState:
    return {
        "status": "AUTO_APPROVED",
        "trace": state.get("trace", []) + ["approve"],
    }


def escalate_node(state: IncidentState) -> IncidentState:
    response = state.get("response", "")
    warning = "Human review required because retrieval confidence is low."
    return {
        "status": "HUMAN_REVIEW",
        "response": f"{response}\n\n{warning}".strip(),
        "trace": state.get("trace", []) + ["escalate"],
    }


def build_graph():
    builder = StateGraph(IncidentState)
    builder.add_node("retrieve", retrieve_node)
    builder.add_node("draft", draft_node)
    builder.add_node("assess", assess_node)
    builder.add_node("approve", approve_node)
    builder.add_node("escalate", escalate_node)
    builder.add_edge(START, "retrieve")
    builder.add_edge("retrieve", "draft")
    builder.add_edge("draft", "assess")
    builder.add_conditional_edges(
        "assess", route_after_assessment, ["approve", "escalate"]
    )
    builder.add_edge("approve", END)
    builder.add_edge("escalate", END)
    return builder.compile()


TRIAGE_GRAPH = build_graph()


def run_incident(query: str) -> IncidentState:
    return TRIAGE_GRAPH.invoke({"query": query, "trace": []})


def print_graph_result(result: IncidentState) -> None:
    print("Status:", result["status"])
    print("Confidence:", round(result["confidence"], 3))
    print("Trace:", " -> ".join(result["trace"]))
    print(result["response"])


TEST_CASES = [
    {
        "query": "Database processor usage is pinned and reads are sluggish",
        "reference_ids": {"db_cpu"},
    },
    {
        "query": "Checkout calls are taking too long",
        "reference_ids": {"payment_latency"},
    },
    {
        "query": "Users cannot sign in after repeated credential errors",
        "reference_ids": {"auth_failures"},
    },
    {
        "query": "The service cannot write files because the disk is full",
        "reference_ids": {"storage_full"},
    },
]


def retrieval_metrics(
    retrieved_ids: list[str], reference_ids: set[str]
) -> tuple[float, float]:
    retrieved_set = set(retrieved_ids)
    relevant_retrieved = retrieved_set & reference_ids
    precision = len(relevant_retrieved) / len(retrieved_set) if retrieved_set else 0.0
    recall = len(relevant_retrieved) / len(reference_ids) if reference_ids else 0.0
    return precision, recall


def evaluate_retrieval(top_k: int = 2) -> None:
    precision_total = 0.0
    recall_total = 0.0
    for case in TEST_CASES:
        hits = semantic_retrieve(case["query"], top_k=top_k)
        retrieved_ids = [hit["id"] for hit in hits]
        precision, recall = retrieval_metrics(retrieved_ids, case["reference_ids"])
        precision_total += precision
        recall_total += recall
        print(
            f"Query={case['query']!r}\n"
            f"Retrieved={retrieved_ids} "
            f"Precision={precision:.2f} Recall={recall:.2f}\n"
        )
    case_count = len(TEST_CASES)
    print(
        "AVERAGE",
        f"Precision={precision_total / case_count:.2f}",
        f"Recall={recall_total / case_count:.2f}",
    )


def demonstrate_keyword_failure() -> None:
    for query in [
        "payment latency spike",
        "Checkout requests are taking too long",
    ]:
        hits = keyword_retrieve(query)
        title = hits[0]["title"] if hits else "NO MATCH"
        print(f"Keyword query: {query!r} -> {title}")


def main() -> None:
    print("\n1. KEYWORD BASELINE")
    demonstrate_keyword_failure()
    print("\n2. SEMANTIC RETRIEVAL")
    semantic_hit = semantic_retrieve(
        "Checkout requests are taking too long", top_k=1
    )[0]
    print(semantic_hit["title"], round(semantic_hit["score"], 3))
    print("\n3. LANGGRAPH WORKFLOW")
    clear_result = run_incident(
        "Payment service response latency is high and timeouts increased"
    )
    print_graph_result(clear_result)
    print()
    uncertain_result = run_incident("Purple banana quantum orchestra")
    print_graph_result(uncertain_result)
    print("\n4. RETRIEVAL EVALUATION")
    evaluate_retrieval(top_k=2)


if __name__ == "__main__":
    main()

This reference combines the four runbooks, both retrieval methods, the safety graph, the labelled evaluation, plus the program entry point. Your saved file should match it exactly.

You have turned the incident triage agent into an evaluated workflow. Its retrieval quality is now visible through labelled evidence instead of a hand-picked demonstration.

Secret mission

Add a Cache Incident and Rebuild the Index

Your incident triage agent only searches the knowledge captured when its vector index was built. In this mission, you will add a cache incident, rebuild the FAISS index, and prove the new runbook works without breaking the original evaluation cases.

Clean Up Your Resources

Clean Up Your Resources

This project uses Google Colab's free hosted runtime without creating paid cloud resources. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Google Colab notebook saved or available to save in Google Drive.
  • Connected Colab runtime version 2026.07 with the required Python packages.
  • Temporary incident_triage_agent.py file stored in the Colab virtual machine.
  • Downloaded sentence-transformers/all-MiniLM-L6-v2 embedding model cache.
  • Five-runbook in-memory FAISS index containing cache_miss_storm.

Keep everything running

No action is needed if you are still testing the incident triage workflow. The connected runtime keeps the temporary file, installed packages, model cache, and in-memory index available.

  • Save the notebook in Google Drive so its cells and explanations remain available.
  • Continue testing new incidents against the five-runbook workflow.

Pause - I'll come back to this later

Pausing ends your current browser session while keeping the notebook available for later.

  • Save the notebook in Google Drive.
  • Close the Colab browser tab.
  • Run the notebook cells from the top when you return with a fresh runtime.

Colab may delete the idle virtual machine with its temporary files. A fresh runtime needs the package installation and index-building cells again.

Delete - I don't want to use this again

Removing all project resources gives you a clean slate. This permanently clears the notebook plus its temporary runtime state.

  • Return to the connected Google Colab notebook.
  • Open the Runtime menu.
  • Select Disconnect and delete runtime to remove the temporary virtual machine state.
  • Open Google Drive in your browser.
  • Locate the Google Colab notebook created for this project.
  • Delete the notebook from Google Drive.
  • Confirm the notebook no longer appears in Google Drive.
  • Confirm Google Colab no longer shows a connected runtime.

Deleting the runtime removes incident_triage_agent.py, the installed packages, the cached embedding model, and the in-memory FAISS index. Deleting the notebook removes your saved cells and explanations.

Nice Work!

Nice Work!

Mission accomplished! You built an evaluation-driven incident triage agent that retrieves runbooks from natural-language alerts.

You've learned how to:

  • Expose the limits of keyword retrieval with a paraphrased incident. Use Sentence Transformers to create normalized embeddings for a FAISS similarity index that retrieves the relevant runbook.
  • Model incident triage as a LangGraph workflow with shared state. Use focused nodes plus conditional routing to escalate weak matches for human review.
  • Measure retrieval quality across labelled cases with transparent retrieval precision and retrieval recall calculations.
  • Secret Mission: Added a cache incident to the knowledge base. Rebuilt the FAISS index. Verified that all five labelled cases still execute.

Ready to quiz yourself?