diff --git a/python/tracing/pinecone/01_upsert_and_query.py b/python/tracing/pinecone/01_upsert_and_query.py index 995677e..493159d 100644 --- a/python/tracing/pinecone/01_upsert_and_query.py +++ b/python/tracing/pinecone/01_upsert_and_query.py @@ -4,17 +4,19 @@ import time from typing import Any -from respan import Respan, workflow - from _shared import ( create_pinecone_index, create_respan, execution_id, finish_respan, + live_configured, + marker, print_result, response_field, + to_jsonable, workflow_attributes, ) +from respan import Respan, workflow WORKFLOW_NAME = "pinecone_upsert_and_query_workflow" @@ -33,29 +35,32 @@ def wait_until_fetchable(index: Any, namespace: str, vector_id: str) -> Any: if vector_id in (response_field(fetched, "vectors", {}) or {}): return fetched time.sleep(1) - raise TimeoutError(f"Pinecone did not expose {vector_id!r} within {timeout:g}s") + raise TimeoutError( + f"Pinecone did not expose the example vector within {timeout:g}s" + ) @workflow(name=WORKFLOW_NAME) -def run_upsert_and_query(run_id: str) -> dict: +def run_upsert_and_query(topic: str, top_k: int) -> dict[str, Any]: index = create_pinecone_index() - namespace = f"respan-example-{run_id}" + execution = execution_id() + namespace = f"respan-example-{execution}" stats = index.describe_index_stats() dimension = int(response_field(stats, "dimension", 0) or 0) if dimension < 1: raise RuntimeError("PINECONE_INDEX_NAME must reference a dense-vector index") - vector_ids = [f"{run_id}-python", f"{run_id}-rust", f"{run_id}-pasta"] + vector_ids = [f"{execution}-python", f"{execution}-rust", f"{execution}-pasta"] vectors = [ { "id": vector_ids[0], "values": basis_vector(dimension, 0), - "metadata": {"topic": "programming", "text": "Python is approachable."}, + "metadata": {"topic": topic, "text": "Python is approachable."}, }, { "id": vector_ids[1], "values": basis_vector(dimension, 1), - "metadata": {"topic": "programming", "text": "Rust emphasizes safety."}, + "metadata": {"topic": topic, "text": "Rust emphasizes safety."}, }, { "id": vector_ids[2], @@ -70,29 +75,35 @@ def run_upsert_and_query(run_id: str) -> dict: queried = index.query( namespace=namespace, vector=basis_vector(dimension, 0), - top_k=2, + top_k=top_k, include_metadata=True, + include_values=True, + ) + return to_jsonable( + { + "index": os.getenv("PINECONE_INDEX_NAME", "loopback-index"), + "mode": "live" if live_configured() else "deterministic", + "namespace": namespace, + "dimension": dimension, + "upsert": upserted, + "fetch": fetched, + "query": queried, + } ) - return { - "index": os.environ["PINECONE_INDEX_NAME"], - "namespace": namespace, - "dimension": dimension, - "upsert": upserted, - "fetch": fetched, - "query": queried, - } finally: - # Only remove IDs created by this run; the existing index is never modified. index.delete(ids=vector_ids, namespace=namespace) def main() -> None: - run_id = execution_id() - respan = create_respan(WORKFLOW_NAME) + run_marker = marker() + execution = execution_id() + respan = create_respan(WORKFLOW_NAME, run_marker) try: - with Respan.propagate_attributes(**workflow_attributes(WORKFLOW_NAME, run_id)): - result = run_upsert_and_query(run_id) - print_result(WORKFLOW_NAME, result) + with Respan.propagate_attributes( + **workflow_attributes(WORKFLOW_NAME, run_marker, execution) + ): + result = run_upsert_and_query("programming", 2) + print_result(WORKFLOW_NAME, result, run_marker) finally: finish_respan(respan) diff --git a/python/tracing/pinecone/02_async_fetch.py b/python/tracing/pinecone/02_async_fetch.py new file mode 100644 index 0000000..c5c8658 --- /dev/null +++ b/python/tracing/pinecone/02_async_fetch.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import asyncio +from typing import Any + +from _shared import ( + create_async_pinecone_index, + create_respan, + execution_id, + finish_respan, + marker, + print_result, + to_jsonable, + workflow_attributes, +) +from respan import Respan, workflow + +WORKFLOW_NAME = "pinecone_async_fetch_workflow" + + +@workflow(name=WORKFLOW_NAME) +async def run_async_fetch(vector_ids: list[str], namespace: str) -> dict[str, Any]: + index = create_async_pinecone_index() + try: + result = await index.fetch(ids=vector_ids, namespace=namespace) + return to_jsonable({"ids": vector_ids, "namespace": namespace, "fetch": result}) + finally: + await index.close() + + +async def main() -> None: + run_marker = marker() + execution = execution_id() + respan = create_respan(WORKFLOW_NAME, run_marker) + try: + with Respan.propagate_attributes( + **workflow_attributes(WORKFLOW_NAME, run_marker, execution) + ): + result = await run_async_fetch(["trace-doc"], "respan-example") + print_result(WORKFLOW_NAME, result, run_marker) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/tracing/pinecone/03_expected_error.py b/python/tracing/pinecone/03_expected_error.py new file mode 100644 index 0000000..a68de30 --- /dev/null +++ b/python/tracing/pinecone/03_expected_error.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from _shared import ( + create_pinecone_index, + create_respan, + execution_id, + finish_respan, + marker, + print_result, + workflow_attributes, +) +from pinecone.exceptions import PineconeApiException +from respan import Respan, workflow + +WORKFLOW_NAME = "pinecone_expected_error_workflow" + + +@workflow(name=WORKFLOW_NAME) +def run_expected_error(namespace: str) -> None: + create_pinecone_index().delete(ids=["missing-vector"], namespace=namespace) + + +def main() -> None: + run_marker = marker() + execution = execution_id() + respan = create_respan(WORKFLOW_NAME, run_marker) + try: + try: + with Respan.propagate_attributes( + **workflow_attributes(WORKFLOW_NAME, run_marker, execution) + ): + run_expected_error("error") + except PineconeApiException as exc: + result = { + "expected_error": type(exc).__name__, + "message": "deterministic Pinecone outage", + } + else: + raise AssertionError("the deterministic Pinecone failure did not occur") + print_result(WORKFLOW_NAME, result, run_marker) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/pinecone/README.md b/python/tracing/pinecone/README.md index 7fd53d9..7c2e876 100644 --- a/python/tracing/pinecone/README.md +++ b/python/tracing/pinecone/README.md @@ -1,30 +1,48 @@ -# Pinecone Tracing Example +# Pinecone tracing examples -This example traces a concise `describe_index_stats` -> `upsert` -> `fetch` -> -`query` flow against an existing dense-vector Pinecone index. It never creates, -configures, or deletes an index, and it cleans up only the unique IDs written by -the current run. +These examples use the real Pinecone Python SDK and local editable Respan +instrumentation. Without Pinecone credentials they run against a bounded local +protocol fixture, so sync, async, success, and service-error paths remain +repeatable. When both `PINECONE_API_KEY` and `PINECONE_INDEX_NAME` are set, the +round-trip example uses that existing dense-vector index and deletes only its +own unique IDs. -Add these values to the repo-root `.env`: +## Setup + +```bash +cd python/tracing/pinecone +pip install -r requirements.txt +``` + +For local instrumentation development, install the package from the sibling +checkout before running the examples: + +```bash +pip install -e ../../../../respan/python-sdks/instrumentations/respan-instrumentation-pinecone +``` + +Required in the repository-root `.env`: ```dotenv RESPAN_API_KEY=... +``` + +Optional live Pinecone settings: + +```dotenv PINECONE_API_KEY=... PINECONE_INDEX_NAME=your-existing-index -# Recommended when known; avoids resolving the host by index name. PINECONE_INDEX_HOST=your-index-host ``` -The emitted workflow name is `pinecone_upsert_and_query_workflow`. +`PINECONE_INDEX_HOST` is required for the async live example. The expected-error +example always uses the deterministic fixture and never mutates a live index. ## Run ```bash -cd python/tracing/pinecone -pip install -r requirements.txt -python run_all.py +RESPAN_EXAMPLE_RUN_ID=my-exact-marker python run_all.py ``` -Set `PINECONE_INGEST_TIMEOUT_SECONDS` to change the default 30-second fetch -polling window. When running before the instrumentor is published, include its -local `src` directory and the local Respan packages on `PYTHONPATH`. +The runner preserves the exact marker for all three processes, applies a +per-process timeout, continues after failures, and reports them together. diff --git a/python/tracing/pinecone/_loopback.py b/python/tracing/pinecone/_loopback.py new file mode 100644 index 0000000..ba54448 --- /dev/null +++ b/python/tracing/pinecone/_loopback.py @@ -0,0 +1,121 @@ +"""Deterministic protocol fixture used by the real Pinecone SDK examples.""" + +from __future__ import annotations + +import json +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from threading import RLock, Thread +from typing import Any +from urllib.parse import parse_qs, urlsplit + +_lock = RLock() +_server: ThreadingHTTPServer | None = None +_thread: Thread | None = None + + +class _Handler(BaseHTTPRequestHandler): + server_version = "PineconeExampleFixture/1.0" + + def log_message(self, *_args: Any) -> None: + return + + def _reply(self, status: int, value: object) -> None: + payload = json.dumps(value, allow_nan=False).encode("utf-8") + self.send_response(status) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def _body(self) -> dict[str, Any]: + length = int(self.headers.get("content-length", "0")) + raw = self.rfile.read(length) if length else b"{}" + try: + value = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + return {} + return value if isinstance(value, dict) else {} + + def do_GET(self) -> None: + if self.path.startswith("/vectors/fetch"): + query = parse_qs(urlsplit(self.path).query) + vector_id = (query.get("ids") or ["trace-doc"])[0] + namespace = (query.get("namespace") or ["respan-example"])[0] + self._reply( + 200, + { + "namespace": namespace, + "vectors": { + vector_id: { + "id": vector_id, + "values": [1.0, 0.0, 0.0, 0.0], + "metadata": {"topic": "tracing"}, + } + }, + }, + ) + return + self._reply(404, {"message": "not found"}) + + def do_POST(self) -> None: + body = self._body() + if self.path == "/describe_index_stats": + self._reply( + 200, + { + "dimension": 4, + "indexFullness": 0.0, + "namespaces": {"respan-example": {"vectorCount": 3}}, + "totalVectorCount": 3, + }, + ) + elif self.path == "/vectors/upsert": + self._reply(200, {"upsertedCount": len(body.get("vectors", []))}) + elif self.path == "/query": + self._reply( + 200, + { + "namespace": body.get("namespace", "respan-example"), + "matches": [ + { + "id": "trace-doc", + "score": 0.99, + "values": [1.0, 0.0, 0.0, 0.0], + "metadata": { + "topic": "tracing", + "text": "Pinecone instrumentation is active.", + }, + } + ], + }, + ) + elif self.path == "/vectors/delete" and body.get("namespace") == "error": + self._reply(503, {"message": "deterministic Pinecone outage"}) + elif self.path == "/vectors/delete": + self._reply(200, {}) + else: + self._reply(404, {"message": "not found"}) + + +def loopback_host() -> str: + global _server, _thread + with _lock: + if _server is None: + _server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + _thread = Thread(target=_server.serve_forever, daemon=True) + _thread.start() + host, port = _server.server_address + return f"http://{host}:{port}" + + +def shutdown_loopback() -> None: + global _server, _thread + with _lock: + server, thread = _server, _thread + _server = None + _thread = None + if server is not None: + server.shutdown() + server.server_close() + if thread is not None: + thread.join(timeout=2) diff --git a/python/tracing/pinecone/_shared.py b/python/tracing/pinecone/_shared.py index 01ba7da..b6ea198 100644 --- a/python/tracing/pinecone/_shared.py +++ b/python/tracing/pinecone/_shared.py @@ -8,8 +8,10 @@ from typing import Any from uuid import uuid4 +from _loopback import loopback_host, shutdown_loopback from dotenv import load_dotenv -from pinecone import Pinecone +from pinecone import Index +from pinecone.async_client.async_index import AsyncIndex from respan import Respan from respan_instrumentation_pinecone import PineconeInstrumentor @@ -20,48 +22,83 @@ def load_example_env() -> None: - env_path = REPO_ROOT / ".env" - load_dotenv(env_path, override=True) - required = ["RESPAN_API_KEY", "PINECONE_API_KEY", "PINECONE_INDEX_NAME"] - missing = [name for name in required if not os.getenv(name)] - if missing: - raise RuntimeError(f"Missing {', '.join(missing)} in {env_path}") + load_dotenv(REPO_ROOT / ".env", override=False) + if not os.getenv("RESPAN_API_KEY"): + raise RuntimeError(f"RESPAN_API_KEY is required in {REPO_ROOT / '.env'}") os.environ.setdefault("RESPAN_BASE_URL", RESPAN_BASE_URL) -def create_respan(workflow_name: str) -> Respan: +def marker() -> str: + return os.getenv("RESPAN_EXAMPLE_RUN_ID", "").strip() or ( + f"pinecone-{uuid4().hex[:10]}" + ) + + +def execution_id() -> str: + return uuid4().hex[:10] + + +def live_configured() -> bool: + return bool(os.getenv("PINECONE_API_KEY") and os.getenv("PINECONE_INDEX_NAME")) + + +def create_respan(workflow_name: str, run_marker: str) -> Respan: load_example_env() return Respan( api_key=os.environ["RESPAN_API_KEY"], base_url=os.getenv("RESPAN_BASE_URL", RESPAN_BASE_URL), app_name=workflow_name, - metadata={"example_set": EXAMPLE_SET, "workflow_name": workflow_name}, + metadata={ + "example_set": EXAMPLE_SET, + "workflow_name": workflow_name, + "example_run_id": run_marker, + "run_id": run_marker, + }, instrumentations=[PineconeInstrumentor()], is_batching_enabled=False, log_level=os.getenv("RESPAN_LOG_LEVEL", "WARNING"), ) -def create_pinecone_index(): - client = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) - host = os.getenv("PINECONE_INDEX_HOST") - return client.Index(host=host) if host else client.Index(os.environ["PINECONE_INDEX_NAME"]) - - -def execution_id() -> str: - prefix = os.getenv("RESPAN_EXAMPLE_RUN_ID", "run") - safe_prefix = "".join(char if char.isalnum() else "-" for char in prefix) - return f"{safe_prefix[:24]}-{uuid4().hex[:8]}" +def create_pinecone_index() -> Index: + if live_configured(): + host = os.getenv("PINECONE_INDEX_HOST") + if host: + return Index(host=host, api_key=os.environ["PINECONE_API_KEY"]) + from pinecone import Pinecone + + return Pinecone(api_key=os.environ["PINECONE_API_KEY"]).Index( + os.environ["PINECONE_INDEX_NAME"] + ) + return Index(host=loopback_host(), api_key="local-pinecone-key", ssl_verify=False) + + +def create_async_pinecone_index() -> AsyncIndex: + if live_configured(): + host = os.getenv("PINECONE_INDEX_HOST") + if not host: + raise RuntimeError( + "PINECONE_INDEX_HOST is required for the async live example" + ) + return AsyncIndex(host=host, api_key=os.environ["PINECONE_API_KEY"]) + return AsyncIndex( + host=loopback_host(), api_key="local-pinecone-key", ssl_verify=False + ) -def workflow_attributes(workflow_name: str, run_id: str) -> dict[str, object]: +def workflow_attributes( + workflow_name: str, run_marker: str, execution: str +) -> dict[str, object]: return { "trace_group_identifier": workflow_name, - "custom_identifier": f"{workflow_name}-{run_id}", + "custom_identifier": f"{workflow_name}-{execution}", "metadata": { "example_set": EXAMPLE_SET, "workflow_name": workflow_name, - "example_run_id": run_id, + "example_run_id": run_marker, + "run_id": run_marker, + "execution_id": execution, + "mode": "live" if live_configured() else "deterministic", }, } @@ -72,13 +109,32 @@ def response_field(response: Any, name: str, default: Any = None) -> Any: return getattr(response, name, default) -def print_result(label: str, value: Any) -> None: +def to_jsonable(value: Any) -> Any: + if value is None or isinstance(value, (str, bool, int, float)): + return value + if isinstance(value, dict): + return {str(key): to_jsonable(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [to_jsonable(item) for item in value] + if type(value).__module__.startswith("pinecone"): + for method_name in ("to_dict", "model_dump"): + method = getattr(value, method_name, None) + if callable(method): + return to_jsonable(method()) + return {"type": type(value).__name__} + + +def print_result(label: str, value: Any, run_marker: str) -> None: + print(f"RESPAN_EXAMPLE_RUN_ID={run_marker}") print(f"\n== {label} ==") - print(json.dumps(value, default=str, indent=2, sort_keys=True)) + print(json.dumps(to_jsonable(value), allow_nan=False, indent=2, sort_keys=True)) def finish_respan(respan: Respan) -> None: try: respan.flush() finally: - respan.shutdown() + try: + respan.shutdown() + finally: + shutdown_loopback() diff --git a/python/tracing/pinecone/requirements.txt b/python/tracing/pinecone/requirements.txt index 006a71d..a9b84b0 100644 --- a/python/tracing/pinecone/requirements.txt +++ b/python/tracing/pinecone/requirements.txt @@ -1,4 +1,4 @@ -pinecone>=5.1.0,<10.0.0 +pinecone>=9.1.0,<10.0.0 python-dotenv>=1.0.0 -respan-ai>=2.17.0 +respan-ai>=4.1.0 respan-instrumentation-pinecone>=0.1.0 diff --git a/python/tracing/pinecone/run_all.py b/python/tracing/pinecone/run_all.py index d345b7c..4adca0c 100644 --- a/python/tracing/pinecone/run_all.py +++ b/python/tracing/pinecone/run_all.py @@ -1,17 +1,47 @@ from __future__ import annotations +import os import subprocess import sys +from datetime import datetime, timezone from pathlib import Path EXAMPLE_DIR = Path(__file__).resolve().parent -SCRIPTS = ["01_upsert_and_query.py"] +SCRIPTS = ( + "01_upsert_and_query.py", + "02_async_fetch.py", + "03_expected_error.py", +) +DEFAULT_TIMEOUT_SECONDS = 90.0 def main() -> None: + marker = os.getenv("RESPAN_EXAMPLE_RUN_ID", "").strip() or ( + "pinecone-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + ) + timeout = float( + os.getenv("RESPAN_EXAMPLE_TIMEOUT_SECONDS", DEFAULT_TIMEOUT_SECONDS) + ) + env = os.environ.copy() + env["RESPAN_EXAMPLE_RUN_ID"] = marker + failures: list[str] = [] for script in SCRIPTS: - print(f"\n### Running {script}") - subprocess.run([sys.executable, str(EXAMPLE_DIR / script)], check=True) + print(f"\n### Running {script}", flush=True) + try: + completed = subprocess.run( + [sys.executable, str(EXAMPLE_DIR / script)], + check=False, + env=env, + timeout=timeout, + ) + except subprocess.TimeoutExpired: + failures.append(f"{script}: timed out after {timeout:g}s") + continue + if completed.returncode: + failures.append(f"{script}: exited {completed.returncode}") + if failures: + raise SystemExit("Pinecone example failures:\n- " + "\n- ".join(failures)) + print(f"\nCompleted Pinecone examples: RESPAN_EXAMPLE_RUN_ID={marker}") if __name__ == "__main__": diff --git a/python/tracing/pinecone/test_contract.py b/python/tracing/pinecone/test_contract.py new file mode 100644 index 0000000..f5e730e --- /dev/null +++ b/python/tracing/pinecone/test_contract.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +import ast +import json +import os +import subprocess +from pathlib import Path +from types import SimpleNamespace + +import _shared +import pytest +import run_all + +EXAMPLE_DIR = Path(__file__).resolve().parent + + +def _workflow_signature(filename: str) -> list[str]: + module = ast.parse((EXAMPLE_DIR / filename).read_text()) + workflows = [ + node + for node in ast.walk(module) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and any( + isinstance(decorator, ast.Call) + and isinstance(decorator.func, ast.Name) + and decorator.func.id == "workflow" + for decorator in node.decorator_list + ) + ] + assert len(workflows) == 1 + return [argument.arg for argument in workflows[0].args.args] + + +def test_workflow_roots_accept_only_semantic_inputs(): + assert _workflow_signature("01_upsert_and_query.py") == ["topic", "top_k"] + assert _workflow_signature("02_async_fetch.py") == ["vector_ids", "namespace"] + assert _workflow_signature("03_expected_error.py") == ["namespace"] + + +def test_shell_marker_survives_dotenv(monkeypatch): + monkeypatch.setenv("RESPAN_EXAMPLE_RUN_ID", "exact-shell-marker") + + def fake_load_dotenv(_path, *, override): + assert override is False + os.environ.setdefault("RESPAN_EXAMPLE_RUN_ID", "dotenv-marker") + os.environ.setdefault("RESPAN_API_KEY", "test-key") + + monkeypatch.setattr(_shared, "load_dotenv", fake_load_dotenv) + _shared.load_example_env() + assert _shared.marker() == "exact-shell-marker" + + +def test_respan_and_workflow_metadata_use_exact_marker(monkeypatch): + captured: dict[str, object] = {} + + class FakeRespan: + def __init__(self, **kwargs): + captured.update(kwargs) + + monkeypatch.setattr(_shared, "load_example_env", lambda: None) + monkeypatch.setattr(_shared, "Respan", FakeRespan) + monkeypatch.setenv("RESPAN_API_KEY", "test-key") + _shared.create_respan("pinecone_contract", "exact-marker") + + assert captured["metadata"] == { + "example_set": "pinecone", + "workflow_name": "pinecone_contract", + "example_run_id": "exact-marker", + "run_id": "exact-marker", + } + attrs = _shared.workflow_attributes( + "pinecone_contract", "exact-marker", "execution-1" + ) + assert attrs["metadata"]["example_run_id"] == "exact-marker" + assert attrs["metadata"]["run_id"] == "exact-marker" + assert attrs["metadata"]["execution_id"] == "execution-1" + + +def test_runner_continues_and_aggregates_failures(monkeypatch): + scripts = ("first.py", "timeout.py", "last.py") + calls: list[str] = [] + + def fake_run(command, **kwargs): + script = Path(command[-1]).name + calls.append(script) + assert kwargs["env"]["RESPAN_EXAMPLE_RUN_ID"] == "runner-marker" + if script == "timeout.py": + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + return SimpleNamespace(returncode=1 if script == "first.py" else 0) + + monkeypatch.setattr(run_all, "SCRIPTS", scripts) + monkeypatch.setattr(run_all.subprocess, "run", fake_run) + monkeypatch.setenv("RESPAN_EXAMPLE_RUN_ID", "runner-marker") + with pytest.raises(SystemExit) as caught: + run_all.main() + assert calls == list(scripts) + assert "first.py: exited 1" in str(caught.value) + assert "timeout.py: timed out" in str(caught.value) + + +def test_runner_contains_complete_committed_set(): + assert run_all.SCRIPTS == ( + "01_upsert_and_query.py", + "02_async_fetch.py", + "03_expected_error.py", + ) + for script in run_all.SCRIPTS: + assert (EXAMPLE_DIR / script).is_file() + + +def test_print_result_is_json_native(capsys): + _shared.print_result("contract", {"vector": [0.1, 0.2]}, "exact-marker") + output = capsys.readouterr().out + payload = output.split("== contract ==\n", 1)[1] + assert json.loads(payload) == {"vector": [0.1, 0.2]} diff --git a/python/tracing/pipecat/01_offline_pipeline.py b/python/tracing/pipecat/01_offline_pipeline.py index 0dca4a6..47f2bd0 100644 --- a/python/tracing/pipecat/01_offline_pipeline.py +++ b/python/tracing/pipecat/01_offline_pipeline.py @@ -1,117 +1,51 @@ -"""Run a local Pipecat pipeline and export Respan spans.""" +"""Run a deterministic current-Pipecat pipeline and export Respan spans.""" from __future__ import annotations import asyncio from pathlib import Path -from pipecat.frames.frames import ( - EndFrame, - Frame, - LLMContextFrame, - LLMFullResponseEndFrame, - LLMFullResponseStartFrame, - LLMTextFrame, +from _pipeline import OfflineLLMService, run_pipeline +from _shared import ( + create_respan, + execution_id, + finish_respan, + marker, + print_result, + workflow_attributes, ) -from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask -from pipecat.processors.aggregators.llm_context import LLMContext -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.services.llm_service import LLMService, LLMSettings -from respan import workflow - -from _shared import create_respan +from respan import Respan, workflow SCRIPT_NAME = Path(__file__).name - - -class OfflineLLMService(LLMService): - """Small local LLM service that emits Pipecat LLM frames without network.""" - - def __init__(self) -> None: - super().__init__( - name="offline_llm", - settings=LLMSettings( - model="offline-pipecat-demo", - system_instruction=None, - temperature=None, - max_tokens=None, - top_p=None, - top_k=None, - frequency_penalty=None, - presence_penalty=None, - seed=None, - filter_incomplete_user_turns=False, - user_turn_completion_config=None, - ), - ) - - async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: - await super().process_frame(frame, direction) - - if isinstance(frame, LLMContextFrame): - await self.push_frame(LLMFullResponseStartFrame()) - await self.push_frame(LLMTextFrame("Pipecat instrumentation is active.")) - await self.push_frame(LLMFullResponseEndFrame()) - else: - await self.push_frame(frame, direction) - - -class TextCollector(FrameProcessor): - def __init__(self) -> None: - super().__init__(name="text_collector", enable_direct_mode=True) - self.text: list[str] = [] - self.done = asyncio.Event() - - async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: - await super().process_frame(frame, direction) - if isinstance(frame, LLMTextFrame): - self.text.append(frame.text) - elif isinstance(frame, LLMFullResponseEndFrame): - self.done.set() - await self.push_frame(frame, direction) - - -@workflow(name=SCRIPT_NAME) -async def run_offline_pipeline() -> str: - collector = TextCollector() - pipeline = Pipeline([OfflineLLMService(), collector]) - task = PipelineTask( - pipeline, - cancel_on_idle_timeout=False, - enable_rtvi=False, - conversation_id="offline-pipecat-session", - ) - - async def push_frames() -> None: - await asyncio.sleep(0.05) - context = LLMContext( - messages=[ - { - "role": "user", - "content": "Confirm that the Pipecat pipeline is traced.", - } - ] - ) - await task.queue_frame(LLMContextFrame(context)) - await asyncio.wait_for(collector.done.wait(), timeout=10) - await asyncio.sleep(0.2) - await task.queue_frame(EndFrame()) - - runner = PipelineRunner(handle_sigint=False) - await asyncio.gather(runner.run(task), push_frames()) - return "".join(collector.text) +WORKFLOW_NAME = "pipecat_offline_pipeline" async def main() -> None: - respan, run_id = create_respan(SCRIPT_NAME, mode="offline") + run_marker = marker() + execution = execution_id() + respan = create_respan(WORKFLOW_NAME, run_marker) try: - result = await run_offline_pipeline() - print(f"run_id={run_id}") - print(result) + + @workflow(name=WORKFLOW_NAME) + async def trace_pipeline(prompt: str) -> dict[str, str]: + result = await run_pipeline( + OfflineLLMService(response="Pipecat instrumentation is active."), + prompt=prompt, + conversation_id=f"offline-{execution}", + ) + return {"response": result.text, "status": "completed"} + + with Respan.propagate_attributes( + **workflow_attributes( + WORKFLOW_NAME, run_marker, execution, mode="deterministic" + ) + ): + result = await trace_pipeline( + "Confirm that the Pipecat pipeline is traced." + ) + print_result(SCRIPT_NAME, result, run_marker) finally: - await asyncio.sleep(1) + finish_respan(respan) if __name__ == "__main__": diff --git a/python/tracing/pipecat/02_gateway_llm_pipeline.py b/python/tracing/pipecat/02_gateway_llm_pipeline.py index 090a771..21c071a 100644 --- a/python/tracing/pipecat/02_gateway_llm_pipeline.py +++ b/python/tracing/pipecat/02_gateway_llm_pipeline.py @@ -1,90 +1,68 @@ -"""Run a Pipecat LLM pipeline through the Respan gateway.""" +"""Run a real Pipecat OpenAI service through the configured Respan gateway.""" from __future__ import annotations import asyncio from pathlib import Path -from pipecat.frames.frames import ( - EndFrame, - Frame, - LLMContextFrame, - LLMFullResponseEndFrame, - LLMTextFrame, +from _pipeline import OfflineLLMService, run_pipeline +from _shared import ( + create_respan, + execution_id, + finish_respan, + gateway_config, + load_example_env, + marker, + print_result, + workflow_attributes, ) -from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask -from pipecat.processors.aggregators.llm_context import LLMContext -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.openai.llm import OpenAILLMService -from respan import workflow - -from _shared import create_respan, load_example_env +from respan import Respan, workflow SCRIPT_NAME = Path(__file__).name - - -class TextCollector(FrameProcessor): - def __init__(self) -> None: - super().__init__(name="text_collector", enable_direct_mode=True) - self.text: list[str] = [] - self.done = asyncio.Event() - - async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: - await super().process_frame(frame, direction) - if isinstance(frame, LLMTextFrame): - self.text.append(frame.text) - elif isinstance(frame, LLMFullResponseEndFrame): - self.done.set() - await self.push_frame(frame, direction) - - -@workflow(name=SCRIPT_NAME) -async def run_gateway_pipeline() -> str: - env = load_example_env() - collector = TextCollector() - llm = OpenAILLMService( - api_key=env["gateway_api_key"], - base_url=env["gateway_base_url"], - settings=OpenAILLMService.Settings(model=env["model"]), - ) - pipeline = Pipeline([llm, collector]) - task = PipelineTask( - pipeline, - cancel_on_idle_timeout=False, - enable_rtvi=False, - conversation_id="gateway-pipecat-session", - ) - - async def push_frames() -> None: - await asyncio.sleep(0.05) - context = LLMContext( - messages=[ - { - "role": "user", - "content": "Reply in one short sentence about Pipecat tracing.", - } - ] - ) - await task.queue_frame(LLMContextFrame(context)) - await asyncio.wait_for(collector.done.wait(), timeout=30) - await asyncio.sleep(0.2) - await task.queue_frame(EndFrame()) - - runner = PipelineRunner(handle_sigint=False) - await asyncio.gather(runner.run(task), push_frames()) - return "".join(collector.text) +WORKFLOW_NAME = "pipecat_gateway_pipeline" async def main() -> None: - respan, run_id = create_respan(SCRIPT_NAME, mode="gateway") + load_example_env() + run_marker = marker() + execution = execution_id() + config = gateway_config() + mode = "live" if config else "deterministic-fallback" + respan = create_respan(WORKFLOW_NAME, run_marker) try: - result = await run_gateway_pipeline() - print(f"run_id={run_id}") - print(result) + + @workflow(name=WORKFLOW_NAME) + async def trace_gateway(prompt: str) -> dict[str, str]: + service = ( + OpenAILLMService( + api_key=config["api_key"], + base_url=config["base_url"], + settings=OpenAILLMService.Settings( + model=config["model"], max_completion_tokens=32 + ), + ) + if config + else OfflineLLMService(response="Pipecat gateway fallback is active.") + ) + result = await run_pipeline( + service, + prompt=prompt, + conversation_id=f"gateway-{execution}", + ) + if result.error: + raise RuntimeError(result.error) + return {"response": result.text, "status": "completed"} + + with Respan.propagate_attributes( + **workflow_attributes(WORKFLOW_NAME, run_marker, execution, mode=mode) + ): + result = await trace_gateway( + "Reply with exactly: Pipecat gateway tracing works." + ) + print_result(SCRIPT_NAME, result, run_marker) finally: - await asyncio.sleep(1) + finish_respan(respan) if __name__ == "__main__": diff --git a/python/tracing/pipecat/03_expected_error.py b/python/tracing/pipecat/03_expected_error.py new file mode 100644 index 0000000..6fea864 --- /dev/null +++ b/python/tracing/pipecat/03_expected_error.py @@ -0,0 +1,64 @@ +"""Export a deterministic real-Pipecat provider error path.""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +from _pipeline import OfflineLLMService, ProviderHTTPError, run_pipeline +from _shared import ( + create_respan, + execution_id, + finish_respan, + marker, + print_result, + workflow_attributes, +) +from respan import Respan, workflow + +SCRIPT_NAME = Path(__file__).name +WORKFLOW_NAME = "pipecat_expected_error" + + +async def main() -> None: + run_marker = marker() + execution = execution_id() + respan = create_respan(WORKFLOW_NAME, run_marker) + + @workflow(name=WORKFLOW_NAME) + async def trace_expected_error(prompt: str) -> None: + result = await run_pipeline( + OfflineLLMService(response="", fail_status=401), + prompt=prompt, + conversation_id=f"error-{execution}", + ) + raise ProviderHTTPError( + result.error or "deterministic provider authorization failure", + status_code=401, + ) + + try: + try: + with Respan.propagate_attributes( + **workflow_attributes( + WORKFLOW_NAME, + run_marker, + execution, + mode="deterministic-error", + ) + ): + await trace_expected_error( + "Exercise the provider authorization failure path." + ) + except ProviderHTTPError as exc: + print_result( + SCRIPT_NAME, + {"expected_error": type(exc).__name__, "status_code": exc.status_code}, + run_marker, + ) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/tracing/pipecat/README.md b/python/tracing/pipecat/README.md index e4c17d5..2524997 100644 --- a/python/tracing/pipecat/README.md +++ b/python/tracing/pipecat/README.md @@ -1,59 +1,28 @@ -# Pipecat Respan Examples +# Pipecat tracing -These examples trace Pipecat pipelines with `respan-instrumentation-pipecat`. -They load environment variables from the `respan-example-projects/.env` file. -Each script wraps the run in a Respan workflow whose name matches the script -filename, so the platform result is easy to map back to the example. +These examples validate `respan-instrumentation-pipecat` with Pipecat's current +`PipelineWorker` and `WorkerRunner` lifecycle. -## Setup +- `01_offline_pipeline.py` runs a deterministic real Pipecat pipeline. +- `02_gateway_llm_pipeline.py` uses Pipecat's real OpenAI service through the + configured Respan gateway (and has a deterministic fallback if gateway + configuration is absent). +- `03_expected_error.py` emits a deterministic provider-style 401 `ErrorFrame`. -```bash -cd python/tracing/pipecat -pip install -r requirements.txt -``` - -For local development against a checkout: +From this directory: ```bash -pip install -e /home/yuyang/KeywordsAI/respan/python-sdks/respan-sdk \ - -e /home/yuyang/KeywordsAI/respan/python-sdks/respan-tracing \ - -e /home/yuyang/KeywordsAI/respan/python-sdks/respan \ - -e /home/yuyang/KeywordsAI/respan/python-sdks/instrumentations/respan-instrumentation-pipecat \ - -r requirements.txt +python -m pip install -r requirements.txt +RESPAN_EXAMPLE_RUN_ID=pipecat-check python run_all.py ``` -## Environment - -The scripts read `/home/yuyang/KeywordsAI/respan-example-projects/.env`. - -Required: - -```bash -RESPAN_API_KEY=... -``` - -Optional gateway settings: - -```bash -RESPAN_GATEWAY_API_KEY=... -RESPAN_GATEWAY_BASE_URL=https://api.respan.ai/api -RESPAN_MODEL=gpt-4.1-nano -``` - -## Examples - -| Example | Description | -|---------|-------------| -| `01_offline_pipeline.py` | Local Pipecat pipeline that emits LLM frames without a provider call. | -| `02_gateway_llm_pipeline.py` | Pipecat `OpenAILLMService` routed through the Respan gateway. | - -Run: +For local instrumentation development, link the package after installing the +registry requirements: ```bash -python 01_offline_pipeline.py -python 02_gateway_llm_pipeline.py +python -m pip install -e ../../../../respan/python-sdks/instrumentations/respan-instrumentation-pipecat ``` -Each script prints a `run_id` and exports spans with -`customer_identifier=pipecat-example`, `environment=example`, workflow name, and -metadata for the script name. +The runner preserves the exact shell marker, runs every committed scenario, +continues after failures/timeouts, and reports an aggregate result. Every +script flushes and shuts down Respan explicitly. diff --git a/python/tracing/pipecat/_pipeline.py b/python/tracing/pipecat/_pipeline.py new file mode 100644 index 0000000..18744a0 --- /dev/null +++ b/python/tracing/pipecat/_pipeline.py @@ -0,0 +1,130 @@ +"""Current Pipecat worker fixtures shared by the runnable examples.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass + +from pipecat.frames.frames import ( + EndFrame, + ErrorFrame, + Frame, + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, +) +from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.worker import PipelineParams, PipelineWorker +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.services.llm_service import LLMService, LLMSettings +from pipecat.workers.runner import WorkerRunner + + +class ProviderHTTPError(RuntimeError): + def __init__(self, message: str, *, status_code: int) -> None: + super().__init__(message) + self.status_code = status_code + + +class OfflineLLMService(LLMService): + def __init__(self, *, response: str, fail_status: int | None = None) -> None: + super().__init__( + name="OfflineLLMService", + settings=LLMSettings( + model="offline-pipecat-demo", + system_instruction=None, + temperature=None, + max_tokens=None, + top_p=None, + top_k=None, + frequency_penalty=None, + presence_penalty=None, + seed=None, + filter_incomplete_user_turns=False, + user_turn_completion_config=None, + ), + ) + self._response = response + self._fail_status = fail_status + + def can_generate_metrics(self) -> bool: + return True + + async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: + await super().process_frame(frame, direction) + if not isinstance(frame, LLMContextFrame): + await self.push_frame(frame, direction) + return + await self.push_frame(LLMFullResponseStartFrame()) + if self._fail_status is not None: + error = ProviderHTTPError( + "deterministic provider authorization failure", + status_code=self._fail_status, + ) + await self.push_frame( + ErrorFrame( + error="deterministic provider authorization failure", + exception=error, + processor=self, + ) + ) + return + await self.push_frame(LLMTextFrame(self._response)) + await self.start_llm_usage_metrics( + LLMTokenUsage(prompt_tokens=8, completion_tokens=6, total_tokens=14) + ) + await self.push_frame(LLMFullResponseEndFrame()) + + +class TextCollector(FrameProcessor): + def __init__(self) -> None: + super().__init__(name="text_collector", enable_direct_mode=True) + self.text: list[str] = [] + self.error: str | None = None + self.done = asyncio.Event() + + async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: + await super().process_frame(frame, direction) + if isinstance(frame, LLMTextFrame): + self.text.append(frame.text) + elif isinstance(frame, ErrorFrame): + self.error = frame.error + self.done.set() + elif isinstance(frame, LLMFullResponseEndFrame): + self.done.set() + await self.push_frame(frame, direction) + + +@dataclass(frozen=True) +class PipelineResult: + text: str + error: str | None + + +async def run_pipeline( + service: LLMService, *, prompt: str, conversation_id: str +) -> PipelineResult: + collector = TextCollector() + worker = PipelineWorker( + Pipeline([service, collector]), + cancel_on_idle_timeout=False, + enable_rtvi=False, + conversation_id=conversation_id, + params=PipelineParams(enable_metrics=True, enable_usage_metrics=True), + ) + runner = WorkerRunner(handle_sigint=False) + await runner.add_workers(worker) + + async def drive() -> None: + await asyncio.sleep(0.05) + await worker.queue_frame( + LLMContextFrame(LLMContext(messages=[{"role": "user", "content": prompt}])) + ) + await asyncio.wait_for(collector.done.wait(), timeout=30) + await worker.queue_frame(EndFrame()) + + await asyncio.wait_for(asyncio.gather(runner.run(), drive()), timeout=40) + return PipelineResult(text="".join(collector.text), error=collector.error) diff --git a/python/tracing/pipecat/_shared.py b/python/tracing/pipecat/_shared.py index 4174e16..c1e032e 100644 --- a/python/tracing/pipecat/_shared.py +++ b/python/tracing/pipecat/_shared.py @@ -1,66 +1,100 @@ -"""Shared setup for Pipecat Respan examples.""" +"""Shared helpers for Pipecat tracing examples.""" from __future__ import annotations +import json import os -import uuid from pathlib import Path from typing import Any +from uuid import uuid4 from dotenv import load_dotenv from respan import Respan from respan_instrumentation_pipecat import PipecatInstrumentor -REPO_ROOT = Path(__file__).resolve().parents[3] +EXAMPLE_DIR = Path(__file__).resolve().parent +REPO_ROOT = EXAMPLE_DIR.parents[2] DEFAULT_BASE_URL = "https://api.respan.ai/api" DEFAULT_MODEL = "gpt-4.1-nano" +EXAMPLE_SET = "pipecat" -def load_example_env() -> dict[str, str]: - """Load the example repo-root .env and return normalized settings.""" - load_dotenv(REPO_ROOT / ".env", override=True) +def load_example_env() -> None: + load_dotenv(REPO_ROOT / ".env", override=False) + if not os.getenv("RESPAN_API_KEY"): + raise RuntimeError(f"RESPAN_API_KEY is required in {REPO_ROOT / '.env'}") + os.environ.setdefault("RESPAN_BASE_URL", DEFAULT_BASE_URL) - respan_api_key = os.getenv("RESPAN_API_KEY") - gateway_api_key = os.getenv("RESPAN_GATEWAY_API_KEY") or respan_api_key - respan_base_url = os.getenv("RESPAN_BASE_URL", DEFAULT_BASE_URL) - gateway_base_url = os.getenv("RESPAN_GATEWAY_BASE_URL") or respan_base_url - model = os.getenv("RESPAN_MODEL", DEFAULT_MODEL) - if not respan_api_key: - raise RuntimeError("RESPAN_API_KEY must be set in respan-example-projects/.env") +def marker() -> str: + return os.getenv("RESPAN_EXAMPLE_RUN_ID", "").strip() or ( + f"pipecat-{uuid4().hex[:10]}" + ) + + +def execution_id() -> str: + return uuid4().hex[:10] - if gateway_api_key: - os.environ["OPENAI_API_KEY"] = gateway_api_key - os.environ["OPENAI_BASE_URL"] = gateway_base_url +def gateway_config() -> dict[str, str] | None: + api_key = os.getenv("RESPAN_GATEWAY_API_KEY") or os.getenv("RESPAN_API_KEY") + base_url = os.getenv("RESPAN_GATEWAY_BASE_URL") or os.getenv("RESPAN_BASE_URL") + if not api_key or not base_url: + return None return { - "respan_api_key": respan_api_key, - "respan_base_url": respan_base_url, - "gateway_api_key": gateway_api_key or "", - "gateway_base_url": gateway_base_url, - "model": model, + "api_key": api_key, + "base_url": base_url, + "model": os.getenv("RESPAN_MODEL", DEFAULT_MODEL), } -def create_respan(example_name: str, **metadata: Any) -> tuple[Respan, str]: - """Create Respan with Pipecat instrumentation and searchable metadata.""" - env = load_example_env() - run_id = metadata.pop("run_id", uuid.uuid4().hex[:12]) - - respan = Respan( - api_key=env["respan_api_key"], - base_url=env["respan_base_url"], - app_name=f"pipecat-{example_name}", - instrumentations=[PipecatInstrumentor()], - is_batching_enabled=False, - customer_identifier="pipecat-example", - thread_identifier=f"pipecat-{run_id}", +def create_respan(workflow_name: str, run_marker: str) -> Respan: + load_example_env() + return Respan( + api_key=os.environ["RESPAN_API_KEY"], + base_url=os.getenv("RESPAN_BASE_URL", DEFAULT_BASE_URL), + app_name=workflow_name, metadata={ - "example": "pipecat", - "script": example_name, - "run_id": run_id, - **metadata, + "example_set": EXAMPLE_SET, + "workflow_name": workflow_name, + "example_run_id": run_marker, + "run_id": run_marker, }, - environment="example", + instrumentations=[PipecatInstrumentor()], + is_batching_enabled=False, + log_level=os.getenv("RESPAN_LOG_LEVEL", "WARNING"), ) - return respan, run_id + + +def workflow_attributes( + workflow_name: str, + run_marker: str, + execution: str, + *, + mode: str, +) -> dict[str, object]: + return { + "trace_group_identifier": workflow_name, + "custom_identifier": f"{workflow_name}-{execution}", + "metadata": { + "example_set": EXAMPLE_SET, + "workflow_name": workflow_name, + "example_run_id": run_marker, + "run_id": run_marker, + "execution_id": execution, + "mode": mode, + }, + } + + +def print_result(label: str, value: Any, run_marker: str) -> None: + print(f"RESPAN_EXAMPLE_RUN_ID={run_marker}") + print(f"\n== {label} ==") + print(json.dumps(value, allow_nan=False, indent=2, sort_keys=True)) + + +def finish_respan(respan: Respan) -> None: + try: + respan.flush() + finally: + respan.shutdown() diff --git a/python/tracing/pipecat/requirements.txt b/python/tracing/pipecat/requirements.txt index cc7a26f..0385c3a 100644 --- a/python/tracing/pipecat/requirements.txt +++ b/python/tracing/pipecat/requirements.txt @@ -1,3 +1,5 @@ -python-dotenv -respan-ai -respan-instrumentation-pipecat +pipecat-ai>=1.7,<2 +openinference-instrumentation-pipecat>=2.0.1,<3 +python-dotenv>=1,<2 +respan-ai>=4.1,<5 +respan-instrumentation-pipecat>=0.1,<1 diff --git a/python/tracing/pipecat/run_all.py b/python/tracing/pipecat/run_all.py new file mode 100644 index 0000000..f6b2229 --- /dev/null +++ b/python/tracing/pipecat/run_all.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import os +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path + +EXAMPLE_DIR = Path(__file__).resolve().parent +SCRIPTS = ( + "01_offline_pipeline.py", + "02_gateway_llm_pipeline.py", + "03_expected_error.py", +) +DEFAULT_TIMEOUT_SECONDS = 120.0 + + +def main() -> None: + run_marker = os.getenv("RESPAN_EXAMPLE_RUN_ID", "").strip() or ( + "pipecat-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + ) + timeout = float( + os.getenv("RESPAN_EXAMPLE_TIMEOUT_SECONDS", DEFAULT_TIMEOUT_SECONDS) + ) + env = os.environ.copy() + env["RESPAN_EXAMPLE_RUN_ID"] = run_marker + failures: list[str] = [] + for script in SCRIPTS: + print(f"\n### Running {script}", flush=True) + try: + completed = subprocess.run( + [sys.executable, str(EXAMPLE_DIR / script)], + check=False, + env=env, + timeout=timeout, + ) + except subprocess.TimeoutExpired: + failures.append(f"{script}: timed out after {timeout:g}s") + continue + if completed.returncode: + failures.append(f"{script}: exited {completed.returncode}") + if failures: + raise SystemExit("Pipecat example failures:\n- " + "\n- ".join(failures)) + print(f"\nCompleted Pipecat examples: RESPAN_EXAMPLE_RUN_ID={run_marker}") + + +if __name__ == "__main__": + main() diff --git a/python/tracing/pipecat/test_contract.py b/python/tracing/pipecat/test_contract.py new file mode 100644 index 0000000..c51de10 --- /dev/null +++ b/python/tracing/pipecat/test_contract.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import ast +import os +import subprocess +from pathlib import Path +from types import SimpleNamespace + +import _shared +import pytest +import run_all + +EXAMPLE_DIR = Path(__file__).resolve().parent + + +def _workflow_signatures(filename: str) -> list[list[str]]: + module = ast.parse((EXAMPLE_DIR / filename).read_text()) + workflows = [ + node + for node in ast.walk(module) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and any( + isinstance(decorator, ast.Call) + and isinstance(decorator.func, ast.Name) + and decorator.func.id == "workflow" + for decorator in node.decorator_list + ) + ] + return [[argument.arg for argument in node.args.args] for node in workflows] + + +def test_roots_capture_only_bounded_semantic_inputs(): + assert _workflow_signatures("01_offline_pipeline.py") == [["prompt"]] + assert _workflow_signatures("02_gateway_llm_pipeline.py") == [["prompt"]] + assert _workflow_signatures("03_expected_error.py") == [["prompt"]] + + +def test_current_worker_api_replaces_deprecated_task_and_runner(): + for filename in run_all.SCRIPTS: + source = (EXAMPLE_DIR / filename).read_text() + assert "PipelineTask" not in source + assert "PipelineRunner" not in source + shared_source = (EXAMPLE_DIR / "_pipeline.py").read_text() + assert "PipelineWorker" in shared_source + assert "WorkerRunner" in shared_source + assert "add_workers" in shared_source + + +def test_shell_marker_survives_dotenv(monkeypatch): + monkeypatch.setenv("RESPAN_EXAMPLE_RUN_ID", "exact-shell-marker") + + def fake_load_dotenv(_path, *, override): + assert override is False + os.environ.setdefault("RESPAN_EXAMPLE_RUN_ID", "dotenv-marker") + os.environ.setdefault("RESPAN_API_KEY", "test-key") + + monkeypatch.setattr(_shared, "load_dotenv", fake_load_dotenv) + _shared.load_example_env() + assert _shared.marker() == "exact-shell-marker" + + +def test_respan_and_workflow_metadata_use_exact_marker(monkeypatch): + captured: dict[str, object] = {} + + class FakeRespan: + def __init__(self, **kwargs): + captured.update(kwargs) + + monkeypatch.setattr(_shared, "load_example_env", lambda: None) + monkeypatch.setattr(_shared, "Respan", FakeRespan) + monkeypatch.setenv("RESPAN_API_KEY", "test-key") + _shared.create_respan("pipecat_contract", "exact-marker") + + assert captured["metadata"] == { + "example_set": "pipecat", + "workflow_name": "pipecat_contract", + "example_run_id": "exact-marker", + "run_id": "exact-marker", + } + attrs = _shared.workflow_attributes( + "pipecat_contract", "exact-marker", "execution-1", mode="deterministic" + ) + assert attrs["metadata"]["example_run_id"] == "exact-marker" + assert attrs["metadata"]["run_id"] == "exact-marker" + + +def test_runner_continues_and_aggregates_failures(monkeypatch): + scripts = ("first.py", "timeout.py", "last.py") + calls: list[str] = [] + + def fake_run(command, **kwargs): + script = Path(command[-1]).name + calls.append(script) + assert kwargs["env"]["RESPAN_EXAMPLE_RUN_ID"] == "runner-marker" + if script == "timeout.py": + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + return SimpleNamespace(returncode=1 if script == "first.py" else 0) + + monkeypatch.setattr(run_all, "SCRIPTS", scripts) + monkeypatch.setattr(run_all.subprocess, "run", fake_run) + monkeypatch.setenv("RESPAN_EXAMPLE_RUN_ID", "runner-marker") + with pytest.raises(SystemExit) as caught: + run_all.main() + assert calls == list(scripts) + assert "first.py: exited 1" in str(caught.value) + assert "timeout.py: timed out" in str(caught.value) + + +def test_runner_contains_complete_committed_set(): + assert run_all.SCRIPTS == ( + "01_offline_pipeline.py", + "02_gateway_llm_pipeline.py", + "03_expected_error.py", + ) + for script in run_all.SCRIPTS: + assert (EXAMPLE_DIR / script).is_file() diff --git a/python/tracing/portkey/01_chat_completion.py b/python/tracing/portkey/01_chat_completion.py index e637b4d..27b8af2 100644 --- a/python/tracing/portkey/01_chat_completion.py +++ b/python/tracing/portkey/01_chat_completion.py @@ -1,50 +1,46 @@ from __future__ import annotations -from respan import workflow - from _shared import ( example_attributes, + execution_id, + finish_respan, make_client, - make_custom_identifier, make_respan, + marker, model_name, print_result, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "chat-completion" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _chat_completion_workflow(client) -> str: - response = client.chat.completions.create( - model=model_name(), - messages=[ - { - "role": "user", - "content": "Reply with one concise sentence about AI gateways.", - } - ], - ) - return response.choices[0].message.content or "" - - -def run_chat_completion() -> None: - respan = make_respan(EXAMPLE_NAME) +def trace_chat(prompt: str) -> dict[str, str]: client = make_client() - custom_identifier = make_custom_identifier(EXAMPLE_NAME) - text = "" - try: - with example_attributes(EXAMPLE_NAME, custom_identifier): - print(f"custom_identifier={custom_identifier}", flush=True) - print(f"workflow_name={workflow_name(EXAMPLE_NAME)}", flush=True) - text = _chat_completion_workflow(client) + response = client.chat.completions.create( + model=model_name(), messages=[{"role": "user", "content": prompt}] + ) + return {"response": response.choices[0].message.content or ""} finally: - respan.shutdown() + client.close() - print_result(EXAMPLE_NAME, custom_identifier, text) + +def main() -> None: + run_marker = marker() + execution = execution_id() + respan = make_respan(EXAMPLE_NAME, run_marker) + try: + with example_attributes( + EXAMPLE_NAME, run_marker, execution, mode="deterministic" + ): + result = trace_chat("Explain AI gateways in one concise sentence.") + print_result(EXAMPLE_NAME, run_marker, result) + finally: + finish_respan(respan) if __name__ == "__main__": - run_chat_completion() + main() diff --git a/python/tracing/portkey/02_async_chat_completion.py b/python/tracing/portkey/02_async_chat_completion.py index 3ed68a1..60819b7 100644 --- a/python/tracing/portkey/02_async_chat_completion.py +++ b/python/tracing/portkey/02_async_chat_completion.py @@ -2,52 +2,47 @@ import asyncio -from respan import workflow - from _shared import ( example_attributes, + execution_id, + finish_respan, make_async_client, - make_custom_identifier, make_respan, + marker, model_name, print_result, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "async-chat-completion" @workflow(name=workflow_name(EXAMPLE_NAME)) -async def _async_chat_completion_workflow(client) -> str: - response = await client.chat.completions.create( - model=model_name(), - messages=[ - { - "role": "user", - "content": "Reply with one concise sentence about tracing LLM apps.", - } - ], - ) - return response.choices[0].message.content or "" - - -async def run_async_chat_completion() -> None: - respan = make_respan(EXAMPLE_NAME) +async def trace_async_chat(prompt: str) -> dict[str, str]: client = make_async_client() - custom_identifier = make_custom_identifier(EXAMPLE_NAME) - text = "" - try: - with example_attributes(EXAMPLE_NAME, custom_identifier): - print(f"custom_identifier={custom_identifier}", flush=True) - print(f"workflow_name={workflow_name(EXAMPLE_NAME)}", flush=True) - text = await _async_chat_completion_workflow(client) + response = await client.chat.completions.create( + model=model_name(), messages=[{"role": "user", "content": prompt}] + ) + return {"response": response.choices[0].message.content or ""} finally: await client.close() - respan.shutdown() - print_result(EXAMPLE_NAME, custom_identifier, text) + +async def main() -> None: + run_marker = marker() + execution = execution_id() + respan = make_respan(EXAMPLE_NAME, run_marker) + try: + with example_attributes( + EXAMPLE_NAME, run_marker, execution, mode="deterministic" + ): + result = await trace_async_chat("Explain tracing in one concise sentence.") + print_result(EXAMPLE_NAME, run_marker, result) + finally: + finish_respan(respan) if __name__ == "__main__": - asyncio.run(run_async_chat_completion()) + asyncio.run(main()) diff --git a/python/tracing/portkey/03_streaming_chat.py b/python/tracing/portkey/03_streaming_chat.py new file mode 100644 index 0000000..1d57a75 --- /dev/null +++ b/python/tracing/portkey/03_streaming_chat.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from _shared import ( + example_attributes, + execution_id, + finish_respan, + make_client, + make_respan, + marker, + model_name, + print_result, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "streaming-chat" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def trace_stream(prompt: str) -> dict[str, str]: + client = make_client() + content: list[str] = [] + try: + stream = client.chat.completions.create( + model=model_name(), + messages=[{"role": "user", "content": prompt}], + stream=True, + stream_options={"include_usage": True}, + ) + for chunk in stream: + if chunk.choices and chunk.choices[0].delta.content: + content.append(chunk.choices[0].delta.content) + return {"response": "".join(content)} + finally: + client.close() + + +def main() -> None: + run_marker = marker() + execution = execution_id() + respan = make_respan(EXAMPLE_NAME, run_marker) + try: + with example_attributes( + EXAMPLE_NAME, run_marker, execution, mode="deterministic" + ): + result = trace_stream("Stream a short Portkey tracing confirmation.") + print_result(EXAMPLE_NAME, run_marker, result) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/portkey/04_tool_calling.py b/python/tracing/portkey/04_tool_calling.py new file mode 100644 index 0000000..fcee806 --- /dev/null +++ b/python/tracing/portkey/04_tool_calling.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import json + +from _shared import ( + example_attributes, + execution_id, + finish_respan, + make_client, + make_respan, + marker, + model_name, + print_result, + workflow_name, +) +from respan import tool, workflow + +EXAMPLE_NAME = "tool-calling" + + +@tool(name="get_weather") +def get_weather(city: str) -> str: + return f"{city} is sunny and 72F." + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def trace_tool(city: str) -> dict[str, str]: + client = make_client() + messages: list[dict[str, object]] = [ + {"role": "user", "content": f"What is the weather in {city}?"} + ] + tools = [ + { + "type": "function", + "function": { + "name": "get_weather", + "description": "Return deterministic weather.", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"], + }, + }, + } + ] + try: + first = client.chat.completions.create( + model=model_name(), messages=messages, tools=tools, tool_choice="auto" + ) + assistant = first.choices[0].message + call = assistant.tool_calls[0] + arguments = json.loads(call.function.arguments) + result = get_weather(arguments["city"]) + messages.extend( + [ + assistant.model_dump(exclude_none=True), + {"role": "tool", "tool_call_id": call.id, "content": result}, + ] + ) + second = client.chat.completions.create( + model=model_name(), messages=messages, tools=tools + ) + return { + "tool_result": result, + "response": second.choices[0].message.content or "", + } + finally: + client.close() + + +def main() -> None: + run_marker = marker() + execution = execution_id() + respan = make_respan(EXAMPLE_NAME, run_marker) + try: + with example_attributes( + EXAMPLE_NAME, run_marker, execution, mode="deterministic" + ): + result = trace_tool("Tokyo") + print_result(EXAMPLE_NAME, run_marker, result) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/portkey/05_expected_error.py b/python/tracing/portkey/05_expected_error.py new file mode 100644 index 0000000..3e7eadf --- /dev/null +++ b/python/tracing/portkey/05_expected_error.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +from typing import Any + +from _shared import ( + example_attributes, + execution_id, + finish_respan, + make_client, + make_respan, + marker, + print_result, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "expected-error" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def trace_expected_error(prompt: str) -> None: + client = make_client() + try: + client.chat.completions.create( + model="error-401", messages=[{"role": "user", "content": prompt}] + ) + finally: + client.close() + + +def _status_code(exc: BaseException) -> int: + value: Any = getattr(exc, "status_code", None) + return value if isinstance(value, int) else 500 + + +def main() -> None: + run_marker = marker() + execution = execution_id() + respan = make_respan(EXAMPLE_NAME, run_marker) + try: + try: + with example_attributes( + EXAMPLE_NAME, + run_marker, + execution, + mode="deterministic-error", + ): + trace_expected_error("Exercise the Portkey provider failure path.") + except Exception as exc: # noqa: BLE001 - this is the expected SDK failure path + print_result( + EXAMPLE_NAME, + run_marker, + { + "expected_error": type(exc).__name__, + "status_code": _status_code(exc), + }, + ) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/portkey/06_live_portkey.py b/python/tracing/portkey/06_live_portkey.py new file mode 100644 index 0000000..4cf4cd7 --- /dev/null +++ b/python/tracing/portkey/06_live_portkey.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +from _shared import ( + example_attributes, + execution_id, + finish_respan, + live_configured, + make_client, + make_respan, + marker, + model_name, + print_result, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "live-portkey" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def trace_live(prompt: str) -> dict[str, str]: + client = make_client(live=True) + try: + response = client.chat.completions.create( + model=model_name(live=True), messages=[{"role": "user", "content": prompt}] + ) + return {"response": response.choices[0].message.content or ""} + finally: + client.close() + + +def main() -> None: + if not live_configured(): + print("SKIP: PORTKEY_API_KEY is not configured") + return + run_marker = marker() + execution = execution_id() + respan = make_respan(EXAMPLE_NAME, run_marker) + try: + with example_attributes(EXAMPLE_NAME, run_marker, execution, mode="live"): + result = trace_live("Reply with exactly: live Portkey tracing works.") + print_result(EXAMPLE_NAME, run_marker, result) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/portkey/README.md b/python/tracing/portkey/README.md index 2e80ee9..9a59700 100644 --- a/python/tracing/portkey/README.md +++ b/python/tracing/portkey/README.md @@ -1,48 +1,22 @@ -# Portkey tracing examples +# Portkey tracing -These examples trace the official `portkey-ai` Python SDK with Respan. They load environment variables from the repository root `.env` file. - -Required for exporting traces: - -```bash -RESPAN_API_KEY=... -``` - -For Portkey calls, use one of these options: - -```bash -# Direct Portkey Gateway calls -PORTKEY_API_KEY=... -PORTKEY_PROVIDER=@your-provider -# or -PORTKEY_CONFIG=pc-... -``` - -If `PORTKEY_API_KEY` is not set, the examples use a local OpenAI-compatible test endpoint so traces are runnable without third-party model credentials. To force a live provider fallback, set one of these in the root `.env` file: - -```bash -PORTKEY_EXAMPLE_USE_OPENAI=1 -OPENAI_API_KEY=... -# or -PORTKEY_EXAMPLE_USE_LIVE_GATEWAY=1 -RESPAN_GATEWAY_API_KEY=... -RESPAN_GATEWAY_BASE_URL=... -``` - -Optional environment variables: +The deterministic suite uses the real current Portkey SDK against a bounded +OpenAI-compatible protocol fixture. It covers sync, async, streaming, a +connected two-turn tool execution, and a provider-style 401. The optional live +script runs only when `PORTKEY_API_KEY` is configured. ```bash -RESPAN_BASE_URL=https://api.respan.ai/api -PORTKEY_BASE_URL=https://api.portkey.ai/v1 -PORTKEY_MODEL=gpt-4o-mini -RESPAN_MODEL=gpt-4.1-nano +python -m pip install -r requirements.txt +RESPAN_EXAMPLE_RUN_ID=portkey-check python run_all.py ``` -Run one script at a time: +For local package development, install the registry requirements first and +then link the instrumentation explicitly: ```bash -python 01_chat_completion.py -python 02_async_chat_completion.py +python -m pip install -e ../../../../respan/python-sdks/instrumentations/respan-instrumentation-portkey ``` -Each script prints both a `custom_identifier` and `workflow_name`. The workflow name is also used as the Respan trace group identifier so the run is easy to find in traces and MCP lookups. +Every script preserves the exact shell marker, uses only bounded semantic +workflow arguments, closes its Portkey client, and flushes/shuts down Respan. +The aggregate runner continues through failures and timeouts. diff --git a/python/tracing/portkey/_local_gateway.py b/python/tracing/portkey/_local_gateway.py index 38d7cda..de0d2db 100644 --- a/python/tracing/portkey/_local_gateway.py +++ b/python/tracing/portkey/_local_gateway.py @@ -31,9 +31,88 @@ def do_POST(self) -> None: model = request.get("model") or "local-portkey-model" messages = request.get("messages") or [] + if model == "error-401": + payload = json.dumps( + {"error": {"message": "deterministic Portkey authorization failure"}} + ).encode() + self.send_response(401) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + return last_message = messages[-1] if messages else {} - user_text = last_message.get("content") if isinstance(last_message, dict) else "" + user_text = ( + last_message.get("content") if isinstance(last_message, dict) else "" + ) + tools = request.get("tools") or [] + has_tool_result = any( + isinstance(message, dict) and message.get("role") == "tool" + for message in messages + ) + if request.get("stream"): + chunks = ["Portkey ", "streaming ", "works."] + self.send_response(200) + self.send_header("content-type", "text/event-stream") + self.end_headers() + for index, content in enumerate(chunks): + event = { + "id": "chatcmpl-portkey-stream", + "object": "chat.completion.chunk", + "created": int(time.time()), + "model": model, + "choices": [ + { + "index": 0, + "delta": { + "role": "assistant" if index == 0 else None, + "content": content, + }, + "finish_reason": None, + } + ], + } + self.wfile.write(f"data: {json.dumps(event)}\n\n".encode()) + self.wfile.flush() + terminal = { + "id": "chatcmpl-portkey-stream", + "object": "chat.completion.chunk", + "created": int(time.time()), + "model": model, + "choices": [], + "usage": { + "prompt_tokens": 7, + "completion_tokens": 5, + "total_tokens": 12, + }, + } + self.wfile.write(f"data: {json.dumps(terminal)}\n\n".encode()) + self.wfile.write(b"data: [DONE]\n\n") + self.wfile.flush() + return + content = f"Local Portkey-compatible response for: {user_text}" + message: dict[str, Any] = {"role": "assistant", "content": content} + finish_reason = "stop" + if tools and not has_tool_result: + message = { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "portkey-weather-1", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"city":"Tokyo"}', + }, + } + ], + } + finish_reason = "tool_calls" + elif has_tool_result: + content = "Tokyo is sunny and 72F." + message = {"role": "assistant", "content": content} response = { "id": "chatcmpl-portkey-local", "object": "chat.completion", @@ -42,8 +121,8 @@ def do_POST(self) -> None: "choices": [ { "index": 0, - "message": {"role": "assistant", "content": content}, - "finish_reason": "stop", + "message": message, + "finish_reason": finish_reason, } ], "usage": { diff --git a/python/tracing/portkey/_shared.py b/python/tracing/portkey/_shared.py index e3f8dc1..05a8598 100644 --- a/python/tracing/portkey/_shared.py +++ b/python/tracing/portkey/_shared.py @@ -1,159 +1,136 @@ +"""Shared Portkey example setup with exact marker propagation.""" + from __future__ import annotations +import json import os from contextlib import contextmanager from pathlib import Path +from typing import Any from uuid import uuid4 +from _local_gateway import local_gateway_base_url, shutdown_local_gateway from dotenv import load_dotenv from portkey_ai import AsyncPortkey, Portkey from respan import Respan, propagate_attributes from respan_instrumentation_portkey import PortkeyInstrumentor -from _local_gateway import local_gateway_base_url - -PROJECT_ROOT = Path(__file__).resolve().parents[3] +EXAMPLE_DIR = Path(__file__).resolve().parent +REPO_ROOT = EXAMPLE_DIR.parents[2] DEFAULT_RESPAN_BASE_URL = "https://api.respan.ai/api" DEFAULT_MODEL = "gpt-4.1-nano" +EXAMPLE_SET = "portkey" def load_root_env() -> None: - load_dotenv(PROJECT_ROOT / ".env", override=True) - - -def env_value(name: str) -> str | None: - value = os.getenv(name) - if not value: - return None - value = value.strip() - if not value or value.upper() == name: - return None - if value.lower() in {"none", "null", "your_api_key_here"}: - return None - return value - - -def require_env(name: str) -> str: - value = env_value(name) - if not value: - raise RuntimeError(f"{name} must be set in the repo root .env file") - return value - - -def respan_api_key() -> str: - load_root_env() - return require_env("RESPAN_API_KEY") - - -def respan_base_url() -> str: - return os.getenv("RESPAN_BASE_URL", DEFAULT_RESPAN_BASE_URL).rstrip("/") - - -def gateway_api_key() -> str: - return respan_api_key() or require_env("RESPAN_GATEWAY_API_KEY") + load_dotenv(REPO_ROOT / ".env", override=False) + if not os.getenv("RESPAN_API_KEY"): + raise RuntimeError(f"RESPAN_API_KEY is required in {REPO_ROOT / '.env'}") -def gateway_base_url() -> str: - return ( - os.getenv("RESPAN_GATEWAY_BASE_URL") - or os.getenv("RESPAN_BASE_URL") - or DEFAULT_RESPAN_BASE_URL - ).rstrip("/") +def marker() -> str: + return os.getenv("RESPAN_EXAMPLE_RUN_ID", "").strip() or ( + f"portkey-{uuid4().hex[:10]}" + ) -def model_name() -> str: - return os.getenv("PORTKEY_MODEL") or os.getenv("RESPAN_MODEL", DEFAULT_MODEL) +def execution_id() -> str: + return uuid4().hex[:10] def workflow_name(example_name: str) -> str: - normalized_name = example_name.replace("-", "_") - return f"portkey_{normalized_name}" - + return f"portkey_{example_name.replace('-', '_')}" -def make_custom_identifier(example_name: str) -> str: - return f"portkey-{example_name}-{uuid4().hex[:8]}" - -def make_respan(example_name: str) -> Respan: +def make_respan(example_name: str, run_marker: str) -> Respan: + load_root_env() return Respan( - api_key=respan_api_key(), - base_url=respan_base_url(), - app_name="portkey-examples", + api_key=os.environ["RESPAN_API_KEY"], + base_url=os.getenv("RESPAN_BASE_URL", DEFAULT_RESPAN_BASE_URL), + app_name=workflow_name(example_name), instrumentations=[PortkeyInstrumentor()], - environment=os.getenv("RESPAN_ENVIRONMENT", "example"), - metadata={"integration": "portkey", "example": example_name}, + is_batching_enabled=False, + metadata={ + "example_set": EXAMPLE_SET, + "workflow_name": workflow_name(example_name), + "example_run_id": run_marker, + "run_id": run_marker, + }, + log_level=os.getenv("RESPAN_LOG_LEVEL", "WARNING"), ) -def _portkey_client_kwargs() -> dict[str, object]: - load_root_env() - portkey_api_key = env_value("PORTKEY_API_KEY") - if portkey_api_key: - kwargs: dict[str, object] = {"api_key": portkey_api_key} - if portkey_base_url := env_value("PORTKEY_BASE_URL"): - kwargs["base_url"] = portkey_base_url.rstrip("/") - if portkey_provider := env_value("PORTKEY_PROVIDER"): - kwargs["provider"] = portkey_provider - if portkey_config := env_value("PORTKEY_CONFIG"): - kwargs["config"] = portkey_config - return kwargs - - if env_value("PORTKEY_EXAMPLE_USE_LIVE_GATEWAY"): - return { - "api_key": gateway_api_key(), - "base_url": gateway_base_url(), - } +def live_configured() -> bool: + return bool(os.getenv("PORTKEY_API_KEY")) - if openai_api_key := env_value("OPENAI_API_KEY"): - if env_value("PORTKEY_EXAMPLE_USE_OPENAI"): - return { - "api_key": openai_api_key, - "base_url": env_value("OPENAI_BASE_URL") or "https://api.openai.com/v1", - } +def _client_kwargs(*, live: bool) -> dict[str, object]: + load_root_env() + if live: + if not live_configured(): + raise RuntimeError( + "PORTKEY_API_KEY is required for the optional live example" + ) + kwargs: dict[str, object] = {"api_key": os.environ["PORTKEY_API_KEY"]} + if base_url := os.getenv("PORTKEY_BASE_URL"): + kwargs["base_url"] = base_url.rstrip("/") + if provider := os.getenv("PORTKEY_PROVIDER"): + kwargs["provider"] = provider + if config := os.getenv("PORTKEY_CONFIG"): + kwargs["config"] = config + return kwargs return { "api_key": "local-portkey-example-key", "base_url": local_gateway_base_url(), } -def make_client() -> Portkey: - return Portkey(**_portkey_client_kwargs()) +def make_client(*, live: bool = False) -> Portkey: + return Portkey(**_client_kwargs(live=live)) -def make_async_client() -> AsyncPortkey: - return AsyncPortkey(**_portkey_client_kwargs()) +def make_async_client(*, live: bool = False) -> AsyncPortkey: + return AsyncPortkey(**_client_kwargs(live=live)) -def client_mode() -> str: - if env_value("PORTKEY_API_KEY"): - return "portkey-gateway" - if env_value("PORTKEY_EXAMPLE_USE_LIVE_GATEWAY"): - return "respan-gateway" - if env_value("PORTKEY_EXAMPLE_USE_OPENAI") and env_value("OPENAI_API_KEY"): - return "openai-api" - return "local-openai-compatible" +def model_name(*, live: bool = False) -> str: + if live: + return os.getenv("PORTKEY_MODEL") or os.getenv("RESPAN_MODEL", DEFAULT_MODEL) + return "local-portkey-model" @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, run_marker: str, execution: str, *, mode: str +): current_workflow_name = workflow_name(example_name) with propagate_attributes( - custom_identifier=custom_identifier, + custom_identifier=f"{current_workflow_name}-{execution}", trace_group_identifier=current_workflow_name, metadata={ + "example_set": EXAMPLE_SET, "example": example_name, - "run_id": custom_identifier, "workflow_name": current_workflow_name, + "example_run_id": run_marker, + "run_id": run_marker, + "execution_id": execution, + "mode": mode, }, ): - yield custom_identifier + yield + + +def print_result(example_name: str, run_marker: str, result: Any) -> None: + print(f"RESPAN_EXAMPLE_RUN_ID={run_marker}") + print(f"\n== {example_name} ==") + print(json.dumps(result, allow_nan=False, indent=2, sort_keys=True)) -def print_result(example_name: str, custom_identifier: str, text: str) -> None: - print(f"example={example_name}") - print(f"custom_identifier={custom_identifier}") - 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: + try: + respan.shutdown() + finally: + shutdown_local_gateway() diff --git a/python/tracing/portkey/requirements.txt b/python/tracing/portkey/requirements.txt index e5b192f..ea85dd2 100644 --- a/python/tracing/portkey/requirements.txt +++ b/python/tracing/portkey/requirements.txt @@ -1,4 +1,6 @@ -respan-ai -respan-instrumentation-portkey -portkey-ai -python-dotenv +portkey-ai>=2.3.4,<3 +openinference-instrumentation-portkey>=0.1.13,<0.2 +python-dotenv>=1,<2 +respan-ai>=4.1,<5 +respan-instrumentation-openinference>=1.2.2,<2 +respan-instrumentation-portkey>=0.1,<1 diff --git a/python/tracing/portkey/run_all.py b/python/tracing/portkey/run_all.py new file mode 100644 index 0000000..2a36342 --- /dev/null +++ b/python/tracing/portkey/run_all.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +import os +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path + +EXAMPLE_DIR = Path(__file__).resolve().parent +SCRIPTS = ( + "01_chat_completion.py", + "02_async_chat_completion.py", + "03_streaming_chat.py", + "04_tool_calling.py", + "05_expected_error.py", + "06_live_portkey.py", +) +DEFAULT_TIMEOUT_SECONDS = 120.0 + + +def main() -> None: + run_marker = os.getenv("RESPAN_EXAMPLE_RUN_ID", "").strip() or ( + "portkey-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + ) + timeout = float( + os.getenv("RESPAN_EXAMPLE_TIMEOUT_SECONDS", DEFAULT_TIMEOUT_SECONDS) + ) + env = os.environ.copy() + env["RESPAN_EXAMPLE_RUN_ID"] = run_marker + failures: list[str] = [] + for script in SCRIPTS: + print(f"\n### Running {script}", flush=True) + try: + completed = subprocess.run( + [sys.executable, str(EXAMPLE_DIR / script)], + check=False, + env=env, + timeout=timeout, + ) + except subprocess.TimeoutExpired: + failures.append(f"{script}: timed out after {timeout:g}s") + continue + if completed.returncode: + failures.append(f"{script}: exited {completed.returncode}") + if failures: + raise SystemExit("Portkey example failures:\n- " + "\n- ".join(failures)) + print(f"\nCompleted Portkey examples: RESPAN_EXAMPLE_RUN_ID={run_marker}") + + +if __name__ == "__main__": + main() diff --git a/python/tracing/portkey/test_contract.py b/python/tracing/portkey/test_contract.py new file mode 100644 index 0000000..37a2ecf --- /dev/null +++ b/python/tracing/portkey/test_contract.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import ast +import os +import subprocess +from pathlib import Path +from types import SimpleNamespace + +import _shared +import pytest +import run_all + +EXAMPLE_DIR = Path(__file__).resolve().parent + + +def _signatures(filename: str) -> list[list[str]]: + module = ast.parse((EXAMPLE_DIR / filename).read_text()) + return [ + [argument.arg for argument in node.args.args] + for node in ast.walk(module) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and any( + isinstance(decorator, ast.Call) + and isinstance(decorator.func, ast.Name) + and decorator.func.id == "workflow" + for decorator in node.decorator_list + ) + ] + + +def test_roots_accept_only_semantic_inputs(): + for script in run_all.SCRIPTS: + assert _signatures(script) == [["prompt"]] or ( + script == "04_tool_calling.py" and _signatures(script) == [["city"]] + ) + + +def test_exact_marker_survives_dotenv(monkeypatch): + monkeypatch.setenv("RESPAN_EXAMPLE_RUN_ID", "shell-marker") + + def fake_load(_path, *, override): + assert override is False + os.environ.setdefault("RESPAN_EXAMPLE_RUN_ID", "dotenv-marker") + os.environ.setdefault("RESPAN_API_KEY", "test-key") + + monkeypatch.setattr(_shared, "load_dotenv", fake_load) + _shared.load_root_env() + assert _shared.marker() == "shell-marker" + + +def test_metadata_uses_exact_marker(monkeypatch): + captured: dict[str, object] = {} + + class FakeRespan: + def __init__(self, **kwargs): + captured.update(kwargs) + + monkeypatch.setattr(_shared, "load_root_env", lambda: None) + monkeypatch.setattr(_shared, "Respan", FakeRespan) + monkeypatch.setenv("RESPAN_API_KEY", "test-key") + _shared.make_respan("contract", "exact-marker") + assert captured["metadata"]["example_run_id"] == "exact-marker" + assert captured["metadata"]["run_id"] == "exact-marker" + + +def test_deterministic_clients_ignore_live_credentials(monkeypatch): + monkeypatch.setattr(_shared, "load_root_env", lambda: None) + monkeypatch.setenv("PORTKEY_API_KEY", "must-not-be-used") + monkeypatch.setattr(_shared, "local_gateway_base_url", lambda: "http://127.0.0.1:9") + assert _shared._client_kwargs(live=False) == { + "api_key": "local-portkey-example-key", + "base_url": "http://127.0.0.1:9", + } + + +def test_runner_continues_and_aggregates(monkeypatch): + calls: list[str] = [] + + def fake_run(command, **kwargs): + name = Path(command[-1]).name + calls.append(name) + assert kwargs["env"]["RESPAN_EXAMPLE_RUN_ID"] == "runner-marker" + if name == "timeout.py": + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + return SimpleNamespace(returncode=1 if name == "first.py" else 0) + + monkeypatch.setattr(run_all, "SCRIPTS", ("first.py", "timeout.py", "last.py")) + monkeypatch.setattr(run_all.subprocess, "run", fake_run) + monkeypatch.setenv("RESPAN_EXAMPLE_RUN_ID", "runner-marker") + with pytest.raises(SystemExit) as caught: + run_all.main() + assert calls == ["first.py", "timeout.py", "last.py"] + assert "first.py: exited 1" in str(caught.value) + assert "timeout.py: timed out" in str(caught.value) + + +def test_runner_has_complete_set(): + assert len(run_all.SCRIPTS) == 6 + for script in run_all.SCRIPTS: + assert (EXAMPLE_DIR / script).is_file()