Tutorials
Extract signals for trace analytics
Optimize structured workflow-signal extraction with Seldon, then validate evidence and cluster representations in your application.
Agent traces contain conversation, tool results, and operational noise. This tutorial uses a Seldon Function to turn each visible trace into structured signals. Your application still prepares the traces, validates evidence, creates vector representations, and clusters them.
Paste the Python blocks in order into trace_tutorial.py. Complete the shared setup first. The default execution is an offline fixture check, not a model benchmark. Our six fictional traces are deliberately small enough to inspect by hand.
1. Define the extraction contract and fixtures
The function receives only visible messages. The application keeps group labels separately for scoring, so the extractor cannot read the answer. Production trace normalization should use an explicit allowlist of fields and remove secrets before hosted inference.
import json
import os
import time
import httpx
from jsonschema import validate
from jsonschema.exceptions import ValidationError
SYSTEM = (
"Read the visible trace and return JSON with intent, actions, outcome, "
"failure_signals, and evidence. intent is a short task description; actions "
"is an array of short action strings. outcome is success, failure, or unknown. "
"failure_signals is an array of observed problems, empty if none are shown. "
"evidence contains one or more exact, nonempty quotations copied from the "
"trace supporting your interpretation. Do not infer an unobserved result."
)
SCHEMA = {
"type": "object",
"properties": {
"intent": {"type": "string", "minLength": 1},
"actions": {"type": "array", "items": {"type": "string"}},
"outcome": {"type": "string", "enum": ["success", "failure", "unknown"]},
"failure_signals": {"type": "array", "items": {"type": "string"}},
"evidence": {"type": "array", "minItems": 1,
"items": {"type": "string", "minLength": 1}},
},
"required": ["intent", "actions", "outcome", "failure_signals", "evidence"],
"additionalProperties": False,
}
def normalize_trace(trace):
return "\n".join(message["role"] + ": " + message["content"]
for message in trace["messages"])
FIXTURES = []
for group, request, reply, intent, action in [
("refund", "Refund order 101.", "Refund issued.", "refund order", "issue refund"),
("refund", "Return payment for order 102.", "Refund issued.", "refund order", "issue refund"),
("refund", "Please refund order 103.", "Refund issued.", "refund order", "issue refund"),
("address", "Change order 201 delivery address.", "Address updated.", "change address", "update address"),
("address", "Ship order 202 to a new address.", "Address updated.", "change address", "update address"),
("address", "Update delivery address for order 203.", "Address updated.", "change address", "update address"),
]:
trace = {"messages": [{"role": "user", "content": request},
{"role": "tool", "content": reply}]}
FIXTURES.append({
"trace": trace, "text": normalize_trace(trace), "group": group,
"expected": {"intent": intent, "actions": [action], "outcome": "success",
"failure_signals": [], "evidence": [reply]},
})The group and expected fields never enter normalize_trace. This protects these fixture labels from accidental inclusion in the inference request. Real traces need cases with failure and unknown outcomes as well; these six examples only demonstrate the wiring.
2. Build the original call and importable request
def make_request(text, model):
return {
"model": model,
"messages": [{"role": "system", "content": SYSTEM},
{"role": "user", "content": text}],
"response_format": {
"type": "json_schema",
"json_schema": {"name": "workflow_signals", "strict": True,
"schema": SCHEMA},
},
}
def validate_signals(text, output):
validate(output, SCHEMA)
if any(not quote.strip() or quote not in text for quote in output["evidence"]):
raise ValueError("Evidence must be copied from the visible trace")
return output
def invoke(text, model):
started = time.perf_counter()
result = {"output": None, "error": None, "usage": {}, "seldon": {}}
try:
# Zero automatic retries; a timed-out request may still incur usage.
with httpx.Client(timeout=httpx.Timeout(65.0, connect=10.0)) as client:
response = client.post(
os.environ["SELDON_BASE_URL"].rstrip("/") + "/chat/completions",
headers={"Authorization": "Bearer " + os.environ["SELDON_API_KEY"]},
json=make_request(text, model),
)
if response.status_code in {401, 402, 403, 429}:
raise RuntimeError(f"Batch stopped: authentication, credit, or rate limit error (HTTP {response.status_code})")
response.raise_for_status()
body = response.json()
if not isinstance(body, dict):
raise ValueError("Expected a chat-completion object")
result["usage"] = body.get("usage", {})
result["seldon"] = body.get("seldon", {})
output = json.loads(body["choices"][0]["message"]["content"])
result["output"] = validate_signals(text, output)
except (httpx.HTTPError, ValueError, TypeError, KeyError, IndexError,
ValidationError) as exc:
result["error"] = type(exc).__name__
result["latency_seconds"] = time.perf_counter() - started
return resultEvidence grounding detects invented quotations. It does not prove that a real quotation supports the claimed intent or outcome; that needs semantic evaluation and human review.
3. Compare alternatives in Functions
The completed script prints an importable request. Set SELDON_REFERENCE_MODEL to a routable model, run offline, and paste that JSON into Functions → New function → Import. To measure the original call, explicitly set RUN_PAID=1 before running without a function alias.
# Assemble every Python block on this page before running.
RUN_PAID=0 python trace_tutorial.py
# Six paid original-model calls; ignore any existing function alias.
RUN_PAID=1 env -u SELDON_FUNCTION_MODEL python trace_tutorial.pyUse the shared dashboard walkthrough. Confirm that the last user message is the variable trace. Review the Evaluation criteria for unsupported outcomes, omitted failure signals, and invented evidence. Generated evaluation inputs are not your external task-group labels.
Choose a priority, inspect the compilation cap and preview, and authorize optimization. Inspect reliability, measurement coverage, failures, and fallback alongside cost and latency in Results. A single smaller model can be the best measured implementation; the historical trace demo selected Luna rather than its more complex candidate.
Before activation, use Try on a fresh fictional failed trace, such as a tool reporting that a requested address update was rejected. Inspect whether the result remains grounded. Try calls charge usage. Keep the original if no alternative has adequate evidence or application quality.
4. Validate and cluster in the application
These steps are outside Seldon. Failed or missing predictions remain in completion coverage; clustering reports exactly how many records it used. We do not silently count an incomplete subset as a successful full run.
def offline_records():
return [{"output": fixture["expected"], "error": None} for fixture in FIXTURES]
def valid_rows(records):
rows = []
for index, fixture in enumerate(FIXTURES):
record = records[index] if index < len(records) else {}
try:
if record.get("error") or record.get("output") is None:
raise ValueError("Missing prediction")
output = validate_signals(fixture["text"], record["output"])
rows.append((fixture, output))
except (ValueError, TypeError, KeyError, ValidationError):
continue
return rows
def score(records):
valid = len(valid_rows(records))
return {"total": len(FIXTURES), "valid": valid,
"execution_or_grounding_failures": len(FIXTURES) - valid,
"completion_coverage": valid / len(FIXTURES) if FIXTURES else None}
def cluster_report(records):
# Optional application dependencies; no Seldon call is made here.
from sklearn.cluster import KMeans
from sklearn.feature_extraction.text import HashingVectorizer
from sklearn.metrics import adjusted_rand_score
rows = valid_rows(records)
if len(rows) < 3:
return {"records_used": len(rows), "task_group_ari": None,
"reason": "Need at least three valid records for two clusters"}
representations = [
" ".join([output["intent"], *output["actions"], output["outcome"],
*output["failure_signals"]])
for _, output in rows
]
if os.getenv("USE_SEMANTIC_EMBEDDINGS") == "1":
from sentence_transformers import SentenceTransformer
encoder = SentenceTransformer(
"sentence-transformers/all-MiniLM-L6-v2",
revision="c9745ed1d9f207416be6d2e6f8de32d1f16199bf",
)
vectors = encoder.encode(representations, normalize_embeddings=True)
embedding = "MiniLM"
else:
vectors = HashingVectorizer(n_features=1024, alternate_sign=False).transform(
representations
)
embedding = "lexical hashing (offline wiring only)"
clusters = KMeans(n_clusters=2, random_state=42, n_init=10).fit_predict(vectors)
groups = [fixture["group"] for fixture, _ in rows]
return {"records_used": len(rows), "embedding": embedding,
"task_group_ari": float(adjusted_rand_score(groups, clusters)),
"scope": "Valid-output subset only; inspect completion coverage too"}Install scikit-learn to enable clustering. Set RUN_CLUSTERING=1 for the local vector-and-clustering check. Lexical hashing is the offline default, not a semantic embedding model. Optionally install sentence-transformers and set USE_SEMANTIC_EMBEDDINGS=1; the first use downloads the pinned MiniLM weights and needs network access and disk space. This is separate from Seldon usage.
python -m pip install scikit-learn
RUN_PAID=0 RUN_CLUSTERING=1 USE_SEMANTIC_EMBEDDINGS=0 python trace_tutorial.py
# Optional semantic embeddings: downloads model weights on first use.
python -m pip install sentence-transformers
RUN_PAID=0 RUN_CLUSTERING=1 USE_SEMANTIC_EMBEDDINGS=1 python trace_tutorial.pyK-means uses two clusters because our toy fixture has two known groups. It is not an automatically discovered production taxonomy. The larger demo used MiniLM and density-based HDBSCAN, with optional dimensionality reduction. Adjusted Rand index (ARI) compares grouping assignments; task labels are only a proxy for usefulness, and a high score on six hand-written traces is not evidence of general quality.
5. Invoke the chosen function
Copy the activated alias or pinned model value from Use & monitor to SELDON_FUNCTION_MODEL. This imported single-call function expects the same fixed system message plus a new last user message. Do not switch to the input-object workflow calling convention.
def main():
reference = os.getenv("SELDON_REFERENCE_MODEL", "MODEL_FROM_MODELS_PAGE")
print("Import request (no credentials):")
print(json.dumps(make_request(FIXTURES[0]["text"], reference), indent=2))
if os.getenv("RUN_PAID") != "1":
print("Offline fixture wiring check; NOT model measurements.")
records = offline_records()
print(score(records))
if os.getenv("RUN_CLUSTERING") == "1":
print(cluster_report(records))
return
if reference == "MODEL_FROM_MODELS_PAGE":
raise ValueError("Set SELDON_REFERENCE_MODEL before paid execution")
models = {"reference": reference}
if os.getenv("SELDON_FUNCTION_MODEL"):
models["candidate"] = os.environ["SELDON_FUNCTION_MODEL"]
for name, model in models.items():
records = [invoke(fixture["text"], model) for fixture in FIXTURES]
print(name, score(records))
if os.getenv("RUN_CLUSTERING") == "1":
print(name, cluster_report(records))
for record in records:
print({k: record[k] for k in ("error", "latency_seconds", "usage", "seldon")})
if __name__ == "__main__":
main()After activation, replace the placeholder with the model value from your function:
export SELDON_FUNCTION_MODEL="seldon/function/YOUR_FUNCTION_ID"
RUN_PAID=1 RUN_CLUSTERING=1 USE_SEMANTIC_EMBEDDINGS=0 python trace_tutorial.pyWith both model values set, this makes twelve paid inference requests, separate from compilation and trial calls. Compare the same trace inputs and the same downstream representation settings. Retain execution receipts and inspect fallback metadata. Measure application-level results on a larger, independently held-out set containing successes, failures, and unknown outcomes before adopting a replacement.
Long production traces need chunking and an explicit policy for combining conflicting or repeated signals. This tutorial uses short traces and does not implement chunk merging. Continue with payment risk for a labeled application scoring example.