diff --git a/python/tracing/mirascope/01_call_and_tool.py b/python/tracing/mirascope/01_call_and_tool.py new file mode 100644 index 0000000..ab85885 --- /dev/null +++ b/python/tracing/mirascope/01_call_and_tool.py @@ -0,0 +1,65 @@ +"""Run a real Mirascope Model.call and Toolkit.execute with deterministic data.""" + +from __future__ import annotations + +import json + +from _shared import ( + close_model_provider, + create_deterministic_model, + create_respan, + finish_respan, + workflow_attributes, +) +from mirascope import llm +from respan import workflow + +WORKFLOW_NAME = "mirascope-call-and-tool" + + +@llm.tool +def lookup_weather(city: str) -> dict[str, object]: + """Return deterministic weather for a city.""" + return {"city": city, "temperature_c": 18, "conditions": "sunny"} + + +def create_runner(model: llm.Model): + @workflow(name=WORKFLOW_NAME) + def run_call_and_tool(city: str) -> dict[str, object]: + response = model.call( + f"Use lookup_weather for {city}.", + tools=[lookup_weather], + ) + outputs = response.execute_tools() + return { + "assistant_tool_calls": [ + {"id": call.id, "name": call.name, "args": json.loads(call.args)} + for call in response.tool_calls + ], + "tool_results": [output.result for output in outputs], + } + + return run_call_and_tool + + +def main() -> None: + respan = create_respan(WORKFLOW_NAME) + model: llm.Model | None = None + try: + model = create_deterministic_model() + runner = create_runner(model) + with respan.propagate_attributes( + **workflow_attributes(WORKFLOW_NAME, "01_call_and_tool.py") + ): + result = runner("Paris") + print(json.dumps(result, sort_keys=True)) + finally: + try: + if model is not None: + close_model_provider(model) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/mirascope/02_sync_async_stream.py b/python/tracing/mirascope/02_sync_async_stream.py new file mode 100644 index 0000000..f1637c2 --- /dev/null +++ b/python/tracing/mirascope/02_sync_async_stream.py @@ -0,0 +1,65 @@ +"""Consume real Mirascope sync and async stream response objects.""" + +from __future__ import annotations + +import asyncio +import json + +from _shared import ( + close_model_provider, + create_deterministic_model, + create_respan, + finish_respan, + workflow_attributes, +) +from mirascope import llm +from respan import workflow + +SYNC_WORKFLOW = "mirascope-sync-stream" +ASYNC_WORKFLOW = "mirascope-async-stream" + + +def create_sync_runner(model: llm.Model): + @workflow(name=SYNC_WORKFLOW) + def run_sync_stream(prompt: str) -> str: + response = model.stream(prompt) + return "".join(response.text_stream()).strip() + + return run_sync_stream + + +def create_async_runner(model: llm.Model): + @workflow(name=ASYNC_WORKFLOW) + async def run_async_stream(prompt: str) -> str: + response = await model.stream_async(prompt) + return "".join([part async for part in response.text_stream()]).strip() + + return run_async_stream + + +async def main() -> None: + respan = create_respan("mirascope-sync-async-stream") + model: llm.Model | None = None + try: + model = create_deterministic_model() + sync_runner = create_sync_runner(model) + async_runner = create_async_runner(model) + with respan.propagate_attributes( + **workflow_attributes(SYNC_WORKFLOW, "02_sync_async_stream.py") + ): + sync_result = sync_runner("Stream the deterministic sync reply.") + with respan.propagate_attributes( + **workflow_attributes(ASYNC_WORKFLOW, "02_sync_async_stream.py") + ): + async_result = await async_runner("Stream the deterministic async reply.") + print(json.dumps({"sync": sync_result, "async": async_result}, sort_keys=True)) + finally: + try: + if model is not None: + close_model_provider(model) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/tracing/mirascope/03_expected_error.py b/python/tracing/mirascope/03_expected_error.py new file mode 100644 index 0000000..7bf558a --- /dev/null +++ b/python/tracing/mirascope/03_expected_error.py @@ -0,0 +1,52 @@ +"""Raise a deterministic Mirascope provider error across the workflow boundary.""" + +from __future__ import annotations + +from _shared import ( + close_model_provider, + create_deterministic_model, + create_respan, + finish_respan, + workflow_attributes, +) +from mirascope import llm +from respan import workflow + +WORKFLOW_NAME = "mirascope-expected-provider-error" + + +def create_runner(model: llm.Model): + @workflow(name=WORKFLOW_NAME) + def run_expected_error(prompt: str) -> None: + model.call(prompt) + + return run_expected_error + + +def main() -> None: + respan = create_respan(WORKFLOW_NAME) + model: llm.Model | None = None + try: + model = create_deterministic_model(fail_status=503) + runner = create_runner(model) + try: + with respan.propagate_attributes( + **workflow_attributes(WORKFLOW_NAME, "03_expected_error.py") + ): + runner("Raise the deterministic provider error.") + except llm.ServerError as exc: + if exc.status_code != 503: + raise AssertionError(f"unexpected status: {exc.status_code}") from exc + print(f"expected failure ({exc.status_code}): {exc}") + else: + raise AssertionError("expected deterministic Mirascope failure") + finally: + try: + if model is not None: + close_model_provider(model) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/mirascope/04_privacy.py b/python/tracing/mirascope/04_privacy.py new file mode 100644 index 0000000..93bbd61 --- /dev/null +++ b/python/tracing/mirascope/04_privacy.py @@ -0,0 +1,53 @@ +"""Run a Mirascope call with content capture disabled.""" + +from __future__ import annotations + +import json + +from _shared import ( + close_model_provider, + create_deterministic_model, + create_respan, + finish_respan, + workflow_attributes, +) +from mirascope import llm +from respan import workflow + +WORKFLOW_NAME = "mirascope-content-disabled" + + +def create_runner(model: llm.Model): + @workflow(name=WORKFLOW_NAME) + def run_private_call(scenario: str) -> dict[str, object]: + response = model.call("private-example-content") + return { + "capture_content": False, + "response_received": bool(response.text()), + "scenario": scenario, + } + + return run_private_call + + +def main() -> None: + respan = create_respan(WORKFLOW_NAME, capture_content=False) + model: llm.Model | None = None + try: + model = create_deterministic_model() + runner = create_runner(model) + with respan.propagate_attributes( + **workflow_attributes(WORKFLOW_NAME, "04_privacy.py") + ): + result = runner("content-capture-disabled") + print(json.dumps(result, sort_keys=True)) + finally: + try: + if model is not None: + close_model_provider(model) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/mirascope/05_live_gateway.py b/python/tracing/mirascope/05_live_gateway.py new file mode 100644 index 0000000..3ae23c0 --- /dev/null +++ b/python/tracing/mirascope/05_live_gateway.py @@ -0,0 +1,52 @@ +"""Run Mirascope's OpenAI provider against the configured Respan gateway.""" + +from __future__ import annotations + +from _shared import ( + close_model_provider, + create_live_model, + create_respan, + finish_respan, + live_example_enabled, + workflow_attributes, +) +from mirascope import llm +from respan import workflow + +WORKFLOW_NAME = "mirascope-live-gateway" + + +def create_runner(model: llm.Model): + @workflow(name=WORKFLOW_NAME) + def run_live_call(prompt: str) -> str: + response = model.call(prompt) + return response.text() + + return run_live_call + + +def main() -> None: + if not live_example_enabled(): + print("skipped live gateway; RESPAN_MIRASCOPE_RUN_LIVE=0") + return + + respan = create_respan(WORKFLOW_NAME) + model: llm.Model | None = None + try: + model = create_live_model() + runner = create_runner(model) + with respan.propagate_attributes( + **workflow_attributes(WORKFLOW_NAME, "05_live_gateway.py") + ): + result = runner("Reply with exactly: Mirascope live tracing works.") + print(result) + finally: + try: + if model is not None: + close_model_provider(model) + finally: + finish_respan(respan) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/mirascope/README.md b/python/tracing/mirascope/README.md new file mode 100644 index 0000000..cd98e77 --- /dev/null +++ b/python/tracing/mirascope/README.md @@ -0,0 +1,43 @@ +# Mirascope tracing examples + +These examples exercise Mirascope 2.x model calls, sync and async streams, +tool-call execution, privacy mode, deterministic provider errors, and an +OpenAI-compatible live call through the Respan gateway. + +The scripts load `RESPAN_API_KEY`, `RESPAN_BASE_URL`, and optional gateway +overrides from the repository-root `.env`. Every script adds the exact +`RESPAN_EXAMPLE_RUN_ID` to its trace metadata and explicitly flushes and shuts +down Respan. + +Install the dependencies: + +```bash +cd python/tracing/mirascope +pip install -r requirements.txt +``` + +When validating an unpublished branch, link the local packages: + +```bash +pip install -e ../../../../respan/python-sdks/respan-sdk +pip install -e ../../../../respan/python-sdks/respan-tracing +pip install -e ../../../../respan/python-sdks/respan +pip install -e ../../../../respan/python-sdks/instrumentations/respan-instrumentation-mirascope +``` + +Run the complete set with one marker: + +```bash +RESPAN_EXAMPLE_RUN_ID=otel2-fix-py-group-19-YYYYMMDDTHHMMSSZ python run_all.py +``` + +`05_live_gateway.py` runs by default with the repository credentials. Set +`RESPAN_MIRASCOPE_RUN_LIVE=0` to skip only that optional provider-backed call. + +| Script | Coverage | +| --- | --- | +| `01_call_and_tool.py` | Real `Model.call`, `ToolCall`, and `Toolkit.execute` objects | +| `02_sync_async_stream.py` | Consumed real sync and async stream objects plus usage | +| `03_expected_error.py` | Exact provider 503 escaping the workflow boundary | +| `04_privacy.py` | Content capture disabled while model, usage, and status remain | +| `05_live_gateway.py` | Mirascope OpenAI provider through the Respan gateway | diff --git a/python/tracing/mirascope/_shared.py b/python/tracing/mirascope/_shared.py new file mode 100644 index 0000000..56f4c2f --- /dev/null +++ b/python/tracing/mirascope/_shared.py @@ -0,0 +1,269 @@ +"""Shared setup and deterministic Mirascope 2.x provider for the examples.""" + +from __future__ import annotations + +import inspect +import os +from collections.abc import AsyncIterator, Iterator, Sequence +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from uuid import uuid4 + +from dotenv import load_dotenv +from mirascope import llm +from respan import Respan +from respan_instrumentation_mirascope import MirascopeInstrumentor + +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" +DEFAULT_RUN_ID = datetime.now(timezone.utc).strftime("mirascope-%Y%m%dT%H%M%SZ") + + +def load_example_env() -> tuple[str, str, str, str]: + """Load Respan export and OpenAI-compatible gateway settings.""" + load_dotenv(REPO_ROOT / ".env", override=False) + api_key = os.getenv("RESPAN_API_KEY") + if not api_key: + raise RuntimeError(f"RESPAN_API_KEY is required in {REPO_ROOT / '.env'}") + base_url = os.getenv("RESPAN_BASE_URL", DEFAULT_RESPAN_BASE_URL) + gateway_key = os.getenv("RESPAN_GATEWAY_API_KEY", api_key) + gateway_url = os.getenv("RESPAN_GATEWAY_BASE_URL", base_url) + return api_key, base_url, gateway_key, gateway_url + + +def create_respan(app_name: str, *, capture_content: bool = True) -> Respan: + """Create a Respan runtime linked to the local Mirascope instrumentor.""" + api_key, base_url, _gateway_key, _gateway_url = load_example_env() + run_id = os.getenv("RESPAN_EXAMPLE_RUN_ID", DEFAULT_RUN_ID) + return Respan( + api_key=api_key, + base_url=base_url, + app_name=app_name, + instrumentations=[ + MirascopeInstrumentor(capture_content=capture_content), + ], + metadata={ + "integration": "mirascope", + "example": app_name, + "example_run_id": run_id, + "run_id": run_id, + }, + environment="examples", + is_batching_enabled=False, + ) + + +def workflow_attributes(workflow_name: str, script_name: str) -> dict[str, object]: + """Return stable grouping plus a unique root identifier for one example.""" + run_id = os.getenv("RESPAN_EXAMPLE_RUN_ID", DEFAULT_RUN_ID) + return { + "trace_group_identifier": workflow_name, + "custom_identifier": f"{workflow_name}-{uuid4().hex[:8]}", + "metadata": { + "example": "mirascope", + "example_run_id": run_id, + "run_id": run_id, + "script": script_name, + "workflow_name": workflow_name, + }, + } + + +def finish_respan(respan: Respan) -> None: + """Force pending spans out and release instrumentation patches.""" + try: + respan.flush() + finally: + respan.shutdown() + + +def _assistant_message( + *, + model_id: str, + content: str | llm.ToolCall, +) -> Any: + return llm.messages.assistant( + content, + provider_id="deterministic", + model_id=model_id, + provider_model_name=model_id.split("/", 1)[-1], + ) + + +class DeterministicProvider: + """Small provider that returns real Mirascope response and stream objects.""" + + id = "deterministic" + default_scope = "deterministic/" + + def __init__(self, *, fail_status: int | None = None) -> None: + self.fail_status = fail_status + self.client = None + + def _raise_if_requested(self) -> None: + if self.fail_status is not None: + raise llm.ServerError( + "deterministic Mirascope provider failure", + provider="deterministic", + status_code=self.fail_status, + ) + + def call( + self, + *, + model_id: str, + messages: Sequence[Any], + toolkit: Any, + format: Any = None, + **params: Any, + ) -> llm.Response: + self._raise_if_requested() + content: str | llm.ToolCall + if toolkit.tools: + content = llm.ToolCall( + id="weather-call-1", + name="lookup_weather", + args='{"city":"Paris"}', + ) + else: + content = "Mirascope deterministic response." + return llm.Response( + raw={"fixture": "deterministic"}, + provider_id="deterministic", + model_id=model_id, + provider_model_name=model_id.split("/", 1)[-1], + params=params, + tools=toolkit, + format=format, + input_messages=messages, + assistant_message=_assistant_message(model_id=model_id, content=content), + finish_reason=None, + usage=llm.Usage(input_tokens=24, output_tokens=10), + ) + + async def call_async(self, **kwargs: Any) -> llm.AsyncResponse: + response = self.call(**kwargs) + return llm.AsyncResponse( + raw=response.raw, + provider_id=response.provider_id, + model_id=response.model_id, + provider_model_name=response.provider_model_name, + params=response.params, + tools=response.toolkit, + format=response.format, + input_messages=response.messages[:-1], + assistant_message=response.messages[-1], + finish_reason=response.finish_reason, + usage=response.usage, + ) + + def stream( + self, + *, + model_id: str, + messages: Sequence[Any], + toolkit: Any, + format: Any = None, + **params: Any, + ) -> llm.StreamResponse: + self._raise_if_requested() + chunks: Iterator[Any] = iter( + [ + llm.TextStartChunk(), + llm.TextChunk(delta="Mirascope streaming works."), + llm.TextEndChunk(), + llm.UsageDeltaChunk(input_tokens=12, output_tokens=6), + ] + ) + return llm.StreamResponse( + provider_id="deterministic", + model_id=model_id, + provider_model_name=model_id.split("/", 1)[-1], + params=params, + tools=toolkit, + format=format, + input_messages=messages, + chunk_iterator=chunks, + ) + + async def stream_async( + self, + *, + model_id: str, + messages: Sequence[Any], + toolkit: Any, + format: Any = None, + **params: Any, + ) -> llm.AsyncStreamResponse: + self._raise_if_requested() + + async def chunks() -> AsyncIterator[Any]: + yield llm.TextStartChunk() + yield llm.TextChunk(delta="Async Mirascope streaming works.") + yield llm.TextEndChunk() + yield llm.UsageDeltaChunk(input_tokens=8, output_tokens=4) + + return llm.AsyncStreamResponse( + provider_id="deterministic", + model_id=model_id, + provider_model_name=model_id.split("/", 1)[-1], + params=params, + tools=toolkit, + format=format, + input_messages=messages, + chunk_iterator=chunks(), + ) + + +def create_deterministic_model(*, fail_status: int | None = None) -> llm.Model: + """Register the deterministic provider and return a real Mirascope model.""" + provider = DeterministicProvider(fail_status=fail_status) + llm.register_provider(provider, scope="deterministic/") + return llm.Model("deterministic/mirascope-2x") + + +def create_live_model() -> llm.Model: + """Register Mirascope's OpenAI provider against the configured gateway.""" + from mirascope.llm.providers import OpenAIProvider + + _api_key, _base_url, gateway_key, gateway_url = load_example_env() + configured_model = os.getenv("RESPAN_MODEL", DEFAULT_MODEL) + model_id = ( + configured_model if "/" in configured_model else f"openai/{configured_model}" + ) + if not model_id.endswith(":completions"): + model_id = f"{model_id}:completions" + provider = OpenAIProvider(api_key=gateway_key, base_url=gateway_url) + llm.register_provider(provider, scope="openai/") + return llm.Model(model_id, max_tokens=64, temperature=0) + + +def live_example_enabled() -> bool: + """Allow the live gateway call to be disabled explicitly.""" + load_example_env() + return os.getenv("RESPAN_MIRASCOPE_RUN_LIVE", "1").strip().lower() not in { + "0", + "false", + "no", + } + + +def close_model_provider(model: llm.Model) -> None: + """Close Mirascope provider clients when they expose synchronous close hooks.""" + provider = model.provider + candidates = [ + getattr(provider, "client", None), + getattr(getattr(provider, "_completions_provider", None), "client", None), + getattr(getattr(provider, "_responses_provider", None), "client", None), + ] + seen: set[int] = set() + for candidate in candidates: + if candidate is None or id(candidate) in seen: + continue + seen.add(id(candidate)) + close = getattr(candidate, "close", None) + if callable(close) and not inspect.iscoroutinefunction(close): + close() diff --git a/python/tracing/mirascope/requirements.txt b/python/tracing/mirascope/requirements.txt new file mode 100644 index 0000000..5b8f230 --- /dev/null +++ b/python/tracing/mirascope/requirements.txt @@ -0,0 +1,4 @@ +mirascope[openai]>=2.5.0,<3.0.0 +python-dotenv>=1.0.0 +respan-ai>=4.1.0 +respan-instrumentation-mirascope>=0.1.0 diff --git a/python/tracing/mirascope/run_all.py b/python/tracing/mirascope/run_all.py new file mode 100644 index 0000000..2652291 --- /dev/null +++ b/python/tracing/mirascope/run_all.py @@ -0,0 +1,47 @@ +"""Run every Mirascope tracing example with one exact batch marker.""" + +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_call_and_tool.py", + "02_sync_async_stream.py", + "03_expected_error.py", + "04_privacy.py", + "05_live_gateway.py", +) + + +def main() -> None: + env = os.environ.copy() + env.setdefault( + "RESPAN_EXAMPLE_RUN_ID", + datetime.now(timezone.utc).strftime("mirascope-suite-%Y%m%dT%H%M%SZ"), + ) + print(f"RESPAN_EXAMPLE_RUN_ID={env['RESPAN_EXAMPLE_RUN_ID']}", flush=True) + failures: list[tuple[str, int]] = [] + for script in SCRIPTS: + print(f"\n=== {script} ===", flush=True) + result = subprocess.run( + [sys.executable, str(EXAMPLE_DIR / script)], + cwd=EXAMPLE_DIR, + env=env, + check=False, + ) + print(f"exit_code={result.returncode} script={script}", flush=True) + if result.returncode: + failures.append((script, result.returncode)) + + if failures: + rendered = ", ".join(f"{script} ({code})" for script, code in failures) + raise SystemExit(f"Mirascope example failures: {rendered}") + + +if __name__ == "__main__": + main() diff --git a/python/tracing/mistralai/01_chat_completion.py b/python/tracing/mistralai/01_chat_completion.py index aae8f5f..dfb84f4 100644 --- a/python/tracing/mistralai/01_chat_completion.py +++ b/python/tracing/mistralai/01_chat_completion.py @@ -1,35 +1,35 @@ from __future__ import annotations -from respan import workflow - from _shared import ( content_to_text, example_attributes, + finish_respan, make_client, make_custom_identifier, make_respan, - model_name, print_result, + print_start, + root_request, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "chat-completion" +PROMPT = "Reply with one concise sentence about tracing Mistral AI apps." + +def _build_chat_completion_workflow(client): + @workflow(name=workflow_name(EXAMPLE_NAME)) + def run(request: dict[str, str]) -> str: + response = client.chat.complete( + model=request["model"], + messages=[{"role": "user", "content": request["prompt"]}], + temperature=0.1, + max_tokens=80, + ) + return content_to_text(response.choices[0].message.content) -@workflow(name=workflow_name(EXAMPLE_NAME)) -def _chat_completion_workflow(client) -> str: - response = client.chat.complete( - model=model_name(), - messages=[ - { - "role": "user", - "content": "Reply with one concise sentence about tracing Mistral AI apps.", - } - ], - temperature=0.1, - max_tokens=80, - ) - return content_to_text(response.choices[0].message.content) + return run def run_chat_completion() -> None: @@ -38,13 +38,16 @@ def run_chat_completion() -> None: text = "" try: - with make_client() as client: - 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) + with ( + make_client() as client, + example_attributes(EXAMPLE_NAME, custom_identifier), + ): + print_start(EXAMPLE_NAME, custom_identifier) + text = _build_chat_completion_workflow(client)( + root_request(EXAMPLE_NAME, PROMPT) + ) finally: - respan.shutdown() + finish_respan(respan) print_result(EXAMPLE_NAME, custom_identifier, text) diff --git a/python/tracing/mistralai/02_multi_turn_chat.py b/python/tracing/mistralai/02_multi_turn_chat.py index 0e6ef5f..6631a7e 100644 --- a/python/tracing/mistralai/02_multi_turn_chat.py +++ b/python/tracing/mistralai/02_multi_turn_chat.py @@ -1,47 +1,49 @@ from __future__ import annotations -from respan import workflow - from _shared import ( content_to_text, example_attributes, + finish_respan, make_client, make_custom_identifier, make_respan, - model_name, print_result, + print_start, + root_request, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "multi-turn-chat" +PROMPT = "Now make that advice specific to Mistral AI apps." + +def _build_multi_turn_chat_workflow(client): + @workflow(name=workflow_name(EXAMPLE_NAME)) + def run(request: dict[str, str]) -> str: + response = client.chat.complete( + model=request["model"], + messages=[ + { + "role": "system", + "content": "You answer with concise observability advice.", + }, + { + "role": "user", + "content": "Name one reason traces help LLM applications.", + }, + { + "role": "assistant", + "content": "They reveal where latency, errors, and token use happen.", + }, + {"role": "user", "content": request["prompt"]}, + ], + temperature=0.1, + max_tokens=100, + ) + return content_to_text(response.choices[0].message.content) -@workflow(name=workflow_name(EXAMPLE_NAME)) -def _multi_turn_chat_workflow(client) -> str: - response = client.chat.complete( - model=model_name(), - messages=[ - { - "role": "system", - "content": "You answer with concise observability advice.", - }, - { - "role": "user", - "content": "Name one reason traces help LLM applications.", - }, - { - "role": "assistant", - "content": "They reveal where latency, errors, and token use happen.", - }, - { - "role": "user", - "content": "Now make that advice specific to Mistral AI apps.", - }, - ], - temperature=0.1, - max_tokens=100, - ) - return content_to_text(response.choices[0].message.content) + return run def run_multi_turn_chat() -> None: @@ -50,13 +52,20 @@ def run_multi_turn_chat() -> None: text = "" try: - with make_client() as client: - 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 = _multi_turn_chat_workflow(client) + with ( + make_client() as client, + example_attributes(EXAMPLE_NAME, custom_identifier), + ): + print_start(EXAMPLE_NAME, custom_identifier) + text = _build_multi_turn_chat_workflow(client)( + root_request( + EXAMPLE_NAME, + PROMPT, + prior_turns=3, + ) + ) finally: - respan.shutdown() + finish_respan(respan) print_result(EXAMPLE_NAME, custom_identifier, text) diff --git a/python/tracing/mistralai/03_async_chat_completion.py b/python/tracing/mistralai/03_async_chat_completion.py index 5a79c71..54e17a2 100644 --- a/python/tracing/mistralai/03_async_chat_completion.py +++ b/python/tracing/mistralai/03_async_chat_completion.py @@ -2,36 +2,36 @@ import asyncio -from respan import workflow - from _shared import ( content_to_text, example_attributes, + finish_respan, make_client, make_custom_identifier, make_respan, - model_name, print_result, + print_start, + root_request, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "async-chat-completion" +PROMPT = "Reply with one concise sentence about async tracing." + +def _build_async_chat_completion_workflow(client): + @workflow(name=workflow_name(EXAMPLE_NAME)) + async def run(request: dict[str, str]) -> str: + response = await client.chat.complete_async( + model=request["model"], + messages=[{"role": "user", "content": request["prompt"]}], + temperature=0.1, + max_tokens=80, + ) + return content_to_text(response.choices[0].message.content) -@workflow(name=workflow_name(EXAMPLE_NAME)) -async def _async_chat_completion_workflow(client) -> str: - response = await client.chat.complete_async( - model=model_name(), - messages=[ - { - "role": "user", - "content": "Reply with one concise sentence about async tracing.", - } - ], - temperature=0.1, - max_tokens=80, - ) - return content_to_text(response.choices[0].message.content) + return run async def run_async_chat_completion() -> None: @@ -42,11 +42,12 @@ async def run_async_chat_completion() -> None: try: async with make_client() as client: 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) + print_start(EXAMPLE_NAME, custom_identifier) + text = await _build_async_chat_completion_workflow(client)( + root_request(EXAMPLE_NAME, PROMPT) + ) finally: - respan.shutdown() + finish_respan(respan) print_result(EXAMPLE_NAME, custom_identifier, text) diff --git a/python/tracing/mistralai/04_sync_streaming.py b/python/tracing/mistralai/04_sync_streaming.py new file mode 100644 index 0000000..3caee34 --- /dev/null +++ b/python/tracing/mistralai/04_sync_streaming.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import json + +from _shared import ( + deterministic_stream_response, + example_attributes, + finish_respan, + make_custom_identifier, + make_mock_sync_client, + make_respan, + print_result, + print_start, + root_request, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "sync-streaming" +PROMPT = "Stream a concise sentence about Mistral observability." + + +def _response(request): + payload = json.loads(request.content) + if payload.get("stream") is not True: + raise RuntimeError("sync streaming fixture expected stream=true") + return deterministic_stream_response( + request, + fragments=("Sync Mistral ", "streaming stays observable."), + prompt_tokens=19, + completion_tokens=7, + ) + + +def _build_sync_streaming_workflow(client): + @workflow(name=workflow_name(EXAMPLE_NAME)) + def run(request: dict[str, str]) -> dict[str, object]: + stream = client.chat.stream( + model=request["model"], + messages=[{"role": "user", "content": request["prompt"]}], + max_tokens=80, + temperature=0, + ) + fragments = [ + event.data.choices[0].delta.content or "" + for event in stream + if event.data.choices + ] + return { + "content": "".join(fragments), + "fragments": len(fragments), + } + + return run + + +def run_sync_streaming() -> None: + respan = make_respan(EXAMPLE_NAME) + custom_identifier = make_custom_identifier(EXAMPLE_NAME) + result: dict[str, object] = {} + + try: + with ( + make_mock_sync_client(_response) as client, + example_attributes(EXAMPLE_NAME, custom_identifier), + ): + print_start(EXAMPLE_NAME, custom_identifier, "deterministic-current-sdk") + result = _build_sync_streaming_workflow(client)( + root_request(EXAMPLE_NAME, PROMPT, stream=True) + ) + finally: + finish_respan(respan) + + print_result( + EXAMPLE_NAME, + custom_identifier, + result, + "deterministic-current-sdk", + ) + + +if __name__ == "__main__": + run_sync_streaming() diff --git a/python/tracing/mistralai/05_async_streaming.py b/python/tracing/mistralai/05_async_streaming.py new file mode 100644 index 0000000..cd7936c --- /dev/null +++ b/python/tracing/mistralai/05_async_streaming.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import asyncio +import json + +from _shared import ( + deterministic_stream_response, + example_attributes, + finish_respan, + make_custom_identifier, + make_mock_async_client, + make_respan, + print_result, + print_start, + root_request, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "async-streaming" +PROMPT = "Stream a concise sentence about async Mistral tracing." + + +def _response(request): + payload = json.loads(request.content) + if payload.get("stream") is not True: + raise RuntimeError("async streaming fixture expected stream=true") + return deterministic_stream_response( + request, + fragments=("Async Mistral ", "streaming keeps complete telemetry."), + prompt_tokens=23, + completion_tokens=8, + ) + + +def _build_async_streaming_workflow(client): + @workflow(name=workflow_name(EXAMPLE_NAME)) + async def run(request: dict[str, str]) -> dict[str, object]: + stream = await client.chat.stream_async( + model=request["model"], + messages=[{"role": "user", "content": request["prompt"]}], + max_tokens=80, + temperature=0, + ) + fragments = [] + async for event in stream: + if event.data.choices: + fragments.append(event.data.choices[0].delta.content or "") + return { + "content": "".join(fragments), + "fragments": len(fragments), + } + + return run + + +async def run_async_streaming() -> None: + respan = make_respan(EXAMPLE_NAME) + custom_identifier = make_custom_identifier(EXAMPLE_NAME) + result: dict[str, object] = {} + + try: + async with make_mock_async_client(_response) as client: + with example_attributes(EXAMPLE_NAME, custom_identifier): + print_start( + EXAMPLE_NAME, custom_identifier, "deterministic-current-sdk" + ) + result = await _build_async_streaming_workflow(client)( + root_request(EXAMPLE_NAME, PROMPT, stream=True) + ) + finally: + finish_respan(respan) + + print_result( + EXAMPLE_NAME, + custom_identifier, + result, + "deterministic-current-sdk", + ) + + +if __name__ == "__main__": + asyncio.run(run_async_streaming()) diff --git a/python/tracing/mistralai/06_tool_calling.py b/python/tracing/mistralai/06_tool_calling.py new file mode 100644 index 0000000..7d9cf96 --- /dev/null +++ b/python/tracing/mistralai/06_tool_calling.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +import json +from typing import Any + +from _shared import ( + deterministic_chat_response, + example_attributes, + finish_respan, + make_custom_identifier, + make_mock_sync_client, + make_respan, + print_result, + print_start, + root_request, + workflow_name, +) +from respan import tool, workflow + +EXAMPLE_NAME = "tool-calling" +PROMPT = "What is the weather in Paris? Use the available tool." +TOOL_SCHEMA = { + "type": "function", + "function": { + "name": "get_weather", + "description": "Return deterministic weather for a city.", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"], + }, + }, +} + + +@tool(name="get_weather") +def get_weather(city: str) -> dict[str, Any]: + """Return deterministic weather for a city.""" + return {"city": city, "condition": "sunny", "temperature_c": 22} + + +def _response(request): + payload = json.loads(request.content) + if not payload.get("tools"): + raise RuntimeError("tool fixture expected a tool definition") + return deterministic_chat_response( + request, + content="", + prompt_tokens=27, + completion_tokens=9, + tool_calls=[ + { + "id": "call_weather_paris", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"city":"Paris"}', + }, + } + ], + ) + + +def _build_tool_calling_workflow(client): + @workflow(name=workflow_name(EXAMPLE_NAME)) + def run(request: dict[str, str]) -> dict[str, Any]: + response = client.chat.complete( + model=request["model"], + messages=[{"role": "user", "content": request["prompt"]}], + tools=[TOOL_SCHEMA], + tool_choice="auto", + temperature=0, + ) + tool_call = response.choices[0].message.tool_calls[0] + arguments = json.loads(tool_call.function.arguments) + tool_result = get_weather(**arguments) + return { + "tool_call": { + "id": tool_call.id, + "name": tool_call.function.name, + "arguments": arguments, + }, + "tool_result": tool_result, + } + + return run + + +def run_tool_calling() -> None: + respan = make_respan(EXAMPLE_NAME) + custom_identifier = make_custom_identifier(EXAMPLE_NAME) + result: dict[str, Any] = {} + + try: + with ( + make_mock_sync_client(_response) as client, + example_attributes(EXAMPLE_NAME, custom_identifier), + ): + print_start(EXAMPLE_NAME, custom_identifier, "deterministic-current-sdk") + result = _build_tool_calling_workflow(client)( + root_request(EXAMPLE_NAME, PROMPT, available_tools=["get_weather"]) + ) + finally: + finish_respan(respan) + + print_result( + EXAMPLE_NAME, + custom_identifier, + result, + "deterministic-current-sdk", + ) + + +if __name__ == "__main__": + run_tool_calling() diff --git a/python/tracing/mistralai/07_expected_provider_failure.py b/python/tracing/mistralai/07_expected_provider_failure.py new file mode 100644 index 0000000..6ca1740 --- /dev/null +++ b/python/tracing/mistralai/07_expected_provider_failure.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import httpx +from _shared import ( + example_attributes, + finish_respan, + make_custom_identifier, + make_mock_sync_client, + make_respan, + print_result, + print_start, + root_request, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "expected-provider-failure" +PROMPT = "Exercise a deterministic provider authentication failure." + + +def _response(request): + return httpx.Response( + 401, + json={"message": "invalid api key", "request_id": "fixture_request"}, + request=request, + ) + + +def _build_provider_failure_workflow(client): + @workflow(name=workflow_name(EXAMPLE_NAME)) + def run(request: dict[str, str]) -> None: + client.chat.complete( + model=request["model"], + messages=[{"role": "user", "content": request["prompt"]}], + ) + raise AssertionError("provider failure fixture unexpectedly succeeded") + + return run + + +def run_expected_provider_failure() -> None: + respan = make_respan(EXAMPLE_NAME) + custom_identifier = make_custom_identifier(EXAMPLE_NAME) + result: dict[str, object] = {} + + try: + with ( + make_mock_sync_client(_response) as client, + example_attributes(EXAMPLE_NAME, custom_identifier), + ): + print_start(EXAMPLE_NAME, custom_identifier, "deterministic-current-sdk") + try: + _build_provider_failure_workflow(client)( + root_request(EXAMPLE_NAME, PROMPT, expected_status=401) + ) + except Exception as exc: + status_code = getattr(exc, "status_code", None) + if status_code != 401: + raise + result = { + "expected": True, + "error_type": type(exc).__name__, + "status_code": status_code, + } + finally: + finish_respan(respan) + + print_result( + EXAMPLE_NAME, + custom_identifier, + result, + "deterministic-current-sdk", + ) + + +if __name__ == "__main__": + run_expected_provider_failure() diff --git a/python/tracing/mistralai/08_expected_application_failure.py b/python/tracing/mistralai/08_expected_application_failure.py new file mode 100644 index 0000000..46e7692 --- /dev/null +++ b/python/tracing/mistralai/08_expected_application_failure.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +from _shared import ( + example_attributes, + finish_respan, + make_custom_identifier, + make_mock_sync_client, + make_respan, + print_result, + print_start, + root_request, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "expected-application-failure" +PROMPT = "Exercise a deterministic Mistral application transport failure." + + +def _response(_request): + raise RuntimeError("deterministic Mistral transport failure") + + +def _build_application_failure_workflow(client): + @workflow(name=workflow_name(EXAMPLE_NAME)) + def run(request: dict[str, str]) -> None: + client.chat.complete( + model=request["model"], + messages=[{"role": "user", "content": request["prompt"]}], + ) + raise AssertionError("application failure fixture unexpectedly succeeded") + + return run + + +def run_expected_application_failure() -> None: + respan = make_respan(EXAMPLE_NAME) + custom_identifier = make_custom_identifier(EXAMPLE_NAME) + result: dict[str, object] = {} + + try: + with ( + make_mock_sync_client(_response) as client, + example_attributes(EXAMPLE_NAME, custom_identifier), + ): + print_start(EXAMPLE_NAME, custom_identifier, "deterministic-application") + try: + _build_application_failure_workflow(client)( + root_request(EXAMPLE_NAME, PROMPT, expected_status=500) + ) + except RuntimeError as exc: + result = { + "expected": True, + "error_type": type(exc).__name__, + "message": str(exc), + "status_code": 500, + } + finally: + finish_respan(respan) + + print_result( + EXAMPLE_NAME, + custom_identifier, + result, + "deterministic-application", + ) + + +if __name__ == "__main__": + run_expected_application_failure() diff --git a/python/tracing/mistralai/README.md b/python/tracing/mistralai/README.md index b48491d..77b38a6 100644 --- a/python/tracing/mistralai/README.md +++ b/python/tracing/mistralai/README.md @@ -9,6 +9,23 @@ Required for exporting traces: RESPAN_API_KEY=... ``` +Install the registry dependencies from this directory: + +```bash +python -m pip install -r requirements.txt +``` + +For repository development and validation, link the local packages after that +registry install: + +```bash +python -m pip install -e ../../../../respan/python-sdks/respan-sdk +python -m pip install -e ../../../../respan/python-sdks/respan-tracing +python -m pip install -e ../../../../respan/python-sdks/respan +python -m pip install -e ../../../../respan/python-sdks/instrumentations/respan-instrumentation-openinference +python -m pip install -e ../../../../respan/python-sdks/instrumentations/respan-instrumentation-mistralai +``` + For Mistral calls, use one of these options: ```bash @@ -37,8 +54,28 @@ Run one script at a time: python 01_chat_completion.py python 02_multi_turn_chat.py python 03_async_chat_completion.py +python 04_sync_streaming.py +python 05_async_streaming.py +python 06_tool_calling.py +python 07_expected_provider_failure.py +python 08_expected_application_failure.py ``` -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. +Or run the complete committed suite with one exact marker: + +```bash +RESPAN_EXAMPLE_RUN_ID=otel2-fix-py-group-19-YYYYMMDDTHHMMSSZ python run_all.py +``` + +When the variable is omitted, `run_all.py` creates and prints one parent marker, +then passes that same marker to all eight child processes. + +The first three examples use the configured live gateway (or direct Mistral +credentials). The stream, tool, and failure examples use the current Mistral +SDK with deterministic HTTP fixtures so their content, usage, tool calls, and +errors are repeatable while the resulting spans are still exported to Respan. + +Each script preserves `RESPAN_EXAMPLE_RUN_ID` in metadata as +`example_run_id`, and also emits a unique per-scenario `custom_identifier`. +Every workflow accepts bounded JSON-native scenario input; live SDK clients are +captured outside decorated signatures. diff --git a/python/tracing/mistralai/_shared.py b/python/tracing/mistralai/_shared.py index 0ce1e9a..9fdeb7c 100644 --- a/python/tracing/mistralai/_shared.py +++ b/python/tracing/mistralai/_shared.py @@ -1,11 +1,14 @@ from __future__ import annotations +import json import os +from collections.abc import Callable, Iterator from contextlib import contextmanager from pathlib import Path from typing import Any from uuid import uuid4 +import httpx from dotenv import load_dotenv from mistralai.client import Mistral from respan import Respan, propagate_attributes @@ -14,10 +17,12 @@ PROJECT_ROOT = Path(__file__).resolve().parents[3] DEFAULT_RESPAN_BASE_URL = "https://api.respan.ai/api" DEFAULT_MISTRAL_MODEL = "mistral/mistral-small" +EXAMPLE_SET = "mistralai" +DEFAULT_RUN_ID = f"mistralai-{uuid4().hex[:12]}" 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: @@ -47,13 +52,19 @@ def model_name() -> str: def make_respan(example_name: str) -> Respan: api_key = require_respan_api_key() + run_id = example_run_id() return Respan( api_key=api_key, base_url=respan_base_url(), app_name="mistralai-examples", instrumentations=[MistralAIInstrumentor()], environment=os.getenv("RESPAN_ENVIRONMENT", "example"), - metadata={"integration": "mistralai", "example": example_name}, + metadata={ + "integration": EXAMPLE_SET, + "example": example_name, + "example_run_id": run_id, + }, + is_batching_enabled=False, ) @@ -72,6 +83,109 @@ def make_client() -> Mistral: ) +def make_mock_sync_client( + handler: Callable[[httpx.Request], httpx.Response], +) -> Mistral: + """Create a current Mistral SDK client backed by a repeatable HTTP fixture.""" + return Mistral( + api_key="deterministic-example-key", + client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + + +def make_mock_async_client( + handler: Callable[[httpx.Request], httpx.Response], +) -> Mistral: + """Create an async current Mistral SDK client backed by an HTTP fixture.""" + + async def async_handler(request: httpx.Request) -> httpx.Response: + return handler(request) + + return Mistral( + api_key="deterministic-example-key", + async_client=httpx.AsyncClient(transport=httpx.MockTransport(async_handler)), + ) + + +def deterministic_chat_response( + request: httpx.Request, + *, + content: str, + prompt_tokens: int, + completion_tokens: int, + tool_calls: list[dict[str, Any]] | None = None, +) -> httpx.Response: + message: dict[str, Any] = {"role": "assistant", "content": content} + if tool_calls is not None: + message["tool_calls"] = tool_calls + return httpx.Response( + 200, + json={ + "id": "mistralai_example_completion", + "object": "chat.completion", + "model": model_name(), + "created": 1_710_000_000, + "usage": { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": prompt_tokens + completion_tokens, + }, + "choices": [ + { + "index": 0, + "message": message, + "finish_reason": "tool_calls" if tool_calls else "stop", + } + ], + }, + request=request, + ) + + +def deterministic_stream_response( + request: httpx.Request, + *, + fragments: tuple[str, ...], + prompt_tokens: int, + completion_tokens: int, +) -> httpx.Response: + chunks = [] + for index, fragment in enumerate(fragments): + final = index == len(fragments) - 1 + chunk: dict[str, Any] = { + "id": "mistralai_example_stream", + "object": "chat.completion.chunk", + "model": model_name(), + "created": 1_710_000_000, + "choices": [ + { + "index": 0, + "delta": { + **({"role": "assistant"} if index == 0 else {}), + "content": fragment, + }, + "finish_reason": "stop" if final else None, + } + ], + } + if final: + chunk["usage"] = { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": prompt_tokens + completion_tokens, + } + chunks.append(chunk) + + body = "".join(f"data: {json.dumps(chunk)}\n\n" for chunk in chunks) + body += "data: [DONE]\n\n" + return httpx.Response( + 200, + content=body, + headers={"content-type": "text/event-stream"}, + request=request, + ) + + def workflow_name(example_name: str) -> str: normalized_name = example_name.replace("-", "_") return f"mistralai_{normalized_name}" @@ -81,16 +195,28 @@ def make_custom_identifier(example_name: str) -> str: return f"mistralai-{example_name}-{uuid4().hex[:8]}" +def example_run_id() -> str: + """Return the exact shell marker, with a unique fallback for ad-hoc runs.""" + return os.getenv("RESPAN_EXAMPLE_RUN_ID") or DEFAULT_RUN_ID + + @contextmanager -def example_attributes(example_name: str, custom_identifier: str | None = None): +def example_attributes( + example_name: str, + custom_identifier: str | None = None, +) -> Iterator[str]: custom_identifier = custom_identifier or make_custom_identifier(example_name) current_workflow_name = workflow_name(example_name) + run_id = example_run_id() with propagate_attributes( custom_identifier=custom_identifier, trace_group_identifier=current_workflow_name, metadata={ "example": example_name, - "run_id": custom_identifier, + "example_set": EXAMPLE_SET, + "example_run_id": run_id, + "run_id": run_id, + "scenario_id": custom_identifier, "workflow_name": current_workflow_name, }, ): @@ -117,10 +243,46 @@ def content_to_text(content: Any) -> str: return str(content) -def print_result(example_name: str, custom_identifier: str, text: str) -> None: +def root_request(example_name: str, prompt: str, **details: Any) -> dict[str, Any]: + """Build bounded, JSON-native input for a decorated workflow boundary.""" + return { + "scenario": example_name, + "model": model_name(), + "prompt": prompt, + **details, + } + + +def print_start( + example_name: str, custom_identifier: str, mode: str | None = None +) -> None: + print(f"example_run_id={example_run_id()}", flush=True) + print(f"example={example_name}", flush=True) + print(f"workflow_name={workflow_name(example_name)}", flush=True) + print(f"custom_identifier={custom_identifier}", flush=True) + print(f"client_mode={mode or client_mode()}", flush=True) + + +def print_result( + example_name: str, + custom_identifier: str, + result: Any, + mode: str | None = None, +) -> None: print(f"example={example_name}") + print(f"example_run_id={example_run_id()}") print(f"workflow_name={workflow_name(example_name)}") print(f"custom_identifier={custom_identifier}") - print(f"client_mode={client_mode()}") + print(f"client_mode={mode or client_mode()}") print(f"model={model_name()}") - print(text.strip()) + if isinstance(result, str): + print(result.strip()) + else: + print(json.dumps(result, default=str, indent=2, sort_keys=True)) + + +def finish_respan(respan: Respan) -> None: + try: + respan.flush() + finally: + respan.shutdown() diff --git a/python/tracing/mistralai/requirements.txt b/python/tracing/mistralai/requirements.txt index cc933e2..bbfbaaa 100644 --- a/python/tracing/mistralai/requirements.txt +++ b/python/tracing/mistralai/requirements.txt @@ -1,8 +1,6 @@ --e ../../../../respan/python-sdks/respan-sdk --e ../../../../respan/python-sdks/respan-tracing --e ../../../../respan/python-sdks/respan --e ../../../../respan/python-sdks/instrumentations/respan-instrumentation-openinference --e ../../../../respan/python-sdks/instrumentations/respan-instrumentation-mistralai -mistralai>=2.0.0 -openinference-instrumentation-mistralai>=2.0.0 -python-dotenv +httpx>=0.28.0,<1.0.0 +mistralai>=2.9.3,<3.0.0 +openinference-instrumentation-mistralai>=2.0.6,<3.0.0 +python-dotenv>=1.0.0,<2.0.0 +respan-ai>=4.1.0 +respan-instrumentation-mistralai>=0.1.0 diff --git a/python/tracing/mistralai/run_all.py b/python/tracing/mistralai/run_all.py new file mode 100644 index 0000000..676ffd8 --- /dev/null +++ b/python/tracing/mistralai/run_all.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path +from uuid import uuid4 + +EXAMPLE_DIR = Path(__file__).resolve().parent +EXAMPLES = ( + "01_chat_completion.py", + "02_multi_turn_chat.py", + "03_async_chat_completion.py", + "04_sync_streaming.py", + "05_async_streaming.py", + "06_tool_calling.py", + "07_expected_provider_failure.py", + "08_expected_application_failure.py", +) + + +def main() -> None: + parent_marker = os.getenv("RESPAN_EXAMPLE_RUN_ID") or ( + f"mistralai-{uuid4().hex[:12]}" + ) + child_environment = os.environ.copy() + child_environment["RESPAN_EXAMPLE_RUN_ID"] = parent_marker + print(f"example_run_id={parent_marker}", flush=True) + + failures = [] + for filename in EXAMPLES: + print(f"\n=== {filename} ===", flush=True) + result = subprocess.run( + [sys.executable, str(EXAMPLE_DIR / filename)], + cwd=EXAMPLE_DIR, + env=child_environment, + check=False, + ) + if result.returncode: + failures.append((filename, result.returncode)) + + if failures: + rendered = ", ".join(f"{name} ({code})" for name, code in failures) + raise SystemExit(f"Mistral example failures: {rendered}") + + +if __name__ == "__main__": + main() diff --git a/python/tracing/ollama/01_chat.py b/python/tracing/ollama/01_chat.py index 8b15323..b5db504 100644 --- a/python/tracing/ollama/01_chat.py +++ b/python/tracing/ollama/01_chat.py @@ -1,9 +1,8 @@ from __future__ import annotations -from respan import workflow - from _shared import ( example_attributes, + flush_and_shutdown, make_client, make_custom_identifier, make_respan, @@ -12,24 +11,26 @@ response_message_content, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "chat" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _chat_workflow(client) -> str: - response = client.chat( - model=model_name(), - messages=[ - {"role": "user", "content": "Reply with one concise tracing sentence."} - ], - ) - return response_message_content(response) +def _chat_workflow(prompt: str) -> str: + client = make_client() + try: + response = client.chat( + model=model_name(), + messages=[{"role": "user", "content": prompt}], + ) + return response_message_content(response) + finally: + client.close() def run_chat() -> None: respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) text = "" @@ -37,9 +38,9 @@ def run_chat() -> None: 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_workflow(client) + text = _chat_workflow("Reply with one concise tracing sentence.") finally: - respan.shutdown() + flush_and_shutdown(respan) print_result(EXAMPLE_NAME, custom_identifier, text) diff --git a/python/tracing/ollama/02_stream_generate.py b/python/tracing/ollama/02_stream_generate.py index f0aee45..ca88807 100644 --- a/python/tracing/ollama/02_stream_generate.py +++ b/python/tracing/ollama/02_stream_generate.py @@ -1,9 +1,8 @@ from __future__ import annotations -from respan import workflow - from _shared import ( example_attributes, + flush_and_shutdown, make_client, make_custom_identifier, make_respan, @@ -11,24 +10,28 @@ print_result, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "stream-generate" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _stream_generate_workflow(client) -> str: - chunks = client.generate( - model=model_name(), - prompt="Write a five word observability slogan.", - system="Be concise.", - stream=True, - ) - return "".join(chunk["response"] for chunk in chunks) +def _stream_generate_workflow(prompt: str) -> str: + client = make_client() + try: + chunks = client.generate( + model=model_name(), + prompt=prompt, + system="Be concise.", + stream=True, + ) + return "".join(chunk["response"] for chunk in chunks) + finally: + client.close() def run_stream_generate() -> None: respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) text = "" @@ -36,9 +39,9 @@ def run_stream_generate() -> None: 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 = _stream_generate_workflow(client) + text = _stream_generate_workflow("Write a five word observability slogan.") finally: - respan.shutdown() + flush_and_shutdown(respan) print_result(EXAMPLE_NAME, custom_identifier, text) diff --git a/python/tracing/ollama/03_tool_calling.py b/python/tracing/ollama/03_tool_calling.py index 239879c..70a027f 100644 --- a/python/tracing/ollama/03_tool_calling.py +++ b/python/tracing/ollama/03_tool_calling.py @@ -2,10 +2,9 @@ from typing import Any -from respan import workflow - from _shared import ( example_attributes, + flush_and_shutdown, make_client, make_custom_identifier, make_respan, @@ -17,10 +16,12 @@ tool_call_name, workflow_name, ) +from respan import tool, workflow EXAMPLE_NAME = "tool-calling" +@tool(name="get_weather") def get_weather(city: str) -> str: """Return deterministic weather for a city.""" return f"sunny and 22 C in {city}" @@ -30,39 +31,44 @@ def get_weather(city: str) -> str: @workflow(name=workflow_name(EXAMPLE_NAME)) -def _tool_calling_workflow(client) -> str: - messages: list[dict[str, Any]] = [ - { - "role": "user", - "content": "Use the weather tool for Tokyo and answer briefly.", - } - ] - first_response = client.chat(model=model_name(), messages=messages, tools=_TOOLS) - tool_calls = response_tool_calls(first_response) - if not tool_calls: - return response_message_content(first_response) - - messages.append( - { - "role": "assistant", - "content": response_message_content(first_response), - "tool_calls": tool_calls, - } - ) - for tool_call in tool_calls: - name = tool_call_name(tool_call) - arguments = tool_call_arguments(tool_call) - if name == "get_weather": - result = get_weather(**arguments) - messages.append({"role": "tool", "tool_name": name, "content": result}) - - final_response = client.chat(model=model_name(), messages=messages) - return response_message_content(final_response) +def _tool_calling_workflow(city: str) -> str: + client = make_client() + try: + messages: list[dict[str, Any]] = [ + { + "role": "user", + "content": f"Use the weather tool for {city} and answer briefly.", + } + ] + first_response = client.chat( + model=model_name(), messages=messages, tools=_TOOLS + ) + tool_calls = response_tool_calls(first_response) + if not tool_calls: + return response_message_content(first_response) + + messages.append( + { + "role": "assistant", + "content": response_message_content(first_response), + "tool_calls": tool_calls, + } + ) + for tool_call in tool_calls: + name = tool_call_name(tool_call) + arguments = tool_call_arguments(tool_call) + if name == "get_weather": + result = get_weather(**arguments) + messages.append({"role": "tool", "tool_name": name, "content": result}) + + final_response = client.chat(model=model_name(), messages=messages) + return response_message_content(final_response) + finally: + client.close() def run_tool_calling() -> None: respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) text = "" @@ -70,9 +76,9 @@ def run_tool_calling() -> None: 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 = _tool_calling_workflow(client) + text = _tool_calling_workflow("Tokyo") finally: - respan.shutdown() + flush_and_shutdown(respan) print_result(EXAMPLE_NAME, custom_identifier, text) diff --git a/python/tracing/ollama/04_embeddings.py b/python/tracing/ollama/04_embeddings.py index efcc757..79b4fec 100644 --- a/python/tracing/ollama/04_embeddings.py +++ b/python/tracing/ollama/04_embeddings.py @@ -1,9 +1,8 @@ from __future__ import annotations -from respan import workflow - from _shared import ( example_attributes, + flush_and_shutdown, make_client, make_custom_identifier, make_respan, @@ -11,24 +10,28 @@ print_result, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "embeddings" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _embeddings_workflow(client) -> str: - response = client.embed( - model=model_name(), - input="Trace local model calls with Respan.", - ) - embeddings = response["embeddings"] - vector_count = len(embeddings or []) - return f"embedding_vectors={vector_count}" +def _embeddings_workflow(text: str) -> str: + client = make_client() + try: + response = client.embed( + model=model_name(), + input=text, + ) + embeddings = response["embeddings"] + vector_count = len(embeddings or []) + return f"embedding_vectors={vector_count}" + finally: + client.close() def run_embeddings() -> None: respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) text = "" @@ -36,9 +39,9 @@ def run_embeddings() -> None: 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 = _embeddings_workflow(client) + text = _embeddings_workflow("Trace local model calls with Respan.") finally: - respan.shutdown() + flush_and_shutdown(respan) print_result(EXAMPLE_NAME, custom_identifier, text) diff --git a/python/tracing/ollama/05_expected_error.py b/python/tracing/ollama/05_expected_error.py new file mode 100644 index 0000000..7b09427 --- /dev/null +++ b/python/tracing/ollama/05_expected_error.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +from _shared import ( + example_attributes, + flush_and_shutdown, + make_client, + make_custom_identifier, + make_respan, + model_name, + print_result, + workflow_name, +) +from ollama import ResponseError +from respan import workflow + +EXAMPLE_NAME = "expected-error" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def _expected_error_workflow(prompt: str) -> str: + client = make_client(force_compat_server=True) + try: + response = client.chat( + model=model_name(), + messages=[ + { + "role": "user", + "content": prompt, + } + ], + ) + return str(response) + finally: + client.close() + + +def run_expected_error() -> None: + respan = make_respan(EXAMPLE_NAME) + 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) + try: + _expected_error_workflow("force expected provider error") + except ResponseError as exc: + if exc.status_code != 503: + raise + text = f"expected_status={exc.status_code} error={exc.error}" + else: + raise AssertionError("The compatibility server should return HTTP 503") + finally: + flush_and_shutdown(respan) + + print_result(EXAMPLE_NAME, custom_identifier, text) + + +if __name__ == "__main__": + run_expected_error() diff --git a/python/tracing/ollama/README.md b/python/tracing/ollama/README.md index 9276f59..14f5a58 100644 --- a/python/tracing/ollama/README.md +++ b/python/tracing/ollama/README.md @@ -8,10 +8,9 @@ The examples load Respan credentials from the repo-root `.env`. If `OLLAMA_HOST` ```bash cd python/tracing/ollama -python 01_chat.py -python 02_stream_generate.py -python 03_tool_calling.py -python 04_embeddings.py +RESPAN_EXAMPLE_RUN_ID=otel2-fix-py-group-NN-YYYYMMDDTHHMMSSZ python run_all.py ``` -Each script prints `workflow_name` and `custom_identifier` so the exported trace can be found in Respan. +`run_all.py` preserves the exact invocation `RESPAN_EXAMPLE_RUN_ID`, runs all five scripts in isolated processes, reports every exit code, and fails after the suite if any script fails. Each script records that marker as `example_run_id` and prints it together with a unique per-case `custom_identifier`, so the exported traces can be queried as one batch without losing scenario identity. + +The tool-calling example executes one `@tool`-decorated `get_weather` function between its two Ollama chat turns. The expected-error example always uses the local compatibility server and verifies an HTTP 503 span. diff --git a/python/tracing/ollama/_shared.py b/python/tracing/ollama/_shared.py index 487cc18..4f80851 100644 --- a/python/tracing/ollama/_shared.py +++ b/python/tracing/ollama/_shared.py @@ -1,14 +1,14 @@ from __future__ import annotations import atexit -from contextlib import contextmanager import json import os +from contextlib import contextmanager +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path -from threading import Thread +from threading import Thread, current_thread from typing import Any from uuid import uuid4 -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from dotenv import load_dotenv from ollama import Client @@ -18,12 +18,14 @@ PROJECT_ROOT = Path(__file__).resolve().parents[3] DEFAULT_RESPAN_BASE_URL = "https://api.respan.ai/api" DEFAULT_MODEL = "llama3.2" +DEFAULT_RUN_ID = f"ollama-{uuid4().hex[:10]}" _FAKE_SERVER: ThreadingHTTPServer | None = None _FAKE_SERVER_THREAD: Thread | None = None def load_root_env() -> None: - load_dotenv(PROJECT_ROOT / ".env", override=True) + # Invocation-scoped values, especially the exact QA marker, take precedence. + load_dotenv(PROJECT_ROOT / ".env", override=False) def require_respan_api_key() -> str: @@ -44,26 +46,37 @@ def model_name() -> str: ) +def example_run_id() -> str: + load_root_env() + return os.getenv("RESPAN_EXAMPLE_RUN_ID", DEFAULT_RUN_ID) + + def make_respan(example_name: str) -> Respan: api_key = require_respan_api_key() + run_id = example_run_id() return Respan( api_key=api_key, base_url=respan_base_url(), app_name="ollama-examples", instrumentations=[OllamaInstrumentor()], environment=os.getenv("RESPAN_ENVIRONMENT", "example"), - metadata={"integration": "ollama", "example": example_name}, + metadata={ + "integration": "ollama", + "example": example_name, + "example_run_id": run_id, + }, + is_batching_enabled=False, ) -def make_client() -> Client: +def make_client(*, force_compat_server: bool = False) -> Client: load_root_env() - return Client(host=ollama_host()) + return Client(host=ollama_host(force_compat_server=force_compat_server)) -def ollama_host() -> str | None: +def ollama_host(*, force_compat_server: bool = False) -> str | None: configured_host = os.getenv("OLLAMA_HOST") - if configured_host: + if configured_host and not force_compat_server: return configured_host return _start_fake_ollama_server() @@ -81,12 +94,14 @@ def make_custom_identifier(example_name: str) -> str: def example_attributes(example_name: str, custom_identifier: str | None = None): custom_identifier = custom_identifier or make_custom_identifier(example_name) current_workflow_name = workflow_name(example_name) + run_id = example_run_id() with propagate_attributes( custom_identifier=custom_identifier, trace_group_identifier=current_workflow_name, metadata={ "example": example_name, - "run_id": custom_identifier, + "example_run_id": run_id, + "case_id": custom_identifier, "workflow_name": current_workflow_name, }, ): @@ -128,11 +143,22 @@ def tool_call_arguments(tool_call: Any) -> dict[str, Any]: 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"RESPAN_EXAMPLE_RUN_ID={example_run_id()}") print(f"workflow_name={workflow_name(example_name)}") print(f"client_mode={client_mode()}") print(text.strip()) +def flush_and_shutdown(respan: Respan) -> None: + try: + respan.flush() + finally: + try: + respan.shutdown() + finally: + _stop_fake_ollama_server() + + def _field(value: Any, name: str, default: Any = None) -> Any: if value is None: return default @@ -142,8 +168,8 @@ def _field(value: Any, name: str, default: Any = None) -> Any: if callable(getter): try: return getter(name, default) - except Exception: - pass + except Exception: # noqa: BLE001 - fall back to attribute access + return getattr(value, name, default) return getattr(value, name, default) @@ -159,11 +185,17 @@ def _start_fake_ollama_server() -> str: def _stop_fake_ollama_server() -> None: - global _FAKE_SERVER - if _FAKE_SERVER is not None: - _FAKE_SERVER.shutdown() - _FAKE_SERVER.server_close() - _FAKE_SERVER = None + global _FAKE_SERVER, _FAKE_SERVER_THREAD + server = _FAKE_SERVER + thread = _FAKE_SERVER_THREAD + _FAKE_SERVER = None + _FAKE_SERVER_THREAD = None + if server is None: + return + server.shutdown() + server.server_close() + if thread is not None and thread is not current_thread(): + thread.join(timeout=2) class _FakeOllamaHandler(BaseHTTPRequestHandler): @@ -196,6 +228,16 @@ def log_message(self, format: str, *args: Any) -> None: def _handle_chat(self, payload: dict[str, Any]) -> None: messages = payload.get("messages") or [] + if any( + "force expected provider error" in str(message.get("content", "")) + for message in messages + if isinstance(message, dict) + ): + self._write_json( + {"error": "Ollama compatibility server unavailable"}, + status_code=503, + ) + return if any( message.get("role") == "tool" for message in messages @@ -266,9 +308,9 @@ def _handle_generate(self, payload: dict[str, Any]) -> None: } ) - def _write_json(self, payload: dict[str, Any]) -> None: + def _write_json(self, payload: dict[str, Any], status_code: int = 200) -> None: body = json.dumps(payload).encode("utf-8") - self.send_response(200) + self.send_response(status_code) self.send_header("content-type", "application/json") self.send_header("content-length", str(len(body))) self.end_headers() diff --git a/python/tracing/ollama/run_all.py b/python/tracing/ollama/run_all.py new file mode 100644 index 0000000..78ccb51 --- /dev/null +++ b/python/tracing/ollama/run_all.py @@ -0,0 +1,48 @@ +"""Run every Ollama tracing example with one exact batch marker.""" + +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.py", + "02_stream_generate.py", + "03_tool_calling.py", + "04_embeddings.py", + "05_expected_error.py", +) + + +def main() -> None: + env = os.environ.copy() + run_id = env.get("RESPAN_EXAMPLE_RUN_ID") or datetime.now(timezone.utc).strftime( + "ollama-suite-%Y%m%dT%H%M%SZ" + ) + env["RESPAN_EXAMPLE_RUN_ID"] = run_id + print(f"RESPAN_EXAMPLE_RUN_ID={run_id}", flush=True) + + failures: list[tuple[str, int]] = [] + for script in SCRIPTS: + print(f"\n=== {script} ===", flush=True) + result = subprocess.run( + [sys.executable, str(EXAMPLE_DIR / script)], + cwd=EXAMPLE_DIR, + env=env, + check=False, + ) + print(f"PROCESS_EXIT script={script} code={result.returncode}", flush=True) + if result.returncode: + failures.append((script, result.returncode)) + + if failures: + rendered = ", ".join(f"{name} ({code})" for name, code in failures) + raise SystemExit(f"Ollama example failures: {rendered}") + + +if __name__ == "__main__": + main() diff --git a/python/tracing/ollama/test_example_contract.py b/python/tracing/ollama/test_example_contract.py new file mode 100644 index 0000000..871167b --- /dev/null +++ b/python/tracing/ollama/test_example_contract.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +from types import SimpleNamespace + +import _shared +import pytest +import run_all + + +def test_repo_dotenv_does_not_override_exact_invocation_marker( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +) -> None: + (tmp_path / ".env").write_text( + "RESPAN_EXAMPLE_RUN_ID=dotenv-marker\n", + encoding="utf-8", + ) + monkeypatch.setattr(_shared, "PROJECT_ROOT", tmp_path) + monkeypatch.setenv("RESPAN_EXAMPLE_RUN_ID", "shell-marker") + + assert _shared.example_run_id() == "shell-marker" + + +def test_compat_server_cleanup_joins_and_resets( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[str] = [] + + class FakeServer: + def shutdown(self) -> None: + calls.append("shutdown") + + def server_close(self) -> None: + calls.append("server_close") + + class FakeThread: + def join(self, timeout: int) -> None: + calls.append(f"join:{timeout}") + + monkeypatch.setattr(_shared, "_FAKE_SERVER", FakeServer()) + monkeypatch.setattr(_shared, "_FAKE_SERVER_THREAD", FakeThread()) + + _shared._stop_fake_ollama_server() + + assert calls == ["shutdown", "server_close", "join:2"] + assert _shared._FAKE_SERVER is None + assert _shared._FAKE_SERVER_THREAD is None + + +def test_run_all_preserves_marker_and_reports_every_exit( + monkeypatch: pytest.MonkeyPatch, +) -> None: + return_codes = iter([0, 2, 0, 3, 0]) + calls: list[tuple[str, str]] = [] + + def run(command, *, cwd, env, check): + calls.append((command[-1], env["RESPAN_EXAMPLE_RUN_ID"])) + assert cwd == run_all.EXAMPLE_DIR + assert check is False + return SimpleNamespace(returncode=next(return_codes)) + + monkeypatch.setattr(run_all.subprocess, "run", run) + monkeypatch.setenv("RESPAN_EXAMPLE_RUN_ID", "shell-marker") + + with pytest.raises(SystemExit, match="02_stream_generate.py.*04_embeddings.py"): + run_all.main() + + assert len(calls) == 5 + assert {marker for _, marker in calls} == {"shell-marker"}