📓 Debugging Recurring Trace Failures¶

You are looking at one bad trace and you want three answers quickly: has this happened before, what recurring failure mode does it belong to, and where in the trace does that failure diverge from a run that worked.

This notebook answers those three questions and stops there. It runs on a small synthetic fixture that ships next to it, so the default path needs no API key, no Snowflake account and no network access. One optional cell shows where to swap in your own records.

In [ ]:

# !pip install trulens pandas scikit-learn matplotlib

# !pip install trulens pandas scikit-learn matplotlib

Setup¶

Configuration¶

Everything tunable lives in this one cell. FAILURE_METRIC names the evaluation metric that defines failure, QUERY_RECORD_ID is the bad record you start from, and the rest control the clustering and the size of the result tables.

In [ ]:

FAILURE_METRIC = "Groundedness"
FAILURE_THRESHOLD = 0.5
QUERY_RECORD_ID = "record-retrieval-003"
N_CLUSTERS = 3
TOP_K = 5

FAILURE_METRIC = "Groundedness" FAILURE_THRESHOLD = 0.5 QUERY_RECORD_ID = "record-retrieval-003" N_CLUSTERS = 3 TOP_K = 5

The fixture¶

data/trace_debugging_fixture.jsonl holds 34 records from a support copilot across two app versions. Each record carries the question, the answer, any error, evaluation scores with their direction, model and tool metadata, and an ordered list of spans with durations. Three recurring failure modes are present, each with passing records nearby.

In [ ]:

import json
from pathlib import Path

import pandas as pd

FIXTURE_PATH = Path("data/trace_debugging_fixture.jsonl")
if not FIXTURE_PATH.exists():
    # Running from the repository root rather than the notebook directory.
    FIXTURE_PATH = Path("examples/expositional/use_cases") / FIXTURE_PATH

records = [
    json.loads(line)
    for line in FIXTURE_PATH.read_text().splitlines()
    if line.strip()
]

# Missing errors are null in the file; empty strings are easier downstream.
for record in records:
    record["error"] = record["error"] or ""
    for span in record["spans"]:
        span["error"] = span["error"] or ""

df = pd.DataFrame(records)
print(f"{len(df)} records, app versions {sorted(df['app_version'].unique())}")
df[["record_id", "app_version", "model", "input"]].head()

import json from pathlib import Path import pandas as pd FIXTURE_PATH = Path("data/trace_debugging_fixture.jsonl") if not FIXTURE_PATH.exists(): # Running from the repository root rather than the notebook directory. FIXTURE_PATH = Path("examples/expositional/use_cases") / FIXTURE_PATH records = [ json.loads(line) for line in FIXTURE_PATH.read_text().splitlines() if line.strip() ] # Missing errors are null in the file; empty strings are easier downstream. for record in records: record["error"] = record["error"] or "" for span in record["spans"]: span["error"] = span["error"] or "" df = pd.DataFrame(records) print(f"{len(df)} records, app versions {sorted(df['app_version'].unique())}") df[["record_id", "app_version", "model", "input"]].head()

Four small helpers do all the derived work. failure_document is the one that matters most: it collapses a record into a short piece of text that the similarity model can compare, built from the question, the answer, the first error, the evaluation explanation and the ordered span types.

In [ ]:

def span_path(spans):
    """Ordered `span_type/name` steps of one trace."""
    return [f"{span['span_type']}/{span['name']}" for span in spans]


def first_error(record):
    """Record-level error if there is one, otherwise the first span error."""
    if record["error"]:
        return record["error"]
    for span in record["spans"]:
        if span["error"]:
            return span["error"]
    return ""


def is_failure(record):
    """Compare the score to the threshold in the metric's own direction."""
    score = record["metrics"][FAILURE_METRIC]
    if record["metric_directions"][FAILURE_METRIC] == "higher_is_better":
        return score < FAILURE_THRESHOLD
    return score > FAILURE_THRESHOLD


def failure_document(record):
    """Collapse one record into the short text used for similarity."""
    parts = [
        record["input"],
        record["output"],
        first_error(record),
        record["metric_explanations"].get(FAILURE_METRIC, ""),
        " ".join(span["span_type"] for span in record["spans"]),
    ]
    return " ".join(part for part in parts if part)


df["score"] = [record["metrics"][FAILURE_METRIC] for record in records]
df["failed"] = [is_failure(record) for record in records]
df["first_error"] = [first_error(record) for record in records]
df["span_path"] = [span_path(record["spans"]) for record in records]
df["failure_document"] = [failure_document(record) for record in records]

print(f"{int(df['failed'].sum())} of {len(df)} records fail on {FAILURE_METRIC}")
df.loc[df["failed"], "failure_document"].iloc[0]

def span_path(spans): """Ordered `span_type/name` steps of one trace.""" return [f"{span['span_type']}/{span['name']}" for span in spans] def first_error(record): """Record-level error if there is one, otherwise the first span error.""" if record["error"]: return record["error"] for span in record["spans"]: if span["error"]: return span["error"] return "" def is_failure(record): """Compare the score to the threshold in the metric's own direction.""" score = record["metrics"][FAILURE_METRIC] if record["metric_directions"][FAILURE_METRIC] == "higher_is_better": return score < FAILURE_THRESHOLD return score > FAILURE_THRESHOLD def failure_document(record): """Collapse one record into the short text used for similarity.""" parts = [ record["input"], record["output"], first_error(record), record["metric_explanations"].get(FAILURE_METRIC, ""), " ".join(span["span_type"] for span in record["spans"]), ] return " ".join(part for part in parts if part) df["score"] = [record["metrics"][FAILURE_METRIC] for record in records] df["failed"] = [is_failure(record) for record in records] df["first_error"] = [first_error(record) for record in records] df["span_path"] = [span_path(record["spans"]) for record in records] df["failure_document"] = [failure_document(record) for record in records] print(f"{int(df['failed'].sum())} of {len(df)} records fail on {FAILURE_METRIC}") df.loc[df["failed"], "failure_document"].iloc[0]

Using your own records¶

The three steps below only need the records list to hold dictionaries with the keys the fixture uses. The cell below sketches where those come from in a live deployment. It stays commented out so the default path keeps running offline.

In [ ]:

# from trulens.core import TruSession
#
# session = TruSession()
# records_df, feedback_columns = session.get_records_and_feedback()
# events_df = session.get_events()
#
# Reshape those two frames into one dictionary per record with the same keys the
# fixture uses: record_id, app_version, model, tools, input, output, error,
# metrics, metric_explanations, metric_directions, and spans, where each span has
# span_type, name, duration_ms and error. Group events_df by record id to build
# the span lists in start-time order. Assign the result to `records` and re-run
# the two cells above.

# from trulens.core import TruSession # # session = TruSession() # records_df, feedback_columns = session.get_records_and_feedback() # events_df = session.get_events() # # Reshape those two frames into one dictionary per record with the same keys the # fixture uses: record_id, app_version, model, tools, input, output, error, # metrics, metric_explanations, metric_directions, and spans, where each span has # span_type, name, duration_ms and error. Group events_df by record id to build # the span lists in start-time order. Assign the result to `records` and re-run # the two cells above.

Step 1: Find similar failures¶

Has this happened before, and where?

A TF-IDF representation reduced with SVD is enough here, and it keeps the notebook free of any embedding service. The query is either the record id you started from or a short description typed in plain text.

In [ ]:

import numpy as np
from sklearn.decomposition import TruncatedSVD
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.preprocessing import normalize

vectorizer = TfidfVectorizer(stop_words="english", sublinear_tf=True)
tfidf = vectorizer.fit_transform(df["failure_document"])

n_components = min(20, tfidf.shape[1] - 1, len(df) - 1)
svd = TruncatedSVD(n_components=n_components, random_state=0)
vectors = normalize(svd.fit_transform(tfidf))

terms = np.array(vectorizer.get_feature_names_out())
failed_positions = np.flatnonzero(df["failed"].to_numpy())

print(f"{tfidf.shape[1]} terms reduced to {n_components} components")
print(f"{len(failed_positions)} failed records are searchable")

import numpy as np from sklearn.decomposition import TruncatedSVD from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.preprocessing import normalize vectorizer = TfidfVectorizer(stop_words="english", sublinear_tf=True) tfidf = vectorizer.fit_transform(df["failure_document"]) n_components = min(20, tfidf.shape[1] - 1, len(df) - 1) svd = TruncatedSVD(n_components=n_components, random_state=0) vectors = normalize(svd.fit_transform(tfidf)) terms = np.array(vectorizer.get_feature_names_out()) failed_positions = np.flatnonzero(df["failed"].to_numpy()) print(f"{tfidf.shape[1]} terms reduced to {n_components} components") print(f"{len(failed_positions)} failed records are searchable")

In [ ]:

def shared_top_terms(query_tfidf, position, n=4):
    """Terms carrying weight in both the query and the match."""
    overlap = query_tfidf.multiply(tfidf[position]).toarray().ravel()
    ranked = np.argsort(-overlap)[:n]
    return ", ".join(terms[k] for k in ranked if overlap[k] > 0)


def similar_failures(query, top_k=TOP_K):
    """Rank failed records against a record id or a free-text description."""
    matches = np.flatnonzero((df["record_id"] == query).to_numpy())
    if matches.size:
        position = int(matches[0])
        query_tfidf = tfidf[position]
        query_vector = vectors[position]
        skip = {position}
    else:
        query_tfidf = vectorizer.transform([query])
        query_vector = normalize(svd.transform(query_tfidf))[0]
        skip = set()

    scored = [
        (int(position), float(vectors[position] @ query_vector))
        for position in failed_positions
        if position not in skip
    ]
    scored.sort(key=lambda pair: -pair[1])

    rows = []
    for position, similarity in scored[:top_k]:
        record = df.iloc[position]
        rows.append({
            "record_id": record["record_id"],
            "similarity": round(similarity, 3),
            "app_version": record["app_version"],
            "metric": FAILURE_METRIC,
            "score": record["score"],
            "first_error": record["first_error"][:55],
            "model": record["model"],
            "tools": ", ".join(record["tools"]),
            "shared_terms": shared_top_terms(query_tfidf, position),
        })
    return pd.DataFrame(rows)


similar_failures(QUERY_RECORD_ID)

def shared_top_terms(query_tfidf, position, n=4): """Terms carrying weight in both the query and the match.""" overlap = query_tfidf.multiply(tfidf[position]).toarray().ravel() ranked = np.argsort(-overlap)[:n] return ", ".join(terms[k] for k in ranked if overlap[k] > 0) def similar_failures(query, top_k=TOP_K): """Rank failed records against a record id or a free-text description.""" matches = np.flatnonzero((df["record_id"] == query).to_numpy()) if matches.size: position = int(matches[0]) query_tfidf = tfidf[position] query_vector = vectors[position] skip = {position} else: query_tfidf = vectorizer.transform([query]) query_vector = normalize(svd.transform(query_tfidf))[0] skip = set() scored = [ (int(position), float(vectors[position] @ query_vector)) for position in failed_positions if position not in skip ] scored.sort(key=lambda pair: -pair[1]) rows = [] for position, similarity in scored[:top_k]: record = df.iloc[position] rows.append({ "record_id": record["record_id"], "similarity": round(similarity, 3), "app_version": record["app_version"], "metric": FAILURE_METRIC, "score": record["score"], "first_error": record["first_error"][:55], "model": record["model"], "tools": ", ".join(record["tools"]), "shared_terms": shared_top_terms(query_tfidf, position), }) return pd.DataFrame(rows) similar_failures(QUERY_RECORD_ID)

shared_terms is why each row is here. It lists the terms that carry weight in both the query record and the match, which is usually enough to see whether a result is the same failure or a coincidence.

The same function takes free text, which is what you want when you are working from a bug report rather than a record id.

In [ ]:

similar_failures("the answer invented a price after the pricing tool failed")

similar_failures("the answer invented a price after the pricing tool failed")

Step 2: Find recurring failure modes¶

What failure patterns are affecting us most?

Step 1 answers a question about one record. This step steps back and groups every failed record, so you can tell a one-off from a pattern worth fixing. Records are selected using the metric direction and threshold from the configuration cell, then clustered with a fixed seed so the notebook gives the same answer twice.

In [ ]:

from collections import Counter

from sklearn.cluster import KMeans

failed = df.loc[df["failed"]].copy()
failed_vectors = vectors[failed_positions]

kmeans = KMeans(n_clusters=N_CLUSTERS, random_state=0, n_init=10)
failed["cluster"] = kmeans.fit_predict(failed_vectors)


def cluster_centroid(positions):
    return normalize(vectors[positions].mean(axis=0).reshape(1, -1))[0]


rows = []
for cluster in range(N_CLUSTERS):
    member_mask = (failed["cluster"] == cluster).to_numpy()
    members = failed.loc[member_mask]
    positions = failed_positions[member_mask]

    centroid = cluster_centroid(positions)
    medoid = members.iloc[int(np.argmax(vectors[positions] @ centroid))]

    errors = [error for error in members["first_error"] if error]
    weights = np.asarray(tfidf[positions].mean(axis=0)).ravel()

    rows.append({
        "cluster": cluster,
        "records": len(members),
        "app_versions": ", ".join(sorted(members["app_version"].unique())),
        "medoid": medoid["record_id"],
        "common_error": Counter(errors).most_common(1)[0][0] if errors else "",
        "metric_range": (
            f"{members['score'].min():.2f} - {members['score'].max():.2f}"
        ),
        "tools": ", ".join(sorted({t for ts in members["tools"] for t in ts})),
        "top_terms": ", ".join(terms[k] for k in np.argsort(-weights)[:6]),
    })

summary = pd.DataFrame(rows).sort_values("records", ascending=False)
summary

from collections import Counter from sklearn.cluster import KMeans failed = df.loc[df["failed"]].copy() failed_vectors = vectors[failed_positions] kmeans = KMeans(n_clusters=N_CLUSTERS, random_state=0, n_init=10) failed["cluster"] = kmeans.fit_predict(failed_vectors) def cluster_centroid(positions): return normalize(vectors[positions].mean(axis=0).reshape(1, -1))[0] rows = [] for cluster in range(N_CLUSTERS): member_mask = (failed["cluster"] == cluster).to_numpy() members = failed.loc[member_mask] positions = failed_positions[member_mask] centroid = cluster_centroid(positions) medoid = members.iloc[int(np.argmax(vectors[positions] @ centroid))] errors = [error for error in members["first_error"] if error] weights = np.asarray(tfidf[positions].mean(axis=0)).ravel() rows.append({ "cluster": cluster, "records": len(members), "app_versions": ", ".join(sorted(members["app_version"].unique())), "medoid": medoid["record_id"], "common_error": Counter(errors).most_common(1)[0][0] if errors else "", "metric_range": ( f"{members['score'].min():.2f} - {members['score'].max():.2f}" ), "tools": ", ".join(sorted({t for ts in members["tools"] for t in ts})), "top_terms": ", ".join(terms[k] for k in np.argsort(-weights)[:6]), }) summary = pd.DataFrame(rows).sort_values("records", ascending=False) summary

Clusters are ranked by how many records they affect, so the top row is the pattern costing you the most. The medoid is the record closest to the centre of its cluster, which makes it the least arbitrary example to open first.

The projection below is for navigation only. PCA runs on the same vectors after clustering; it never feeds the clustering.

In [ ]:

import matplotlib.pyplot as plt
from sklearn.decomposition import PCA

projection = PCA(n_components=2, random_state=0).fit_transform(failed_vectors)

fig, ax = plt.subplots(figsize=(8, 5))
for cluster in range(N_CLUSTERS):
    member_mask = (failed["cluster"] == cluster).to_numpy()
    ax.scatter(
        projection[member_mask, 0],
        projection[member_mask, 1],
        s=70,
        label=f"cluster {cluster}",
    )

for (x, y), record_id in zip(projection, failed["record_id"]):
    ax.annotate(
        record_id.replace("record-", ""),
        (x, y),
        xytext=(4, 3),
        textcoords="offset points",
        fontsize=7,
        alpha=0.75,
    )

ax.set_title(f"Failed records, {FAILURE_METRIC} below {FAILURE_THRESHOLD}")
ax.set_xlabel("component 1")
ax.set_ylabel("component 2")
ax.legend()
plt.show()

import matplotlib.pyplot as plt from sklearn.decomposition import PCA projection = PCA(n_components=2, random_state=0).fit_transform(failed_vectors) fig, ax = plt.subplots(figsize=(8, 5)) for cluster in range(N_CLUSTERS): member_mask = (failed["cluster"] == cluster).to_numpy() ax.scatter( projection[member_mask, 0], projection[member_mask, 1], s=70, label=f"cluster {cluster}", ) for (x, y), record_id in zip(projection, failed["record_id"]): ax.annotate( record_id.replace("record-", ""), (x, y), xytext=(4, 3), textcoords="offset points", fontsize=7, alpha=0.75, ) ax.set_title(f"Failed records, {FAILURE_METRIC} below {FAILURE_THRESHOLD}") ax.set_xlabel("component 1") ax.set_ylabel("component 2") ax.legend() plt.show()

Step 3: Inspect one failure group¶

Where in the trace does this failure diverge?

The largest cluster is selected by default. Two representative failures are shown against the passing record nearest the cluster centre, because the fastest way to read a broken trace is to put a working one beside it.

In [ ]:

target_cluster = int(summary.iloc[0]["cluster"])

member_mask = (failed["cluster"] == target_cluster).to_numpy()
members = failed.loc[member_mask]
member_positions = failed_positions[member_mask]

centroid = cluster_centroid(member_positions)
ranked = np.argsort(-(vectors[member_positions] @ centroid))
representatives = members.iloc[ranked[:2]]

passing_positions = np.flatnonzero(~df["failed"].to_numpy())
nearest = int(np.argmax(vectors[passing_positions] @ centroid))
nearby_success = df.iloc[passing_positions[nearest]]

print(f"Cluster {target_cluster}, {len(members)} records")
print(f"Representatives: {', '.join(representatives['record_id'])}")
print(f"Nearest passing record: {nearby_success['record_id']}")

target_cluster = int(summary.iloc[0]["cluster"]) member_mask = (failed["cluster"] == target_cluster).to_numpy() members = failed.loc[member_mask] member_positions = failed_positions[member_mask] centroid = cluster_centroid(member_positions) ranked = np.argsort(-(vectors[member_positions] @ centroid)) representatives = members.iloc[ranked[:2]] passing_positions = np.flatnonzero(~df["failed"].to_numpy()) nearest = int(np.argmax(vectors[passing_positions] @ centroid)) nearby_success = df.iloc[passing_positions[nearest]] print(f"Cluster {target_cluster}, {len(members)} records") print(f"Representatives: {', '.join(representatives['record_id'])}") print(f"Nearest passing record: {nearby_success['record_id']}")

In [ ]:

def slow_spans(spans):
    """Spans at or above the within-trace p90, ignoring the root span."""
    inner = [span for span in spans if span["span_type"] != "RECORD_ROOT"]
    if not inner:
        return []
    cutoff = float(np.percentile([s["duration_ms"] for s in inner], 90))
    return [
        f"{span['span_type']}/{span['name']} ({span['duration_ms']} ms)"
        for span in inner
        if span["duration_ms"] >= cutoff
    ]


def first_error_span(spans):
    """The first span in trace order that recorded an error."""
    for span in spans:
        if span["error"]:
            return f"{span['span_type']}/{span['name']}: {span['error']}"
    return ""


def column(record, role):
    return {
        "role": role,
        "app version": record["app_version"],
        FAILURE_METRIC: record["score"],
        "input": record["input"],
        "output": record["output"],
        "explanation": record["metric_explanations"][FAILURE_METRIC],
        "span path": " > ".join(record["span_path"]),
        "first error span": first_error_span(record["spans"]) or "none",
        "slow spans": "; ".join(slow_spans(record["spans"])) or "none",
    }


comparison = {
    record["record_id"]: column(record, "representative failure")
    for _, record in representatives.iterrows()
}
comparison[nearby_success["record_id"]] = column(
    nearby_success, "nearest passing record"
)

with pd.option_context("display.max_colwidth", 90):
    display(pd.DataFrame(comparison))

def slow_spans(spans): """Spans at or above the within-trace p90, ignoring the root span.""" inner = [span for span in spans if span["span_type"] != "RECORD_ROOT"] if not inner: return [] cutoff = float(np.percentile([s["duration_ms"] for s in inner], 90)) return [ f"{span['span_type']}/{span['name']} ({span['duration_ms']} ms)" for span in inner if span["duration_ms"] >= cutoff ] def first_error_span(spans): """The first span in trace order that recorded an error.""" for span in spans: if span["error"]: return f"{span['span_type']}/{span['name']}: {span['error']}" return "" def column(record, role): return { "role": role, "app version": record["app_version"], FAILURE_METRIC: record["score"], "input": record["input"], "output": record["output"], "explanation": record["metric_explanations"][FAILURE_METRIC], "span path": " > ".join(record["span_path"]), "first error span": first_error_span(record["spans"]) or "none", "slow spans": "; ".join(slow_spans(record["spans"])) or "none", } comparison = { record["record_id"]: column(record, "representative failure") for _, record in representatives.iterrows() } comparison[nearby_success["record_id"]] = column( nearby_success, "nearest passing record" ) with pd.option_context("display.max_colwidth", 90): display(pd.DataFrame(comparison))

The last thing to pin down is the point where the two traces stop agreeing. Span paths are compared step by step on span_type/name, so a step that is missing and a step that was replaced both show up.

In [ ]:

def first_divergence(failing_path, passing_path):
    """First step where two span paths stop agreeing, or None."""
    for step, (failing, passing) in enumerate(zip(failing_path, passing_path)):
        if failing != passing:
            return step, failing, passing
    if len(failing_path) != len(passing_path):
        step = min(len(failing_path), len(passing_path))
        return (
            step,
            failing_path[step] if step < len(failing_path) else "(trace ends)",
            passing_path[step] if step < len(passing_path) else "(trace ends)",
        )
    return None


representative = representatives.iloc[0]
divergence = first_divergence(
    representative["span_path"], nearby_success["span_path"]
)
divergence

def first_divergence(failing_path, passing_path): """First step where two span paths stop agreeing, or None.""" for step, (failing, passing) in enumerate(zip(failing_path, passing_path)): if failing != passing: return step, failing, passing if len(failing_path) != len(passing_path): step = min(len(failing_path), len(passing_path)) return ( step, failing_path[step] if step < len(failing_path) else "(trace ends)", passing_path[step] if step < len(passing_path) else "(trace ends)", ) return None representative = representatives.iloc[0] divergence = first_divergence( representative["span_path"], nearby_success["span_path"] ) divergence

Diagnostic summary¶

The text below is assembled from the numbers above with no model in the loop, so it says the same thing on every run. It reports what co-occurs in the cluster and leaves the causal claim to you.

In [ ]:

def diagnostic_summary():
    """Deterministic description of what this cluster has in common."""
    lines = [
        f"Cluster {target_cluster} holds {len(members)} of {len(failed)} failed"
        f" records, on app versions"
        f" {', '.join(sorted(members['app_version'].unique()))}.",
        f"{FAILURE_METRIC} runs {members['score'].min():.2f} to"
        f" {members['score'].max():.2f} across the cluster, median"
        f" {members['score'].median():.2f}, against a threshold of"
        f" {FAILURE_THRESHOLD}.",
    ]

    errors = [error for error in members["first_error"] if error]
    if errors:
        error, count = Counter(errors).most_common(1)[0]
        lines.append(
            f"{count} of {len(members)} records report the same first error:"
            f" {error}"
        )
    else:
        lines.append("No record in the cluster recorded an error.")

    slow = slow_spans(representative["spans"])
    if slow:
        lines.append(
            f"In {representative['record_id']}, the spans at or above the"
            f" within-trace p90 are: {'; '.join(slow)}."
        )

    if divergence:
        step, failing, passing = divergence
        lines.append(
            f"Against {nearby_success['record_id']}, the span path first"
            f" differs at step {step}: {failing} in the failing trace,"
            f" {passing} in the passing one."
        )
    else:
        lines.append(
            f"The span path of {representative['record_id']} matches"
            f" {nearby_success['record_id']} step for step."
        )

    lines.append(
        "These signals co-occur in the cluster. Confirming a cause takes a"
        " change to the app and a rerun."
    )
    return "\n".join(lines)


print(diagnostic_summary())

def diagnostic_summary(): """Deterministic description of what this cluster has in common.""" lines = [ f"Cluster {target_cluster} holds {len(members)} of {len(failed)} failed" f" records, on app versions" f" {', '.join(sorted(members['app_version'].unique()))}.", f"{FAILURE_METRIC} runs {members['score'].min():.2f} to" f" {members['score'].max():.2f} across the cluster, median" f" {members['score'].median():.2f}, against a threshold of" f" {FAILURE_THRESHOLD}.", ] errors = [error for error in members["first_error"] if error] if errors: error, count = Counter(errors).most_common(1)[0] lines.append( f"{count} of {len(members)} records report the same first error:" f" {error}" ) else: lines.append("No record in the cluster recorded an error.") slow = slow_spans(representative["spans"]) if slow: lines.append( f"In {representative['record_id']}, the spans at or above the" f" within-trace p90 are: {'; '.join(slow)}." ) if divergence: step, failing, passing = divergence lines.append( f"Against {nearby_success['record_id']}, the span path first" f" differs at step {step}: {failing} in the failing trace," f" {passing} in the passing one." ) else: lines.append( f"The span path of {representative['record_id']} matches" f" {nearby_success['record_id']} step for step." ) lines.append( "These signals co-occur in the cluster. Confirming a cause takes a" " change to the app and a rerun." ) return "\n".join(lines) print(diagnostic_summary())

That is the whole loop: one bad record, the pattern it belongs to, and the step where that pattern parts company with a run that worked. Point the configuration cell at a different metric, threshold or record id and the three steps run again over the same data.