#!/usr/bin/env python3 """Atelier synthétique Nymphar.ai. Aucun appel LLM, réseau ou export cloud. Les durées, coûts et évaluations sont des fixtures, pas des mesures de performance. Export JSONL pédagogique du SDK : ce fichier n'est pas un message OTLP. """ import argparse import copy import html import json import math import tempfile from collections import defaultdict from datetime import datetime, timezone from pathlib import Path from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExporter, SpanExportResult from opentelemetry.trace import Status, StatusCode class JsonlExporter(SpanExporter): def __init__(self, path): self.path = path def export(self, spans): with self.path.open("a", encoding="utf-8") as output: for span in spans: output.write(json.dumps(json.loads(span.to_json()), ensure_ascii=False) + "\n") return SpanExportResult.SUCCESS def shutdown(self): pass def generate(scenarios, path): provider = TracerProvider() provider.add_span_processor(SimpleSpanProcessor(JsonlExporter(path))) tracer = provider.get_tracer("nymphar.synthetic-observability", "1.0") # Horodatage fixe de démonstration, sans lien avec le temps d'exécution réel. base = int(datetime(2026, 10, 3, 7, tzinfo=timezone.utc).timestamp()) * 10**9 for number, case in enumerate(scenarios): start = base + number * 10 * 10**9 end = start + case["latency_ms"] * 10**6 root = tracer.start_span("support.answer", start_time=start, attributes={ "app.kind": "workflow", "app.request_id": case["id"], "app.environment": "synthetic", "app.prompt_version": "support-v3", "app.technical_status": case["technical_status"], "app.quality": case["quality"], "app.quality_rubric": "policy-correct-v1", "app.document_version": case["document_version"], }) context = trace.set_span_in_context(root) retrieval = tracer.start_span("retrieve.policy", context=context, start_time=start, attributes={"app.kind": "retrieval"}) retrieval.end(start + 100 * 10**6) cursor = start + 100 * 10**6 duration = (end - cursor) // len(case["attempts"]) for i, attempt in enumerate(case["attempts"], 1): attributes = {"app.kind": "model_attempt", "app.attempt": i, "app.model": "fictional-model", "app.cost_source": "fixture", "app.cost_known": attempt["cost_usd"] is not None} if attempt["cost_usd"] is not None: attributes["app.cost_usd"] = attempt["cost_usd"] span = tracer.start_span("model.attempt", context=context, start_time=cursor, attributes=attributes) span.set_status(Status(StatusCode.ERROR if attempt["error"] else StatusCode.OK)) span.end(cursor + duration) cursor += duration root.set_status(Status(StatusCode.ERROR if case["technical_status"] == "error" else StatusCode.OK)) root.end(end) provider.shutdown() def aggregate(spans): """Dénominateurs : workflows pour qualité/erreurs, tentatives pour coûts connus.""" groups = defaultdict(list) identities = set() for span in spans: identity = (span["context"]["trace_id"], span["context"]["span_id"]) if identity in identities: raise ValueError("Span dupliqué : agrégation refusée") identities.add(identity) groups[identity[0]].append(span) rows = [] request_ids = set() for trace_id, members in groups.items(): roots = [s for s in members if s["attributes"].get("app.kind") == "workflow"] if len(roots) != 1 or roots[0].get("parent_id") is not None: raise ValueError("Une racine workflow par trace est nécessaire") root = roots[0] root_id = root["context"]["span_id"] # Ce validateur porte sur le graphe simple de cet atelier uniquement. if any(s.get("parent_id") != root_id for s in members if s is not root): raise ValueError("Parent inattendu dans le graphe de cet atelier") attrs = root["attributes"] request_id = attrs["app.request_id"] if request_id in request_ids: raise ValueError("Identifiant de requête dupliqué") request_ids.add(request_id) attempts = [s for s in members if s["attributes"].get("app.kind") == "model_attempt"] costs = [s["attributes"].get("app.cost_usd") for s in attempts] known = [c for c in costs if c is not None] if any(not isinstance(c, (int, float)) or not math.isfinite(c) or c < 0 for c in known): raise ValueError("Coût invalide") quality = attrs["app.quality"] if quality not in ("accepted", "rejected", "unreviewed"): raise ValueError("Évaluation inconnue") latency = (datetime.fromisoformat(root["end_time"]) - datetime.fromisoformat(root["start_time"])).total_seconds() * 1000 rows.append({"request_id": request_id, "trace_id": trace_id, "technical_status": attrs["app.technical_status"], "quality": quality, "document_version": attrs["app.document_version"], "latency_ms": round(latency, 3), "attempts": len(attempts), "failed_attempts": sum(s["status"]["status_code"] == "ERROR" for s in attempts), "known_cost_attempts": len(known), "known_cost_usd": round(sum(known), 6), "model_cost_usd": round(sum(known), 6) if costs and len(known) == len(costs) else None}) n = len(rows) accepted = sum(r["quality"] == "accepted" for r in rows) reviewed = sum(r["quality"] != "unreviewed" for r in rows) total_attempts = sum(r["attempts"] for r in rows) known_attempts = sum(r["known_cost_attempts"] for r in rows) known_cost = round(sum(r["known_cost_usd"] for r in rows), 6) total_cost = known_cost if n and total_attempts and known_attempts == total_attempts else None return {"synthetic": True, "spans": len(spans), "workflows": n, "technical_ok": sum(r["technical_status"] == "ok" for r in rows), "technical_errors": sum(r["technical_status"] == "error" for r in rows), "reviewed": reviewed, "accepted": accepted, "rejected": reviewed - accepted, "unreviewed": n - reviewed, "acceptance_among_reviewed": accepted / reviewed if reviewed else None, "review_coverage": reviewed / n if n else None, "model_attempts": total_attempts, "workflows_with_retry": sum(r["attempts"] > 1 for r in rows), "known_cost_attempts": known_attempts, "known_model_cost_usd": known_cost, "total_model_cost_usd": total_cost, "model_cost_per_accepted_workflow_usd": total_cost / accepted if total_cost is not None and accepted else None, "p95_latency_ms_nearest_rank": sorted(r["latency_ms"] for r in rows)[math.ceil(.95 * n) - 1] if n else None, "rows": sorted(rows, key=lambda r: r["request_id"])} def dashboard(report): cards = [("Parcours", str(report["workflows"])), ("Évalués", f'{report["reviewed"]} / {report["workflows"]}'), ("Acceptés parmi les évalués", f'{report["accepted"]} / {report["reviewed"]}'), ("Coûts connus des tentatives", f'{report["known_cost_attempts"]} / {report["model_attempts"]}')] cards_html = "".join(f'
{k}{v}
' for k, v in cards) labels = {"accepted": "Acceptée", "rejected": "Refusée", "unreviewed": "Non évaluée"} rows_html = "" for row in report["rows"]: cost = "Inconnu" if row["model_cost_usd"] is None else f'{row["model_cost_usd"]:.4f} $' values = [row["request_id"], row["technical_status"], labels[row["quality"]], row["document_version"], str(row["attempts"]), str(row["latency_ms"]), cost] rows_html += '' + ''.join(f'{html.escape(x)}' for x in values) + '' return f''' LLM observability : atelier synthétique | Nymphar.ai

Nymphar.ai · Atelier local

Une réponse rapide peut être incorrecte.

Données entièrement fictives. Les traces sont exportées par le SDK OpenTelemetry ; les durées, évaluations et montants sont imposés par le scénario. Aucun modèle n’a été appelé.

{cards_html}
{rows_html}
RequêteTechniqueQualitéDocumentTentativesDurée (ms)Coût modèle

Trois contrôles avant de conclure

  1. La requête t2 réussit techniquement mais sa réponse est refusée : le document utilisé est ancien.
  2. La requête t3 est acceptée après deux tentatives. La tentative en erreur reste comptée dans son coût.
  3. Une tentative n’a aucun coût renseigné. Le sous-total connu vaut {report['known_model_cost_usd']:.4f} $. Le total et le coût par parcours accepté restent inconnus.

Le p95 calculé par rang supérieur vaut {report['p95_latency_ms_nearest_rank']} ms sur ces six cas synthétiques. Il sert à vérifier la formule, pas à estimer une performance en production. Les coûts d’outils, d’évaluation et de supervision ne sont pas inclus.

Lire le guide et reproduire l’atelier

''' def self_test(cases): with tempfile.TemporaryDirectory() as folder: path = Path(folder) / "traces.jsonl" generate(cases, path) spans = [json.loads(line) for line in path.read_text().splitlines()] result = aggregate(spans) assert result["spans"] == 19 and result["workflows"] == 6 assert result["model_attempts"] == 7 and result["known_cost_attempts"] == 6 assert result["known_model_cost_usd"] == .009 assert result["total_model_cost_usd"] is None and result["model_cost_per_accepted_workflow_usd"] is None assert (result["reviewed"], result["accepted"], result["technical_ok"]) == (3, 2, 5) assert result["review_coverage"] == .5 and result["acceptance_among_reviewed"] == 2 / 3 assert result["p95_latency_ms_nearest_rank"] == 2200 retry = next(r for r in result["rows"] if r["request_id"] == "t3") assert retry["model_cost_usd"] == .003 and retry["failed_attempts"] == 1 empty = aggregate([]) assert empty["review_coverage"] is None and empty["total_model_cost_usd"] is None assert empty["p95_latency_ms_nearest_rank"] is None # Le coût complet devient calculable quand le coût manquant est réellement renseigné. complete = copy.deepcopy(spans) for span in complete: attrs = span["attributes"] if attrs.get("app.kind") == "model_attempt" and "app.cost_usd" not in attrs: attrs["app.cost_usd"] = .004 assert aggregate(complete)["model_cost_per_accepted_workflow_usd"] == .0065 broken_parent = copy.deepcopy(spans) next(s for s in broken_parent if s.get("parent_id"))["parent_id"] = "0x0000000000000000" for invalid in [spans + [spans[0]], broken_parent, [s for s in spans if s["attributes"].get("app.kind") != "workflow"]]: try: aggregate(invalid) except ValueError: pass else: raise AssertionError("Un graphe invalide a été accepté") print("OK : agrégations, dénominateurs, retries, coût inconnu, jeu vide, doublons et parents.") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--scenarios", type=Path, default=Path(__file__).with_name("scenarios.json")) parser.add_argument("--output", type=Path, default=Path("resultats-observabilite")) parser.add_argument("--self-test", action="store_true") args = parser.parse_args() cases = json.loads(args.scenarios.read_text()) if args.self_test: self_test(cases) return if args.output.exists() and any(args.output.iterdir()): raise SystemExit("Le dossier de sortie doit être vide. Choisissez un nouveau chemin --output.") args.output.mkdir(parents=True, exist_ok=True) path = args.output / "traces.jsonl" generate(cases, path) report = aggregate([json.loads(line) for line in path.read_text().splitlines()]) (args.output / "resultat.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n") (args.output / "tableau-de-bord.html").write_text(dashboard(report)) print(json.dumps({k: v for k, v in report.items() if k != "rows"}, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()