fix(examples-py): validate Ragas, Replicate, and Restate OTel 2.x spans

python/tracing/ragas/01_modern_metrics.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import asyncio
+
+from _shared import create_respan, example_context, finish_respan
+from ragas.metrics.collections import ExactMatch
+from respan import workflow
+
+CASE = "modern_metrics"
+
+
+@workflow(name="ragas_modern_metrics")
+def metric_workflow(reference: str, response: str) -> dict[str, object]:
+    metric = ExactMatch()
+    sync_value = metric.score(reference=reference, response=response).value
+    async_value = asyncio.run(
+        metric.ascore(reference=reference, response=response)
+    ).value
+    batch = metric.batch_score(
+        [
+            {"reference": reference, "response": response},
+            {"reference": "Rome", "response": "Milan"},
+        ]
+    )
+    return {
+        "sync": sync_value,
+        "async": async_value,
+        "batch": [item.value for item in batch],
+    }
+
+
+def main() -> None:
+    respan = create_respan()
+    try:
+        with example_context(CASE):
+            result = metric_workflow("Paris", "Paris")
+            print(result, flush=True)
+    finally:
+        finish_respan(respan)
+
+
+if __name__ == "__main__":
+    main()

python/tracing/ragas/02_evaluate.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import ragas
+from _shared import create_respan, example_context, finish_respan
+from ragas import EvaluationDataset
+from ragas.metrics import ExactMatch
+from respan import workflow
+
+CASE = "evaluate"
+
+
+@workflow(name="ragas_evaluate")
+def evaluation_workflow(question: str, answer: str) -> dict[str, object]:
+    dataset = EvaluationDataset.from_list(
+        [{"user_input": question, "response": answer, "reference": "Paris"}]
+    )
+    result = ragas.evaluate(
+        dataset,
+        metrics=[ExactMatch()],
+        experiment_name="offline-exact-match",
+        show_progress=False,
+    )
+    return {"exact_match": list(result["exact_match"])}
+
+
+def main() -> None:
+    respan = create_respan()
+    try:
+        with example_context(CASE):
+            result = evaluation_workflow("What is France's capital?", "Paris")
+            print(result, flush=True)
+    finally:
+        finish_respan(respan)
+
+
+if __name__ == "__main__":
+    main()

python/tracing/ragas/03_experiment.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import asyncio
+
+import ragas
+from _shared import create_respan, example_context, finish_respan
+from ragas.backends.inmemory import InMemoryBackend
+from ragas.dataset import Dataset
+from respan import workflow
+
+CASE = "experiment"
+BACKEND = InMemoryBackend()
+
+
+@ragas.experiment(backend=BACKEND, name_prefix="offline")
+async def answer_row(row: dict[str, str]) -> dict[str, str]:
+    return {"answer": row["answer"].upper()}
+
+
+@workflow(name="ragas_experiment")
+async def experiment_workflow(dataset_name: str) -> dict[str, object]:
+    dataset = Dataset(
+        name=dataset_name,
+        backend=BACKEND,
+        data=[{"answer": "Paris"}, {"answer": "Rome"}],
+    )
+    result = await answer_row.arun(dataset, name="two-rows")
+    return {"experiment": result.name, "rows": len(result)}
+
+
+async def main() -> None:
+    respan = create_respan()
+    try:
+        with example_context(CASE):
+            result = await experiment_workflow("capital-answers")
+            print(result, flush=True)
+    finally:
+        finish_respan(respan)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/tracing/ragas/README.md

@@ -0,0 +1,15 @@
+# Ragas tracing examples
+
+These examples validate current Ragas 0.4 evaluation, collection metrics, and
+experiment APIs with Respan OTel 2.x instrumentation. They are deterministic
+and need only the repository `RESPAN_API_KEY`.
+
+For local instrumentation development, install the Ragas package from the
+adjacent `respan` checkout in editable mode, then run:
+
+```bash
+RESPAN_EXAMPLE_RUN_ID=otel2-ragas-check python run_all.py
+```
+
+The runner preserves an existing marker, runs every process with a timeout,
+continues after failures, and returns nonzero when any example fails.

python/tracing/ragas/_shared.py

@@ -0,0 +1,78 @@
+"""Shared lifecycle and marker helpers for Ragas tracing examples."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any
+
+from dotenv import load_dotenv
+from respan import Respan, propagate_attributes
+from respan_instrumentation_ragas import RagasInstrumentor
+
+EXAMPLE_DIR = Path(__file__).resolve().parent
+REPO_ROOT = EXAMPLE_DIR.parents[2]
+EXAMPLE_SET = "ragas"
+DEFAULT_RESPAN_BASE_URL = "https://api.respan.ai/api"
+
+
+def load_env() -> None:
+    load_dotenv(REPO_ROOT / ".env", override=False)
+    if not os.getenv("RESPAN_API_KEY"):
+        raise RuntimeError("RESPAN_API_KEY must be set in the repository .env")
+
+
+def run_id() -> str:
+    value = os.getenv("RESPAN_EXAMPLE_RUN_ID")
+    if not value:
+        raise RuntimeError("RESPAN_EXAMPLE_RUN_ID must be supplied by run_all.py")
+    return value
+
+
+def create_respan() -> Respan:
+    load_env()
+    marker = run_id()
+    return Respan(
+        api_key=os.environ["RESPAN_API_KEY"],
+        base_url=os.getenv("RESPAN_BASE_URL", DEFAULT_RESPAN_BASE_URL),
+        app_name="ragas-examples",
+        metadata={
+            "example_set": EXAMPLE_SET,
+            "example_run_id": marker,
+            "run_id": marker,
+        },
+        instrumentations=[RagasInstrumentor()],
+        is_batching_enabled=False,
+        log_level=os.getenv("RESPAN_LOG_LEVEL", "WARNING"),
+    )
+
+
+@contextmanager
+def example_context(case: str) -> Iterator[None]:
+    marker = run_id()
+    with propagate_attributes(
+        custom_identifier=f"{EXAMPLE_SET}-{case}-{marker}",
+        trace_group_identifier=f"ragas_{case}",
+        metadata={
+            "example_set": EXAMPLE_SET,
+            "example_case": case,
+            "example_run_id": marker,
+            "run_id": marker,
+        },
+    ):
+        yield
+
+
+def finish_respan(respan: Respan) -> None:
+    try:
+        respan.flush()
+    finally:
+        respan.shutdown()
+
+
+def bounded_result(value: Any) -> Any:
+    if hasattr(value, "to_pandas"):
+        return {"rows": len(value)}
+    return value

python/tracing/ragas/requirements.txt

@@ -0,0 +1,4 @@
+ragas>=0.4.3,<0.5.0
+respan-ai>=4,<5
+respan-instrumentation-ragas>=0.1.0,<0.2.0
+python-dotenv>=1,<2

python/tracing/ragas/run_all.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+
+EXAMPLE_DIR = Path(__file__).resolve().parent
+SCRIPTS = ["01_modern_metrics.py", "02_evaluate.py", "03_experiment.py"]
+TIMEOUT_SECONDS = 120
+
+
+def marker() -> str:
+    existing = os.getenv("RESPAN_EXAMPLE_RUN_ID")
+    if existing:
+        return existing
+    return f"otel2-ragas-{datetime.now(UTC).strftime('%Y%m%dT%H%M%SZ')}"
+
+
+def main() -> None:
+    env = dict(os.environ)
+    env["RESPAN_EXAMPLE_RUN_ID"] = marker()
+    failures: list[str] = []
+    print(f"RESPAN_EXAMPLE_RUN_ID={env['RESPAN_EXAMPLE_RUN_ID']}", flush=True)
+    for script in SCRIPTS:
+        try:
+            result = subprocess.run(
+                [sys.executable, str(EXAMPLE_DIR / script)],
+                cwd=EXAMPLE_DIR,
+                env=env,
+                timeout=TIMEOUT_SECONDS,
+                check=False,
+            )
+        except subprocess.TimeoutExpired:
+            failures.append(f"{script}: timed out")
+            continue
+        if result.returncode:
+            failures.append(f"{script}: exit {result.returncode}")
+    if failures:
+        raise SystemExit("; ".join(failures))
+
+
+if __name__ == "__main__":
+    main()

python/tracing/ragas/test_contract.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+EXAMPLE_DIR = Path(__file__).resolve().parent
+
+
+def test_runner_covers_every_numbered_example() -> None:
+    tree = ast.parse((EXAMPLE_DIR / "run_all.py").read_text())
+    assignment = next(
+        node
+        for node in tree.body
+        if isinstance(node, ast.Assign)
+        and any(
+            isinstance(target, ast.Name) and target.id == "SCRIPTS"
+            for target in node.targets
+        )
+    )
+    assert ast.literal_eval(assignment.value) == sorted(
+        path.name for path in EXAMPLE_DIR.glob("[0-9][0-9]_*.py")
+    )
+
+
+def test_workflow_roots_accept_bounded_semantic_arguments() -> None:
+    for path in EXAMPLE_DIR.glob("[0-9][0-9]_*.py"):
+        tree = ast.parse(path.read_text())
+        workflows = [
+            node
+            for node in ast.walk(tree)
+            if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef)
+            and any(
+                isinstance(decorator, ast.Call)
+                and getattr(decorator.func, "id", None) == "workflow"
+                for decorator in node.decorator_list
+            )
+        ]
+        assert workflows
+        assert all(function.args.args for function in workflows)
+
+
+def test_env_loading_preserves_shell_values() -> None:
+    source = (EXAMPLE_DIR / "_shared.py").read_text()
+    assert "override=False" in source

python/tracing/replicate/01_run_prediction.py

@@ -1,43 +1,44 @@
 from __future__ import annotations
 
-from respan import workflow
-
 from _shared import (
     example_attributes,
+    finish_respan,
     make_client,
-    make_custom_identifier,
     make_respan,
     model_name,
     print_result,
     text_from_output,
     workflow_name,
 )
+from respan import workflow
 
 EXAMPLE_NAME = "run-prediction"
 
 
 @workflow(name=workflow_name(EXAMPLE_NAME))
-def _run_prediction_workflow(client) -> str:
+def _run_prediction_workflow(prompt: str) -> str:
+    client = make_client()
     output = client.run(
         model_name(),
-        input={"prompt": "Reply with one concise sentence about Replicate tracing."},
+        input={"prompt": prompt},
     )
     return text_from_output(output)
 
 
 def run_prediction() -> None:
     respan = make_respan(EXAMPLE_NAME)
-    client = make_client()
-    custom_identifier = make_custom_identifier(EXAMPLE_NAME)
+    custom_identifier = ""
     text = ""
 
     try:
-        with example_attributes(EXAMPLE_NAME, custom_identifier):
+        with example_attributes(EXAMPLE_NAME) as custom_identifier:
             print(f"custom_identifier={custom_identifier}", flush=True)
             print(f"workflow_name={workflow_name(EXAMPLE_NAME)}", flush=True)
-            text = _run_prediction_workflow(client)
+            text = _run_prediction_workflow(
+                "Reply with one concise sentence about Replicate tracing."
+            )
     finally:
-        respan.shutdown()
+        finish_respan(respan)
 
     print_result(EXAMPLE_NAME, custom_identifier, text)