Observe Agent Workflows with Langfuse

Introduction

30 Second Summary

A polished answer can look trustworthy even when the journey behind it was slow or expensive. Without evidence from each step, a hidden failure can pass as success.

In this project, you will use Langfuse to trace every command in Part 2's multi-agent LangGraph triage workflow. You will turn those traces into a quality-and-cost dashboard with a webhook-backed alert.

What You'll Build

You will open one Langfuse session where the full path from alert to scored diagnosis is visible.

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

• A traceable incident session that reveals the nested specialist and tool observations behind every start or resume command.
• A searchable failure trail ending at metric_window with Metrics backend timed out. as the exact status message.
• A quality-and-cost dashboard with a webhook-backed alert for falling root-cause accuracy.
• Secret Mission: Use Langfuse prompt management to compare two production versions of the analyst prompt by operational metrics.

Are there any prerequisites?

You need Python and terminal experience with the completed Part 2 triage workflow. Your Google Cloud account needs billing enabled plus permission to enable Vertex AI.

Before We Start

This step defines the evidence your observability layer must provide before you change the Part 2 workflow. The three production questions cover what the agents and tools did during a run, what each incident cost and how long it took, plus whether the final diagnosis was objectively correct.

Set Up Cloud Shell and Langfuse

Your observability layer needs a known-good triage workflow as its baseline. This step prepares that baseline inside Google Cloud Shell.

You will run the original workflow before adding instrumentation. Its output reveals which production questions the terminal cannot answer.

In this step, get ready to:

• Prepare a billed Google Cloud project for Vertex AI.
• Run the uninstrumented Part 2 workflow in Cloud Shell.
• Create a Langfuse project with credentials loaded into your environment.

Prepare Cloud Shell and the starter

The workflow calls a model through Google Cloud Vertex AI. Your selected Google Cloud project needs active billing before that call can succeed.

Selecting a billed project permits model usage charges. This project keeps the expected usage small.

• Sign in to the Google Cloud console with the account that has billing access.
• Open the project selector in the console header to choose or create the billed project for this lab.

Choose the path that matches your current Google Cloud account.

✔️ I have a billed project

• Select the billed project you want to use.
• Confirm its active billing account from Billing.
• Record its project ID here: your Google Cloud project ID.

ⓧ I need a billed project

• Create a billed Google Cloud project named Langfuse Triage Lab from Manage resources.
• Record the generated project ID here: your Google Cloud project ID.

• Click Activate Cloud Shell in the Google Cloud console header.
• Wait for the terminal prompt to appear in the bottom panel.
• Select your project by replacing YOUR_PROJECT_ID with your recorded project ID in the first command below:

gcloud config set project YOUR_PROJECT_ID
gcloud services enable aiplatform.googleapis.com

What do these commands do?

• The first command makes your billed project the active project for later Google Cloud commands.
• The second command enables the Vertex AI service used by the triage workflow.

The terminal prints a line ending in finished successfully. once the service is enabled. Your Cloud Shell session is now authenticated to the project.

Project selection or API enablement failed?

Confirm that you replaced YOUR_PROJECT_ID with the project ID instead of the display name. Check that your account can enable services in that project.

If billing is disabled, return to Billing in the Google Cloud console. Link an active billing account before retrying.

Help me diagnose my Google Cloud project or Vertex AI enablement problem.

The starter lives on the main branch of a public GitHub repository. The solution branch is only for comparing the completed project.

• Clone the starter repository into your Cloud Shell home directory by running these commands:

git clone https://github.com/nextwork-projects/langgraph-triage-observability.git ~/langgraph-triage-observability
cd ~/langgraph-triage-observability

What do these commands do?

• The first command downloads the starter into ~/langgraph-triage-observability.
• The second command moves your terminal into that folder.
• The clone keeps main as the active starter branch.

• Create an isolated Python environment by running:

python3 -m venv .venv
source .venv/bin/activate

Why use a virtual environment?

A virtual environment keeps this project's Python packages inside .venv. Activating it makes later installs use those isolated packages.

You should see (.venv) at the start of your terminal prompt.

• Install the starter dependencies by running:

python3 -m pip install -r requirements.txt

What does this command install?

The requirements file pins the LangGraph workflow packages to the versions tested for this project. The virtual environment receives those packages without changing Cloud Shell's system Python.

The first dependency install can take a few minutes while Python downloads the packages. A stream of package names in the terminal means the installation is progressing.

Dependency installation failed?

Check that your prompt starts with (.venv). Confirm that your terminal is inside ~/langgraph-triage-observability.

Help me troubleshoot the pinned dependency installation.

Run the uninstrumented smoke test

The starter already routes incidents through specialist agents. This run gives you a reference point before any observability code exists.

Before you run it, do you think the terminal can identify the slowest specialist or prove that the diagnosis is correct?

• Run the uninstrumented INC-2042 workflow by entering:

python3 app.py start --thread smoke-001 --alert INC-2042

What does this command run?

• The start command begins a new incident workflow.
• The thread ID smoke-001 identifies this durable incident run.
• The alert ID INC-2042 selects the search-latency fixture.

This run usually takes about two to three minutes. The specialists are making model calls while the terminal is quiet between status lines.

The terminal may first print two LangChainBetaWarning lines and a Warning: there are non-text parts in the response line. These warnings are expected, and every later run prints them too.

You will eventually see WORKFLOW COMPLETE followed by Runbook: runbooks/smoke-001.md.

The workflow published its runbook successfully. The terminal printed no trace URL or per-agent cost.

It also omitted per-agent latency and an objective root-cause score. That blind spot is the designed result of this baseline run.

Smoke run stopped before publishing?

A newly enabled project can briefly report 403 PERMISSION_DENIED. Wait a few minutes before retrying.

A Vertex AI rate limit or timeout can also interrupt the first run. Retry the saved thread without starting over by running:

python3 app.py resume --thread smoke-001

This command resumes the checkpointed workflow for smoke-001.

Help me troubleshoot my interrupted smoke run.

You now have a working baseline plus a concrete observability gap. The next setup adds the destination where future traces and scores will live.

Connect Langfuse and load the environment

Langfuse Cloud stores the traces that you add in the next step. Its Hobby plan provides the project features needed here without requiring a credit card.

• Create or open a project on the Hobby plan in Langfuse Cloud US.
• Open API Keys from the project's Settings.

Creating credentials deserves a careful pause. The secret key appears only once after creation.

• Create a new project API key pair.
• Keep the reveal page open while you configure Cloud Shell.
• Close the reveal page only after both keys are stored securely.

That is the sensitive part complete. Your Langfuse project now has a public key plus a one-time secret key for its SDK connection.

The starter requirements currently contain the four Part 2 packages. You will add the pinned Langfuse SDK without changing the existing versions.

• Open requirements.txt in the Cloud Shell Editor.
• Find this current package list:

langgraph==1.2.14
langgraph-checkpoint-sqlite==3.1.1
langchain-google-genai==4.4.0
langchain==1.4.4

What is already installed?

These four pins reproduce the Part 2 workflow environment. Keeping them unchanged protects the known-good baseline you just ran.

• Add langfuse==4.17.0 beneath the final line.
• Compare your updated file with this version:

langgraph==1.2.14
langgraph-checkpoint-sqlite==3.1.1
langchain-google-genai==4.4.0
langchain==1.4.4
langfuse==4.17.0

Why pin Langfuse?

The project uses Langfuse Python SDK 4.17.0 because the tracing code and callback behavior are tested against that release. The pin prevents a later SDK release from changing the walkthrough underneath you.

• Save requirements.txt.
• Install the updated pinned requirements from the Cloud Shell terminal by running:

python3 -m pip install -r requirements.txt

What changed in the environment?

Python keeps the four installed workflow packages at their pinned versions. It adds the Langfuse SDK plus the dependencies that SDK requires.

The installation completes when the terminal returns to the (.venv) prompt without an error.

Langfuse installation failed?

Confirm that langfuse==4.17.0 appears on its own line in requirements.txt. Check that the (.venv) prefix is still visible.

Help me troubleshoot the Langfuse SDK installation.

The example environment file defines every configuration name the project uses. It contains placeholders only, so it is safe to compare on screen.

• Open .env.example in the Cloud Shell Editor.
• Replace its contents with this exact configuration:

GOOGLE_CLOUD_PROJECT=your-project-id
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_BASE_URL=https://us.cloud.langfuse.com
LANGFUSE_TRACING_ENVIRONMENT=dev

What does this configuration define?

• The Google Cloud project value tells the model client which billed project to use.
• The Langfuse key pair authenticates the SDK to your Langfuse project.
• The base URL keeps trace data in the Langfuse Cloud US region.
• The tracing environment labels future telemetry as dev.

• Save .env.example.
• Create the private environment file from the Cloud Shell terminal by running:

cp .env.example .env

Why create a separate environment file?

The command copies the safe template into .env. That private file receives your real project ID plus your Langfuse credentials.

• Open .env in the Cloud Shell Editor.
• Replace your-project-id with your Google Cloud project ID.
• Replace pk-lf-... with the public key from Langfuse.
• Replace sk-lf-... with the secret key from Langfuse.
• Keep LANGFUSE_BASE_URL=https://us.cloud.langfuse.com and LANGFUSE_TRACING_ENVIRONMENT=dev unchanged.
• Save .env.

Keep the secret key private

Do not upload .env or show it in a screenshot. The public .env.example file remains the safe configuration reference.

The environment file is ready. Loading it exports the five values into the current Cloud Shell session.

Before you run the command, what do you expect the terminal to show if the file is valid?

• Load the environment variables from the Cloud Shell terminal by running:

set -a && . ./.env && set +a

How is the file loaded?

• The first set -a makes subsequently defined shell variables available to child processes.
• The middle command reads the assignments from .env.
• The final set +a restores the shell's normal export behavior.

A clean return to the prompt confirms that the shell parsed the file. Your credentials remain hidden because the command prints none of their values.

Environment file did not load?

Check that every line in .env uses one variable name followed by one equals sign. Remove accidental spaces around the equals signs.

Help me troubleshoot my environment file without exposing secrets.

Use this comparison if you want to confirm the two safe project files. The secret-bearing .env stays off screen.

✔️ Awesome, I've got everything!

Your pinned requirements and safe environment template are saved. Your active shell has loaded the private values from .env.

ⓧ I'd like to double check the full code

Compare your complete requirements.txt with this file.

langgraph==1.2.14
langgraph-checkpoint-sqlite==3.1.1
langchain-google-genai==4.4.0
langchain==1.4.4
langfuse==4.17.0

This file contains all five pinned dependencies used by the project.

Compare your complete .env.example with this safe template.

GOOGLE_CLOUD_PROJECT=your-project-id
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_BASE_URL=https://us.cloud.langfuse.com
LANGFUSE_TRACING_ENVIRONMENT=dev

This template contains placeholders only. Your real credentials belong in .env.

Your baseline workflow runs inside its pinned environment. Next, you will turn each command into a Langfuse trace that exposes the nested agent work.

Trace Every Workflow Run

The smoke test proved that your triage workflow can publish a runbook. However, its terminal output hides the agent calls that consumed time and model tokens.

A LangChain callback sends that nested work to Langfuse as a trace. A shared session groups every command from the same incident thread.

In this step, get ready to:

• Create a tracing module for Langfuse callbacks and trace links.
• Connect every start and resume command to the tracing module.
• Inspect one incident session containing nested agent and tool observations.

Create the tracing module

The tracing module keeps observability code outside the workflow graph. This preserves the Part 2 design while giving every command the same tracing path.

• Use the Cloud Shell editor to create telemetry.py inside ~/langgraph-triage-observability.
• Add the Langfuse client helpers by pasting this code into telemetry.py:

"""Langfuse tracing for the triage workflow."""
import os

from specialists import load_json

_langfuse = None


def tracing_enabled() -> bool:
    return bool(os.environ.get("LANGFUSE_PUBLIC_KEY"))


def langfuse_client():
    global _langfuse
    if _langfuse is None:
        from langfuse import get_client

        _langfuse = get_client()
    return _langfuse


def owning_team(alert_id: str) -> str:
    alert = {a["id"]: a for a in load_json("alerts.json")}[alert_id]
    return load_json("services.json")[alert["service"]]["team"]

What do these helpers do?

• The tracing_enabled() check keeps the workflow runnable when no Langfuse public key is loaded.
• The langfuse_client() function creates one client for the command.
• The owning_team() function maps an alert to the team responsible for its service.

• Save telemetry.py.
• Confirm the editor shows telemetry.py beside app.py.

Can't find the new file?

Check that telemetry.py is inside ~/langgraph-triage-observability. A file outside that folder cannot be imported by app.py.

Check that the filename ends with .py.

Help me find why app.py cannot import telemetry.py.

Each CLI command needs a fresh callback handler. That handler captures nested model calls without changing the graph nodes.

• Add the callback configuration below owning_team() by pasting this code:

def trace_config(
    thread_id: str, command: str, alert_id: str, skip: list[str] = ()
) -> tuple[dict, object | None]:
    if not tracing_enabled():
        return {}, None
    from langfuse.langchain import CallbackHandler

    langfuse_client()
    handler = CallbackHandler()
    tags = [command, alert_id] + [f"skip:{name}" for name in skip]
    return {
        "callbacks": [handler],
        "run_name": f"triage {command}",
        "metadata": {
            "langfuse_session_id": thread_id,
            "langfuse_user_id": owning_team(alert_id),
            "langfuse_tags": tags,
        },
    }, handler


def trace_url(trace_id: str) -> str | None:
    return langfuse_client().get_trace_url(trace_id=trace_id)


def shutdown():
    if _langfuse is not None:
        _langfuse.flush()

How does tracing work?

• A new CallbackHandler captures the nested work from one command.
• The run_name value creates names such as triage start.
• The langfuse_session_id value groups commands by workflow thread.
• The trace_url() helper converts the completed trace ID into a link.
• The shutdown() helper flushes queued events before the CLI exits.

• Save telemetry.py.
• Compare the complete file with the reference below.

✔️ Awesome, I've got everything!

Your tracing module now creates callbacks and trace links. It also assigns session metadata before flushing queued events.

ⓧ I'd like to double check the full code

Compare your complete telemetry.py file with this reference.

"""Langfuse tracing for the triage workflow."""
import os

from specialists import load_json

_langfuse = None


def tracing_enabled() -> bool:
    return bool(os.environ.get("LANGFUSE_PUBLIC_KEY"))


def langfuse_client():
    global _langfuse
    if _langfuse is None:
        from langfuse import get_client

        _langfuse = get_client()
    return _langfuse


def owning_team(alert_id: str) -> str:
    alert = {a["id"]: a for a in load_json("alerts.json")}[alert_id]
    return load_json("services.json")[alert["service"]]["team"]


def trace_config(
    thread_id: str, command: str, alert_id: str, skip: list[str] = ()
) -> tuple[dict, object | None]:
    if not tracing_enabled():
        return {}, None
    from langfuse.langchain import CallbackHandler

    langfuse_client()
    handler = CallbackHandler()
    tags = [command, alert_id] + [f"skip:{name}" for name in skip]
    return {
        "callbacks": [handler],
        "run_name": f"triage {command}",
        "metadata": {
            "langfuse_session_id": thread_id,
            "langfuse_user_id": owning_team(alert_id),
            "langfuse_tags": tags,
        },
    }, handler


def trace_url(trace_id: str) -> str | None:
    return langfuse_client().get_trace_url(trace_id=trace_id)


def shutdown():
    if _langfuse is not None:
        _langfuse.flush()

How to use this reference

Check every function name against your file. Pay close attention to the metadata keys inside trace_config().

Connect tracing to the CLI

The workflow already sends graph activity through graph.stream_events(...). Passing the callback through the same configuration lets Langfuse observe the existing workflow.

• Return to app.py in the Cloud Shell editor.
• Add the telemetry import below the existing specialists import using these lines:

from specialists import SPECIALISTS, alert_brief, build_model, load_json, run_specialist
import telemetry

Why import telemetry here?

The import gives the CLI access to the callback configuration. It also gives main() access to the final flush.

• Add the saved skip context at the end of IncidentState using these lines:

    runbook_path: str
    skip_specialists: list[str]

Why save skip context?

A resume command has no start arguments. Persisting the value lets future resume traces recover the same command context from the checkpoint.

• Find the start parser inside parse_args().
• Add the optional skip argument below the existing alert argument using this line:

    start_parser.add_argument("--skip", nargs="*", default=[], choices=list(SPECIALISTS))

What does this argument preserve?

The argument records any omitted specialist names as command context. Its default keeps the current workflow behavior unchanged.

• Find the start command's graph_input dictionary.
• Add the skip state below audit_log using these lines:

                "audit_log": [],
                "skip_specialists": args.skip,

How does this reach a resume?

The SQLite checkpoint stores skip_specialists with the incident state. A later resume can read it without receiving the start arguments again.

• Find the print(f"Router: ...") line inside route_incident().
• Add the skip filter directly above that line using these lines:

    skipped = state.get("skip_specialists", [])
    decision["specialists"] = [name for name in decision["specialists"] if name not in skipped]

What does the filter change?

The router can still choose a skipped specialist. The filter removes it before dispatch, so --skip deploys really runs without deployment evidence.

• Rename the current workflow-running main() function to run_command(args, config).
• Remove the argument parsing lines from the renamed function.
• Keep the existing SqliteSaver block inside the renamed function.
• Add this new main() function immediately above run_command():

def main():
    args = parse_args()
    config = {"configurable": {"thread_id": args.thread}}
    try:
        run_command(args, config)
    finally:
        telemetry.shutdown()

Why use finally?

• The try block runs the requested workflow command.
• The finally block flushes telemetry after successful commands and failed commands.

• Add before = graph.get_state(config).values directly below graph = build_graph(checkpointer), so the renamed function begins with these lines:

def run_command(args, config):
    with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
        graph = build_graph(checkpointer)
        before = graph.get_state(config).values

What does before contain?

The before value captures the saved state before the command runs. Resume commands use that state to recover the alert and skip context.

The final integration creates one callback immediately before streaming events. It merges that callback with the existing LangGraph thread configuration.

• Find the existing graph.stream_events(...) call inside run_command().
• Replace the existing stream and failure-handling tail with this code:

        alert_id = args.alert if args.command == "start" else before["alert"]["id"]
        skip = args.skip if args.command == "start" else before.get("skip_specialists", [])
        trace_extra, handler = telemetry.trace_config(args.thread, args.command, alert_id, skip)
        try:
            stream = graph.stream_events(graph_input, config={**config, **trace_extra}, durability="sync", version="v3")
            display_stream(stream)
        except Exception as error:
            display_failure(graph, config, error)
        values = graph.get_state(config).values
        if handler and handler.last_trace_id:
            print(f"Trace: {telemetry.trace_url(handler.last_trace_id)}")

What changes for each command?

• A start command reads its alert and skip values from the CLI arguments.
• A resume command reads the same values from the saved workflow state.
• The merged configuration gives LangGraph its thread ID while giving Langfuse its callback.
• The completed handler exposes last_trace_id for the printed trace link.

• Save app.py.
• Confirm the original graph nodes remain in the file.
• Confirm the existing SQLite workflow functions remain in the file.

Seeing a Python error after the edit?

Check that import telemetry is at the top level. An indented import can become part of another block.

Check that before is assigned before the resume path reads the saved alert.

Help me compare my tracing edits with the required integration.

Run and inspect one incident session

One CLI command produces one top-level trace. The workflow's agents and tools become nested observations beneath it.

This first traced run can take about 30 seconds to 4 minutes while Vertex AI completes the model calls. A quiet terminal during that wait does not mean the command has stopped.

Before you run the command, do you expect the workflow pause to prevent Langfuse from recording a trace?

• Start incident thread obs-001 by running this command:

python3 app.py start --thread obs-001 --alert INC-2041

What should you see?

The workflow pauses for team approvals. The terminal then prints a Trace: link.

That link confirms the callback recorded a triage start trace even though the workflow reached interrupt().

🙋‍♀️ Did Vertex AI return a limit or timeout error?

The SQLite checkpoint keeps completed workflow state. Retry the unfinished work without supplying an approval decision by running:

python3 app.py resume --thread obs-001

This decision-free resume continues from the saved checkpoint. It creates a triage resume trace in the same session.

Help me interpret my failed Vertex AI run and retry it safely.

• Approve the payments action by running this command:

python3 app.py resume --thread obs-001 --team payments --decision approve

What does this resume record?

The command records the payments decision. Its triage resume trace uses the existing obs-001 session ID.

• Approve the platform-data action by running this command:

python3 app.py resume --thread obs-001 --team platform-data --decision approve

What completes this thread?

The remaining approval lets the workflow publish its runbook. The terminal prints another Trace: link for the final resume command.

Before you inspect Langfuse, do you expect three separate sessions or one session containing every command trace?

• Open session obs-001 from the Sessions view in your Langfuse project.
• Confirm that one triage start trace and the triage resume traces belong to this session.
• Click Open trace on the triage start trace, then open the Tree to see its nested observations for routing, specialist work, and analysis.
• Open the last triage resume trace the same way to see approval_gate and publish_runbook.

You should see nested observations for routing and investigation. You should also see analysis and policy work from the start command.

Session structure does not match?

Confirm that you opened session obs-001 instead of an individual trace. The session view should group the start and resume commands together.

Help me verify the obs-001 session structure.

That is the blind spot closed: session obs-001 now preserves the work behind every start and resume command.

Your workflow now leaves a traceable history across starts and resumes. Next, you can use those observations to follow an intentional failure to its exact source.

Attribute Traces and Debug Failures

Your obs-001 session in Langfuse now groups each command into named traces. Its nested observations show which agents and tools ran.

Execution alone does not identify the team responsible for an incident. In this step, you will confirm that every trace carries ownership plus incident context before using an intentional crash to find the exact failing tool.

In this step, get ready to:

• Confirm each trace carries owning-team attribution plus incident context.
• Separate development traces with the dev environment.
• Trigger the planted metrics failure and locate its exact error observation.

Confirm trace attribution wiring

Trace attribution answers who owns a run. User IDs map activity to a team. Tags make the command plus alert searchable.

• Switch back to telemetry.py from the previous step.
• Locate trace_config().
• Confirm langfuse_user_id uses owning_team(alert_id).
• Confirm langfuse_tags uses the tags list.

How does attribution help?

The langfuse_user_id value lets the Users view aggregate activity by owning team.

The langfuse_tags value records the command plus alert ID. It also supports skip:NAME tags when a later run omits a specialist.

Resume commands have no --alert argument. The saved SQLite state supplies the alert ID plus skipped specialists for consistent attribution.

• Switch back to app.py from the previous step.
• Locate run_command().
• Confirm resume commands read the alert ID from before["alert"]["id"].
• Confirm resume commands read skipped roles from before.get("skip_specialists", []).

These saved values let every command keep the same owner plus incident context throughout its session.

✔️ Awesome, I've got everything!

Your existing tracing code already carries the session, owner, command, alert, plus optional skip tags.

ⓧ I'd like to double check the full code

"""Langfuse tracing for the triage workflow."""
import os

from specialists import load_json

_langfuse = None


def tracing_enabled() -> bool:
    return bool(os.environ.get("LANGFUSE_PUBLIC_KEY"))


def langfuse_client():
    global _langfuse
    if _langfuse is None:
        from langfuse import get_client

        _langfuse = get_client()
    return _langfuse


def owning_team(alert_id: str) -> str:
    alert = {a["id"]: a for a in load_json("alerts.json")}[alert_id]
    return load_json("services.json")[alert["service"]]["team"]


def trace_config(
    thread_id: str, command: str, alert_id: str, skip: list[str] = ()
) -> tuple[dict, object | None]:
    if not tracing_enabled():
        return {}, None
    from langfuse.langchain import CallbackHandler

    langfuse_client()
    handler = CallbackHandler()
    tags = [command, alert_id] + [f"skip:{name}" for name in skip]
    return {
        "callbacks": [handler],
        "run_name": f"triage {command}",
        "metadata": {
            "langfuse_session_id": thread_id,
            "langfuse_user_id": owning_team(alert_id),
            "langfuse_tags": tags,
        },
    }, handler


def trace_url(trace_id: str) -> str | None:
    return langfuse_client().get_trace_url(trace_id=trace_id)


def shutdown():
    if _langfuse is not None:
        _langfuse.flush()

What does this file provide?

• The owning_team() function resolves an alert to the team responsible for its service.
• The trace_config() function creates one callback for each CLI command.
• The metadata keeps the session ID, owning team, plus searchable tags attached to the trace.
• The shutdown() function flushes queued telemetry before the CLI exits.

Separate development traces

A tracing environment separates development activity from other stages. Loading dev gives every new trace that boundary.

• In the Cloud Shell Editor, set LANGFUSE_TRACING_ENVIRONMENT to dev in .env.
• Reload .env in the active Cloud Shell terminal with the environment-loading command from Step 1.

Why separate development traces?

Environment separation keeps test incidents out of production views. Your next trace should display dev as its environment.

Debug the planted metrics failure

The fixture can force the metrics tool to crash inside a real workflow run. Its trace preserves the complete path from the root command to the failing tool.

Before you run this, which specialist do you expect the red error path to end on?

• Create the failing obs-002 run by running this command:

TRIAGE_FAIL=metrics python3 app.py start --thread obs-002 --alert INC-2041

This failure is intentional

You will see WORKFLOW FAILED: Metrics backend timed out.

The SQLite checkpoint keeps completed workflow state. Langfuse records the failing observation path for inspection.

• Open the printed trace URL in Langfuse.
• Click the ⋮ menu above the observation tree, then set Min Level to ERROR.
• Open metric_window at the end of the red path, then check its Preview tab for the timeout status.

You will see the exact status message Metrics backend timed out. The agent retries the tool, so up to three metric_window calls can show the same error.

• Check the trace header for the user payments and the tags start and INC-2041.
• Click route_incident, then open its Attributes tab to confirm that the environment is dev.

You found the failure at its source. The trace now identifies the responsible team, incident context, environment, plus exact broken tool.

Cannot find the error path?

• Check that the trace belongs to obs-002.
• Clear the tree search box.
• Set Min Level back to ERROR from the ⋮ menu.

Help me find the planted metrics failure in my Langfuse observations.

Before you retry, do you expect the workflow to repeat the metrics error or continue from its saved state?

• Retry the failed workflow without an approval decision by running:

python3 app.py resume --thread obs-002

What does this retry prove?

The first command set TRIAGE_FAIL only for that process. This retry re-enters the saved workflow without the injected failure.

The new triage resume trace keeps the same session plus attribution metadata.

You will see the workflow move beyond the previous metrics error. It stops at WORKFLOW PAUSED: 2 team approvals pending and prints a fresh Trace: URL.

Leave both approvals pending. This step only needs the retry trace, so obs-002 can stay paused.

Temporary model error?

• Wait briefly if Vertex AI returns a temporary service error.
• Rerun the decision-free resume command above.

Help me recover my saved triage workflow after a temporary model error.

Your traces now explain who owned each run plus where the planted failure occurred. Next, you will turn diagnosis correctness into a Boolean score that Langfuse can chart.

Score Root-Cause Quality

Your attributed Langfuse traces now reveal the path each incident followed. They still cannot prove that the final diagnosis matches the planted root cause.

This step adds a deterministic quality check that compares each diagnosis with the fixture's expected deployment ID. It gives every newly analyzed trace an objective signal without another model call.

In this step, get ready to:

• Create a deterministic root-cause scoring module.
• Score only the trace that first creates an analysis.
• Compare a complete investigation with a skipped-deploys investigation.

Create a deterministic quality score

A Boolean score records one of two outcomes for each diagnosis. The fixture earns a true result only when the diagnosis contains the expected deployment ID.

• Use the new-file control in the Cloud Shell file editor to create score.py inside ~/langgraph-triage-observability.
• Paste this scoring logic into score.py.

ROOT_CAUSES = {"INC-2041": "d-1042", "INC-2042": "d-1039"}


def root_cause_correct(alert_id: str, analysis: dict) -> bool:
    expected = ROOT_CAUSES.get(alert_id)
    return bool(expected) and expected in analysis.get("likely_cause", "")


def score_run(langfuse, trace_id: str, alert_id: str, analysis: dict) -> bool:
    correct = root_cause_correct(alert_id, analysis)
    langfuse.create_score(
        name="root_cause_correct",
        value=1 if correct else 0,
        trace_id=trace_id,
        data_type="BOOLEAN",
        comment=f"Expected {ROOT_CAUSES.get(alert_id)} in: {analysis.get('likely_cause', '')}",
    )
    return correct

What does this scoring logic do?

• The ROOT_CAUSES dictionary connects each fixture alert to its planted deployment ID.
• The root_cause_correct() function checks whether the diagnosis names that deployment ID.
• The score_run() function attaches the result to the current trace as root_cause_correct.
• The numeric values 1 and 0 represent true and false for the Boolean score.

• Save score.py.
• Confirm that the editor's file explorer lists score.py beside app.py.

Good progress. Your project now has a repeatable answer key for diagnosis quality.

Can't find score.py?

Check that you created the file inside ~/langgraph-triage-observability. A similarly named file outside that folder cannot be imported by app.py.

Ask for help with the file location or scoring code: Help me check why app.py cannot find my score.py module.

✔️ Awesome, I've got everything!

Your saved score.py file is ready to connect to the triage workflow.

ⓧ I'd like to double check the full code

ROOT_CAUSES = {"INC-2041": "d-1042", "INC-2042": "d-1039"}


def root_cause_correct(alert_id: str, analysis: dict) -> bool:
    expected = ROOT_CAUSES.get(alert_id)
    return bool(expected) and expected in analysis.get("likely_cause", "")


def score_run(langfuse, trace_id: str, alert_id: str, analysis: dict) -> bool:
    correct = root_cause_correct(alert_id, analysis)
    langfuse.create_score(
        name="root_cause_correct",
        value=1 if correct else 0,
        trace_id=trace_id,
        data_type="BOOLEAN",
        comment=f"Expected {ROOT_CAUSES.get(alert_id)} in: {analysis.get('likely_cause', '')}",
    )
    return correct

What should match?

Compare the answer-key values and both function definitions with your saved file. The score name and data type must also match exactly.

Score only new analysis

Approval resumes reuse the analysis saved in the SQLite checkpoint. The scoring guard compares state from before and after each command so those later resumes do not create duplicate scores.

• In app.py, find the existing import telemetry line.
• Add the scoring import directly below it so the local imports look like this:

import telemetry
from score import score_run

Why import score_run?

The workflow keeps score creation in score.py. Importing score_run gives app.py one function for evaluating the diagnosis and writing the result.

• In run_command(), find the block that prints the trace URL after values = graph.get_state(config).values.
• Replace that trace-output block with this guarded scoring block:

        if handler and handler.last_trace_id:
            print(f"Trace: {telemetry.trace_url(handler.last_trace_id)}")
            if "analysis" in values and "analysis" not in before:
                correct = score_run(telemetry.langfuse_client(), handler.last_trace_id, alert_id, values["analysis"])
                print(f"Score root_cause_correct: {correct}")

How does the guard prevent duplicates?

• The outer condition confirms that the command produced a traced run with a trace ID.
• The inner condition confirms that analysis exists after the command.
• The same condition confirms that analysis did not exist before the command.
• Only the command that crosses that state boundary receives the quality score.

• Save app.py.

✔️ Awesome, I've got everything!

Your CLI now scores the trace that first creates an analysis.

ⓧ I'd like to double check the updated code

import argparse
import json
import operator
import os
from pathlib import Path
from typing import Annotated, TypedDict

os.environ.setdefault("LANGGRAPH_STRICT_MSGPACK", "true")

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, Send, interrupt

from specialists import SPECIALISTS, alert_brief, build_model, load_json, run_specialist
import telemetry
from score import score_run

DB_PATH = "triage_checkpoints.db"
SEVERITY_ORDER = ["SEV4", "SEV3", "SEV2", "SEV1"]
RUNBOOK_DIR = Path("runbooks")
RESTRICTED_TERMS = ("delete", "disable", "drop", "production", "restart", "rotate")

ANALYSIS_SCHEMA = {
    "title": "IncidentPlan",
    "type": "object",
    "properties": {
        "severity": {"type": "string", "enum": ["SEV1", "SEV2", "SEV3", "SEV4"]},
        "summary": {"type": "string"},
        "likely_cause": {"type": "string"},
        "rationale": {"type": "string"},
        "actions": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {"service": {"type": "string"}, "action": {"type": "string"}},
                "required": ["service", "action"],
            },
        },
    },
    "required": ["severity", "summary", "likely_cause", "rationale", "actions"],
}

ROUTER_SCHEMA = {
    "title": "TriageDecision",
    "type": "object",
    "properties": {
        "severity": {"type": "string", "enum": ["SEV1", "SEV2", "SEV3", "SEV4"]},
        "specialists": {"type": "array", "items": {"type": "string", "enum": list(SPECIALISTS)}},
        "reason": {"type": "string"},
    },
    "required": ["severity", "specialists", "reason"],
}


def highest_severity(current: str, new: str) -> str:
    if current not in SEVERITY_ORDER:
        return new
    return max(current, new, key=SEVERITY_ORDER.index)


class IncidentState(TypedDict, total=False):
    incident_id: str
    incident: str
    alert: dict
    severity: Annotated[str, highest_severity]
    specialists: list[str]
    findings: Annotated[list[dict], operator.add]
    analysis: dict
    actions: list[dict]
    approvals: Annotated[list[dict], operator.add]
    status: str
    audit_log: list[str]
    runbook_path: str
    skip_specialists: list[str]


class SpecialistTask(TypedDict):
    specialist: str
    brief: str


class ActionTask(TypedDict):
    incident_id: str
    team: str
    service: str
    action: str
    policy_reason: str


def build_analyst():
    return build_model().with_structured_output(schema=ANALYSIS_SCHEMA, method="json_schema")


def build_router():
    return build_model().with_structured_output(schema=ROUTER_SCHEMA, method="json_schema")


def safe_incident_id(incident_id: str) -> str:
    safe_id = "".join(c for c in incident_id if c.isalnum() or c in {"-", "_"})
    if not any(c.isalnum() for c in safe_id):
        raise ValueError("Incident ID must contain a letter or number.")
    return safe_id


def incident_brief(state: IncidentState) -> str:
    return alert_brief(state["alert"]["id"])


def route_incident(state: IncidentState):
    decision = build_router().invoke([
        ("system", "You are the incident commander. Set a first severity and choose which specialists investigate: logs, deploys, metrics. For SEV1 or SEV2, send every specialist. For SEV3 or SEV4, send only the one or two most likely to explain the symptom."),
        ("human", incident_brief(state)),
    ])
    skipped = state.get("skip_specialists", [])
    decision["specialists"] = [name for name in decision["specialists"] if name not in skipped]
    print(f"Router: {decision['severity']}, sending {', '.join(decision['specialists'])}")
    return {
        "severity": decision["severity"],
        "specialists": decision["specialists"],
        "status": "routed",
        "audit_log": state["audit_log"] + [f"Router chose {decision['specialists']}: {decision['reason']}"],
    }


def dispatch_specialists(state: IncidentState):
    brief = incident_brief(state)
    return [Send("investigate", {"specialist": name, "brief": brief}) for name in state["specialists"]]


def investigate(task: SpecialistTask):
    name = task["specialist"]
    print(f"  {name} specialist investigating...")
    finding = run_specialist(name, task["brief"])
    print(f"  {name} specialist done: {finding['severity']}, {finding['suspected_cause']}")
    return {"findings": [{"specialist": name, **finding}], "severity": finding["severity"]}


def analyze_incident(state: IncidentState):
    findings = json.dumps(state["findings"], indent=2)
    analysis = build_analyst().invoke([
        ("system", "You are an SRE incident analyst. Combine the specialists' findings into one diagnosis and name the most likely cause with its evidence. Propose one reversible remediation step for each service the evidence implicates: the alerting service and every dependency named in the findings' evidence, at most three. Each service's owning team must approve its own step. Keep every field concise."),
        ("human", f"{state['incident']}\n\nFindings:\n{findings}"),
    ])
    return {
        "analysis": analysis,
        "severity": analysis["severity"],
        "status": "analyzed",
        "audit_log": state["audit_log"] + [f"Model combined {len(state['findings'])} findings into one analysis."],
    }


def policy_check(state: IncidentState):
    services = load_json("services.json")
    severity_requires_review = state["severity"] in {"SEV1", "SEV2"}
    actions, approvals = [], []
    for proposed in state["analysis"]["actions"]:
        service = services.get(proposed["service"], {"team": "unowned"})
        searchable_text = f"{state['incident']} {proposed['action']}".lower()
        requires_approval = severity_requires_review or any(term in searchable_text for term in RESTRICTED_TERMS)
        action = {
            "incident_id": state["incident_id"],
            "team": service["team"],
            "service": proposed["service"],
            "action": proposed["action"],
            "policy_reason": "High severity or restricted operational language detected." if requires_approval else "No high-risk policy rule matched.",
        }
        actions.append(action | {"requires_approval": requires_approval})
        if not requires_approval:
            approvals.append(action | {"decision": "not required"})
    gated = sum(action["requires_approval"] for action in actions)
    return {
        "actions": actions,
        "approvals": approvals,
        "status": "policy_checked",
        "audit_log": state["audit_log"] + [f"Policy check: {gated} of {len(actions)} actions need team approval."],
    }


def policy_route(state: IncidentState):
    gated = [action for action in state["actions"] if action["requires_approval"]]
    if not gated:
        return "publish_runbook"
    return [Send("approval_gate", {k: v for k, v in action.items() if k != "requires_approval"}) for action in gated]


def approval_gate(task: ActionTask):
    decision = interrupt({
        "question": f"Approve this action for team {task['team']}?",
        "incident_id": task["incident_id"],
        "team": task["team"],
        "service": task["service"],
        "action": task["action"],
        "policy_reason": task["policy_reason"],
        "allowed_decisions": ["approve", "reject"],
    })
    if decision not in ("approve", "reject"):
        raise ValueError("Decision must be 'approve' or 'reject'.")
    return {"approvals": [task | {"decision": decision}]}


def publish_runbook(state: IncidentState):
    approvals = state.get("approvals", [])
    approved = [a for a in approvals if a["decision"] in ("approve", "not required")]
    decisions = [f"Team {a['team']} decision for {a['service']}: {a['decision']}." for a in approvals]
    if not approved:
        return {"status": "rejected_without_publish", "audit_log": state["audit_log"] + decisions + ["Every action was rejected. Publishing was skipped."]}
    RUNBOOK_DIR.mkdir(exist_ok=True)
    path = RUNBOOK_DIR / f"{safe_incident_id(state['incident_id'])}.md"
    analysis = state["analysis"]
    audit_log = state["audit_log"] + decisions + [f"Published runbook to {path}."]
    finding_lines = "\n".join(f"- {f['specialist']} ({f['severity']}): {f['summary']}" for f in state["findings"])
    action_lines = "\n".join(f"- {a['team']} / {a['service']}: {a['action']} ({'approved by team' if a['decision'] == 'approve' else 'approval not required'})" for a in approved)
    audit_lines = "\n".join(f"- {entry}" for entry in audit_log)
    path.write_text(
        "# Policy-Controlled Incident Runbook\n\n"
        f"Incident ID: {state['incident_id']}\n\nIncident: {state['incident']}\n\n"
        f"Severity: {state['severity']}\n\nSummary: {analysis['summary']}\n\n"
        f"Likely cause: {analysis['likely_cause']}\n\nRationale: {analysis['rationale']}\n\n"
        f"Findings:\n{finding_lines}\n\nApproved actions:\n{action_lines}\n\nAudit log:\n{audit_lines}\n",
        encoding="utf-8",
    )
    return {"status": "published", "runbook_path": str(path), "audit_log": audit_log}


def build_graph(checkpointer: SqliteSaver):
    builder = StateGraph(IncidentState)
    builder.add_node("route_incident", route_incident)
    builder.add_node("investigate", investigate)
    builder.add_node("analyze_incident", analyze_incident, defer=True)
    builder.add_node("policy_check", policy_check)
    builder.add_node("approval_gate", approval_gate)
    builder.add_node("publish_runbook", publish_runbook, defer=True)
    builder.add_edge(START, "route_incident")
    builder.add_conditional_edges("route_incident", dispatch_specialists, ["investigate"])
    builder.add_edge("investigate", "analyze_incident")
    builder.add_edge("analyze_incident", "policy_check")
    builder.add_conditional_edges("policy_check", policy_route, ["approval_gate", "publish_runbook"])
    builder.add_edge("approval_gate", "publish_runbook")
    builder.add_edge("publish_runbook", END)
    return builder.compile(checkpointer=checkpointer)


def display_stream(stream):
    final_state = stream.output
    if stream.interrupted:
        print(f"WORKFLOW PAUSED: {len(stream.interrupts)} team approvals pending")
        for pending in stream.interrupts:
            request = pending.value
            print(f"- {request['team']} ({request['service']}): {request['action']}")
        return
    print("WORKFLOW COMPLETE")
    print(f"Status: {final_state['status']}")
    print(f"Runbook: {final_state.get('runbook_path', 'not published')}")


def display_failure(graph, config, error):
    snapshot = graph.get_state(config)
    saved = [finding["specialist"] for finding in snapshot.values.get("findings", [])]
    print(f"WORKFLOW FAILED: {error}")
    print(f"Saved findings: {', '.join(saved) or 'none'}")
    print(f"Waiting to re-run: {', '.join(snapshot.next)}")


def parse_args():
    parser = argparse.ArgumentParser(description="Run a resumable incident-triage workflow.")
    subparsers = parser.add_subparsers(dest="command", required=True)
    start_parser = subparsers.add_parser("start")
    start_parser.add_argument("--thread", required=True)
    start_parser.add_argument("--alert", required=True)
    start_parser.add_argument("--skip", nargs="*", default=[], choices=list(SPECIALISTS))
    resume_parser = subparsers.add_parser("resume")
    resume_parser.add_argument("--thread", required=True)
    resume_parser.add_argument("--decision", choices=["approve", "reject"], help="Leave out to retry a failed run.")
    resume_parser.add_argument("--team", help="The team whose action you decide.")
    args = parser.parse_args()
    if args.command == "resume" and args.decision and not args.team:
        parser.error("--decision needs --team.")
    return args


def main():
    args = parse_args()
    config = {"configurable": {"thread_id": args.thread}}
    try:
        run_command(args, config)
    finally:
        telemetry.shutdown()


def run_command(args, config):
    with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
        graph = build_graph(checkpointer)
        before = graph.get_state(config).values
        if args.command == "start":
            alerts = {alert["id"]: alert for alert in load_json("alerts.json")}
            alert = alerts[args.alert]
            graph_input = {
                "incident_id": args.thread,
                "incident": f"{alert['title']}: {alert['description']}",
                "alert": alert,
                "status": "new",
                "audit_log": [],
                "skip_specialists": args.skip,
            }
        elif args.decision:
            pending = graph.get_state(config).interrupts
            team_requests = {request.id: args.decision for request in pending if request.value["team"] == args.team}
            if not team_requests:
                teams = sorted({request.value["team"] for request in pending})
                print(f"No pending approval for team {args.team}. Waiting on: {', '.join(teams) or 'nobody'}")
                return
            graph_input = Command(resume=team_requests)
            print(f"Recorded {args.decision} for {args.team}.")
        else:
            graph_input = None
        alert_id = args.alert if args.command == "start" else before["alert"]["id"]
        skip = args.skip if args.command == "start" else before.get("skip_specialists", [])
        trace_extra, handler = telemetry.trace_config(args.thread, args.command, alert_id, skip)
        try:
            stream = graph.stream_events(graph_input, config={**config, **trace_extra}, durability="sync", version="v3")
            display_stream(stream)
        except Exception as error:
            display_failure(graph, config, error)
        values = graph.get_state(config).values
        if handler and handler.last_trace_id:
            print(f"Trace: {telemetry.trace_url(handler.last_trace_id)}")
            if "analysis" in values and "analysis" not in before:
                correct = score_run(telemetry.langfuse_client(), handler.last_trace_id, alert_id, values["analysis"])
                print(f"Score root_cause_correct: {correct}")


if __name__ == "__main__":
    main()

What should you compare?

Focus on the score_run import and the guarded score call near the end of run_command(). The existing tracing and checkpoint behavior remains unchanged.

Compare complete and skipped runs

The existing skip control creates a controlled weak investigation by removing one specialist from routing. This lets you test whether the score distinguishes a complete diagnosis from one missing deployment evidence.

The complete run can take roughly 30 seconds to 4 minutes while the agents investigate. A quiet terminal during that window does not mean the workflow has stalled.

Before you run the complete investigation, do you expect its diagnosis to contain the planted deployment ID?

• Start the complete INC-2041 investigation on a fresh thread by running:

python3 app.py start --thread obs-003 --alert INC-2041

What does this run test?

This command starts a fresh traced investigation with the normal specialist routing. The new analysis is compared with the expected root cause before the CLI exits.

You should see Score root_cause_correct: True after the trace link. That result proves the complete investigation found the planted deployment evidence.

Temporary model error?

If the start prints WORKFLOW FAILED: The read operation timed out, retry the saved thread without a decision by running:

python3 app.py resume --thread obs-003

The score then appears after the resume's Trace: link, on the triage resume trace.

Help me recover obs-003 after a temporary model error.

Before you run the weaker investigation, do you think the diagnosis can still identify the expected deployment without the deploys specialist?

• Start the skipped-deploys investigation on another fresh thread by running:

python3 app.py start --thread obs-004 --alert INC-2041 --skip deploys

What changes in this run?

The --skip deploys option removes the deploys specialist from routing. The trace also receives the skip:deploys tag so the degraded run stays searchable.

You should see Score root_cause_correct: False after the second trace link. The deterministic check has caught the deliberately weak diagnosis.

Both runs stop at WORKFLOW PAUSED: 2 team approvals pending. Leave them paused, because the scores are already recorded.

• Open the Scores view in your Langfuse project.
• Click Columns, then turn on Trace Tags and Trace Name.
• Confirm that the root_cause_correct row without a skip tag is true and the row with the skip:deploys tag is false.

Missing or unexpected scores?

If no score appears, confirm that your Langfuse credentials remain loaded before rerunning a fresh thread. The scoring block only runs when a traced command creates analysis for the first time.

If the complete run is false, inspect whether its diagnosis includes d-1042. A model or service error can leave the investigation without the fixture's expected evidence.

Get targeted help here: Help me debug my missing or incorrect Langfuse root-cause scores.

Your traces now show execution evidence and objective diagnosis quality. Next, you'll turn those individual results into an aggregate view of cost, latency, errors, and accuracy.

Build a Quality and Cost Dashboard

Your traces now explain each incident run. Your deterministic Boolean scores show whether each diagnosis found the planted root cause.

Operations teams need trends across many incidents. In this step, you will turn your signals into a Langfuse dashboard and connect a quality alert to a webhook.

In this step, get ready to:

• Build a four-widget dashboard for cost, latency, quality, and errors.
• Connect a webhook.site endpoint to a Langfuse automation.
• Create an alert that detects a drop in root-cause accuracy.

Create the four-widget dashboard

A custom dashboard combines signals from many workflow runs into one operational view. You will use the traces and scores already stored in the dev project.

• Open Dashboards, click New dashboard, then create one named Triage quality and cost.
• Set the time range to Past 7 days in the dashboard header.
• For each widget below, click Add Widget, then Custom Chart.
• Add Cost per triage thread using Observations and Total Cost (Sum). Pick Horizontal Bar Chart first, then add a Session Id breakdown.
• Add P95 latency per agent using Observations, Latency (P 95), a Name breakdown, and a Horizontal Bar Chart. Filter Observation Name to logs_specialist, metrics_specialist, deploys_specialist, route_incident, and analyze_incident.
• Add Root-cause accuracy using Scores Boolean, Value (Avg), and the Vertical Bar Chart under Time Series. In the Score Name filter, type root_cause_correct in Enter custom value and tick it.
• Add Errors by step using Observations, Count, a Name breakdown, and a Horizontal Bar Chart filtered to status ERROR.
• Drag the widgets by their handles into a 2 × 2 grid, so all four fit on one screen.

Why These Four Widgets?

The session breakdown turns model cost into cost per triage thread. The latency breakdown highlights slow agents across separate runs.

The Boolean average shows the share of correct diagnoses. The error count reveals which workflow observations carry failures.

💡 Keep the Tested Chart Shapes

Use Session Id for the cost breakdown because Trace Name grouped the demo cost under n/a. Keep accuracy as a time-series Vertical Bar so quiet days do not look like zero accuracy.

Parent observations can share the failing child's count because an error propagates up its execution path.

You should now see four populated widgets covering cost, latency, accuracy, and errors. That single view turns separate incident runs into an operational trend.

Dashboard does not match?

Check each widget's data source, aggregation, breakdown, visualization, and filter against the specifications above.

If the cost widget shows High cardinality dimension(s) 'sessionId' cannot be used with timeDimension., change its chart type to Horizontal Bar Chart.

Help me check the four dashboard widgets.

Connect the webhook automation

Langfuse alerts deliver notifications through automations. You need to create the automation before saving the alert that uses it.

• Open webhook.site in a new browser tab.
• Copy the generated HTTPS URL into this project value: your webhook.site HTTPS URL.
• In Langfuse, open Alerts, then click Connect Webhooks.
• Create an Alert automation named Triage quality webhook with a Webhook action that uses your webhook.site HTTPS URL.
• Save the automation.
• Close the one-time Webhook Secret Created dialog. This project does not verify webhook signatures, so you do not need the secret.

Why Create the Automation First?

Each alert needs at least one delivery automation. Creating the webhook automation first prevents the alert setup from stopping with At least one automation is required.

💡 This Endpoint Is Temporary

A free webhook.site URL expires after 7 days. It accepts at most 100 requests during that period.

A browser that used webhook.site before reopens its last URL. Use that URL while it is still valid, or create a new one on the page.

The saved automation now gives Langfuse a destination for future alert notifications. Your workflow data remains unchanged.

Automation Not Saving?

• Confirm that the event source is set to Alert.
• Confirm that the action is set to Webhook.
• Check that the endpoint begins with HTTPS.

Get help with connecting the Langfuse webhook automation.

Create and test the accuracy alert

The accuracy widget helps you investigate trends manually. The alert adds an automated signal whenever the average correctness score drops below perfect accuracy.

• Open Alerts, then click Create Alert.
• Enter Root-cause accuracy dropped as the Title.
• Choose the Scores (boolean) view, the Value measure, and the Avg aggregation.
• In the Score Name filter, type root_cause_correct in Enter custom value and tick it.
• Set the condition to below with an ALERT threshold of 1.
• Set Over the past to 1 hour.
• Open Advanced Options, then set When there is no data to Show severity NO DATA.
• Check that Triage quality webhook is ticked under Automations in Notifications.
• Save the alert.

You should see the saved alert with the Boolean average condition, the one-hour window, the no-data behavior, and the attached automation. The new alert starts at NO DATA until a fresh score arrives.

This model-backed run can take roughly 30 seconds to 4 minutes. A quiet terminal during that window is normal.

Before you run this, do you expect a diagnosis without the deploys specialist to earn a true or false score?

• Generate a fresh weak diagnosis for the alert by running this command:

python3 app.py start --thread obs-005 --alert INC-2041 --skip deploys

What Does This Command Test?

• The obs-005 thread creates a fresh checkpoint and Langfuse session.
• The INC-2041 fixture has the planted deploy root cause used by the deterministic scorer.
• The --skip deploys option removes the deploys specialist and adds the skip:deploys trace tag.

The terminal should print Score root_cause_correct: False and a Trace: URL. The false score supplies fresh data for the alert condition.

Alert Still Unchanged?

• Refresh the alert if its state remains NO DATA or OK.
• Confirm that root_cause_correct is ticked in the score-name filter.
• Confirm that the webhook automation is still attached to the alert.

Get help with checking why the accuracy alert has not updated.

• Refresh the saved alert and confirm that its no-data behavior reads Show severity NO DATA.
• Confirm that all four widgets on the Triage quality and cost dashboard contain data.
• Check your webhook.site tab for an alert request.

Within about a minute, the alert changes to ALERT and your webhook.site tab receives a POST request from Langfuse. Its body includes "type": "monitor-alert".

You now have an aggregate view of workflow cost, agent latency, diagnosis accuracy, and errors. Your saved alert can turn a quality drop into an automated notification.

Secret mission

Version the Analyst Prompt

Move the analyst prompt into Langfuse. Publish two production versions without changing your application code. Then compare their latency, tokens, cost, generation count, and diagnosis score.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep your resources available, pause your work, or delete the project entirely. Idle project resources do not create new model-request costs.

Cost warning

Running more incidents creates new Gemini model requests that can add usage charges. Stop running the CLI when you finish experimenting.

The Langfuse Hobby plan remains free within its documented limits. Your free webhook endpoint expires after seven days.

Resources you used:

• A Google Cloud Shell repository at ~/langgraph-triage-observability. This includes the virtual environment, .env, SQLite checkpoints, generated runbooks, and project code.
• A dedicated Google Cloud project with the Vertex AI API enabled, if you created one solely for this project.
• A Langfuse project containing traces, observations, sessions, scores, the Triage quality and cost dashboard, the alert, the webhook automation, API keys, and triage-analyst prompt versions.

Keep everything running

No action is needed. Choose this option if you are still comparing incidents or prompt versions.

• Keep the Cloud Shell repository so you can run more triage incidents.
• Keep the Langfuse project so its traces, scores, dashboard, alert, and prompt metrics remain available.
• Keep the dedicated Google Cloud project only while you are actively using it.
• Limit new CLI runs to experiments you still want to measure.

Your free webhook.site URL expires after seven days. It accepts at most 100 requests before then.

Pause - I'll come back to this later

Stop running processes to prevent new model requests while keeping your code and observability data available.

• Stop running triage CLI commands.
• Close the existing Google Cloud Shell session.
• Leave ~/langgraph-triage-observability in your Cloud Shell home directory.
• Leave the Langfuse project intact so its dashboard and prompt comparisons remain available.
• Leave the Google Cloud project intact so you can resume without repeating the cloud setup.

What remains while paused?

Your Langfuse traces, scores, dashboard, alert, automation, and prompt versions remain available. Your checkpoint database keeps the saved incident threads.

No new model-request cost occurs while the CLI stays idle.

Delete - I don't want to use this again

Deleting these resources is permanent. That finality can feel risky, so use this option only when you no longer need the code or observability history.

Delete the Langfuse resources:

• Revoke each API key from API Keys in the Langfuse project settings.
• Delete the project with the project deletion control in Settings, then complete the confirmation Langfuse shows.
• Confirm that the project no longer appears in your Langfuse organization.

Remove the local project files:

This command permanently removes the repository from Cloud Shell. Check the path carefully before you run it.

• Remove the Cloud Shell repository by running this command:

rm -rf ~/langgraph-triage-observability

What does this remove?

This command recursively removes ~/langgraph-triage-observability. The repository's .venv, .env, triage_checkpoints.db, generated runbooks, and source files disappear with it.

• Confirm that langgraph-triage-observability no longer appears in your Cloud Shell home directory.

Delete a dedicated Google Cloud project:

Only a project created solely for this work belongs in this cleanup. Keep any shared project that supports other workloads.

• Select the dedicated project in the Google Cloud console.
• Schedule the project for deletion from its settings page, then complete the confirmation Google Cloud shows.
• Confirm that the dedicated project is scheduled for deletion.

The free webhook.site URL expires on its own after seven days.

Nice Work!

Nice Work!

Outstanding work! You've built a production-style observability layer for your multi-agent triage CLI using Langfuse.

You've learned how to:

• Instrument every start command plus every resume command with named traces. Inspect the nested path from specialist to tool. Group each incident thread into a session with owning-team attribution.
• Trace the planted metrics failure to metric_window. Verify the exact Metrics backend timed out. status. Apply a deterministic Boolean score through root_cause_correct to catch a weak diagnosis.
• Build a four-widget dashboard that turns individual runs into operational trends. Connect a webhook-backed alert to drops in root_cause_correct.
• Complete the optional Secret Mission by versioning triage-analyst in Langfuse. Link each prompt version to its analyst generation. Compare v1 with v2 in Prompt Metrics.

Ready to quiz yourself?