From fec28a5657275ff5293972f2c787bc091c925cde Mon Sep 17 00:00:00 2001 From: Nightingalelyy Date: Tue, 18 Aug 2026 18:27:27 +0800 Subject: [PATCH] fix(examples-py): validate Together, Vertex AI, and Watson Orchestrate ADK OTel 2.x spans --- python/tracing/together/01_chat_completion.py | 32 +- python/tracing/together/02_stream_chat.py | 52 ++- python/tracing/together/03_async_chat.py | 33 +- python/tracing/together/04_text_completion.py | 20 +- python/tracing/together/05_embeddings.py | 30 +- python/tracing/together/06_rerank.py | 18 +- .../tracing/together/07_image_generation.py | 20 +- python/tracing/together/08_tool_calling.py | 104 +++--- python/tracing/together/09_expected_error.py | 52 +++ python/tracing/together/README.md | 50 ++- python/tracing/together/_shared.py | 352 +++++++++++++----- python/tracing/together/requirements.txt | 10 +- python/tracing/together/run_all.py | 45 +++ python/tracing/together/test_contract.py | 58 +++ .../tracing/vertex-ai/01_generate_content.py | 71 ++-- python/tracing/vertex-ai/02_chat_streaming.py | 59 +-- python/tracing/vertex-ai/03_tool_execution.py | 60 +++ python/tracing/vertex-ai/04_async_generate.py | 38 ++ python/tracing/vertex-ai/05_expected_error.py | 44 +++ python/tracing/vertex-ai/06_live_provider.py | 48 +++ python/tracing/vertex-ai/README.md | 30 +- python/tracing/vertex-ai/_fake_vertexai.py | 121 ------ python/tracing/vertex-ai/_shared.py | 223 +++++++++-- python/tracing/vertex-ai/pyproject.toml | 2 +- python/tracing/vertex-ai/run_all.py | 45 +++ python/tracing/vertex-ai/run_examples.py | 29 -- python/tracing/vertex-ai/test_contract.py | 53 +++ .../01_local_agent_tool.py | 21 +- .../02_live_run_client.py | 42 --- .../watson-orchestrate-adk/02_run_client.py | 39 ++ .../03_live_watsonx_chat.py | 39 -- .../watson-orchestrate-adk/03_watsonx_chat.py | 37 ++ .../watson-orchestrate-adk/04_async_run.py | 40 ++ .../05_expected_error.py | 46 +++ .../06_live_run_client.py | 57 +++ .../07_live_watsonx_chat.py | 46 +++ .../tracing/watson-orchestrate-adk/README.md | 49 +-- .../watson-orchestrate-adk/__init__.py | 1 - .../tracing/watson-orchestrate-adk/_shared.py | 168 ++++++++- .../watson-orchestrate-adk/pyproject.toml | 2 +- .../tracing/watson-orchestrate-adk/run_all.py | 45 +++ .../watson-orchestrate-adk/test_contract.py | 50 +++ 42 files changed, 1677 insertions(+), 704 deletions(-) create mode 100644 python/tracing/together/09_expected_error.py create mode 100644 python/tracing/together/run_all.py create mode 100644 python/tracing/together/test_contract.py create mode 100644 python/tracing/vertex-ai/03_tool_execution.py create mode 100644 python/tracing/vertex-ai/04_async_generate.py create mode 100644 python/tracing/vertex-ai/05_expected_error.py create mode 100644 python/tracing/vertex-ai/06_live_provider.py delete mode 100644 python/tracing/vertex-ai/_fake_vertexai.py create mode 100644 python/tracing/vertex-ai/run_all.py delete mode 100644 python/tracing/vertex-ai/run_examples.py create mode 100644 python/tracing/vertex-ai/test_contract.py delete mode 100644 python/tracing/watson-orchestrate-adk/02_live_run_client.py create mode 100644 python/tracing/watson-orchestrate-adk/02_run_client.py delete mode 100644 python/tracing/watson-orchestrate-adk/03_live_watsonx_chat.py create mode 100644 python/tracing/watson-orchestrate-adk/03_watsonx_chat.py create mode 100644 python/tracing/watson-orchestrate-adk/04_async_run.py create mode 100644 python/tracing/watson-orchestrate-adk/05_expected_error.py create mode 100644 python/tracing/watson-orchestrate-adk/06_live_run_client.py create mode 100644 python/tracing/watson-orchestrate-adk/07_live_watsonx_chat.py delete mode 100644 python/tracing/watson-orchestrate-adk/__init__.py create mode 100644 python/tracing/watson-orchestrate-adk/run_all.py create mode 100644 python/tracing/watson-orchestrate-adk/test_contract.py diff --git a/python/tracing/together/01_chat_completion.py b/python/tracing/together/01_chat_completion.py index 7b7a013..e224606 100644 --- a/python/tracing/together/01_chat_completion.py +++ b/python/tracing/together/01_chat_completion.py @@ -1,7 +1,5 @@ from __future__ import annotations -from respan import workflow - from _shared import ( example_attributes, first_message_text, @@ -13,36 +11,34 @@ print_start, 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 tracing Together AI apps.", - } - ], - max_tokens=80, - temperature=0, - ) - return first_message_text(response) +def _chat_completion_workflow(prompt: str) -> str: + with make_client() as client: + response = client.chat.completions.create( + model=model_name(), + messages=[{"role": "user", "content": prompt}], + max_tokens=80, + temperature=0, + ) + return first_message_text(response) def run_chat_completion() -> None: - respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) + respan = make_respan(EXAMPLE_NAME, custom_identifier) text = "" try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_start(EXAMPLE_NAME, custom_identifier) - text = _chat_completion_workflow(client) + text = _chat_completion_workflow( + "Reply with one concise sentence about tracing Together AI apps." + ) finally: respan.shutdown() diff --git a/python/tracing/together/02_stream_chat.py b/python/tracing/together/02_stream_chat.py index 9ca7094..05dbefa 100644 --- a/python/tracing/together/02_stream_chat.py +++ b/python/tracing/together/02_stream_chat.py @@ -1,7 +1,5 @@ from __future__ import annotations -from respan import workflow - from _shared import ( example_attributes, make_client, @@ -12,46 +10,44 @@ print_start, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "stream-chat" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _stream_chat_workflow(client) -> str: - stream = client.chat.completions.create( - model=model_name(), - messages=[ - { - "role": "user", - "content": "Stream a seven-word sentence about observability.", - } - ], - max_tokens=80, - temperature=0, - stream=True, - ) - parts: list[str] = [] - for chunk in stream: - choices = getattr(chunk, "choices", None) or [] - if not choices: - continue - delta = getattr(choices[0], "delta", None) - content = getattr(delta, "content", None) - if content: - parts.append(content) - return "".join(parts) +def _stream_chat_workflow(prompt: str) -> str: + with make_client() as client: + stream = client.chat.completions.create( + model=model_name(), + messages=[{"role": "user", "content": prompt}], + max_tokens=80, + temperature=0, + stream=True, + ) + parts: list[str] = [] + for chunk in stream: + choices = getattr(chunk, "choices", None) or [] + if not choices: + continue + delta = getattr(choices[0], "delta", None) + content = getattr(delta, "content", None) + if content: + parts.append(content) + return "".join(parts) def run_stream_chat() -> None: - respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) + respan = make_respan(EXAMPLE_NAME, custom_identifier) text = "" try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_start(EXAMPLE_NAME, custom_identifier) - text = _stream_chat_workflow(client) + text = _stream_chat_workflow( + "Stream a short sentence about observable trace data." + ) finally: respan.shutdown() diff --git a/python/tracing/together/03_async_chat.py b/python/tracing/together/03_async_chat.py index dee1030..e6cb5d7 100644 --- a/python/tracing/together/03_async_chat.py +++ b/python/tracing/together/03_async_chat.py @@ -2,8 +2,6 @@ import asyncio -from respan import workflow - from _shared import ( example_attributes, first_message_text, @@ -15,38 +13,35 @@ print_start, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "async-chat" @workflow(name=workflow_name(EXAMPLE_NAME)) -async def _async_chat_workflow(client) -> str: - response = await client.chat.completions.create( - model=model_name(), - messages=[ - { - "role": "user", - "content": "Reply with one concise sentence about async tracing.", - } - ], - max_tokens=80, - temperature=0, - ) - return first_message_text(response) +async def _async_chat_workflow(prompt: str) -> str: + async with make_async_client() as async_client: + response = await async_client.chat.completions.create( + model=model_name(), + messages=[{"role": "user", "content": prompt}], + max_tokens=80, + temperature=0, + ) + return first_message_text(response) async def run_async_chat() -> None: - respan = make_respan(EXAMPLE_NAME) - client = make_async_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) + respan = make_respan(EXAMPLE_NAME, custom_identifier) text = "" try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_start(EXAMPLE_NAME, custom_identifier) - text = await _async_chat_workflow(client) + text = await _async_chat_workflow( + "Reply with one concise sentence about async tracing." + ) finally: - await client.close() respan.shutdown() print_result(EXAMPLE_NAME, custom_identifier, text) diff --git a/python/tracing/together/04_text_completion.py b/python/tracing/together/04_text_completion.py index 49203e4..06c44be 100644 --- a/python/tracing/together/04_text_completion.py +++ b/python/tracing/together/04_text_completion.py @@ -1,7 +1,5 @@ from __future__ import annotations -from respan import workflow - from _shared import ( completion_model_name, example_attributes, @@ -11,38 +9,36 @@ make_respan, print_result, print_start, - SDK_UNAVAILABLE_ERRORS, - unavailable_text, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "text-completion" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _text_completion_workflow(client) -> str: - try: +def _text_completion_workflow(prompt: str) -> str: + with make_client() as client: response = client.completions.create( model=completion_model_name(), - prompt="Complete this sentence in under ten words: Tracing AI calls helps", + prompt=prompt, max_tokens=40, temperature=0, ) return first_text_completion(response) - except SDK_UNAVAILABLE_ERRORS as exc: - return unavailable_text("text completions", exc) def run_text_completion() -> None: - respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) + respan = make_respan(EXAMPLE_NAME, custom_identifier) text = "" try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_start(EXAMPLE_NAME, custom_identifier) - text = _text_completion_workflow(client) + text = _text_completion_workflow( + "Complete this sentence in under ten words: Tracing AI calls helps" + ) finally: respan.shutdown() diff --git a/python/tracing/together/05_embeddings.py b/python/tracing/together/05_embeddings.py index b435c33..79e8513 100644 --- a/python/tracing/together/05_embeddings.py +++ b/python/tracing/together/05_embeddings.py @@ -1,7 +1,5 @@ from __future__ import annotations -from respan import workflow - from _shared import ( embedding_model_name, example_attributes, @@ -10,41 +8,41 @@ make_respan, print_result, print_start, - SDK_UNAVAILABLE_ERRORS, - unavailable_text, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "embeddings" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _embeddings_workflow(client) -> str: - try: +def _embeddings_workflow(texts: list[str]) -> str: + with make_client() as client: response = client.embeddings.create( model=embedding_model_name(), - input=[ - "Respan traces Together AI chat calls.", - "Embeddings should not export vector payloads.", - ], + input=texts, ) data = getattr(response, "data", None) or [] first_embedding = getattr(data[0], "embedding", []) if data else [] - return f"embedding_count={len(data)} embedding_dimensions={len(first_embedding)}" - except SDK_UNAVAILABLE_ERRORS as exc: - return unavailable_text("embeddings", exc) + return ( + f"embedding_count={len(data)} embedding_dimensions={len(first_embedding)}" + ) def run_embeddings() -> None: - respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) + respan = make_respan(EXAMPLE_NAME, custom_identifier) text = "" try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_start(EXAMPLE_NAME, custom_identifier) - text = _embeddings_workflow(client) + text = _embeddings_workflow( + [ + "Respan traces Together AI chat calls.", + "Embeddings retain complete vector data.", + ] + ) finally: respan.shutdown() diff --git a/python/tracing/together/06_rerank.py b/python/tracing/together/06_rerank.py index ddb0276..5154dec 100644 --- a/python/tracing/together/06_rerank.py +++ b/python/tracing/together/06_rerank.py @@ -1,7 +1,5 @@ from __future__ import annotations -from respan import workflow - from _shared import ( example_attributes, make_client, @@ -10,20 +8,19 @@ print_result, print_start, rerank_model_name, - SDK_UNAVAILABLE_ERRORS, - unavailable_text, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "rerank" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _rerank_workflow(client) -> str: - try: +def _rerank_workflow(query: str) -> str: + with make_client() as client: response = client.rerank.create( model=rerank_model_name(), - query="Which document is about observability?", + query=query, documents=[ "Distributed tracing shows how requests move through services.", "Sourdough bread needs flour, water, salt, and patience.", @@ -37,20 +34,17 @@ def _rerank_workflow(client) -> str: return "no rerank results" top = results[0] return f"top_index={top.index} relevance_score={top.relevance_score}" - except SDK_UNAVAILABLE_ERRORS as exc: - return unavailable_text("rerank", exc) def run_rerank() -> None: - respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) + respan = make_respan(EXAMPLE_NAME, custom_identifier) text = "" try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_start(EXAMPLE_NAME, custom_identifier) - text = _rerank_workflow(client) + text = _rerank_workflow("Which document is about observability?") finally: respan.shutdown() diff --git a/python/tracing/together/07_image_generation.py b/python/tracing/together/07_image_generation.py index c5add96..a3660ee 100644 --- a/python/tracing/together/07_image_generation.py +++ b/python/tracing/together/07_image_generation.py @@ -1,7 +1,5 @@ from __future__ import annotations -from respan import workflow - from _shared import ( example_attributes, image_model_name, @@ -10,20 +8,19 @@ make_respan, print_result, print_start, - SDK_UNAVAILABLE_ERRORS, - unavailable_text, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "image-generation" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _image_generation_workflow(client) -> str: - try: +def _image_generation_workflow(prompt: str) -> str: + with make_client() as client: response = client.images.generate( model=image_model_name(), - prompt="A small line-art observability dashboard icon", + prompt=prompt, n=1, width=256, height=256, @@ -36,20 +33,19 @@ def _image_generation_workflow(client) -> str: first = data[0] image_type = getattr(first, "type", "unknown") return f"image_count={len(data)} first_type={image_type}" - except SDK_UNAVAILABLE_ERRORS as exc: - return unavailable_text("image generation", exc) def run_image_generation() -> None: - respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) + respan = make_respan(EXAMPLE_NAME, custom_identifier) text = "" try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_start(EXAMPLE_NAME, custom_identifier) - text = _image_generation_workflow(client) + text = _image_generation_workflow( + "A small line-art observability dashboard icon" + ) finally: respan.shutdown() diff --git a/python/tracing/together/08_tool_calling.py b/python/tracing/together/08_tool_calling.py index ce44fce..0def5ae 100644 --- a/python/tracing/together/08_tool_calling.py +++ b/python/tracing/together/08_tool_calling.py @@ -3,8 +3,6 @@ import json from typing import Any -from respan import workflow - from _shared import ( example_attributes, first_message_text, @@ -16,10 +14,12 @@ print_start, workflow_name, ) +from respan import tool, workflow EXAMPLE_NAME = "tool-calling" +@tool(name="get_weather") def get_weather(city: str) -> str: return f"Sunny and 22 C in {city}" @@ -38,11 +38,11 @@ def _run_tool_call(tool_call: Any) -> dict[str, str]: @workflow(name=workflow_name(EXAMPLE_NAME)) -def _tool_calling_workflow(client) -> str: +def _tool_calling_workflow(city: str) -> str: messages: list[dict[str, Any]] = [ { "role": "user", - "content": "What is the weather in Tokyo? Use the tool when available.", + "content": f"What is the weather in {city}? Use the tool when available.", } ] tools = [ @@ -59,67 +59,67 @@ def _tool_calling_workflow(client) -> str: }, } ] - response = client.chat.completions.create( - model=model_name(), - messages=messages, - tools=tools, - tool_choice="auto", - max_tokens=160, - temperature=0, - ) - first_choice = (getattr(response, "choices", None) or [None])[0] - message = getattr(first_choice, "message", None) - tool_calls = getattr(message, "tool_calls", None) or [] - if not tool_calls: - return first_message_text(response) - - messages.append( - { - "role": "assistant", - "content": getattr(message, "content", "") or "", - "tool_calls": [ - { - "id": tool_call.id, - "type": tool_call.type, - "function": { - "name": tool_call.function.name, - "arguments": tool_call.function.arguments, - }, - } - for tool_call in tool_calls - ], - } - ) - for tool_call in tool_calls: - tool_result = _run_tool_call(tool_call) + with make_client() as client: + response = client.chat.completions.create( + model=model_name(), + messages=messages, + tools=tools, + tool_choice="auto", + max_tokens=160, + temperature=0, + ) + first_choice = (getattr(response, "choices", None) or [None])[0] + message = getattr(first_choice, "message", None) + tool_calls = getattr(message, "tool_calls", None) or [] + if not tool_calls: + return first_message_text(response) + messages.append( { - "role": "tool", - "tool_call_id": tool_call.id, - "content": tool_result["result"], + "role": "assistant", + "content": getattr(message, "content", "") or "", + "tool_calls": [ + { + "id": tool_call.id, + "type": tool_call.type, + "function": { + "name": tool_call.function.name, + "arguments": tool_call.function.arguments, + }, + } + for tool_call in tool_calls + ], } ) - - final_response = client.chat.completions.create( - model=model_name(), - messages=messages, - tools=tools, - max_tokens=120, - temperature=0, - ) - return first_message_text(final_response) + for tool_call in tool_calls: + tool_result = _run_tool_call(tool_call) + messages.append( + { + "role": "tool", + "tool_call_id": tool_call.id, + "content": tool_result["result"], + } + ) + + final_response = client.chat.completions.create( + model=model_name(), + messages=messages, + tools=tools, + max_tokens=120, + temperature=0, + ) + return first_message_text(final_response) def run_tool_calling() -> None: - respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) + respan = make_respan(EXAMPLE_NAME, custom_identifier) text = "" try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_start(EXAMPLE_NAME, custom_identifier) - text = _tool_calling_workflow(client) + text = _tool_calling_workflow("Tokyo") finally: respan.shutdown() diff --git a/python/tracing/together/09_expected_error.py b/python/tracing/together/09_expected_error.py new file mode 100644 index 0000000..c1db380 --- /dev/null +++ b/python/tracing/together/09_expected_error.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +from _shared import ( + example_attributes, + make_client, + make_custom_identifier, + make_respan, + model_name, + print_result, + print_start, + workflow_name, +) +from respan import workflow +from together import APIStatusError + +EXAMPLE_NAME = "expected-error" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def _expected_error_workflow(prompt: str) -> str: + with make_client(error_status=429) as client: + client.chat.completions.create( + model=model_name(), + messages=[{"role": "user", "content": prompt}], + max_tokens=16, + ) + return "unexpected success" + + +def run_expected_error() -> None: + marker = make_custom_identifier(EXAMPLE_NAME) + respan = make_respan(EXAMPLE_NAME, marker) + result: dict[str, object] = {} + try: + with example_attributes(EXAMPLE_NAME, marker): + print_start(EXAMPLE_NAME, marker) + try: + _expected_error_workflow("Trigger deterministic provider throttling.") + except APIStatusError as exc: + result = { + "expected_error": type(exc).__name__, + "status_code": getattr(exc, "status_code", None), + } + else: + raise AssertionError("expected provider failure was not raised") + finally: + respan.shutdown() + print_result(EXAMPLE_NAME, marker, result) + + +if __name__ == "__main__": + run_expected_error() diff --git a/python/tracing/together/README.md b/python/tracing/together/README.md index b6cade2..e17b1dc 100644 --- a/python/tracing/together/README.md +++ b/python/tracing/together/README.md @@ -1,39 +1,35 @@ -# Together AI tracing examples +# Together OTel 2.x tracing examples -These examples exercise `respan-instrumentation-together` against the official -Together Python SDK. +The nine examples exercise the official Together Python SDK with the local +Respan instrumentation. By default, the SDK parses deterministic +`httpx.MockTransport` responses, so chat, streaming, text completion, +embeddings, rerank, image, connected tool execution, and a provider 429 are +repeatable without a Together credential. -## Setup +Set `RESPAN_TOGETHER_LIVE=1` and `TOGETHER_API_KEY` to make the non-error +examples call Together directly. The exact `RESPAN_EXAMPLE_RUN_ID` supplied by +the caller is preserved in both `run_id` and `example_run_id` metadata. -From this directory: +## Registry setup ```bash -uv venv -uv pip install -r requirements.txt +python -m venv .venv +.venv/bin/pip install -r requirements.txt ``` -The examples load `/home/yuyang/KeywordsAI/respan-example-projects/.env`. -They use `TOGETHER_API_KEY` directly when present. Otherwise they use -`RESPAN_GATEWAY_API_KEY` or `RESPAN_API_KEY` with `RESPAN_GATEWAY_BASE_URL` or -`RESPAN_BASE_URL`. +For local instrumentation development, link the sibling checkout after the +registry install: -Optional model overrides: - -- `RESPAN_TOGETHER_MODEL` -- `RESPAN_TOGETHER_COMPLETION_MODEL` -- `RESPAN_TOGETHER_EMBEDDING_MODEL` -- `RESPAN_TOGETHER_RERANK_MODEL` -- `RESPAN_TOGETHER_IMAGE_MODEL` +```bash +.venv/bin/pip install --no-build-isolation --no-deps -e \ + ../../../../respan/python-sdks/instrumentations/respan-instrumentation-together +``` -## Run +## Run the complete set ```bash -python 01_chat_completion.py -python 02_stream_chat.py -python 03_async_chat.py -python 04_text_completion.py -python 05_embeddings.py -python 06_rerank.py -python 07_image_generation.py -python 08_tool_calling.py +RESPAN_EXAMPLE_RUN_ID=my-exact-marker .venv/bin/python run_all.py ``` + +`run_all.py` applies one marker to every subprocess, bounds each process with a +timeout, continues after failures, and returns nonzero if any example fails. diff --git a/python/tracing/together/_shared.py b/python/tracing/together/_shared.py index 5983063..45de864 100644 --- a/python/tracing/together/_shared.py +++ b/python/tracing/together/_shared.py @@ -1,31 +1,30 @@ from __future__ import annotations +import json import os +from collections.abc import 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 respan import Respan, propagate_attributes from respan_instrumentation_together import TogetherInstrumentor -from together import APIConnectionError, APIStatusError, APITimeoutError, AsyncTogether, Together +from together import AsyncTogether, Together EXAMPLE_DIR = Path(__file__).resolve().parent PROJECT_ROOT = EXAMPLE_DIR.parents[2] - DEFAULT_RESPAN_BASE_URL = "https://api.respan.ai/api" -DEFAULT_DIRECT_CHAT_MODEL = "meta-llama/Llama-3.2-3B-Instruct-Turbo" -DEFAULT_GATEWAY_CHAT_MODEL = "openai/gpt-4.1-mini" -DEFAULT_COMPLETION_MODEL = DEFAULT_DIRECT_CHAT_MODEL +DEFAULT_CHAT_MODEL = "meta-llama/Llama-3.3-70B-Instruct-Turbo" DEFAULT_EMBEDDING_MODEL = "BAAI/bge-base-en-v1.5" DEFAULT_RERANK_MODEL = "Salesforce/Llama-Rank-v1" DEFAULT_IMAGE_MODEL = "black-forest-labs/FLUX.1-schnell-Free" -SDK_UNAVAILABLE_ERRORS = (APIConnectionError, APIStatusError, APITimeoutError) def load_root_env() -> None: - load_dotenv(PROJECT_ROOT / ".env", override=True) + load_dotenv(PROJECT_ROOT / ".env", override=False) def require_env(*names: str) -> str: @@ -34,94 +33,263 @@ def require_env(*names: str) -> str: value = os.getenv(name) if value: return value - raise RuntimeError(f"One of {', '.join(names)} must be set in the repo root .env") + raise RuntimeError(f"One of {', '.join(names)} must be set") def respan_api_key() -> str: return require_env("RESPAN_API_KEY", "RESPAN_GATEWAY_API_KEY") -def gateway_api_key() -> str: - return os.getenv("TOGETHER_API_KEY") or require_env( - "RESPAN_GATEWAY_API_KEY", - "RESPAN_API_KEY", - ) - - def respan_base_url() -> str: load_root_env() return os.getenv("RESPAN_BASE_URL", DEFAULT_RESPAN_BASE_URL).rstrip("/") -def gateway_base_url() -> str | None: - load_root_env() - if os.getenv("TOGETHER_API_KEY"): - return os.getenv("TOGETHER_BASE_URL") - return ( - os.getenv("RESPAN_GATEWAY_BASE_URL") - or os.getenv("RESPAN_BASE_URL") - or DEFAULT_RESPAN_BASE_URL - ).rstrip("/") - - -def client_mode() -> str: - return "direct-together" if os.getenv("TOGETHER_API_KEY") else "respan-gateway" - - def model_name() -> str: - load_root_env() - default_model = ( - DEFAULT_DIRECT_CHAT_MODEL - if os.getenv("TOGETHER_API_KEY") - else DEFAULT_GATEWAY_CHAT_MODEL - ) - return os.getenv("RESPAN_TOGETHER_MODEL", default_model) + return os.getenv("RESPAN_TOGETHER_MODEL", DEFAULT_CHAT_MODEL) def completion_model_name() -> str: - load_root_env() - return os.getenv("RESPAN_TOGETHER_COMPLETION_MODEL", DEFAULT_COMPLETION_MODEL) + return os.getenv("RESPAN_TOGETHER_COMPLETION_MODEL", DEFAULT_CHAT_MODEL) def embedding_model_name() -> str: - load_root_env() return os.getenv("RESPAN_TOGETHER_EMBEDDING_MODEL", DEFAULT_EMBEDDING_MODEL) def rerank_model_name() -> str: - load_root_env() return os.getenv("RESPAN_TOGETHER_RERANK_MODEL", DEFAULT_RERANK_MODEL) def image_model_name() -> str: - load_root_env() return os.getenv("RESPAN_TOGETHER_IMAGE_MODEL", DEFAULT_IMAGE_MODEL) -def make_client() -> Together: - base_url = gateway_base_url() - kwargs: dict[str, Any] = {"api_key": gateway_api_key()} - if base_url: - kwargs["base_url"] = base_url - return Together(**kwargs) +def _request_json(request: httpx.Request) -> dict[str, Any]: + try: + value = json.loads(request.content.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + return {} + return value if isinstance(value, dict) else {} + + +def _usage(prompt: int, completion: int) -> dict[str, int]: + return { + "prompt_tokens": prompt, + "completion_tokens": completion, + "total_tokens": prompt + completion, + } + + +def _chat_response(payload: dict[str, Any]) -> dict[str, Any]: + messages = payload.get("messages") or [] + tools = payload.get("tools") or [] + has_tool_result = any( + isinstance(message, dict) and message.get("role") == "tool" + for message in messages + ) + if tools and not has_tool_result: + message: dict[str, Any] = { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "together-weather-1", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"city":"Tokyo"}', + }, + } + ], + } + finish_reason = "tool_calls" + completion = 12 + elif has_tool_result: + message = { + "role": "assistant", + "content": "Tokyo is sunny and 22 C.", + } + finish_reason = "stop" + completion = 8 + else: + message = { + "role": "assistant", + "content": "Together tracing keeps model calls observable.", + } + finish_reason = "stop" + completion = 7 + return { + "id": "together-chat-deterministic", + "object": "chat.completion", + "created": 1_787_000_000, + "model": payload.get("model") or model_name(), + "choices": [ + { + "index": 0, + "message": message, + "finish_reason": finish_reason, + } + ], + "usage": _usage(11, completion), + } + + +def _stream_response(request: httpx.Request, payload: dict[str, Any]) -> httpx.Response: + chunks = [ + { + "id": "together-stream-deterministic", + "object": "chat.completion.chunk", + "created": 1_787_000_000, + "model": payload.get("model") or model_name(), + "choices": [ + { + "index": 0, + "delta": {"role": "assistant", "content": "Trace data "}, + "finish_reason": None, + } + ], + }, + { + "id": "together-stream-deterministic", + "object": "chat.completion.chunk", + "created": 1_787_000_000, + "model": payload.get("model") or model_name(), + "choices": [ + { + "index": 0, + "delta": {"content": "flows clearly."}, + "finish_reason": "stop", + } + ], + "usage": _usage(9, 4), + }, + ] + body = "".join(f"data: {json.dumps(chunk)}\n\n" for chunk in chunks) + body += "data: [DONE]\n\n" + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + text=body, + request=request, + ) + +def _deterministic_response( + request: httpx.Request, *, error_status: int | None +) -> httpx.Response: + if error_status is not None: + return httpx.Response( + error_status, + json={"error": {"message": "deterministic provider limit"}}, + request=request, + ) + payload = _request_json(request) + path = request.url.path + if path.endswith("/chat/completions"): + if payload.get("stream") is True: + return _stream_response(request, payload) + body = _chat_response(payload) + elif path.endswith("/completions"): + body = { + "id": "together-text-deterministic", + "object": "text_completion", + "created": 1_787_000_000, + "model": payload.get("model") or completion_model_name(), + "choices": [ + { + "index": 0, + "text": "Completion tracing is deterministic.", + "finish_reason": "stop", + } + ], + "usage": _usage(6, 5), + } + elif path.endswith("/embeddings"): + body = { + "object": "list", + "model": payload.get("model") or embedding_model_name(), + "data": [ + {"object": "embedding", "index": 0, "embedding": [0.1, 0.2, 0.3]}, + {"object": "embedding", "index": 1, "embedding": [0.4, 0.5, 0.6]}, + ], + "usage": _usage(4, 0), + } + elif path.endswith("/rerank"): + body = { + "id": "together-rerank-deterministic", + "model": payload.get("model") or rerank_model_name(), + "results": [ + { + "index": 1, + "relevance_score": 0.98, + "document": {"text": "Washington, D.C. is the capital."}, + } + ], + "usage": _usage(7, 0), + } + elif path.endswith("/images/generations"): + body = { + "id": "together-image-deterministic", + "model": payload.get("model") or image_model_name(), + "data": [ + { + "index": 0, + "type": "url", + "url": "https://example.invalid/deterministic-image.png", + } + ], + } + else: + body = {"error": {"message": f"unhandled deterministic path {path}"}} + return httpx.Response(404, json=body, request=request) + return httpx.Response(200, json=body, request=request) + + +def _live_mode() -> bool: + return os.getenv("RESPAN_TOGETHER_LIVE") == "1" + + +def make_client(*, error_status: int | None = None) -> Together: + load_root_env() + if _live_mode() and error_status is None: + api_key = require_env("TOGETHER_API_KEY") + base_url = os.getenv("TOGETHER_BASE_URL") + return Together(api_key=api_key, base_url=base_url) + transport = httpx.MockTransport( + lambda request: _deterministic_response(request, error_status=error_status) + ) + return Together( + api_key="deterministic-together-key", + base_url="https://together.invalid/v1", + http_client=httpx.Client(transport=transport), + ) -def make_async_client() -> AsyncTogether: - base_url = gateway_base_url() - kwargs: dict[str, Any] = {"api_key": gateway_api_key()} - if base_url: - kwargs["base_url"] = base_url - return AsyncTogether(**kwargs) +def make_async_client(*, error_status: int | None = None) -> AsyncTogether: + transport = httpx.MockTransport( + lambda request: _deterministic_response(request, error_status=error_status) + ) + return AsyncTogether( + api_key="deterministic-together-key", + base_url="https://together.invalid/v1", + http_client=httpx.AsyncClient(transport=transport), + ) -def make_respan(example_name: str) -> Respan: + +def make_respan(example_name: str, marker: str) -> Respan: return Respan( api_key=respan_api_key(), base_url=respan_base_url(), app_name="together-examples", instrumentations=[TogetherInstrumentor()], environment=os.getenv("RESPAN_ENVIRONMENT", "example"), - metadata={"integration": "together", "example": example_name}, + metadata={ + "integration": "together", + "example": example_name, + "run_id": marker, + "example_run_id": marker, + }, ) @@ -130,72 +298,58 @@ def workflow_name(example_name: str) -> str: def make_custom_identifier(example_name: str) -> str: - return f"together-{example_name}-{uuid4().hex[:8]}" + return ( + os.getenv("RESPAN_EXAMPLE_RUN_ID") + or f"together-{example_name}-{uuid4().hex[:8]}" + ) @contextmanager -def example_attributes(example_name: str, custom_identifier: str | None = None): - custom_identifier = custom_identifier or make_custom_identifier(example_name) +def example_attributes( + example_name: str, custom_identifier: str | None = None +) -> Iterator[str]: + marker = custom_identifier or make_custom_identifier(example_name) current_workflow_name = workflow_name(example_name) with propagate_attributes( - custom_identifier=custom_identifier, + custom_identifier=marker, trace_group_identifier=current_workflow_name, customer_identifier="together-example-user", - thread_identifier=f"{custom_identifier}-thread", + thread_identifier=f"{marker}-{example_name}", metadata={ "example": example_name, - "run_id": custom_identifier, + "run_id": marker, + "example_run_id": marker, "workflow_name": current_workflow_name, - "client_mode": client_mode(), + "example_set": "together", + "client_mode": "live" if _live_mode() else "deterministic", }, ): - yield custom_identifier + yield marker def first_message_text(response: Any) -> str: choices = getattr(response, "choices", None) or [] - if not choices: - return "" - message = getattr(choices[0], "message", None) + message = getattr(choices[0], "message", None) if choices else None content = getattr(message, "content", None) - if isinstance(content, str): - return content - text = getattr(choices[0], "text", None) - return text if isinstance(text, str) else "" + return content if isinstance(content, str) else "" def first_text_completion(response: Any) -> str: choices = getattr(response, "choices", None) or [] - if not choices: - return "" - text = getattr(choices[0], "text", None) + text = getattr(choices[0], "text", None) if choices else None return text if isinstance(text, str) else "" -def print_start(example_name: str, custom_identifier: str) -> None: - print(f"custom_identifier={custom_identifier}", flush=True) - print(f"workflow_name={workflow_name(example_name)}", flush=True) - print(f"client_mode={client_mode()}", flush=True) +def print_start(example_name: str, marker: str) -> None: + print(f"example={example_name} marker={marker}", flush=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 unavailable_text(feature: str, exc: BaseException) -> str: - detail = str(exc).replace("\n", " ") - if len(detail) > 240: - detail = f"{detail[:237]}..." - return f"{feature} unavailable for {client_mode()}: {exc.__class__.__name__}: {detail}" - - -def close_async_client(client: AsyncTogether) -> None: - close = getattr(client, "close", None) - if callable(close): - result = close() - if hasattr(result, "__await__"): - raise RuntimeError("Use await client.close() for async clients") +def print_result(example_name: str, marker: str, result: Any) -> None: + print( + json.dumps( + {"example": example_name, "marker": marker, "result": result}, + ensure_ascii=False, + sort_keys=True, + ), + flush=True, + ) diff --git a/python/tracing/together/requirements.txt b/python/tracing/together/requirements.txt index 417ad0f..ffa94ea 100644 --- a/python/tracing/together/requirements.txt +++ b/python/tracing/together/requirements.txt @@ -1,6 +1,4 @@ --e ../../../../respan/python-sdks/respan-sdk --e ../../../../respan/python-sdks/respan-tracing --e ../../../../respan/python-sdks/respan --e ../../../../respan/python-sdks/instrumentations/respan-instrumentation-together -together>=2.0.0 -python-dotenv +respan-ai +respan-instrumentation-together +together>=2.31,<3 +python-dotenv>=1,<2 diff --git a/python/tracing/together/run_all.py b/python/tracing/together/run_all.py new file mode 100644 index 0000000..ec1de15 --- /dev/null +++ b/python/tracing/together/run_all.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path +from uuid import uuid4 + +EXAMPLE_DIR = Path(__file__).resolve().parent +SCRIPTS = tuple(sorted(EXAMPLE_DIR.glob("[0-9][0-9]_*.py"))) +TIMEOUT_SECONDS = int(os.getenv("RESPAN_EXAMPLE_TIMEOUT_SECONDS", "90")) + + +def main() -> int: + marker = os.getenv("RESPAN_EXAMPLE_RUN_ID") or ( + f"otel2-together-{uuid4().hex[:12]}" + ) + environment = dict(os.environ) + environment["RESPAN_EXAMPLE_RUN_ID"] = marker + environment["PYTHONDONTWRITEBYTECODE"] = "1" + failures: list[str] = [] + print(f"marker={marker} scripts={len(SCRIPTS)}", flush=True) + for script in SCRIPTS: + try: + result = subprocess.run( + [sys.executable, str(script)], + cwd=EXAMPLE_DIR, + env=environment, + check=False, + timeout=TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired: + failures.append(f"{script.name}:timeout") + continue + if result.returncode: + failures.append(f"{script.name}:exit={result.returncode}") + if failures: + print("failures=" + ",".join(failures), flush=True) + return 1 + print(f"completed={len(SCRIPTS)} marker={marker}", flush=True) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python/tracing/together/test_contract.py b/python/tracing/together/test_contract.py new file mode 100644 index 0000000..2d2e07a --- /dev/null +++ b/python/tracing/together/test_contract.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +import ast +from pathlib import Path + +from _shared import ( + example_attributes, + load_root_env, + make_custom_identifier, +) +from run_all import SCRIPTS + +EXAMPLE_DIR = Path(__file__).resolve().parent + + +def test_shell_marker_survives_dotenv_and_propagation(monkeypatch) -> None: + monkeypatch.setenv("RESPAN_EXAMPLE_RUN_ID", "shell-exact-marker") + load_root_env() + assert make_custom_identifier("contract") == "shell-exact-marker" + with example_attributes("contract") as marker: + assert marker == "shell-exact-marker" + + +def test_runner_covers_every_committed_numbered_example() -> None: + expected = tuple(sorted(EXAMPLE_DIR.glob("[0-9][0-9]_*.py"))) + assert SCRIPTS == expected + assert len(SCRIPTS) == 9 + + +def test_workflow_roots_take_bounded_semantic_arguments() -> None: + for script in SCRIPTS: + tree = ast.parse(script.read_text()) + workflows = [ + node + for node in tree.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and any( + isinstance(decorator, ast.Call) + and getattr(decorator.func, "id", None) == "workflow" + for decorator in node.decorator_list + ) + ] + assert workflows, script.name + for workflow in workflows: + names = [argument.arg for argument in workflow.args.args] + assert names + assert all( + blocked not in name.lower() + for name in names + for blocked in ("client", "key", "credential", "token") + ) + + +def test_requirements_are_registry_only() -> None: + requirements = (EXAMPLE_DIR / "requirements.txt").read_text() + assert "-e " not in requirements + assert "file:" not in requirements + assert "../../../../" not in requirements diff --git a/python/tracing/vertex-ai/01_generate_content.py b/python/tracing/vertex-ai/01_generate_content.py index aa38f08..f4dcca6 100644 --- a/python/tracing/vertex-ai/01_generate_content.py +++ b/python/tracing/vertex-ai/01_generate_content.py @@ -1,46 +1,35 @@ from __future__ import annotations -from _shared import model_name, prepare_vertexai_runtime - -prepare_vertexai_runtime() - -from respan import Respan, propagate_attributes, workflow # noqa: E402 -from respan_instrumentation_vertexai import VertexAIInstrumentor # noqa: E402 -from vertexai.generative_models import FunctionDeclaration, GenerativeModel, Tool # noqa: E402 - - -WORKFLOW_NAME = "vertexai_generate_content_example" - - -@workflow(name=WORKFLOW_NAME) -def run_example() -> str: - weather_tool = Tool( - function_declarations=[ - FunctionDeclaration( - name="get_weather", - description="Get a short weather summary for a city.", - parameters={ - "type": "object", - "properties": {"city": {"type": "string"}}, - "required": ["city"], - }, - ) - ] - ) - model = GenerativeModel( - model_name(), - system_instruction="Answer with one concise sentence.", - tools=[weather_tool], - ) - with propagate_attributes( - trace_group_identifier=WORKFLOW_NAME, - metadata={"example": WORKFLOW_NAME}, - ): - response = model.generate_content("Say hello from Vertex AI tracing.") - print(response.text) - return response.text +from _shared import ( + deterministic_model, + deterministic_vertex_runtime, + example_attributes, + make_respan, + marker_for, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "generate-content" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def generate_content(prompt: str) -> str: + model = deterministic_model(system_instruction="Answer concisely.") + return model.generate_content(prompt).text + + +def main() -> None: + marker = marker_for(EXAMPLE_NAME) + with deterministic_vertex_runtime(): + respan = make_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + result = generate_content("Say hello from Vertex AI tracing.") + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) if __name__ == "__main__": - respan = Respan(instrumentations=[VertexAIInstrumentor()]) - run_example() + main() diff --git a/python/tracing/vertex-ai/02_chat_streaming.py b/python/tracing/vertex-ai/02_chat_streaming.py index c0dc5bd..e02e888 100644 --- a/python/tracing/vertex-ai/02_chat_streaming.py +++ b/python/tracing/vertex-ai/02_chat_streaming.py @@ -1,34 +1,39 @@ from __future__ import annotations -from _shared import model_name, prepare_vertexai_runtime - -prepare_vertexai_runtime() - -from respan import Respan, propagate_attributes, workflow # noqa: E402 -from respan_instrumentation_vertexai import VertexAIInstrumentor # noqa: E402 -from vertexai.generative_models import GenerativeModel # noqa: E402 - - -WORKFLOW_NAME = "vertexai_chat_streaming_example" +from _shared import ( + deterministic_chat, + deterministic_model, + deterministic_vertex_runtime, + example_attributes, + make_respan, + marker_for, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "chat-streaming" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def chat_streaming(prompt: str) -> str: + chat = deterministic_chat( + deterministic_model(system_instruction="Keep responses short.") + ) + chunks = list(chat.send_message(prompt, stream=True)) + return "".join(chunk.text for chunk in chunks) -@workflow(name=WORKFLOW_NAME) -def run_example() -> str: - model = GenerativeModel( - model_name(), - system_instruction="Keep responses short and direct.", - ) - chat = model.start_chat() - with propagate_attributes( - trace_group_identifier=WORKFLOW_NAME, - metadata={"example": WORKFLOW_NAME}, - ): - chunks = list(chat.send_message("Stream a short Vertex AI reply.", stream=True)) - text = "".join(chunk.text for chunk in chunks) - print(text) - return text +def main() -> None: + marker = marker_for(EXAMPLE_NAME) + with deterministic_vertex_runtime(): + respan = make_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + result = chat_streaming("Stream a short Vertex AI reply.") + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) if __name__ == "__main__": - respan = Respan(instrumentations=[VertexAIInstrumentor()]) - run_example() + main() diff --git a/python/tracing/vertex-ai/03_tool_execution.py b/python/tracing/vertex-ai/03_tool_execution.py new file mode 100644 index 0000000..d576ac4 --- /dev/null +++ b/python/tracing/vertex-ai/03_tool_execution.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from typing import Any + +from _shared import ( + deterministic_model, + deterministic_vertex_runtime, + example_attributes, + make_respan, + marker_for, + workflow_name, +) +from respan import tool, workflow +from vertexai.generative_models import FunctionDeclaration, Tool + +EXAMPLE_NAME = "tool-execution" + + +@tool(name="get_weather") +def get_weather(city: str) -> str: + return f"Sunny and 22 C in {city}" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def tool_execution(city: str) -> str: + tool_definition = Tool( + function_declarations=[ + FunctionDeclaration( + name="get_weather", + description="Return deterministic weather for a city.", + parameters={ + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"], + }, + ) + ] + ) + model = deterministic_model(tools=[tool_definition]) + first = model.generate_content(f"What is the weather in {city}?") + part: Any = first.candidates[0].content.parts[0] + call = part.function_call + result = get_weather(city=call.args["city"]) + return model.generate_content(f"Tool result: {result}").text + + +def main() -> None: + marker = marker_for(EXAMPLE_NAME) + with deterministic_vertex_runtime(): + respan = make_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + result = tool_execution("Tokyo") + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/vertex-ai/04_async_generate.py b/python/tracing/vertex-ai/04_async_generate.py new file mode 100644 index 0000000..e7690e1 --- /dev/null +++ b/python/tracing/vertex-ai/04_async_generate.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import asyncio + +from _shared import ( + deterministic_model, + deterministic_vertex_runtime, + example_attributes, + make_respan, + marker_for, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "async-generate" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +async def async_generate(prompt: str) -> str: + model = deterministic_model() + response = await model.generate_content_async(prompt) + return response.text + + +async def run() -> None: + marker = marker_for(EXAMPLE_NAME) + with deterministic_vertex_runtime(): + respan = make_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + result = await async_generate("Trace an async Vertex response.") + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) + + +if __name__ == "__main__": + asyncio.run(run()) diff --git a/python/tracing/vertex-ai/05_expected_error.py b/python/tracing/vertex-ai/05_expected_error.py new file mode 100644 index 0000000..2b054cb --- /dev/null +++ b/python/tracing/vertex-ai/05_expected_error.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from _shared import ( + DeterministicVertexError, + deterministic_model, + deterministic_vertex_runtime, + example_attributes, + make_respan, + marker_for, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "expected-error" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def expected_error(prompt: str) -> str: + return deterministic_model().generate_content(prompt).text + + +def main() -> None: + marker = marker_for(EXAMPLE_NAME) + result: dict[str, object] = {} + with deterministic_vertex_runtime(): + respan = make_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + try: + expected_error("Trigger deterministic provider failure.") + except DeterministicVertexError as exc: + result = { + "expected_error": type(exc).__name__, + "status_code": exc.status_code, + } + else: + raise AssertionError("expected provider error was not raised") + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/vertex-ai/06_live_provider.py b/python/tracing/vertex-ai/06_live_provider.py new file mode 100644 index 0000000..d09c3a7 --- /dev/null +++ b/python/tracing/vertex-ai/06_live_provider.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import os + +import vertexai +from _shared import ( + example_attributes, + load_repo_env, + make_respan, + marker_for, + model_name, + workflow_name, +) +from respan import workflow +from vertexai.generative_models import GenerativeModel + +EXAMPLE_NAME = "live-provider" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def live_provider(prompt: str) -> str: + return GenerativeModel(model_name()).generate_content(prompt).text + + +def main() -> None: + load_repo_env() + required = ("GOOGLE_CLOUD_PROJECT", "GOOGLE_CLOUD_LOCATION") + if not all(os.getenv(name) for name in required): + print( + "live Vertex AI skipped: GOOGLE_CLOUD_PROJECT/LOCATION absent", flush=True + ) + return + marker = marker_for(EXAMPLE_NAME) + vertexai.init( + project=os.environ["GOOGLE_CLOUD_PROJECT"], + location=os.environ["GOOGLE_CLOUD_LOCATION"], + ) + respan = make_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + result = live_provider("Reply exactly: live Vertex verified") + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/vertex-ai/README.md b/python/tracing/vertex-ai/README.md index f0cd212..efb211a 100644 --- a/python/tracing/vertex-ai/README.md +++ b/python/tracing/vertex-ai/README.md @@ -1,21 +1,23 @@ -# Vertex AI tracing examples +# Vertex AI OTel 2.x tracing examples -These examples trace Vertex AI `GenerativeModel` and `ChatSession` calls with -`respan-instrumentation-vertexai`. +The first five scripts patch the public methods on the installed +`google-cloud-aiplatform` 1.164.x classes with deterministic responses before +activating Respan. This verifies the current real class boundary without +requiring Google credentials. The set covers generation, streaming, async, +connected tool execution, and an exact provider 503. -They load environment variables from the repository root `.env`. If -`GOOGLE_CLOUD_PROJECT`, `GOOGLE_CLOUD_LOCATION`, and Google application -credentials are available, the scripts use the real Vertex AI SDK. Otherwise -they install a small local compatibility stub so the tracing pipeline can be run -and verified without GCP credentials. +`06_live_provider.py` uses the real Vertex service when +`GOOGLE_CLOUD_PROJECT` and `GOOGLE_CLOUD_LOCATION` are configured; otherwise +it exits with an explicit skip. -## Run +The exact caller-supplied `RESPAN_EXAMPLE_RUN_ID` is retained in `run_id` and +`example_run_id` metadata on every deterministic record. + +Run all examples with one marker: ```bash -python run_examples.py +RESPAN_EXAMPLE_RUN_ID=my-exact-marker python run_all.py ``` -Every script sets an explicit workflow name: - -- `vertexai_generate_content_example` -- `vertexai_chat_streaming_example` +The runner applies a per-process timeout, continues through failures, and +returns nonzero if any script fails. diff --git a/python/tracing/vertex-ai/_fake_vertexai.py b/python/tracing/vertex-ai/_fake_vertexai.py deleted file mode 100644 index 6b117ce..0000000 --- a/python/tracing/vertex-ai/_fake_vertexai.py +++ /dev/null @@ -1,121 +0,0 @@ -"""Local Vertex AI compatibility objects for runnable tracing examples.""" - -from __future__ import annotations - -import sys -from collections.abc import Iterator -from types import ModuleType -from typing import Any - - -class _Obj: - def __init__(self, **kwargs: Any) -> None: - for key, value in kwargs.items(): - setattr(self, key, value) - - -class FunctionDeclaration: - def __init__( - self, - *, - name: str, - description: str | None = None, - parameters: dict[str, Any] | None = None, - ) -> None: - self.name = name - self.description = description - self.parameters = parameters - - -class Tool: - def __init__(self, *, function_declarations: list[FunctionDeclaration]) -> None: - self.function_declarations = function_declarations - - -class _Usage: - def __init__(self, prompt_tokens: int, completion_tokens: int) -> None: - self.prompt_token_count = prompt_tokens - self.candidates_token_count = completion_tokens - self.total_token_count = prompt_tokens + completion_tokens - - -class _Response: - def __init__( - self, - text: str, - *, - usage: _Usage, - parts: list[Any] | None = None, - ) -> None: - self.text = text - content = _Obj(role="model", parts=parts if parts is not None else [_Obj(text=text)]) - self.candidates = [_Obj(content=content)] - self.usage_metadata = usage - - -class ChatSession: - def __init__(self, model: "GenerativeModel") -> None: - self.model = model - - def send_message(self, content: str, *, stream: bool = False, **_: Any) -> Any: - if stream: - return iter( - [ - _Response("Vertex ", usage=_Usage(0, 0)), - _Response("chat response", usage=_Usage(9, 13)), - ] - ) - return _Response(f"Vertex chat response: {content}", usage=_Usage(9, 13)) - - async def send_message_async(self, content: str, **_: Any) -> _Response: - return _Response(f"Async Vertex chat response: {content}", usage=_Usage(9, 13)) - - -class GenerativeModel: - def __init__( - self, - model_name: str, - *, - system_instruction: str | None = None, - tools: list[Tool] | None = None, - **_: Any, - ) -> None: - self._model_name = model_name - self._system_instruction = system_instruction - self._tools = tools - - def generate_content(self, contents: str, *, stream: bool = False, **_: Any) -> Any: - if stream: - return self._stream_response() - return _Response( - f"Vertex generated answer: {contents}", - usage=_Usage(12, 18), - ) - - async def generate_content_async(self, contents: str, **_: Any) -> _Response: - return _Response(f"Async Vertex generated answer: {contents}", usage=_Usage(12, 18)) - - def start_chat(self, **_: Any) -> ChatSession: - return ChatSession(self) - - def _stream_response(self) -> Iterator[_Response]: - yield _Response("Vertex ", usage=_Usage(0, 0)) - yield _Response("streamed answer", usage=_Usage(12, 18)) - - -def install_fake_vertexai() -> None: - vertexai_module = ModuleType("vertexai") - generative_models_module = ModuleType("vertexai.generative_models") - - def init(**_: Any) -> None: - return None - - vertexai_module.init = init # type: ignore[attr-defined] - vertexai_module.generative_models = generative_models_module # type: ignore[attr-defined] - generative_models_module.GenerativeModel = GenerativeModel # type: ignore[attr-defined] - generative_models_module.ChatSession = ChatSession # type: ignore[attr-defined] - generative_models_module.FunctionDeclaration = FunctionDeclaration # type: ignore[attr-defined] - generative_models_module.Tool = Tool # type: ignore[attr-defined] - - sys.modules["vertexai"] = vertexai_module - sys.modules["vertexai.generative_models"] = generative_models_module diff --git a/python/tracing/vertex-ai/_shared.py b/python/tracing/vertex-ai/_shared.py index 1be94ee..00d06c8 100644 --- a/python/tracing/vertex-ai/_shared.py +++ b/python/tracing/vertex-ai/_shared.py @@ -1,50 +1,215 @@ -"""Shared setup for Vertex AI tracing examples.""" +"""Shared setup for Vertex AI OTel 2.x tracing examples.""" from __future__ import annotations import os +from collections.abc import Iterator +from contextlib import contextmanager from pathlib import Path +from types import SimpleNamespace +from typing import Any +from uuid import uuid4 from dotenv import load_dotenv +from respan import Respan, propagate_attributes +from respan_instrumentation_vertexai import VertexAIInstrumentor +from vertexai.generative_models import ChatSession, GenerativeModel +EXAMPLE_DIR = Path(__file__).resolve().parent +PROJECT_ROOT = EXAMPLE_DIR.parents[2] +DEFAULT_RESPAN_BASE_URL = "https://api.respan.ai/api" +DEFAULT_MODEL = "gemini-2.5-flash" -REPO_ROOT = Path(__file__).resolve().parents[3] -ENV_PATH = REPO_ROOT / ".env" + +class DeterministicVertexError(Exception): + status_code = 503 def load_repo_env() -> None: - load_dotenv(ENV_PATH, override=True) - - -def should_use_real_vertexai() -> bool: - if os.getenv("RESPAN_VERTEXAI_EXAMPLE_MODE", "").lower() == "fake": - return False - return all( - os.getenv(name) - for name in ( - "GOOGLE_CLOUD_PROJECT", - "GOOGLE_CLOUD_LOCATION", - "GOOGLE_APPLICATION_CREDENTIALS", - ) - ) + load_dotenv(PROJECT_ROOT / ".env", override=False) -def prepare_vertexai_runtime() -> bool: +def require_env(*names: str) -> str: load_repo_env() - if not should_use_real_vertexai(): - from _fake_vertexai import install_fake_vertexai + for name in names: + value = os.getenv(name) + if value: + return value + raise RuntimeError(f"One of {', '.join(names)} must be set") - install_fake_vertexai() - return False - import vertexai - - vertexai.init( - project=os.environ["GOOGLE_CLOUD_PROJECT"], - location=os.environ["GOOGLE_CLOUD_LOCATION"], +def marker_for(example_name: str) -> str: + return os.getenv("RESPAN_EXAMPLE_RUN_ID") or ( + f"vertexai-{example_name}-{uuid4().hex[:8]}" ) - return True def model_name() -> str: - return os.getenv("VERTEXAI_MODEL", "gemini-2.0-flash") + return os.getenv("VERTEXAI_MODEL", DEFAULT_MODEL) + + +def _usage(prompt: int, completion: int) -> Any: + return SimpleNamespace( + prompt_token_count=prompt, + candidates_token_count=completion, + thoughts_token_count=0, + total_token_count=prompt + completion, + ) + + +def _response( + text: str, + *, + prompt_tokens: int, + completion_tokens: int, + function_call: Any = None, +) -> Any: + parts = ( + [SimpleNamespace(function_call=function_call)] + if function_call is not None + else [SimpleNamespace(text=text)] + ) + return SimpleNamespace( + text=text, + candidates=[ + SimpleNamespace(content=SimpleNamespace(role="model", parts=parts)) + ], + usage_metadata=_usage(prompt_tokens, completion_tokens), + ) + + +def _generate_content( + self: Any, contents: Any, *, stream: bool = False, **kwargs: Any +) -> Any: + del self, kwargs + text = contents if isinstance(contents, str) else "" + if "provider failure" in text.lower(): + raise DeterministicVertexError("deterministic Vertex provider unavailable") + if stream: + return iter( + [ + _response("Vertex ", prompt_tokens=0, completion_tokens=0), + _response("streamed clearly.", prompt_tokens=10, completion_tokens=4), + ] + ) + if "weather" in text.lower() and "tool result" not in text.lower(): + return _response( + "", + prompt_tokens=12, + completion_tokens=6, + function_call=SimpleNamespace( + id="vertex-weather-1", + name="get_weather", + args={"city": "Tokyo"}, + ), + ) + if "tool result" in text.lower(): + return _response( + "Tokyo is sunny and 22 C.", + prompt_tokens=18, + completion_tokens=7, + ) + return _response( + "Vertex tracing is deterministic.", + prompt_tokens=9, + completion_tokens=5, + ) + + +async def _generate_content_async(self: Any, contents: Any, **kwargs: Any) -> Any: + return _generate_content(self, contents, **kwargs) + + +def _send_message( + self: Any, content: Any, *, stream: bool = False, **kwargs: Any +) -> Any: + return _generate_content(self, content, stream=stream, **kwargs) + + +async def _send_message_async(self: Any, content: Any, **kwargs: Any) -> Any: + return _generate_content(self, content, **kwargs) + + +@contextmanager +def deterministic_vertex_runtime() -> Iterator[None]: + originals = { + (GenerativeModel, "generate_content"): GenerativeModel.generate_content, + ( + GenerativeModel, + "generate_content_async", + ): GenerativeModel.generate_content_async, + (ChatSession, "send_message"): ChatSession.send_message, + (ChatSession, "send_message_async"): ChatSession.send_message_async, + } + GenerativeModel.generate_content = _generate_content + GenerativeModel.generate_content_async = _generate_content_async + ChatSession.send_message = _send_message + ChatSession.send_message_async = _send_message_async + try: + yield + finally: + for (cls, method_name), original in originals.items(): + if getattr(cls, method_name) in { + _generate_content, + _generate_content_async, + _send_message, + _send_message_async, + }: + setattr(cls, method_name, original) + + +def deterministic_model( + *, + system_instruction: str | None = None, + tools: list[Any] | None = None, +) -> GenerativeModel: + model = object.__new__(GenerativeModel) + object.__setattr__(model, "_model_name", model_name()) + object.__setattr__(model, "_system_instruction", system_instruction) + object.__setattr__(model, "_tools", tools) + return model + + +def deterministic_chat(model: GenerativeModel) -> ChatSession: + chat = object.__new__(ChatSession) + object.__setattr__(chat, "model", model) + return chat + + +def make_respan(example_name: str, marker: str) -> Respan: + return Respan( + api_key=require_env("RESPAN_API_KEY", "RESPAN_GATEWAY_API_KEY"), + base_url=os.getenv("RESPAN_BASE_URL", DEFAULT_RESPAN_BASE_URL).rstrip("/"), + app_name="vertexai-examples", + instrumentations=[VertexAIInstrumentor()], + environment=os.getenv("RESPAN_ENVIRONMENT", "example"), + metadata={ + "integration": "vertexai", + "example": example_name, + "run_id": marker, + "example_run_id": marker, + }, + ) + + +def workflow_name(example_name: str) -> str: + return f"vertexai_{example_name.replace('-', '_')}" + + +@contextmanager +def example_attributes(example_name: str, marker: str) -> Iterator[None]: + name = workflow_name(example_name) + with propagate_attributes( + custom_identifier=marker, + trace_group_identifier=name, + customer_identifier="vertexai-example-user", + thread_identifier=f"{marker}-{example_name}", + metadata={ + "example": example_name, + "example_set": "vertexai", + "run_id": marker, + "example_run_id": marker, + "workflow_name": name, + }, + ): + yield diff --git a/python/tracing/vertex-ai/pyproject.toml b/python/tracing/vertex-ai/pyproject.toml index fac4795..5b9699d 100644 --- a/python/tracing/vertex-ai/pyproject.toml +++ b/python/tracing/vertex-ai/pyproject.toml @@ -3,7 +3,7 @@ name = "respan-vertexai-tracing-examples" version = "0.1.0" requires-python = ">=3.11" dependencies = [ - "google-cloud-aiplatform>=1.71.0", + "google-cloud-aiplatform>=1.164,<2", "python-dotenv>=1.0.0", "respan-ai>=2.17.0", "respan-instrumentation-vertexai>=0.1.0", diff --git a/python/tracing/vertex-ai/run_all.py b/python/tracing/vertex-ai/run_all.py new file mode 100644 index 0000000..23681ae --- /dev/null +++ b/python/tracing/vertex-ai/run_all.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path +from uuid import uuid4 + +EXAMPLE_DIR = Path(__file__).resolve().parent +SCRIPTS = tuple(sorted(EXAMPLE_DIR.glob("[0-9][0-9]_*.py"))) +TIMEOUT_SECONDS = int(os.getenv("RESPAN_EXAMPLE_TIMEOUT_SECONDS", "90")) + + +def main() -> int: + marker = os.getenv("RESPAN_EXAMPLE_RUN_ID") or ( + f"otel2-vertexai-{uuid4().hex[:12]}" + ) + environment = dict(os.environ) + environment["RESPAN_EXAMPLE_RUN_ID"] = marker + environment["PYTHONDONTWRITEBYTECODE"] = "1" + failures: list[str] = [] + print(f"marker={marker} scripts={len(SCRIPTS)}", flush=True) + for script in SCRIPTS: + try: + result = subprocess.run( + [sys.executable, str(script)], + cwd=EXAMPLE_DIR, + env=environment, + check=False, + timeout=TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired: + failures.append(f"{script.name}:timeout") + continue + if result.returncode: + failures.append(f"{script.name}:exit={result.returncode}") + if failures: + print("failures=" + ",".join(failures), flush=True) + return 1 + print(f"completed={len(SCRIPTS)} marker={marker}", flush=True) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python/tracing/vertex-ai/run_examples.py b/python/tracing/vertex-ai/run_examples.py deleted file mode 100644 index b56bfe6..0000000 --- a/python/tracing/vertex-ai/run_examples.py +++ /dev/null @@ -1,29 +0,0 @@ -from __future__ import annotations - -import importlib.util -from pathlib import Path - - -EXAMPLE_FILES = ( - "01_generate_content.py", - "02_chat_streaming.py", -) - - -def _load_module(path: Path): - spec = importlib.util.spec_from_file_location(path.stem, path) - if spec is None or spec.loader is None: - raise RuntimeError(f"Could not load example module: {path}") - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return module - - -def main() -> None: - base_dir = Path(__file__).resolve().parent - for filename in EXAMPLE_FILES: - module = _load_module(base_dir / filename) - respan = module.Respan(instrumentations=[module.VertexAIInstrumentor()]) - module.run_example() -if __name__ == "__main__": - main() diff --git a/python/tracing/vertex-ai/test_contract.py b/python/tracing/vertex-ai/test_contract.py new file mode 100644 index 0000000..df8d794 --- /dev/null +++ b/python/tracing/vertex-ai/test_contract.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import ast +from pathlib import Path + +from _shared import load_repo_env, marker_for +from run_all import SCRIPTS + +EXAMPLE_DIR = Path(__file__).resolve().parent + + +def test_shell_marker_survives_dotenv(monkeypatch) -> None: + monkeypatch.setenv("RESPAN_EXAMPLE_RUN_ID", "shell-exact-marker") + load_repo_env() + assert marker_for("contract") == "shell-exact-marker" + + +def test_runner_covers_all_examples() -> None: + assert SCRIPTS == tuple(sorted(EXAMPLE_DIR.glob("[0-9][0-9]_*.py"))) + assert len(SCRIPTS) == 6 + + +def test_deterministic_examples_use_current_sdk_not_fake_modules() -> None: + assert not (EXAMPLE_DIR / "_fake_vertexai.py").exists() + shared = (EXAMPLE_DIR / "_shared.py").read_text() + assert "from vertexai.generative_models import" in shared + assert "sys.modules" not in shared + + +def test_workflow_roots_take_semantic_arguments() -> None: + for script in SCRIPTS: + tree = ast.parse(script.read_text()) + workflows = [ + node + for node in tree.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and any( + isinstance(decorator, ast.Call) + and getattr(decorator.func, "id", None) == "workflow" + for decorator in node.decorator_list + ) + ] + assert workflows + for workflow in workflows: + names = [argument.arg for argument in workflow.args.args] + assert names + assert all("client" not in name.lower() for name in names) + + +def test_project_dependencies_are_registry_only() -> None: + pyproject = (EXAMPLE_DIR / "pyproject.toml").read_text() + assert "-e " not in pyproject + assert "../../../../" not in pyproject diff --git a/python/tracing/watson-orchestrate-adk/01_local_agent_tool.py b/python/tracing/watson-orchestrate-adk/01_local_agent_tool.py index 26ce13d..f236d19 100644 --- a/python/tracing/watson-orchestrate-adk/01_local_agent_tool.py +++ b/python/tracing/watson-orchestrate-adk/01_local_agent_tool.py @@ -2,12 +2,16 @@ from pathlib import Path +from _shared import ( + create_respan, + example_attributes, + marker_for, + workflow_name, +) from ibm_watsonx_orchestrate.agent_builder.agents import Agent from ibm_watsonx_orchestrate.agent_builder.tools import tool from respan import workflow -from _shared import create_respan - SCRIPT_NAME = Path(__file__).name APP_NAME = SCRIPT_NAME.removesuffix(".py") @@ -24,8 +28,8 @@ def always_fails(reason: str) -> str: raise RuntimeError(f"deterministic failure: {reason}") -@workflow(name=SCRIPT_NAME) -def run_local_agent_tool() -> dict[str, object]: +@workflow(name=workflow_name(APP_NAME)) +def run_local_agent_tool(ticket_id: str) -> dict[str, object]: agent = Agent( name="respan_watson_orchestrate_local_agent", description="Local agent spec for Respan instrumentation examples.", @@ -34,7 +38,7 @@ def run_local_agent_tool() -> dict[str, object]: tools=[lookup_ticket], ) - success = lookup_ticket(ticket_id="INC-1001") + success = lookup_ticket(ticket_id=ticket_id) failure = None try: always_fails(reason="example coverage") @@ -52,11 +56,14 @@ def run_local_agent_tool() -> dict[str, object]: def main() -> None: - respan = create_respan(APP_NAME) + marker = marker_for(APP_NAME) + respan = create_respan(APP_NAME, marker) try: - run_local_agent_tool() + with example_attributes(APP_NAME, marker): + result = run_local_agent_tool("INC-1001") finally: respan.shutdown() + print({"example": APP_NAME, "marker": marker, "result": result}, flush=True) if __name__ == "__main__": diff --git a/python/tracing/watson-orchestrate-adk/02_live_run_client.py b/python/tracing/watson-orchestrate-adk/02_live_run_client.py deleted file mode 100644 index 611aba5..0000000 --- a/python/tracing/watson-orchestrate-adk/02_live_run_client.py +++ /dev/null @@ -1,42 +0,0 @@ -"""Live Watson Orchestrate ADK run-client call traced by Respan.""" - -from pathlib import Path - -from ibm_watsonx_orchestrate.client.chat.run_client import RunClient -from respan import workflow - -from _shared import create_respan, optional_env, require_env - -SCRIPT_NAME = Path(__file__).name -APP_NAME = SCRIPT_NAME.removesuffix(".py") - - -@workflow(name=SCRIPT_NAME) -def run_live_agent() -> dict: - client = RunClient( - base_url=require_env("WATSON_ORCHESTRATE_BASE_URL"), - api_key=require_env("WATSON_ORCHESTRATE_API_KEY"), - is_local=optional_env("WATSON_ORCHESTRATE_IS_LOCAL") == "true", - verify=optional_env("WATSON_ORCHESTRATE_VERIFY_SSL"), - ) - response = client.create_run( - message=optional_env("WATSON_ORCHESTRATE_MESSAGE") - or "Reply with one concise sentence from a traced Respan example.", - agent_id=require_env("WATSON_ORCHESTRATE_AGENT_ID"), - thread_id=optional_env("WATSON_ORCHESTRATE_THREAD_ID"), - capture_logs=True, - ) - print(response) - return response - - -def main() -> None: - respan = create_respan(APP_NAME) - try: - run_live_agent() - finally: - respan.shutdown() - - -if __name__ == "__main__": - main() diff --git a/python/tracing/watson-orchestrate-adk/02_run_client.py b/python/tracing/watson-orchestrate-adk/02_run_client.py new file mode 100644 index 0000000..2a71b11 --- /dev/null +++ b/python/tracing/watson-orchestrate-adk/02_run_client.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from _shared import ( + create_respan, + deterministic_run_client, + deterministic_watson_runtime, + example_attributes, + marker_for, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "run-client" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def run_client(message: str) -> dict: + return deterministic_run_client().create_run( + message=message, + agent_id="watson-agent-deterministic", + thread_id="watson-thread-deterministic", + capture_logs=True, + ) + + +def main() -> None: + marker = marker_for(EXAMPLE_NAME) + with deterministic_watson_runtime(): + respan = create_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + result = run_client("Trace a deterministic Watson agent run.") + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/watson-orchestrate-adk/03_live_watsonx_chat.py b/python/tracing/watson-orchestrate-adk/03_live_watsonx_chat.py deleted file mode 100644 index 806472f..0000000 --- a/python/tracing/watson-orchestrate-adk/03_live_watsonx_chat.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Live WatsonxAIClient chat call traced by Respan.""" - -from pathlib import Path - -from ibm_watsonx_orchestrate.client.autodiscover.watsonx_ai.watsonx_ai_client import ( - WatsonxAIClient, -) -from respan import workflow - -from _shared import create_respan, optional_env - -SCRIPT_NAME = Path(__file__).name -APP_NAME = SCRIPT_NAME.removesuffix(".py") - - -@workflow(name=SCRIPT_NAME) -def run_live_watsonx_chat() -> dict: - client = WatsonxAIClient( - model=optional_env("WATSON_ORCHESTRATE_LLM_MODEL"), - ) - response = client.generate_response( - input=optional_env("WATSON_ORCHESTRATE_LLM_PROMPT") - or "Return one sentence explaining why tracing agent runs is useful.", - instructions="Answer in one concise sentence.", - ) - print(response) - return response - - -def main() -> None: - respan = create_respan(APP_NAME) - try: - run_live_watsonx_chat() - finally: - respan.shutdown() - - -if __name__ == "__main__": - main() diff --git a/python/tracing/watson-orchestrate-adk/03_watsonx_chat.py b/python/tracing/watson-orchestrate-adk/03_watsonx_chat.py new file mode 100644 index 0000000..84282f0 --- /dev/null +++ b/python/tracing/watson-orchestrate-adk/03_watsonx_chat.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from _shared import ( + create_respan, + deterministic_chat_client, + deterministic_watson_runtime, + example_attributes, + marker_for, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "watsonx-chat" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def watsonx_chat(prompt: str) -> dict: + return deterministic_chat_client().generate_response( + input=prompt, + instructions="Answer in one concise sentence.", + ) + + +def main() -> None: + marker = marker_for(EXAMPLE_NAME) + with deterministic_watson_runtime(): + respan = create_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + result = watsonx_chat("Explain why tracing agent runs is useful.") + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/watson-orchestrate-adk/04_async_run.py b/python/tracing/watson-orchestrate-adk/04_async_run.py new file mode 100644 index 0000000..38df03f --- /dev/null +++ b/python/tracing/watson-orchestrate-adk/04_async_run.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import asyncio + +from _shared import ( + create_respan, + deterministic_run_client, + deterministic_watson_runtime, + example_attributes, + marker_for, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "async-run" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +async def async_run(run_id: str) -> dict: + return await deterministic_run_client().stream_run_with_websocket( + agent_id="watson-agent-deterministic", + thread_id="watson-thread-deterministic", + run_id=run_id, + ) + + +async def run() -> None: + marker = marker_for(EXAMPLE_NAME) + with deterministic_watson_runtime(): + respan = create_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + result = await async_run("watson-run-deterministic") + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) + + +if __name__ == "__main__": + asyncio.run(run()) diff --git a/python/tracing/watson-orchestrate-adk/05_expected_error.py b/python/tracing/watson-orchestrate-adk/05_expected_error.py new file mode 100644 index 0000000..ccd3a85 --- /dev/null +++ b/python/tracing/watson-orchestrate-adk/05_expected_error.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from _shared import ( + DeterministicWatsonError, + create_respan, + deterministic_chat_client, + deterministic_watson_runtime, + example_attributes, + marker_for, + workflow_name, +) +from respan import workflow + +EXAMPLE_NAME = "expected-error" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def expected_error(prompt: str) -> dict: + return deterministic_chat_client().generate_response(input=prompt) + + +def main() -> None: + marker = marker_for(EXAMPLE_NAME) + result: dict[str, object] = {} + with deterministic_watson_runtime(): + respan = create_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + try: + expected_error("Trigger deterministic provider failure.") + except DeterministicWatsonError as exc: + result = { + "expected_error": type(exc).__name__, + "status_code": exc.status_code, + } + else: + raise AssertionError( + "expected Watson provider error was not raised" + ) + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/watson-orchestrate-adk/06_live_run_client.py b/python/tracing/watson-orchestrate-adk/06_live_run_client.py new file mode 100644 index 0000000..479af4a --- /dev/null +++ b/python/tracing/watson-orchestrate-adk/06_live_run_client.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import os + +from _shared import ( + create_respan, + example_attributes, + load_repo_env, + marker_for, + optional_env, + require_env, + workflow_name, +) +from ibm_watsonx_orchestrate_clients.chat.run_client import RunClient +from respan import workflow + +EXAMPLE_NAME = "live-run-client" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def live_run_client(message: str) -> dict: + client = RunClient( + base_url=require_env("WATSON_ORCHESTRATE_BASE_URL"), + api_key=require_env("WATSON_ORCHESTRATE_API_KEY"), + is_local=optional_env("WATSON_ORCHESTRATE_IS_LOCAL") == "true", + verify=optional_env("WATSON_ORCHESTRATE_VERIFY_SSL"), + ) + return client.create_run( + message=message, + agent_id=require_env("WATSON_ORCHESTRATE_AGENT_ID"), + thread_id=optional_env("WATSON_ORCHESTRATE_THREAD_ID"), + capture_logs=True, + ) + + +def main() -> None: + load_repo_env() + required = ( + "WATSON_ORCHESTRATE_BASE_URL", + "WATSON_ORCHESTRATE_API_KEY", + "WATSON_ORCHESTRATE_AGENT_ID", + ) + if not all(os.getenv(name) for name in required): + print("live Watson run skipped: service credentials absent", flush=True) + return + marker = marker_for(EXAMPLE_NAME) + respan = create_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + result = live_run_client("Return one concise traced response.") + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/watson-orchestrate-adk/07_live_watsonx_chat.py b/python/tracing/watson-orchestrate-adk/07_live_watsonx_chat.py new file mode 100644 index 0000000..87a45ed --- /dev/null +++ b/python/tracing/watson-orchestrate-adk/07_live_watsonx_chat.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import os + +from _shared import ( + create_respan, + example_attributes, + load_repo_env, + marker_for, + optional_env, + workflow_name, +) +from ibm_watsonx_orchestrate.client.autodiscover.watsonx_ai.watsonx_ai_client import ( + WatsonxAIClient, +) +from respan import workflow + +EXAMPLE_NAME = "live-watsonx-chat" + + +@workflow(name=workflow_name(EXAMPLE_NAME)) +def live_watsonx_chat(prompt: str) -> dict: + client = WatsonxAIClient(model=optional_env("WATSON_ORCHESTRATE_LLM_MODEL")) + return client.generate_response( + input=prompt, + instructions="Answer in one concise sentence.", + ) + + +def main() -> None: + load_repo_env() + if not os.getenv("WATSONX_APIKEY") or not os.getenv("WATSONX_SPACE_ID"): + print("live Watsonx chat skipped: WATSONX credentials absent", flush=True) + return + marker = marker_for(EXAMPLE_NAME) + respan = create_respan(EXAMPLE_NAME, marker) + try: + with example_attributes(EXAMPLE_NAME, marker): + result = live_watsonx_chat("Explain why tracing agent runs is useful.") + finally: + respan.shutdown() + print({"example": EXAMPLE_NAME, "marker": marker, "result": result}, flush=True) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/watson-orchestrate-adk/README.md b/python/tracing/watson-orchestrate-adk/README.md index 94e38d7..febe91a 100644 --- a/python/tracing/watson-orchestrate-adk/README.md +++ b/python/tracing/watson-orchestrate-adk/README.md @@ -1,42 +1,21 @@ -# Watson Orchestrate ADK tracing examples +# Watson Orchestrate ADK OTel 2.x examples -These examples trace IBM watsonx Orchestrate ADK activity with Respan. +The first five scripts exercise the installed +`ibm-watsonx-orchestrate` 2.15 public classes without service credentials: +local Python tools, synchronous and asynchronous run clients, chat, and a +precise provider 429. Deterministic methods are installed on the real current +SDK classes before Respan activation and restored after shutdown. -They load environment variables from `respan-example-projects/.env`. +`06_live_run_client.py` and `07_live_watsonx_chat.py` call IBM services only +when their documented credentials are present; otherwise they exit with an +explicit skip. -Required for all scripts: - -- `RESPAN_API_KEY` -- `RESPAN_BASE_URL` (optional, defaults to `https://api.respan.ai/api`) - -Required for the live run-client script: - -- `WATSON_ORCHESTRATE_BASE_URL` -- `WATSON_ORCHESTRATE_API_KEY` -- `WATSON_ORCHESTRATE_AGENT_ID` -- `WATSON_ORCHESTRATE_THREAD_ID` (optional) - -Required for the live watsonx.ai chat script: - -- `WATSONX_APIKEY` -- `WATSONX_SPACE_ID` -- `WATSONX_URL` (optional) -- `WATSON_ORCHESTRATE_LLM_MODEL` (optional) - -Run from this directory after installing local packages: +Every deterministic trace retains the exact caller-supplied +`RESPAN_EXAMPLE_RUN_ID` in both `run_id` and `example_run_id` metadata. ```bash -python 01_local_agent_tool.py -python 02_live_run_client.py -python 03_live_watsonx_chat.py +RESPAN_EXAMPLE_RUN_ID=my-exact-marker python run_all.py ``` -`01_local_agent_tool.py` does not need IBM service credentials. It defines an -ADK tool and agent spec, invokes a successful tool, and invokes a deterministic -failing tool so the instrumentation emits both success and error tool spans. - -`02_live_run_client.py` submits a message to a deployed Orchestrate agent when -the `WATSON_ORCHESTRATE_*` variables are set. - -`03_live_watsonx_chat.py` calls the ADK watsonx.ai autodiscover chat client when -the `WATSONX_*` variables are set. +The runner applies one marker and timeout to every process, continues through +failures, and returns nonzero if any example fails. diff --git a/python/tracing/watson-orchestrate-adk/__init__.py b/python/tracing/watson-orchestrate-adk/__init__.py deleted file mode 100644 index 419399f..0000000 --- a/python/tracing/watson-orchestrate-adk/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Watson Orchestrate ADK tracing examples.""" diff --git a/python/tracing/watson-orchestrate-adk/_shared.py b/python/tracing/watson-orchestrate-adk/_shared.py index 58093c6..2afaeb4 100644 --- a/python/tracing/watson-orchestrate-adk/_shared.py +++ b/python/tracing/watson-orchestrate-adk/_shared.py @@ -3,49 +3,185 @@ from __future__ import annotations import os -from datetime import datetime, timezone +from collections.abc import Iterator +from contextlib import contextmanager from pathlib import Path +from typing import Any +from uuid import uuid4 from dotenv import load_dotenv -from respan import Respan -DEFAULT_RUN_ID = datetime.now(timezone.utc).strftime("watson-orchestrate-adk-%Y%m%d-%H%M%S") - +from ibm_watsonx_orchestrate.client.autodiscover.watsonx_ai.watsonx_ai_client import ( + WatsonxAIClient, +) +from ibm_watsonx_orchestrate_clients.chat.run_client import RunClient +from respan import Respan, propagate_attributes from respan_instrumentation_watson_orchestrate_adk import ( WatsonOrchestrateADKInstrumentor, ) +EXAMPLE_DIR = Path(__file__).resolve().parent +PROJECT_ROOT = EXAMPLE_DIR.parents[2] +DEFAULT_RESPAN_BASE_URL = "https://api.respan.ai/api" +DEFAULT_MODEL = "watsonx/meta-llama/llama-3-3-70b-instruct" + + +class DeterministicWatsonError(Exception): + status_code = 429 + def load_repo_env() -> None: - """Load environment variables from respan-example-projects/.env.""" - repo_root = Path(__file__).resolve().parents[3] - load_dotenv(repo_root / ".env", override=True) + load_dotenv(PROJECT_ROOT / ".env", override=False) def require_env(name: str) -> str: + load_repo_env() value = os.getenv(name) if not value: - raise RuntimeError(f"{name} must be set in respan-example-projects/.env") + raise RuntimeError(f"{name} must be set") return value def optional_env(name: str) -> str | None: - value = os.getenv(name) - return value if value else None + load_repo_env() + return os.getenv(name) or None -def create_respan(app_name: str) -> Respan: - load_repo_env() - run_id = os.getenv("RESPAN_EXAMPLE_RUN_ID", DEFAULT_RUN_ID) +def marker_for(example_name: str) -> str: + return os.getenv("RESPAN_EXAMPLE_RUN_ID") or ( + f"watson-orchestrate-{example_name}-{uuid4().hex[:8]}" + ) + + +def workflow_name(example_name: str) -> str: + return f"watson_orchestrate_{example_name.replace('-', '_')}" + + +def create_respan(app_name: str, marker: str) -> Respan: return Respan( - app_name=app_name, + app_name="watson-orchestrate-adk-examples", api_key=require_env("RESPAN_API_KEY"), - base_url=os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api"), + base_url=os.getenv("RESPAN_BASE_URL", DEFAULT_RESPAN_BASE_URL), instrumentations=[WatsonOrchestrateADKInstrumentor()], is_batching_enabled=False, metadata={ "integration": "watson-orchestrate-adk", "example": app_name, - "run_id": run_id, + "run_id": marker, + "example_run_id": marker, }, environment="examples", ) + + +@contextmanager +def example_attributes(example_name: str, marker: str) -> Iterator[None]: + name = workflow_name(example_name) + with propagate_attributes( + custom_identifier=marker, + trace_group_identifier=name, + customer_identifier="watson-orchestrate-example-user", + thread_identifier=f"{marker}-{example_name}", + metadata={ + "example": example_name, + "example_set": "watson-orchestrate-adk", + "run_id": marker, + "example_run_id": marker, + "workflow_name": name, + }, + ): + yield + + +def _create_run( + self: Any, + message: str, + agent_id: str | None = None, + thread_id: str | None = None, + capture_logs: bool = False, +) -> dict[str, Any]: + del self, capture_logs + if "provider failure" in message.lower(): + raise DeterministicWatsonError("deterministic provider rate limit") + return { + "run_id": "watson-run-deterministic", + "thread_id": thread_id or "watson-thread-deterministic", + "agent_id": agent_id or "watson-agent-deterministic", + "status": "queued", + "message": message, + } + + +async def _stream_run_with_websocket( + self: Any, + agent_id: str, + thread_id: str, + run_id: str, + **kwargs: Any, +) -> dict[str, Any]: + del self, kwargs + return { + "agent_id": agent_id, + "thread_id": thread_id, + "run_id": run_id, + "status": "completed", + } + + +def _generate_response( + self: Any, + input: str, + model: str | None = None, + **kwargs: Any, +) -> dict[str, Any]: + del self, kwargs + if "provider failure" in input.lower(): + raise DeterministicWatsonError("deterministic provider rate limit") + return { + "model": model or DEFAULT_MODEL, + "choices": [ + { + "message": { + "role": "assistant", + "content": "Watson Orchestrate tracing is deterministic.", + } + } + ], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 6, + "total_tokens": 16, + }, + } + + +@contextmanager +def deterministic_watson_runtime() -> Iterator[None]: + originals = { + (RunClient, "create_run"): RunClient.create_run, + (RunClient, "stream_run_with_websocket"): RunClient.stream_run_with_websocket, + (WatsonxAIClient, "generate_response"): WatsonxAIClient.generate_response, + } + RunClient.create_run = _create_run + RunClient.stream_run_with_websocket = _stream_run_with_websocket + WatsonxAIClient.generate_response = _generate_response + try: + yield + finally: + for (cls, method_name), original in originals.items(): + current = getattr(cls, method_name) + if current in { + _create_run, + _stream_run_with_websocket, + _generate_response, + }: + setattr(cls, method_name, original) + + +def deterministic_run_client() -> RunClient: + return object.__new__(RunClient) + + +def deterministic_chat_client() -> WatsonxAIClient: + client = object.__new__(WatsonxAIClient) + object.__setattr__(client, "model", DEFAULT_MODEL) + return client diff --git a/python/tracing/watson-orchestrate-adk/pyproject.toml b/python/tracing/watson-orchestrate-adk/pyproject.toml index b168ad1..a7dc73c 100644 --- a/python/tracing/watson-orchestrate-adk/pyproject.toml +++ b/python/tracing/watson-orchestrate-adk/pyproject.toml @@ -7,7 +7,7 @@ readme = "README.md" [tool.poetry.dependencies] python = ">=3.11,<3.15" -ibm-watsonx-orchestrate = ">=2.12.0,<3.0.0" +ibm-watsonx-orchestrate = ">=2.15.0,<3.0.0" python-dotenv = ">=1.0.0" respan-ai = ">=2.17.0" respan-instrumentation-watson-orchestrate-adk = ">=0.1.0" diff --git a/python/tracing/watson-orchestrate-adk/run_all.py b/python/tracing/watson-orchestrate-adk/run_all.py new file mode 100644 index 0000000..6c0e663 --- /dev/null +++ b/python/tracing/watson-orchestrate-adk/run_all.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path +from uuid import uuid4 + +EXAMPLE_DIR = Path(__file__).resolve().parent +SCRIPTS = tuple(sorted(EXAMPLE_DIR.glob("[0-9][0-9]_*.py"))) +TIMEOUT_SECONDS = int(os.getenv("RESPAN_EXAMPLE_TIMEOUT_SECONDS", "90")) + + +def main() -> int: + marker = os.getenv("RESPAN_EXAMPLE_RUN_ID") or ( + f"otel2-watson-orchestrate-{uuid4().hex[:12]}" + ) + environment = dict(os.environ) + environment["RESPAN_EXAMPLE_RUN_ID"] = marker + environment["PYTHONDONTWRITEBYTECODE"] = "1" + failures: list[str] = [] + print(f"marker={marker} scripts={len(SCRIPTS)}", flush=True) + for script in SCRIPTS: + try: + result = subprocess.run( + [sys.executable, str(script)], + cwd=EXAMPLE_DIR, + env=environment, + check=False, + timeout=TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired: + failures.append(f"{script.name}:timeout") + continue + if result.returncode: + failures.append(f"{script.name}:exit={result.returncode}") + if failures: + print("failures=" + ",".join(failures), flush=True) + return 1 + print(f"completed={len(SCRIPTS)} marker={marker}", flush=True) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python/tracing/watson-orchestrate-adk/test_contract.py b/python/tracing/watson-orchestrate-adk/test_contract.py new file mode 100644 index 0000000..95729ba --- /dev/null +++ b/python/tracing/watson-orchestrate-adk/test_contract.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +import ast +from pathlib import Path + +from _shared import load_repo_env, marker_for +from run_all import SCRIPTS + +EXAMPLE_DIR = Path(__file__).resolve().parent + + +def test_shell_marker_survives_dotenv(monkeypatch) -> None: + monkeypatch.setenv("RESPAN_EXAMPLE_RUN_ID", "shell-exact-marker") + load_repo_env() + assert marker_for("contract") == "shell-exact-marker" + + +def test_runner_covers_all_examples() -> None: + assert SCRIPTS == tuple(sorted(EXAMPLE_DIR.glob("[0-9][0-9]_*.py"))) + assert len(SCRIPTS) == 7 + + +def test_workflow_roots_take_semantic_arguments() -> None: + for script in SCRIPTS: + tree = ast.parse(script.read_text()) + workflows = [ + node + for node in tree.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and any( + isinstance(decorator, ast.Call) + and getattr(decorator.func, "id", None) == "workflow" + for decorator in node.decorator_list + ) + ] + assert workflows + for workflow in workflows: + names = [argument.arg for argument in workflow.args.args] + assert names + assert all( + blocked not in name.lower() + for name in names + for blocked in ("client", "key", "credential", "token") + ) + + +def test_project_dependencies_are_registry_only() -> None: + pyproject = (EXAMPLE_DIR / "pyproject.toml").read_text() + assert "-e " not in pyproject + assert "../../../../" not in pyproject