diff --git a/python/tracing/sagemaker/01_invoke_endpoint_text.py b/python/tracing/sagemaker/01_invoke_endpoint_text.py index 074d350..fbe3d8d 100644 --- a/python/tracing/sagemaker/01_invoke_endpoint_text.py +++ b/python/tracing/sagemaker/01_invoke_endpoint_text.py @@ -1,7 +1,5 @@ from __future__ import annotations -from respan import workflow - from _shared import ( custom_attributes, endpoint_name, @@ -17,15 +15,17 @@ stubbed_response, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "invoke-endpoint-text" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _invoke_text_workflow(client) -> dict: +def _invoke_text_workflow(prompt: str) -> dict: + client = make_client() request_body = json_bytes( { - "inputs": "Reply with one concise sentence about SageMaker observability.", + "inputs": prompt, "parameters": {"max_new_tokens": 32, "temperature": 0.1}, } ) @@ -48,21 +48,25 @@ def _invoke_text_workflow(client) -> dict: "ContentType": "application/json", } - with stubbed_response(client, "invoke_endpoint", response, params): - result = client.invoke_endpoint(**params) - return {"response": read_json_body(result)} + try: + with stubbed_response(client, "invoke_endpoint", response, params): + result = client.invoke_endpoint(**params) + return {"response": read_json_body(result)} + finally: + client.close() def run_invoke_endpoint_text() -> None: respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) result: dict = {} try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_run_header(EXAMPLE_NAME, custom_identifier) - result = _invoke_text_workflow(client) + result = _invoke_text_workflow( + "Reply with one concise sentence about SageMaker observability." + ) finally: respan.shutdown() diff --git a/python/tracing/sagemaker/02_invoke_endpoint_chat_tools.py b/python/tracing/sagemaker/02_invoke_endpoint_chat_tools.py index f8f9dde..ee3670c 100644 --- a/python/tracing/sagemaker/02_invoke_endpoint_chat_tools.py +++ b/python/tracing/sagemaker/02_invoke_endpoint_chat_tools.py @@ -2,8 +2,6 @@ import json -from respan import tool, workflow - from _shared import ( custom_attributes, endpoint_name, @@ -19,6 +17,7 @@ stubbed_response, workflow_name, ) +from respan import tool, workflow EXAMPLE_NAME = "invoke-endpoint-chat-tools" @@ -49,13 +48,13 @@ def _extract_tool_call(response_payload: dict) -> dict: first_choice = choices[0] message = first_choice.get("message") if isinstance(first_choice, dict) else None if not isinstance(message, dict): - raise RuntimeError("SageMaker tool example expected an assistant message.") + raise TypeError("SageMaker tool example expected an assistant message.") tool_calls = message.get("tool_calls") if not isinstance(tool_calls, list) or not tool_calls: raise RuntimeError("SageMaker tool example expected an assistant tool call.") tool_call = tool_calls[0] if not isinstance(tool_call, dict): - raise RuntimeError("SageMaker tool example received an invalid tool call.") + raise TypeError("SageMaker tool example received an invalid tool call.") return tool_call @@ -70,10 +69,11 @@ def _tool_arguments(tool_call: dict) -> dict: @workflow(name=workflow_name(EXAMPLE_NAME)) -def _invoke_chat_tools_workflow(client) -> dict: +def _invoke_chat_tools_workflow(city: str) -> dict: + client = make_client() user_message = { "role": "user", - "content": "What is the weather in Tokyo? Use the tool.", + "content": f"What is the weather in {city}? Use the tool.", } first_body = { "messages": [user_message], @@ -100,7 +100,7 @@ def _invoke_chat_tools_workflow(client) -> dict: "type": "function", "function": { "name": "get_weather", - "arguments": "{\"city\": \"Tokyo\"}", + "arguments": '{"city": "Tokyo"}', }, } ], @@ -117,73 +117,77 @@ def _invoke_chat_tools_workflow(client) -> dict: "ContentType": "application/json", } - with stubbed_response(client, "invoke_endpoint", first_response, first_params): - first_result = read_json_body(client.invoke_endpoint(**first_params)) - - tool_call = _extract_tool_call(first_result) - tool_result = get_weather(**_tool_arguments(tool_call)) - - assistant_message = first_result["choices"][0]["message"] - second_body = { - "messages": [ - user_message, - assistant_message, - { - "role": "tool", - "tool_call_id": tool_call.get("id"), - "content": tool_result, - }, - ], - "tools": [TOOL_SCHEMA], - } - second_params = { - "EndpointName": endpoint_name(), - "Body": json_bytes(second_body), - "ContentType": "application/json", - "Accept": "application/json", - "CustomAttributes": custom_attributes(), - } - second_response = { - "Body": streaming_body( - { - "choices": [ - { - "message": { - "role": "assistant", - "content": "Tokyo is sunny and 22 C.", - } - } - ], - "usage": { - "prompt_tokens": 32, - "completion_tokens": 7, - "total_tokens": 39, + try: + with stubbed_response(client, "invoke_endpoint", first_response, first_params): + first_result = read_json_body(client.invoke_endpoint(**first_params)) + + tool_call = _extract_tool_call(first_result) + tool_result = get_weather(**_tool_arguments(tool_call)) + + assistant_message = first_result["choices"][0]["message"] + second_body = { + "messages": [ + user_message, + assistant_message, + { + "role": "tool", + "tool_call_id": tool_call.get("id"), + "content": tool_result, }, - } - ), - "ContentType": "application/json", - } - - with stubbed_response(client, "invoke_endpoint", second_response, second_params): - final_result = read_json_body(client.invoke_endpoint(**second_params)) - - return { - "tool_call": tool_call, - "tool_result": tool_result, - "final_response": final_result, - } + ], + "tools": [TOOL_SCHEMA], + } + second_params = { + "EndpointName": endpoint_name(), + "Body": json_bytes(second_body), + "ContentType": "application/json", + "Accept": "application/json", + "CustomAttributes": custom_attributes(), + } + second_response = { + "Body": streaming_body( + { + "choices": [ + { + "message": { + "role": "assistant", + "content": f"{city} is sunny and 22 C.", + } + } + ], + "usage": { + "prompt_tokens": 32, + "completion_tokens": 7, + "total_tokens": 39, + }, + } + ), + "ContentType": "application/json", + } + + with stubbed_response( + client, "invoke_endpoint", second_response, second_params + ): + final_result = read_json_body(client.invoke_endpoint(**second_params)) + + return { + "tool_call": tool_call, + "tool_result": tool_result, + "final_response": final_result, + } + finally: + client.close() def run_invoke_endpoint_chat_tools() -> None: respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) result: dict = {} try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_run_header(EXAMPLE_NAME, custom_identifier) - result = _invoke_chat_tools_workflow(client) + result = _invoke_chat_tools_workflow("Tokyo") finally: respan.shutdown() diff --git a/python/tracing/sagemaker/03_invoke_endpoint_stream.py b/python/tracing/sagemaker/03_invoke_endpoint_stream.py index 33617fd..e51c54f 100644 --- a/python/tracing/sagemaker/03_invoke_endpoint_stream.py +++ b/python/tracing/sagemaker/03_invoke_endpoint_stream.py @@ -2,9 +2,8 @@ import json -from respan import workflow - from _shared import ( + collect_stream_text, custom_attributes, endpoint_name, example_attributes, @@ -16,15 +15,16 @@ print_run_header, stubbed_response, workflow_name, - collect_stream_text, ) +from respan import workflow EXAMPLE_NAME = "invoke-endpoint-stream" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _invoke_stream_workflow(client) -> dict: - request_body = json_bytes({"inputs": "Stream a concise SageMaker sentence."}) +def _invoke_stream_workflow(prompt: str) -> dict: + client = make_client() + request_body = json_bytes({"inputs": prompt}) params = { "EndpointName": endpoint_name(), "Body": request_body, @@ -48,26 +48,28 @@ def _invoke_stream_workflow(client) -> dict: "ContentType": "application/json", } - with stubbed_response( - client, - "invoke_endpoint_with_response_stream", - response, - params, - ): - result = client.invoke_endpoint_with_response_stream(**params) - return {"stream_text": collect_stream_text(result)} + try: + with stubbed_response( + client, + "invoke_endpoint_with_response_stream", + response, + params, + ): + result = client.invoke_endpoint_with_response_stream(**params) + return {"stream_text": collect_stream_text(result)} + finally: + client.close() def run_invoke_endpoint_stream() -> None: respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) result: dict = {} try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_run_header(EXAMPLE_NAME, custom_identifier) - result = _invoke_stream_workflow(client) + result = _invoke_stream_workflow("Stream a concise SageMaker sentence.") finally: respan.shutdown() diff --git a/python/tracing/sagemaker/04_invoke_endpoint_async.py b/python/tracing/sagemaker/04_invoke_endpoint_async.py index 77d584a..add4617 100644 --- a/python/tracing/sagemaker/04_invoke_endpoint_async.py +++ b/python/tracing/sagemaker/04_invoke_endpoint_async.py @@ -1,7 +1,5 @@ from __future__ import annotations -from respan import workflow - from _shared import ( custom_attributes, endpoint_name, @@ -14,15 +12,17 @@ stubbed_response, workflow_name, ) +from respan import workflow EXAMPLE_NAME = "invoke-endpoint-async" @workflow(name=workflow_name(EXAMPLE_NAME)) -def _invoke_async_workflow(client) -> dict: +def _invoke_async_workflow(input_location: str) -> dict: + client = make_client() params = { "EndpointName": endpoint_name(), - "InputLocation": "s3://respan-sagemaker-example/input.json", + "InputLocation": input_location, "ContentType": "application/json", "Accept": "application/json", "CustomAttributes": custom_attributes(), @@ -32,24 +32,26 @@ def _invoke_async_workflow(client) -> dict: "OutputLocation": "s3://respan-sagemaker-example/output.json", } - with stubbed_response(client, "invoke_endpoint_async", response, params): - result = client.invoke_endpoint_async(**params) - return { - "inference_id": result.get("InferenceId"), - "output_location": result.get("OutputLocation"), - } + try: + with stubbed_response(client, "invoke_endpoint_async", response, params): + result = client.invoke_endpoint_async(**params) + return { + "inference_id": result.get("InferenceId"), + "output_location": result.get("OutputLocation"), + } + finally: + client.close() def run_invoke_endpoint_async() -> None: respan = make_respan(EXAMPLE_NAME) - client = make_client() custom_identifier = make_custom_identifier(EXAMPLE_NAME) result: dict = {} try: with example_attributes(EXAMPLE_NAME, custom_identifier): print_run_header(EXAMPLE_NAME, custom_identifier) - result = _invoke_async_workflow(client) + result = _invoke_async_workflow("s3://respan-sagemaker-example/input.json") finally: respan.shutdown() diff --git a/python/tracing/sagemaker/README.md b/python/tracing/sagemaker/README.md index 03fb438..c26aeab 100644 --- a/python/tracing/sagemaker/README.md +++ b/python/tracing/sagemaker/README.md @@ -3,7 +3,7 @@ These examples cover Respan's SageMaker Runtime instrumentation package. ```bash -cd /home/yuyang/KeywordsAI/respan-example-projects/python/tracing/sagemaker +cd python/tracing/sagemaker pip install -r requirements.txt python run_all.py ``` @@ -11,14 +11,14 @@ python run_all.py For local package development, install from source instead: ```bash -pip install -e /home/yuyang/KeywordsAI/respan/python-sdks/respan \ - -e /home/yuyang/KeywordsAI/respan/python-sdks/respan-sdk \ - -e /home/yuyang/KeywordsAI/respan/python-sdks/respan-tracing \ - -e /home/yuyang/KeywordsAI/respan/python-sdks/instrumentations/respan-instrumentation-sagemaker \ +pip install -e ../../../../respan/python-sdks/respan \ + -e ../../../../respan/python-sdks/respan-sdk \ + -e ../../../../respan/python-sdks/respan-tracing \ + -e ../../../../respan/python-sdks/instrumentations/respan-instrumentation-sagemaker \ boto3 python-dotenv ``` -The scripts load `/home/yuyang/KeywordsAI/respan-example-projects/.env`. +The scripts load the repository root `.env` without overriding shell values. When `SAGEMAKER_ENDPOINT_NAME` is absent, they run in boto3 `Stubber` mode so the instrumentation can be validated without an AWS endpoint. Set `SAGEMAKER_EXAMPLE_MODE=live` and `SAGEMAKER_ENDPOINT_NAME` to exercise a real @@ -38,3 +38,6 @@ endpoints. | `02_invoke_endpoint_chat_tools.py` | `InvokeEndpoint` two-turn tool flow with a model tool call, decorated local tool execution, tool-result follow-up, and final answer | | `03_invoke_endpoint_stream.py` | `InvokeEndpointWithResponseStream` with event-stream output | | `04_invoke_endpoint_async.py` | `InvokeEndpointAsync` request and output-location metadata | + +`run_all.py` preserves one externally supplied `RESPAN_EXAMPLE_RUN_ID`, runs +all four scenarios with per-process timeouts, and reports every failure. diff --git a/python/tracing/sagemaker/_shared.py b/python/tracing/sagemaker/_shared.py index 3442434..b3ef62e 100644 --- a/python/tracing/sagemaker/_shared.py +++ b/python/tracing/sagemaker/_shared.py @@ -23,7 +23,12 @@ def load_root_env() -> None: - load_dotenv(PROJECT_ROOT / ".env", override=True) + load_dotenv(PROJECT_ROOT / ".env", override=False) + + +def example_run_id() -> str: + load_root_env() + return os.getenv("RESPAN_EXAMPLE_RUN_ID") or f"sagemaker-local-{uuid4().hex[:12]}" def respan_api_key() -> str | None: @@ -41,13 +46,20 @@ def respan_base_url() -> str: def make_respan(example_name: str) -> Respan: + run_id = example_run_id() return Respan( api_key=respan_api_key(), base_url=respan_base_url(), app_name="sagemaker-examples", instrumentations=[SageMakerInstrumentor()], environment=os.getenv("RESPAN_ENVIRONMENT", "example"), - metadata={"integration": "sagemaker", "example": example_name}, + metadata={ + "integration": "sagemaker", + "example_set": "sagemaker", + "example": example_name, + "run_id": run_id, + "example_run_id": run_id, + }, is_batching_enabled=False, log_level=os.getenv("RESPAN_LOG_LEVEL", "WARNING"), ) @@ -71,8 +83,7 @@ def endpoint_name() -> str: if use_live_sagemaker(): if not endpoint: raise RuntimeError( - "SAGEMAKER_ENDPOINT_NAME is required when " - "SAGEMAKER_EXAMPLE_MODE=live." + "SAGEMAKER_ENDPOINT_NAME is required when SAGEMAKER_EXAMPLE_MODE=live." ) return endpoint return endpoint or STUB_ENDPOINT_NAME @@ -121,12 +132,16 @@ def make_custom_identifier(example_name: str) -> str: def example_attributes(example_name: str, custom_identifier: str | None = None): custom_identifier = custom_identifier or make_custom_identifier(example_name) current_workflow_name = workflow_name(example_name) + run_id = example_run_id() with propagate_attributes( custom_identifier=custom_identifier, trace_group_identifier=current_workflow_name, metadata={ "example": example_name, - "run_id": custom_identifier, + "example_set": "sagemaker", + "run_id": run_id, + "example_run_id": run_id, + "execution_id": custom_identifier, "workflow_name": current_workflow_name, "sagemaker_mode": sagemaker_mode(), }, @@ -171,19 +186,24 @@ def collect_stream_text(response: dict[str, Any]) -> str: parts: list[str] = [] for event in response["Body"]: payload_part = event.get("PayloadPart") if isinstance(event, dict) else None - payload_bytes = payload_part.get("Bytes") if isinstance(payload_part, dict) else None + payload_bytes = ( + payload_part.get("Bytes") if isinstance(payload_part, dict) else None + ) if payload_bytes is None: continue payload = json.loads(payload_bytes.decode("utf-8")) token = payload.get("token") if isinstance(payload, dict) else None if isinstance(token, dict) and isinstance(token.get("text"), str): parts.append(token["text"]) - elif isinstance(payload, dict) and isinstance(payload.get("generated_text"), str): + elif isinstance(payload, dict) and isinstance( + payload.get("generated_text"), str + ): parts.append(payload["generated_text"]) return "".join(parts) def print_run_header(example_name: str, custom_identifier: str) -> None: + print(f"example_run_id={example_run_id()}", flush=True) print(f"custom_identifier={custom_identifier}", flush=True) print(f"workflow_name={workflow_name(example_name)}", flush=True) print(f"sagemaker_mode={sagemaker_mode()}", flush=True) @@ -196,4 +216,4 @@ def print_result(example_name: str, custom_identifier: str, result: Any) -> None print(f"workflow_name={workflow_name(example_name)}") print(f"sagemaker_mode={sagemaker_mode()}") print(f"model={model_name()}") - print(json.dumps(result, default=str, indent=2, sort_keys=True)) + print(json.dumps(result, indent=2, sort_keys=True)) diff --git a/python/tracing/sagemaker/run_all.py b/python/tracing/sagemaker/run_all.py index deaceb0..a1f2cc9 100644 --- a/python/tracing/sagemaker/run_all.py +++ b/python/tracing/sagemaker/run_all.py @@ -1,7 +1,9 @@ from __future__ import annotations +import os import subprocess import sys +from datetime import datetime, timezone from pathlib import Path SCRIPTS = [ @@ -14,9 +16,28 @@ def main() -> None: root = Path(__file__).resolve().parent + env = os.environ.copy() + env.setdefault( + "RESPAN_EXAMPLE_RUN_ID", + datetime.now(timezone.utc).strftime("otel2-sagemaker-%Y%m%dT%H%M%SZ"), + ) + failures: list[str] = [] for script in SCRIPTS: print(f"\n=== {script} ===", flush=True) - subprocess.run([sys.executable, str(root / script)], check=True) + try: + result = subprocess.run( + [sys.executable, str(root / script)], + check=False, + env=env, + timeout=120, + ) + except subprocess.TimeoutExpired: + failures.append(f"{script}: timeout") + continue + if result.returncode: + failures.append(f"{script}: exit {result.returncode}") + if failures: + raise SystemExit("; ".join(failures)) if __name__ == "__main__": diff --git a/python/tracing/sagemaker/test_contract.py b/python/tracing/sagemaker/test_contract.py new file mode 100644 index 0000000..2fcb647 --- /dev/null +++ b/python/tracing/sagemaker/test_contract.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +import ast +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +SCRIPTS = tuple(sorted(ROOT.glob("0[1-4]_*.py"))) + + +def _source(path: Path) -> str: + return path.read_text(encoding="utf-8") + + +def test_marker_precedence_and_shared_runner_contract() -> None: + shared = _source(ROOT / "_shared.py") + runner = _source(ROOT / "run_all.py") + assert "override=False" in shared + assert 'os.getenv("RESPAN_EXAMPLE_RUN_ID")' in shared + assert "env.setdefault" in runner + assert "subprocess.TimeoutExpired" in runner + assert "check=False" in runner + assert all(path.name in runner for path in SCRIPTS) + + +def test_workflows_capture_only_bounded_semantic_arguments() -> None: + for path in SCRIPTS: + tree = ast.parse(_source(path)) + workflows = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef) + and any( + isinstance(decorator, ast.Call) + and getattr(decorator.func, "id", None) == "workflow" + for decorator in node.decorator_list + ) + ] + assert len(workflows) == 1 + names = [argument.arg for argument in workflows[0].args.args] + assert names and not {"client", "sdk", "dsn"}.intersection(names) + source = _source(path) + assert "client.close()" in source + assert "respan.shutdown()" in source + + +def test_documentation_is_portable() -> None: + readme = _source(ROOT / "README.md") + requirements = _source(ROOT / "requirements.txt") + assert "/home/" not in readme + assert "/Users/" not in readme + assert " -e " not in requirements diff --git a/python/tracing/semantic-kernel/01_kernel_function.py b/python/tracing/semantic-kernel/01_kernel_function.py index c54ddf2..7e3782b 100644 --- a/python/tracing/semantic-kernel/01_kernel_function.py +++ b/python/tracing/semantic-kernel/01_kernel_function.py @@ -3,11 +3,16 @@ import asyncio from pathlib import Path +from _shared import ( + close_kernel_clients, + create_kernel, + create_respan, + example_attributes, + print_result, +) from respan import workflow from semantic_kernel.functions import KernelArguments, kernel_function -from _shared import create_kernel, create_respan, print_result - SCRIPT_NAME = Path(__file__).name APP_NAME = SCRIPT_NAME.removesuffix(".py") @@ -22,13 +27,13 @@ def normalize_city(self, city: str) -> str: @workflow(name=SCRIPT_NAME) -async def run_kernel_function() -> str: +async def run_kernel_function(city: str) -> str: kernel = create_kernel(with_chat_service=False) kernel.add_plugin(TextPlugin(), plugin_name="Text") result = await kernel.invoke( function_name="normalize_city", plugin_name="Text", - arguments=KernelArguments(city=" san francisco "), + arguments=KernelArguments(city=city), ) output = str(result) print_result("normalized_city", output) @@ -38,9 +43,13 @@ async def run_kernel_function() -> str: async def main() -> None: respan = create_respan(APP_NAME) try: - await run_kernel_function() + with example_attributes(APP_NAME): + await run_kernel_function(" san francisco ") finally: - respan.shutdown() + try: + await close_kernel_clients() + finally: + respan.shutdown() if __name__ == "__main__": diff --git a/python/tracing/semantic-kernel/02_chat_completion.py b/python/tracing/semantic-kernel/02_chat_completion.py index 17410a9..87dffd0 100644 --- a/python/tracing/semantic-kernel/02_chat_completion.py +++ b/python/tracing/semantic-kernel/02_chat_completion.py @@ -3,18 +3,23 @@ import asyncio from pathlib import Path +from _shared import ( + close_kernel_clients, + create_kernel, + create_respan, + example_attributes, + print_result, +) from respan import workflow from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings from semantic_kernel.functions import KernelArguments -from _shared import create_kernel, create_respan, print_result - SCRIPT_NAME = Path(__file__).name APP_NAME = SCRIPT_NAME.removesuffix(".py") @workflow(name=SCRIPT_NAME) -async def run_chat_completion() -> str: +async def run_chat_completion(prompt: str) -> str: kernel = create_kernel() settings = OpenAIChatPromptExecutionSettings( service_id="chat", @@ -22,7 +27,7 @@ async def run_chat_completion() -> str: max_tokens=80, ) result = await kernel.invoke_prompt( - "Reply in one sentence: what does Semantic Kernel help developers build?", + prompt, arguments=KernelArguments(settings=settings), ) output = str(result) @@ -33,9 +38,15 @@ async def run_chat_completion() -> str: async def main() -> None: respan = create_respan(APP_NAME) try: - await run_chat_completion() + with example_attributes(APP_NAME): + await run_chat_completion( + "Reply in one sentence: what does Semantic Kernel help developers build?" + ) finally: - respan.shutdown() + try: + await close_kernel_clients() + finally: + respan.shutdown() if __name__ == "__main__": diff --git a/python/tracing/semantic-kernel/03_plugin_tool_call.py b/python/tracing/semantic-kernel/03_plugin_tool_call.py index b8e0a86..d8f5368 100644 --- a/python/tracing/semantic-kernel/03_plugin_tool_call.py +++ b/python/tracing/semantic-kernel/03_plugin_tool_call.py @@ -3,6 +3,13 @@ import asyncio from pathlib import Path +from _shared import ( + close_kernel_clients, + create_kernel, + create_respan, + example_attributes, + print_result, +) from respan import workflow from semantic_kernel.connectors.ai.function_choice_behavior import ( FunctionChoiceBehavior, @@ -10,8 +17,6 @@ from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings from semantic_kernel.functions import KernelArguments, kernel_function -from _shared import create_kernel, create_respan, print_result - SCRIPT_NAME = Path(__file__).name APP_NAME = SCRIPT_NAME.removesuffix(".py") @@ -26,7 +31,7 @@ def get_weather(self, city: str) -> str: @workflow(name=SCRIPT_NAME) -async def run_plugin_tool_call() -> str: +async def run_plugin_tool_call(city: str) -> str: kernel = create_kernel() kernel.add_plugin(TravelPlugin(), plugin_name="Travel") settings = OpenAIChatPromptExecutionSettings( @@ -36,7 +41,7 @@ async def run_plugin_tool_call() -> str: function_choice_behavior=FunctionChoiceBehavior.Auto(auto_invoke=True), ) result = await kernel.invoke_prompt( - "Use the Travel plugin to check the weather in Tokyo, then summarize it.", + f"Use the Travel plugin to check the weather in {city}, then summarize it.", arguments=KernelArguments(settings=settings), ) output = str(result) @@ -47,9 +52,13 @@ async def run_plugin_tool_call() -> str: async def main() -> None: respan = create_respan(APP_NAME) try: - await run_plugin_tool_call() + with example_attributes(APP_NAME): + await run_plugin_tool_call("Tokyo") finally: - respan.shutdown() + try: + await close_kernel_clients() + finally: + respan.shutdown() if __name__ == "__main__": diff --git a/python/tracing/semantic-kernel/04_function_failure.py b/python/tracing/semantic-kernel/04_function_failure.py index def5d5c..c8f198d 100644 --- a/python/tracing/semantic-kernel/04_function_failure.py +++ b/python/tracing/semantic-kernel/04_function_failure.py @@ -3,12 +3,17 @@ import asyncio from pathlib import Path +from _shared import ( + close_kernel_clients, + create_kernel, + create_respan, + example_attributes, + print_result, +) from respan import workflow from semantic_kernel.exceptions.kernel_exceptions import KernelInvokeException from semantic_kernel.functions import kernel_function -from _shared import create_kernel, create_respan, print_result - SCRIPT_NAME = Path(__file__).name APP_NAME = SCRIPT_NAME.removesuffix(".py") @@ -23,27 +28,31 @@ def fail_deterministically(self) -> str: @workflow(name=SCRIPT_NAME) -async def run_function_failure() -> str: +async def run_function_failure(scenario: str) -> None: kernel = create_kernel(with_chat_service=False) kernel.add_plugin(FailurePlugin(), plugin_name="Failure") - try: - await kernel.invoke( - function_name="fail_deterministically", - plugin_name="Failure", - ) - except (RuntimeError, KernelInvokeException) as exc: - message = f"Caught expected failure: {exc}" - print_result("failure", message) - return message - raise AssertionError("FailurePlugin.fail_deterministically unexpectedly succeeded") + await kernel.invoke( + function_name="fail_deterministically", + plugin_name="Failure", + ) + raise AssertionError(f"FailurePlugin unexpectedly succeeded for {scenario}") async def main() -> None: respan = create_respan(APP_NAME) try: - await run_function_failure() + with example_attributes(APP_NAME): + try: + await run_function_failure("deterministic") + except (RuntimeError, KernelInvokeException) as exc: + print_result( + "failure", f"Caught expected failure: {type(exc).__name__}" + ) finally: - respan.shutdown() + try: + await close_kernel_clients() + finally: + respan.shutdown() if __name__ == "__main__": diff --git a/python/tracing/semantic-kernel/README.md b/python/tracing/semantic-kernel/README.md index 958e372..6a05561 100644 --- a/python/tracing/semantic-kernel/README.md +++ b/python/tracing/semantic-kernel/README.md @@ -12,11 +12,14 @@ The scripts load environment variables from the examples repo root `.env` file: - `RESPAN_GATEWAY_BASE_URL` - `RESPAN_MODEL` -Run from this directory after installing local packages: +Run the complete set from this directory after installing local packages: ```bash -python 01_kernel_function.py -python 02_chat_completion.py -python 03_plugin_tool_call.py -python 04_function_failure.py +python run_all.py ``` + +The runner preserves an externally supplied `RESPAN_EXAMPLE_RUN_ID`, applies a +timeout to every process, continues after individual failures, and reports the +aggregate result. The failure scenario lets the exception escape the decorated +workflow before catching it in `main`, so both native tool and workflow status +are observable. diff --git a/python/tracing/semantic-kernel/_shared.py b/python/tracing/semantic-kernel/_shared.py index 507d090..19b1733 100644 --- a/python/tracing/semantic-kernel/_shared.py +++ b/python/tracing/semantic-kernel/_shared.py @@ -4,19 +4,20 @@ import os from collections.abc import Iterable -from datetime import datetime, timezone +from contextlib import contextmanager from dataclasses import dataclass +from datetime import datetime, timezone from pathlib import Path from dotenv import load_dotenv from openai import AsyncOpenAI -from respan import Respan +from respan import Respan, propagate_attributes from respan_instrumentation_semantic_kernel import SemanticKernelInstrumentor from semantic_kernel import Kernel from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion - DEFAULT_RUN_ID = datetime.now(timezone.utc).strftime("semantic-kernel-%Y%m%d-%H%M%S") +_CLIENTS: list[AsyncOpenAI] = [] @dataclass(frozen=True) @@ -29,7 +30,12 @@ class GatewaySettings: 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(repo_root / ".env", override=False) + + +def example_run_id() -> str: + load_repo_env() + return os.getenv("RESPAN_EXAMPLE_RUN_ID", DEFAULT_RUN_ID) def require_env(name: str) -> str: @@ -55,7 +61,7 @@ def gateway_settings() -> GatewaySettings: def create_respan(app_name: str) -> Respan: load_repo_env() - run_id = os.getenv("RESPAN_EXAMPLE_RUN_ID", DEFAULT_RUN_ID) + run_id = example_run_id() return Respan( app_name=app_name, api_key=require_env("RESPAN_API_KEY"), @@ -64,8 +70,10 @@ def create_respan(app_name: str) -> Respan: is_batching_enabled=False, metadata={ "integration": "semantic-kernel", + "example_set": "semantic-kernel", "example": app_name, "run_id": run_id, + "example_run_id": run_id, }, environment="examples", ) @@ -81,6 +89,7 @@ def create_kernel(*, with_chat_service: bool = True) -> Kernel: api_key=settings.api_key, base_url=settings.base_url, ) + _CLIENTS.append(client) kernel.add_service( OpenAIChatCompletion( ai_model_id=settings.model, @@ -91,6 +100,28 @@ def create_kernel(*, with_chat_service: bool = True) -> Kernel: return kernel +async def close_kernel_clients() -> None: + while _CLIENTS: + client = _CLIENTS.pop() + await client.close() + + +@contextmanager +def example_attributes(app_name: str): + run_id = example_run_id() + with propagate_attributes( + trace_group_identifier=app_name, + metadata={ + "integration": "semantic-kernel", + "example_set": "semantic-kernel", + "example": app_name, + "run_id": run_id, + "example_run_id": run_id, + }, + ): + yield + + def print_result(label: str, value: object) -> None: print(f"{label}: {value}") diff --git a/python/tracing/semantic-kernel/run_all.py b/python/tracing/semantic-kernel/run_all.py new file mode 100644 index 0000000..e8a13ea --- /dev/null +++ b/python/tracing/semantic-kernel/run_all.py @@ -0,0 +1,45 @@ +"""Run every committed Semantic Kernel tracing example.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path + +SCRIPTS = ( + "01_kernel_function.py", + "02_chat_completion.py", + "03_plugin_tool_call.py", + "04_function_failure.py", +) + + +def main() -> None: + root = Path(__file__).resolve().parent + env = os.environ.copy() + env.setdefault( + "RESPAN_EXAMPLE_RUN_ID", + datetime.now(timezone.utc).strftime("otel2-semantic-kernel-%Y%m%dT%H%M%SZ"), + ) + failures: list[str] = [] + for script in SCRIPTS: + try: + result = subprocess.run( + [sys.executable, str(root / script)], + check=False, + env=env, + timeout=180, + ) + except subprocess.TimeoutExpired: + failures.append(f"{script}: timeout") + continue + if result.returncode: + failures.append(f"{script}: exit {result.returncode}") + if failures: + raise SystemExit("; ".join(failures)) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/semantic-kernel/test_contract.py b/python/tracing/semantic-kernel/test_contract.py new file mode 100644 index 0000000..fb90626 --- /dev/null +++ b/python/tracing/semantic-kernel/test_contract.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import ast +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +SCRIPTS = tuple(sorted(ROOT.glob("0[1-4]_*.py"))) + + +def _source(path: Path) -> str: + return path.read_text(encoding="utf-8") + + +def test_exact_marker_and_runner_contract() -> None: + shared = _source(ROOT / "_shared.py") + runner = _source(ROOT / "run_all.py") + assert "override=False" in shared + assert 'os.getenv("RESPAN_EXAMPLE_RUN_ID", DEFAULT_RUN_ID)' in shared + assert "env.setdefault" in runner + assert "subprocess.TimeoutExpired" in runner + assert "check=False" in runner + assert all(path.name in runner for path in SCRIPTS) + + +def test_workflow_roots_have_semantic_inputs_and_nested_teardown() -> None: + for path in SCRIPTS: + tree = ast.parse(_source(path)) + workflows = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef) + and any( + isinstance(decorator, ast.Call) + and getattr(decorator.func, "id", None) == "workflow" + for decorator in node.decorator_list + ) + ] + assert len(workflows) == 1 + assert [argument.arg for argument in workflows[0].args.args] + source = _source(path) + assert "await close_kernel_clients()" in source + assert "respan.shutdown()" in source + + +def test_failure_escapes_decorated_workflow() -> None: + source = _source(ROOT / "04_function_failure.py") + tree = ast.parse(source) + workflow = next( + node + for node in tree.body + if isinstance(node, ast.AsyncFunctionDef) + and node.name == "run_function_failure" + ) + assert not any(isinstance(node, ast.Try) for node in ast.walk(workflow)) + assert "except (RuntimeError, KernelInvokeException)" in source + + +def test_requirements_are_registry_portable() -> None: + requirements = _source(ROOT / "requirements.txt") + assert " -e " not in requirements + assert "file:" not in requirements diff --git a/python/tracing/smolagents/01_code_agent.py b/python/tracing/smolagents/01_code_agent.py index 398ccde..7d92ac0 100644 --- a/python/tracing/smolagents/01_code_agent.py +++ b/python/tracing/smolagents/01_code_agent.py @@ -1,10 +1,9 @@ """Trace a smolagents CodeAgent run with a local tool.""" +from _shared import build_model, build_respan, example_attributes from respan import workflow from smolagents import CodeAgent, tool -from _shared import build_model, build_respan - EXAMPLE_NAME = "code-agent" WORKFLOW_NAME = "smolagents_code_agent_workflow" @@ -25,16 +24,13 @@ def get_city_population(city: str) -> str: @workflow(name=WORKFLOW_NAME) -def execute_code_agent_workflow() -> str: +def execute_code_agent_workflow(prompt: str) -> str: agent = CodeAgent( tools=[get_city_population], model=build_model(), max_steps=3, ) - result = agent.run( - "Use the get_city_population tool for Paris exactly once, then return " - "one sentence with the population fact." - ) + result = agent.run(prompt) print(result) return str(result) @@ -42,7 +38,11 @@ def execute_code_agent_workflow() -> str: def run_code_agent() -> str: respan = build_respan(example_name=EXAMPLE_NAME, workflow_name=WORKFLOW_NAME) try: - return execute_code_agent_workflow() + with example_attributes(EXAMPLE_NAME, WORKFLOW_NAME): + return execute_code_agent_workflow( + "Use the get_city_population tool for Paris exactly once, then return " + "one sentence with the population fact." + ) finally: respan.shutdown() diff --git a/python/tracing/smolagents/02_tool_calling_agent.py b/python/tracing/smolagents/02_tool_calling_agent.py index 2372622..056c720 100644 --- a/python/tracing/smolagents/02_tool_calling_agent.py +++ b/python/tracing/smolagents/02_tool_calling_agent.py @@ -1,10 +1,9 @@ """Trace a smolagents ToolCallingAgent run with a function tool.""" +from _shared import build_model, build_respan, example_attributes from respan import workflow from smolagents import ToolCallingAgent, tool -from _shared import build_model, build_respan - EXAMPLE_NAME = "tool-calling-agent" WORKFLOW_NAME = "smolagents_tool_calling_agent_workflow" @@ -22,16 +21,13 @@ def calculate_invoice_total(unit_price_usd: int, quantity: int) -> str: @workflow(name=WORKFLOW_NAME) -def execute_tool_calling_agent_workflow() -> str: +def execute_tool_calling_agent_workflow(prompt: str) -> str: agent = ToolCallingAgent( tools=[calculate_invoice_total], model=build_model(), max_steps=3, ) - result = agent.run( - "Use the calculate_invoice_total tool for 7 items priced at 9 USD " - "each, then return only the final total sentence." - ) + result = agent.run(prompt) print(result) return str(result) @@ -39,7 +35,11 @@ def execute_tool_calling_agent_workflow() -> str: def run_tool_calling_agent() -> str: respan = build_respan(example_name=EXAMPLE_NAME, workflow_name=WORKFLOW_NAME) try: - return execute_tool_calling_agent_workflow() + with example_attributes(EXAMPLE_NAME, WORKFLOW_NAME): + return execute_tool_calling_agent_workflow( + "Use the calculate_invoice_total tool for 7 items priced at 9 USD " + "each, then return only the final total sentence." + ) finally: respan.shutdown() diff --git a/python/tracing/smolagents/03_expected_tool_failure.py b/python/tracing/smolagents/03_expected_tool_failure.py new file mode 100644 index 0000000..18863eb --- /dev/null +++ b/python/tracing/smolagents/03_expected_tool_failure.py @@ -0,0 +1,39 @@ +"""Trace a deterministic failing smolagents tool call.""" + +from _shared import build_respan, example_attributes +from respan import workflow +from smolagents import tool + +EXAMPLE_NAME = "expected-tool-failure" +WORKFLOW_NAME = "smolagents_expected_tool_failure_workflow" + + +@tool +def fail_city_lookup(city: str) -> str: + """Raise the deterministic failure used by this tracing example. + + Args: + city: City included in the bounded error message. + """ + raise RuntimeError(f"No deterministic population fixture for {city}") + + +@workflow(name=WORKFLOW_NAME) +def execute_expected_failure(city: str) -> None: + fail_city_lookup(city) + + +def main() -> None: + respan = build_respan(EXAMPLE_NAME, WORKFLOW_NAME) + try: + with example_attributes(EXAMPLE_NAME, WORKFLOW_NAME): + try: + execute_expected_failure("Atlantis") + except RuntimeError as exc: + print(f"expected_error={type(exc).__name__}") + finally: + respan.shutdown() + + +if __name__ == "__main__": + main() diff --git a/python/tracing/smolagents/04_streaming_agent.py b/python/tracing/smolagents/04_streaming_agent.py new file mode 100644 index 0000000..58f9c7a --- /dev/null +++ b/python/tracing/smolagents/04_streaming_agent.py @@ -0,0 +1,33 @@ +"""Trace a streamed smolagents ToolCallingAgent run.""" + +from _shared import build_model, build_respan, example_attributes +from respan import workflow +from smolagents import ToolCallingAgent + +EXAMPLE_NAME = "streaming-agent" +WORKFLOW_NAME = "smolagents_streaming_agent_workflow" + + +@workflow(name=WORKFLOW_NAME) +def execute_streaming_agent(prompt: str) -> str: + agent = ToolCallingAgent(tools=[], model=build_model(), max_steps=2) + result = "" + for chunk in agent.run(prompt, stream=True): + value = getattr(chunk, "output", None) + if isinstance(value, str): + result = value + print(result) + return result + + +def main() -> None: + respan = build_respan(EXAMPLE_NAME, WORKFLOW_NAME) + try: + with example_attributes(EXAMPLE_NAME, WORKFLOW_NAME): + execute_streaming_agent("Return exactly: streamed smolagents tracing works") + finally: + respan.shutdown() + + +if __name__ == "__main__": + main() diff --git a/python/tracing/smolagents/README.md b/python/tracing/smolagents/README.md index 6fd2456..dc83118 100644 --- a/python/tracing/smolagents/README.md +++ b/python/tracing/smolagents/README.md @@ -27,8 +27,7 @@ uv venv /tmp/respan-smolagents-example-venv Run the examples: ```bash -python 01_code_agent.py -python 02_tool_calling_agent.py +python run_all.py ``` ## Examples @@ -37,3 +36,9 @@ python 02_tool_calling_agent.py |--------|----------| | `01_code_agent.py` | `CodeAgent` planning, code execution, local tool use, and LLM spans under `smolagents_code_agent_workflow` | | `02_tool_calling_agent.py` | `ToolCallingAgent` function-tool planning and LLM spans under `smolagents_tool_calling_agent_workflow` | +| `03_expected_tool_failure.py` | Deterministic connected tool failure with the exception escaping the workflow wrapper | +| `04_streaming_agent.py` | Streamed `ToolCallingAgent` execution with bounded semantic workflow input/output | + +`run_all.py` preserves one externally supplied `RESPAN_EXAMPLE_RUN_ID`, applies +a timeout to every child process, runs every scenario even after a failure, and +returns a non-zero exit when any scenario fails. diff --git a/python/tracing/smolagents/_shared.py b/python/tracing/smolagents/_shared.py index e64c10d..b8b1adb 100644 --- a/python/tracing/smolagents/_shared.py +++ b/python/tracing/smolagents/_shared.py @@ -3,16 +3,17 @@ from __future__ import annotations import os +from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from dotenv import load_dotenv -from respan import Respan +from respan import Respan, propagate_attributes from respan_instrumentation_smolagents import SmolagentsInstrumentor from smolagents import LiteLLMModel REPO_ROOT = Path(__file__).resolve().parents[3] -load_dotenv(REPO_ROOT / ".env", override=True) +load_dotenv(REPO_ROOT / ".env", override=False) DEFAULT_RESPAN_BASE_URL = "https://api.respan.ai/api" DEFAULT_CUSTOMER_IDENTIFIER = "smolagents-example-user" @@ -72,9 +73,30 @@ def build_respan(example_name: str, workflow_name: str) -> Respan: DEFAULT_CUSTOMER_IDENTIFIER, ), metadata={ + "integration": "smolagents", + "example_set": "smolagents", "example": example_name, "run_id": run_id, + "example_run_id": run_id, "workflow_name": workflow_name, }, environment="examples", + is_batching_enabled=False, ) + + +@contextmanager +def example_attributes(example_name: str, workflow_name: str): + run_id = os.getenv("RESPAN_EXAMPLE_RUN_ID", DEFAULT_RUN_ID) + with propagate_attributes( + trace_group_identifier=workflow_name, + metadata={ + "integration": "smolagents", + "example_set": "smolagents", + "example": example_name, + "run_id": run_id, + "example_run_id": run_id, + "workflow_name": workflow_name, + }, + ): + yield diff --git a/python/tracing/smolagents/run_all.py b/python/tracing/smolagents/run_all.py new file mode 100644 index 0000000..ac8a478 --- /dev/null +++ b/python/tracing/smolagents/run_all.py @@ -0,0 +1,45 @@ +"""Run every committed smolagents tracing example.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path + +SCRIPTS = ( + "01_code_agent.py", + "02_tool_calling_agent.py", + "03_expected_tool_failure.py", + "04_streaming_agent.py", +) + + +def main() -> None: + root = Path(__file__).resolve().parent + env = os.environ.copy() + env.setdefault( + "RESPAN_EXAMPLE_RUN_ID", + datetime.now(timezone.utc).strftime("otel2-smolagents-%Y%m%dT%H%M%SZ"), + ) + failures: list[str] = [] + for script in SCRIPTS: + try: + result = subprocess.run( + [sys.executable, str(root / script)], + check=False, + env=env, + timeout=240, + ) + except subprocess.TimeoutExpired: + failures.append(f"{script}: timeout") + continue + if result.returncode: + failures.append(f"{script}: exit {result.returncode}") + if failures: + raise SystemExit("; ".join(failures)) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/smolagents/test_contract.py b/python/tracing/smolagents/test_contract.py new file mode 100644 index 0000000..aad8d3c --- /dev/null +++ b/python/tracing/smolagents/test_contract.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import ast +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +SCRIPTS = tuple(sorted(ROOT.glob("0[1-4]_*.py"))) + + +def _source(path: Path) -> str: + return path.read_text(encoding="utf-8") + + +def test_exact_marker_and_runner_contract() -> None: + shared = _source(ROOT / "_shared.py") + runner = _source(ROOT / "run_all.py") + assert "override=False" in shared + assert 'os.getenv("RESPAN_EXAMPLE_RUN_ID", DEFAULT_RUN_ID)' in shared + assert "env.setdefault" in runner + assert "subprocess.TimeoutExpired" in runner + assert "check=False" in runner + assert all(path.name in runner for path in SCRIPTS) + + +def test_workflow_roots_use_semantic_inputs_and_shutdown() -> None: + for path in SCRIPTS: + tree = ast.parse(_source(path)) + workflows = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef) + and any( + isinstance(decorator, ast.Call) + and getattr(decorator.func, "id", None) == "workflow" + for decorator in node.decorator_list + ) + ] + assert len(workflows) == 1 + assert [argument.arg for argument in workflows[0].args.args] + assert "respan.shutdown()" in _source(path) + + +def test_failure_and_stream_scenarios_are_contract_shaped() -> None: + failure = _source(ROOT / "03_expected_tool_failure.py") + stream = _source(ROOT / "04_streaming_agent.py") + assert "except RuntimeError" in failure + assert "list(agent.run" not in stream + assert 'getattr(chunk, "output", None)' in stream + + +def test_requirements_are_registry_portable() -> None: + requirements = _source(ROOT / "requirements.txt") + assert " -e " not in requirements + assert "file:" not in requirements