From f602d6fdb2c94e4496bc8efc04bb64cd7b1d0d11 Mon Sep 17 00:00:00 2001 From: Nightingalelyy Date: Tue, 18 Aug 2026 11:34:53 +0800 Subject: [PATCH] fix(examples): validate Ragas, Replicate, and Restate OTel 2.x spans --- python/tracing/ragas/01_modern_metrics.py | 43 +++++++ python/tracing/ragas/02_evaluate.py | 37 ++++++ python/tracing/ragas/03_experiment.py | 42 +++++++ python/tracing/ragas/README.md | 15 +++ python/tracing/ragas/_shared.py | 78 ++++++++++++ python/tracing/ragas/requirements.txt | 4 + python/tracing/ragas/run_all.py | 45 +++++++ python/tracing/ragas/test_contract.py | 44 +++++++ python/tracing/replicate/01_run_prediction.py | 21 ++-- .../tracing/replicate/02_stream_prediction.py | 21 ++-- .../replicate/03_async_run_prediction.py | 21 ++-- .../replicate/04_prediction_lifecycle.py | 21 ++-- python/tracing/replicate/05_expected_error.py | 38 ++++++ python/tracing/replicate/README.md | 14 +-- python/tracing/replicate/_shared.py | 57 ++++++--- python/tracing/replicate/requirements.txt | 8 +- python/tracing/replicate/run_all.py | 30 ++++- python/tracing/replicate/test_contract.py | 52 ++++++++ python/tracing/restate/01_workflow_success.py | 44 +++++++ python/tracing/restate/02_service_success.py | 43 +++++++ python/tracing/restate/03_expected_error.py | 48 ++++++++ python/tracing/restate/README.md | 16 +++ python/tracing/restate/_shared.py | 111 ++++++++++++++++++ python/tracing/restate/requirements.txt | 4 + python/tracing/restate/run_all.py | 41 +++++++ python/tracing/restate/test_contract.py | 47 ++++++++ 26 files changed, 875 insertions(+), 70 deletions(-) create mode 100644 python/tracing/ragas/01_modern_metrics.py create mode 100644 python/tracing/ragas/02_evaluate.py create mode 100644 python/tracing/ragas/03_experiment.py create mode 100644 python/tracing/ragas/README.md create mode 100644 python/tracing/ragas/_shared.py create mode 100644 python/tracing/ragas/requirements.txt create mode 100644 python/tracing/ragas/run_all.py create mode 100644 python/tracing/ragas/test_contract.py create mode 100644 python/tracing/replicate/05_expected_error.py create mode 100644 python/tracing/replicate/test_contract.py create mode 100644 python/tracing/restate/01_workflow_success.py create mode 100644 python/tracing/restate/02_service_success.py create mode 100644 python/tracing/restate/03_expected_error.py create mode 100644 python/tracing/restate/README.md create mode 100644 python/tracing/restate/_shared.py create mode 100644 python/tracing/restate/requirements.txt create mode 100644 python/tracing/restate/run_all.py create mode 100644 python/tracing/restate/test_contract.py diff --git a/python/tracing/ragas/01_modern_metrics.py b/python/tracing/ragas/01_modern_metrics.py new file mode 100644 index 0000000..39186ac --- /dev/null +++ b/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() diff --git a/python/tracing/ragas/02_evaluate.py b/python/tracing/ragas/02_evaluate.py new file mode 100644 index 0000000..8d8ced6 --- /dev/null +++ b/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() diff --git a/python/tracing/ragas/03_experiment.py b/python/tracing/ragas/03_experiment.py new file mode 100644 index 0000000..676f438 --- /dev/null +++ b/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()) diff --git a/python/tracing/ragas/README.md b/python/tracing/ragas/README.md new file mode 100644 index 0000000..6268cb3 --- /dev/null +++ b/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. diff --git a/python/tracing/ragas/_shared.py b/python/tracing/ragas/_shared.py new file mode 100644 index 0000000..c3b1f7a --- /dev/null +++ b/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 diff --git a/python/tracing/ragas/requirements.txt b/python/tracing/ragas/requirements.txt new file mode 100644 index 0000000..818c08e --- /dev/null +++ b/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 diff --git a/python/tracing/ragas/run_all.py b/python/tracing/ragas/run_all.py new file mode 100644 index 0000000..5b59338 --- /dev/null +++ b/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() diff --git a/python/tracing/ragas/test_contract.py b/python/tracing/ragas/test_contract.py new file mode 100644 index 0000000..cd6717f --- /dev/null +++ b/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 diff --git a/python/tracing/replicate/01_run_prediction.py b/python/tracing/replicate/01_run_prediction.py index 9741802..122d149 100644 --- a/python/tracing/replicate/01_run_prediction.py +++ b/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) diff --git a/python/tracing/replicate/02_stream_prediction.py b/python/tracing/replicate/02_stream_prediction.py index 6877cc0..de8b891 100644 --- a/python/tracing/replicate/02_stream_prediction.py +++ b/python/tracing/replicate/02_stream_prediction.py @@ -1,26 +1,26 @@ 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, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "stream-prediction" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _stream_prediction_workflow(client) -> str: +def _stream_prediction_workflow(prompt: str) -> str: + client = make_client() chunks: list[str] = [] for event in client.stream( model_name(), - input={"prompt": "Stream a short sentence about production traces."}, + input={"prompt": prompt}, ): chunks.append(str(getattr(event, "data", event))) return "".join(chunks) @@ -28,17 +28,18 @@ def _stream_prediction_workflow(client) -> str: def run_stream_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 = _stream_prediction_workflow(client) + text = _stream_prediction_workflow( + "Stream a short sentence about production traces." + ) finally: - respan.shutdown() + finish_respan(respan) print_result(EXAMPLE_NAME, custom_identifier, text) diff --git a/python/tracing/replicate/03_async_run_prediction.py b/python/tracing/replicate/03_async_run_prediction.py index 6e5a1f7..55f86b4 100644 --- a/python/tracing/replicate/03_async_run_prediction.py +++ b/python/tracing/replicate/03_async_run_prediction.py @@ -2,44 +2,45 @@ import asyncio -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 = "async-run-prediction" @workflow(name=workflow_name(EXAMPLE_NAME)) -async def _async_run_prediction_workflow(client) -> str: +async def _async_run_prediction_workflow(prompt: str) -> str: + client = make_client() output = await client.async_run( model_name(), - input={"prompt": "Reply with one concise sentence about async tracing."}, + input={"prompt": prompt}, ) return text_from_output(output) async def run_async_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 = await _async_run_prediction_workflow(client) + text = await _async_run_prediction_workflow( + "Reply with one concise sentence about async tracing." + ) finally: - respan.shutdown() + finish_respan(respan) print_result(EXAMPLE_NAME, custom_identifier, text) diff --git a/python/tracing/replicate/04_prediction_lifecycle.py b/python/tracing/replicate/04_prediction_lifecycle.py index 30d3ced..8d76b89 100644 --- a/python/tracing/replicate/04_prediction_lifecycle.py +++ b/python/tracing/replicate/04_prediction_lifecycle.py @@ -1,26 +1,26 @@ 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 = "prediction-lifecycle" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _prediction_lifecycle_workflow(client) -> str: +def _prediction_lifecycle_workflow(prompt: str) -> str: + client = make_client() prediction = client.predictions.create( model=model_name(), - input={"prompt": "Reply with one concise sentence about background jobs."}, + input={"prompt": prompt}, wait=False, ) prediction.wait() @@ -38,17 +38,18 @@ def _prediction_lifecycle_workflow(client) -> str: def run_prediction_lifecycle() -> 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 = _prediction_lifecycle_workflow(client) + text = _prediction_lifecycle_workflow( + "Reply with one concise sentence about background jobs." + ) finally: - respan.shutdown() + finish_respan(respan) print_result(EXAMPLE_NAME, custom_identifier, text) diff --git a/python/tracing/replicate/05_expected_error.py b/python/tracing/replicate/05_expected_error.py new file mode 100644 index 0000000..b372fb3 --- /dev/null +++ b/python/tracing/replicate/05_expected_error.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from _shared import ( + example_attributes, + finish_respan, + make_client, + make_respan, + model_name, + print_result, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "expected-error" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def expected_error_workflow(prompt: str) -> None: + make_client().run(model_name(), input={"prompt": prompt}) + + +def main() -> None: + respan = make_respan(EXAMPLE_NAME) + custom_identifier = "" + outcome = "" + try: + try: + with example_attributes(EXAMPLE_NAME) as custom_identifier: + expected_error_workflow("Expected provider error") + except Exception as exc: # noqa: BLE001 - expected real SDK error. + outcome = f"expected {type(exc).__name__}" + finally: + finish_respan(respan) + print_result(EXAMPLE_NAME, custom_identifier, outcome) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/replicate/README.md b/python/tracing/replicate/README.md index fc043d4..f62dcc9 100644 --- a/python/tracing/replicate/README.md +++ b/python/tracing/replicate/README.md @@ -9,23 +9,20 @@ Required for exporting traces: RESPAN_API_KEY=... ``` -For real Replicate calls, set: +The committed validation path always uses the real Replicate SDK against a +deterministic `httpx.MockTransport`. To opt into billable provider calls, set: ```bash REPLICATE_API_TOKEN=... +RESPAN_REPLICATE_LIVE=1 ``` -If `REPLICATE_API_TOKEN` is not set, the examples use the real Replicate SDK -against a deterministic in-process mock transport. That keeps the examples -runnable with the repo-root `.env` while still exercising the instrumentation -and exporting spans to Respan. - Optional environment variables: ```bash RESPAN_BASE_URL=https://api.respan.ai/api RESPAN_REPLICATE_MODEL=meta/meta-llama-3-8b-instruct -RESPAN_REPLICATE_MOCK=1 +RESPAN_EXAMPLE_RUN_ID=otel2-replicate-check ``` Run one script at a time: @@ -35,12 +32,13 @@ python 01_run_prediction.py python 02_stream_prediction.py python 03_async_run_prediction.py python 04_prediction_lifecycle.py +python 05_expected_error.py ``` Or run the full set: ```bash -python run_all.py +RESPAN_EXAMPLE_RUN_ID=otel2-replicate-check python run_all.py ``` Each script prints a `custom_identifier` and `workflow_name`. The workflow name diff --git a/python/tracing/replicate/_shared.py b/python/tracing/replicate/_shared.py index f1cbeec..debfda3 100644 --- a/python/tracing/replicate/_shared.py +++ b/python/tracing/replicate/_shared.py @@ -4,7 +4,6 @@ import os from contextlib import contextmanager from pathlib import Path -from uuid import uuid4 import httpx import replicate @@ -19,7 +18,7 @@ def load_root_env() -> None: - load_dotenv(PROJECT_ROOT / ".env", override=True) + load_dotenv(PROJECT_ROOT / ".env", override=False) def require_respan_api_key() -> str: @@ -39,10 +38,18 @@ def model_name() -> str: def use_mock_replicate() -> bool: - explicit = os.getenv("RESPAN_REPLICATE_MOCK") - if explicit is not None: - return explicit.lower() not in {"0", "false", "no"} - return not bool(os.getenv("REPLICATE_API_TOKEN")) + return os.getenv("RESPAN_REPLICATE_LIVE", "0").lower() not in { + "1", + "true", + "yes", + } + + +def run_id() -> str: + marker = os.getenv("RESPAN_EXAMPLE_RUN_ID") + if not marker: + raise RuntimeError("RESPAN_EXAMPLE_RUN_ID must be supplied by run_all.py") + return marker def make_respan(example_name: str) -> Respan: @@ -53,7 +60,12 @@ def make_respan(example_name: str) -> Respan: app_name="replicate-examples", instrumentations=[ReplicateInstrumentor()], environment=os.getenv("RESPAN_ENVIRONMENT", "example"), - metadata={"integration": "replicate", "example": example_name}, + metadata={ + "example_set": "replicate", + "example": example_name, + "example_run_id": run_id(), + "run_id": run_id(), + }, ) @@ -88,11 +100,18 @@ def _mock_response(request: httpx.Request) -> httpx.Response: path = request.url.path method = request.method.upper() if method == "POST" and ( - path == "/v1/predictions" or path.startswith("/v1/models/") and path.endswith("/predictions") + path == "/v1/predictions" + or path.startswith("/v1/models/") + and path.endswith("/predictions") ): body = json.loads(request.content.decode() or "{}") prompt = (body.get("input") or {}).get("prompt", "") - prediction_id = f"mock-{uuid4().hex[:8]}" + prediction_id = "mock-prediction" + if "expected provider error" in prompt.lower(): + return httpx.Response( + 429, + json={"detail": "deterministic Replicate rate limit"}, + ) return httpx.Response( 201, json=_mock_prediction( @@ -149,20 +168,19 @@ def workflow_name(example_name: str) -> str: return f"replicate_{normalized_name}" -def make_custom_identifier(example_name: str) -> str: - return f"replicate-{example_name}-{uuid4().hex[:8]}" - - @contextmanager -def example_attributes(example_name: str, custom_identifier: str | None = None): - custom_identifier = custom_identifier or make_custom_identifier(example_name) +def example_attributes(example_name: str): + marker = run_id() + custom_identifier = f"replicate-{example_name}-{marker}" current_workflow_name = workflow_name(example_name) with propagate_attributes( custom_identifier=custom_identifier, trace_group_identifier=current_workflow_name, metadata={ "example": example_name, - "run_id": custom_identifier, + "example_set": "replicate", + "example_run_id": marker, + "run_id": marker, "workflow_name": current_workflow_name, "replicate_client_mode": client_mode(), }, @@ -191,3 +209,10 @@ def print_result(example_name: str, custom_identifier: str, text: str) -> None: print(f"workflow_name={workflow_name(example_name)}") print(f"client_mode={client_mode()}") print(text.strip()) + + +def finish_respan(respan: Respan) -> None: + try: + respan.flush() + finally: + respan.shutdown() diff --git a/python/tracing/replicate/requirements.txt b/python/tracing/replicate/requirements.txt index 88f8b07..bcd7add 100644 --- a/python/tracing/replicate/requirements.txt +++ b/python/tracing/replicate/requirements.txt @@ -1,4 +1,4 @@ -respan-ai -respan-instrumentation-replicate -replicate>=1.0.0 -python-dotenv +replicate>=1.0.0,<2.0.0 +respan-ai>=4,<5 +respan-instrumentation-replicate>=0.1.0,<0.2.0 +python-dotenv>=1,<2 diff --git a/python/tracing/replicate/run_all.py b/python/tracing/replicate/run_all.py index 37b364d..2c2c302 100644 --- a/python/tracing/replicate/run_all.py +++ b/python/tracing/replicate/run_all.py @@ -1,6 +1,9 @@ from __future__ import annotations -import runpy +import os +import subprocess +import sys +from datetime import UTC, datetime from pathlib import Path EXAMPLES = [ @@ -8,14 +11,37 @@ "02_stream_prediction.py", "03_async_run_prediction.py", "04_prediction_lifecycle.py", + "05_expected_error.py", ] +TIMEOUT_SECONDS = 120 def main() -> None: base_dir = Path(__file__).resolve().parent + env = dict(os.environ) + env.setdefault( + "RESPAN_EXAMPLE_RUN_ID", + f"otel2-replicate-{datetime.now(UTC).strftime('%Y%m%dT%H%M%SZ')}", + ) + failures: list[str] = [] + print(f"RESPAN_EXAMPLE_RUN_ID={env['RESPAN_EXAMPLE_RUN_ID']}", flush=True) for script in EXAMPLES: print(f"\n=== {script} ===", flush=True) - runpy.run_path(str(base_dir / script), run_name="__main__") + try: + result = subprocess.run( + [sys.executable, str(base_dir / script)], + cwd=base_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__": diff --git a/python/tracing/replicate/test_contract.py b/python/tracing/replicate/test_contract.py new file mode 100644 index 0000000..c7b25bb --- /dev/null +++ b/python/tracing/replicate/test_contract.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import ast +from pathlib import Path + +EXAMPLE_DIR = Path(__file__).resolve().parent + + +def test_runner_covers_all_examples_and_aggregates_failures() -> None: + source = (EXAMPLE_DIR / "run_all.py").read_text() + tree = ast.parse(source) + assignment = next( + node + for node in tree.body + if isinstance(node, ast.Assign) + and any( + isinstance(target, ast.Name) and target.id == "EXAMPLES" + 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") + ) + assert "TimeoutExpired" in source + assert "failures" in source + + +def test_workflows_accept_semantic_values_not_clients() -> 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 + for function in workflows: + names = [argument.arg for argument in function.args.args] + assert names + assert "client" not in names + + +def test_marker_and_live_mode_are_explicit() -> None: + shared = (EXAMPLE_DIR / "_shared.py").read_text() + assert "override=False" in shared + assert "example_run_id" in shared + assert "RESPAN_REPLICATE_LIVE" in shared diff --git a/python/tracing/restate/01_workflow_success.py b/python/tracing/restate/01_workflow_success.py new file mode 100644 index 0000000..49ff884 --- /dev/null +++ b/python/tracing/restate/01_workflow_success.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +import asyncio + +import restate +from _shared import ( + create_respan, + example_context, + finish_respan, + invoke_registered_handler, +) +from respan import workflow + +CASE = "workflow_success" + + +@workflow(name="restate_workflow_success") +async def workflow_success(order_id: str) -> dict[str, str]: + checkout = restate.Workflow("CheckoutWorkflow") + + @checkout.main(name="run") + async def run(_ctx, request: dict[str, str]) -> dict[str, str]: + return {"order_id": request["order_id"], "status": "accepted"} + + return await invoke_registered_handler( + checkout, + "run", + {"order_id": order_id}, + invocation_id="checkout-workflow-1", + key=order_id, + ) + + +async def main() -> None: + respan = create_respan() + try: + with example_context(CASE): + print(await workflow_success("order-42"), flush=True) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/tracing/restate/02_service_success.py b/python/tracing/restate/02_service_success.py new file mode 100644 index 0000000..1d2fd99 --- /dev/null +++ b/python/tracing/restate/02_service_success.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import asyncio + +import restate +from _shared import ( + create_respan, + example_context, + finish_respan, + invoke_registered_handler, +) +from respan import workflow + +CASE = "service_success" + + +@workflow(name="restate_service_success") +async def service_success(customer: str) -> dict[str, str]: + greeter = restate.Service("GreeterService") + + @greeter.handler(name="greet") + async def greet(_ctx, request: dict[str, str]) -> dict[str, str]: + return {"message": f"Hello, {request['customer']}"} + + return await invoke_registered_handler( + greeter, + "greet", + {"customer": customer}, + invocation_id="greeter-service-1", + ) + + +async def main() -> None: + respan = create_respan() + try: + with example_context(CASE): + print(await service_success("Ada"), flush=True) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/tracing/restate/03_expected_error.py b/python/tracing/restate/03_expected_error.py new file mode 100644 index 0000000..3a98a3e --- /dev/null +++ b/python/tracing/restate/03_expected_error.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import asyncio + +import restate +from _shared import ( + create_respan, + example_context, + finish_respan, + invoke_registered_handler, +) +from respan import workflow + +CASE = "expected_error" + + +@workflow(name="restate_expected_error") +async def expected_error(reason: str) -> None: + failing = restate.Service("FailureService") + + @failing.handler(name="fail") + async def fail(_ctx, request: dict[str, str]) -> None: + raise RuntimeError(request["reason"]) + + await invoke_registered_handler( + failing, + "fail", + {"reason": reason}, + invocation_id="failure-service-1", + ) + + +async def main() -> None: + respan = create_respan() + result = "" + try: + try: + with example_context(CASE): + await expected_error("deterministic Restate handler failure") + except RuntimeError as exc: + result = f"expected {type(exc).__name__}" + finally: + finish_respan(respan) + print(result, flush=True) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/tracing/restate/README.md b/python/tracing/restate/README.md new file mode 100644 index 0000000..45ab444 --- /dev/null +++ b/python/tracing/restate/README.md @@ -0,0 +1,16 @@ +# Restate tracing examples + +These examples register real Restate 1.x workflow and service handlers and +exercise their configured invocation context managers with a deterministic +in-process invocation fixture. This validates instrumentation mapping without +requiring a deployed Restate runtime or protocol VM. + +A true replay/deployment validation still requires an external Restate server; +that service boundary is intentionally not replaced by package behavior. + +```bash +RESPAN_EXAMPLE_RUN_ID=otel2-restate-check python run_all.py +``` + +The runner preserves the supplied marker, times out each child independently, +continues after failures, and exits nonzero if any example fails. diff --git a/python/tracing/restate/_shared.py b/python/tracing/restate/_shared.py new file mode 100644 index 0000000..906f2f3 --- /dev/null +++ b/python/tracing/restate/_shared.py @@ -0,0 +1,111 @@ +"""Shared deterministic Restate handler and Respan lifecycle helpers.""" + +from __future__ import annotations + +import os +from collections.abc import AsyncIterator, Iterator +from contextlib import AsyncExitStack, contextmanager +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +from dotenv import load_dotenv +from respan import Respan, propagate_attributes +from respan_instrumentation_restate import RestateInstrumentor +from restate import server_context + +EXAMPLE_DIR = Path(__file__).resolve().parent +REPO_ROOT = EXAMPLE_DIR.parents[2] +EXAMPLE_SET = "restate" +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: + marker = os.getenv("RESPAN_EXAMPLE_RUN_ID") + if not marker: + raise RuntimeError("RESPAN_EXAMPLE_RUN_ID must be supplied by run_all.py") + return marker + + +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="restate-examples", + metadata={ + "example_set": EXAMPLE_SET, + "example_run_id": marker, + "run_id": marker, + }, + instrumentations=[RestateInstrumentor()], + 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"restate_{case}", + metadata={ + "example_set": EXAMPLE_SET, + "example_case": case, + "example_run_id": marker, + "run_id": marker, + }, + ): + yield + + +async def invoke_registered_handler( + component: Any, + handler_name: str, + payload: Any, + *, + invocation_id: str, + key: str | None = None, +) -> Any: + """Exercise a real registered handler without requiring a Restate deployment.""" + handler = component.handlers[handler_name] + encoded = handler.handler_io.input_serde.serialize(payload) + context = SimpleNamespace( + handler=handler, + invocation=SimpleNamespace( + invocation_id=invocation_id, + input_buffer=encoded, + key=key, + scope=None, + limit_key=None, + idempotency_key=None, + ), + ) + original_current_context = server_context.current_context + original_replaying = server_context.restate_context_is_replaying + server_context.current_context = lambda: context + server_context.restate_context_is_replaying = SimpleNamespace(get=lambda: False) + try: + async with AsyncExitStack() as stack: + managers: list[AsyncIterator[None]] = list(handler.context_managers or ()) + for manager in managers: + await stack.enter_async_context(manager()) + return await handler.fn(None, payload) + finally: + server_context.current_context = original_current_context + server_context.restate_context_is_replaying = original_replaying + + +def finish_respan(respan: Respan) -> None: + try: + respan.flush() + finally: + respan.shutdown() diff --git a/python/tracing/restate/requirements.txt b/python/tracing/restate/requirements.txt new file mode 100644 index 0000000..b690759 --- /dev/null +++ b/python/tracing/restate/requirements.txt @@ -0,0 +1,4 @@ +restate-sdk>=1.0.0,<2.0.0 +respan-ai>=4,<5 +respan-instrumentation-restate>=0.1.0,<0.2.0 +python-dotenv>=1,<2 diff --git a/python/tracing/restate/run_all.py b/python/tracing/restate/run_all.py new file mode 100644 index 0000000..90550bc --- /dev/null +++ b/python/tracing/restate/run_all.py @@ -0,0 +1,41 @@ +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_workflow_success.py", "02_service_success.py", "03_expected_error.py"] +TIMEOUT_SECONDS = 120 + + +def main() -> None: + env = dict(os.environ) + env.setdefault( + "RESPAN_EXAMPLE_RUN_ID", + f"otel2-restate-{datetime.now(UTC).strftime('%Y%m%dT%H%M%SZ')}", + ) + 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() diff --git a/python/tracing/restate/test_contract.py b/python/tracing/restate/test_contract.py new file mode 100644 index 0000000..8938daf --- /dev/null +++ b/python/tracing/restate/test_contract.py @@ -0,0 +1,47 @@ +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_semantic_values() -> None: + for path in EXAMPLE_DIR.glob("[0-9][0-9]_*.py"): + tree = ast.parse(path.read_text()) + roots = [ + 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 roots + assert all(root.args.args for root in roots) + + +def test_marker_and_teardown_contracts() -> None: + shared = (EXAMPLE_DIR / "_shared.py").read_text() + assert "override=False" in shared + assert "example_run_id" in shared + assert "respan.flush()" in shared + assert "respan.shutdown()" in shared