From ac19e7e12071a286759a2efae04d6bae1c4d8280 Mon Sep 17 00:00:00 2001 From: Nikhil Pareek <8347993+nik13@users.noreply.github.com> Date: Sat, 31 Oct 2026 07:04:35 -0700 Subject: [PATCH 1/6] fix(deepseek): preserve provider fields on the shared OpenAI route Retain the deprecated legacy entry point and SDK floor, preserve stream lifetime, and give the parity-capable shared instrumentor single-span ownership. Add network-isolated real SDK and OTLP contracts. Co-Authored-By: Claude Code --- .github/workflows/deepseek-python-tests.yml | 67 +++ python/frameworks/deepseek/README.md | 351 +++------------ .../deepseek/traceai_deepseek/__init__.py | 9 + .../deepseek/traceai_deepseek/_stream.py | 150 +++++++ .../deepseek/traceai_deepseek/_utils.py | 24 +- .../deepseek/traceai_deepseek/_wrappers.py | 85 +--- .../openai/tests/test_deepseek_route.py | 412 ++++++++++++++++++ .../openai/traceai_openai/__init__.py | 4 + .../openai/traceai_openai/_deepseek.py | 141 ++++++ .../openai/traceai_openai/_request.py | 27 +- .../_response_attributes_extractor.py | 14 +- .../openai/traceai_openai/_stream.py | 20 + 12 files changed, 932 insertions(+), 372 deletions(-) create mode 100644 .github/workflows/deepseek-python-tests.yml create mode 100644 python/frameworks/deepseek/traceai_deepseek/_stream.py create mode 100644 python/frameworks/openai/tests/test_deepseek_route.py create mode 100644 python/frameworks/openai/traceai_openai/_deepseek.py diff --git a/.github/workflows/deepseek-python-tests.yml b/.github/workflows/deepseek-python-tests.yml new file mode 100644 index 000000000..353117ab1 --- /dev/null +++ b/.github/workflows/deepseek-python-tests.yml @@ -0,0 +1,67 @@ +name: DeepSeek Python contracts + +on: + pull_request: + branches: [dev, main] + paths: + - 'python/frameworks/deepseek/**' + - 'python/frameworks/openai/**' + - 'python/fi_instrumentation/**' + - '.github/workflows/deepseek-python-tests.yml' + workflow_dispatch: + +permissions: + contents: read + +jobs: + contract: + name: DeepSeek SDK ${{ matrix.openai }} / core ${{ matrix.core }} + runs-on: ubuntu-latest + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + include: + - {python: '3.10', openai: '1.0.0', core: source, shared: false} + - {python: '3.10', openai: '1.69.0', core: source, shared: true} + - {python: '3.12', openai: '3.42.0', core: source, shared: true} + - {python: '3.12', openai: '3.42.0', core: '1.2.2', shared: true} + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + - name: Install isolated contract dependencies + env: + SDK_VERSION: ${{ matrix.openai }} + CORE_VERSION: ${{ matrix.core }} + SHARED: ${{ matrix.shared }} + run: | + python -m venv "$RUNNER_TEMP/deepseek-venv" + PYTHON="$RUNNER_TEMP/deepseek-venv/bin/python" + if [ "$CORE_VERSION" = source ]; then + "$PYTHON" -m pip install -e python + else + "$PYTHON" -m pip install "fi-instrumentation-otel==$CORE_VERSION" + fi + "$PYTHON" -m pip install -e python/frameworks/deepseek "openai==$SDK_VERSION" 'httpx==0.27.2' pytest pytest-asyncio + if [ "$SHARED" = true ]; then + "$PYTHON" -m pip install -e python/frameworks/openai "openai==$SDK_VERSION" + fi + - name: Run with only loopback networking + env: + SHARED: ${{ matrix.shared }} + FI_BASE_URL: http://127.0.0.1:9 + FI_GRPC_URL: http://127.0.0.1:9 + run: | + sudo -E unshare --net bash -euo pipefail -c ' + ip link set lo up + PYTHON="$RUNNER_TEMP/deepseek-venv/bin/python" + "$PYTHON" -I -m pytest -q --import-mode=importlib python/frameworks/openai/tests/test_deepseek_route.py + "$PYTHON" -I -m pytest -q python/frameworks/deepseek/tests + if [ "$SHARED" = true ]; then + "$PYTHON" -I -m pytest -q python/frameworks/openai/tests + fi + ' diff --git a/python/frameworks/deepseek/README.md b/python/frameworks/deepseek/README.md index 04919916c..aae058518 100644 --- a/python/frameworks/deepseek/README.md +++ b/python/frameworks/deepseek/README.md @@ -1,327 +1,96 @@ -# TraceAI DeepSeek Instrumentation +# Trace DeepSeek with the OpenAI integration -OpenTelemetry instrumentation for [DeepSeek](https://www.deepseek.com/) - chat completions, reasoning models, and function calling. +Use `traceAI-openai` to trace DeepSeek Chat Completions and Responses in Future AGI. The shared integration preserves OpenAI's synchronous and asynchronous streaming interfaces and records DeepSeek reasoning and cache fields when the API returns them. -## Installation +## Install -```bash -pip install traceai-deepseek -``` - -## Features - -- Automatic tracing of DeepSeek API calls via OpenAI SDK -- Support for DeepSeek Chat, Coder, and Reasoner (R1) models -- Streaming response support -- Token usage tracking including prompt cache metrics -- Reasoning content extraction for R1 models -- Function/tool calling support -- Full OpenTelemetry semantic conventions compliance - -## Usage - -### Basic Setup - -```python -from openai import OpenAI -from opentelemetry import trace -from opentelemetry.sdk.trace import TracerProvider -from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor - -from traceai_deepseek import DeepSeekInstrumentor - -# Set up tracing -provider = TracerProvider() -provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter())) -trace.set_tracer_provider(provider) - -# Instrument DeepSeek -DeepSeekInstrumentor().instrument(tracer_provider=provider) - -# Use DeepSeek via OpenAI SDK -client = OpenAI( - api_key="your-deepseek-api-key", - base_url="https://api.deepseek.com/v1" -) - -response = client.chat.completions.create( - model="deepseek-chat", - messages=[{"role": "user", "content": "Hello!"}] -) -print(response.choices[0].message.content) -``` - -### DeepSeek Chat - -```python -from openai import OpenAI +Use Python 3.10–3.13 and OpenAI SDK 1.69 or newer. -client = OpenAI( - api_key="your-deepseek-api-key", - base_url="https://api.deepseek.com/v1" -) - -# Simple chat -response = client.chat.completions.create( - model="deepseek-chat", - messages=[ - {"role": "system", "content": "You are a helpful assistant."}, - {"role": "user", "content": "What is machine learning?"} - ], - temperature=0.7, - max_tokens=1024 -) -print(response.choices[0].message.content) +```bash +pip install --upgrade traceAI-openai fi-instrumentation-otel openai ``` -### DeepSeek Reasoner (R1 Models) - -DeepSeek R1 models provide reasoning capabilities with detailed thought processes. - -```python -from openai import OpenAI - -client = OpenAI( - api_key="your-deepseek-api-key", - base_url="https://api.deepseek.com/v1" -) - -# Using DeepSeek Reasoner -response = client.chat.completions.create( - model="deepseek-reasoner", - messages=[ - {"role": "user", "content": "Solve this step by step: If a train travels 120 km in 2 hours, what is its average speed?"} - ] -) - -# Access the reasoning content (thought process) -message = response.choices[0].message -print("Reasoning:", message.reasoning_content) -print("Answer:", message.content) - -# Token usage includes reasoning tokens -print(f"Reasoning tokens: {response.usage.completion_tokens_details.reasoning_tokens}") -``` +## Configure and run -### Streaming Responses +Set `DEEPSEEK_API_KEY`, `FI_API_KEY`, and `FI_SECRET_KEY` in your environment. Set `DEEPSEEK_MODEL` to a model available to your DeepSeek account. Obtain a DeepSeek key from the [DeepSeek platform](https://platform.deepseek.com/). ```python -from openai import OpenAI - -client = OpenAI( - api_key="your-deepseek-api-key", - base_url="https://api.deepseek.com/v1" -) - -# Streaming chat -stream = client.chat.completions.create( - model="deepseek-chat", - messages=[{"role": "user", "content": "Tell me a story"}], - stream=True -) +import os -for chunk in stream: - if chunk.choices[0].delta.content: - print(chunk.choices[0].delta.content, end="", flush=True) -print() - -# Streaming with R1 (includes reasoning) -stream = client.chat.completions.create( - model="deepseek-reasoner", - messages=[{"role": "user", "content": "What is 15% of 240?"}], - stream=True -) - -reasoning_content = "" -content = "" -for chunk in stream: - delta = chunk.choices[0].delta - if hasattr(delta, 'reasoning_content') and delta.reasoning_content: - reasoning_content += delta.reasoning_content - if delta.content: - content += delta.content - -print("Reasoning:", reasoning_content) -print("Answer:", content) -``` - -### Function Calling / Tools - -```python +from fi_instrumentation import TraceConfig, register +from fi_instrumentation.fi_types import ProjectType from openai import OpenAI -import json - -client = OpenAI( - api_key="your-deepseek-api-key", - base_url="https://api.deepseek.com/v1" -) - -tools = [ - { - "type": "function", - "function": { - "name": "get_weather", - "description": "Get the current weather for a location", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "The city name" - }, - "unit": { - "type": "string", - "enum": ["celsius", "fahrenheit"], - "description": "Temperature unit" - } - }, - "required": ["location"] - } - } - } -] +from traceai_openai import OpenAIInstrumentor -response = client.chat.completions.create( - model="deepseek-chat", - messages=[{"role": "user", "content": "What's the weather in Paris?"}], - tools=tools, - tool_choice="auto" -) - -message = response.choices[0].message -if message.tool_calls: - for tool_call in message.tool_calls: - print(f"Function: {tool_call.function.name}") - print(f"Arguments: {tool_call.function.arguments}") +provider = register(project_type=ProjectType.OBSERVE, project_name="deepseek-app") +OpenAIInstrumentor().instrument( + tracer_provider=provider, + config=TraceConfig(hide_inputs=True, hide_outputs=True), +) + +try: + with OpenAI( + api_key=os.environ["DEEPSEEK_API_KEY"], + base_url="https://api.deepseek.com", + ) as client: + response = client.chat.completions.create( + model=os.environ["DEEPSEEK_MODEL"], + messages=[{"role": "user", "content": "What is 15% of 240?"}], + ) + print(response.choices[0].message.content) +finally: + provider.force_flush() + provider.shutdown() ``` -### Async Usage - -```python -import asyncio -from openai import AsyncOpenAI - -async def main(): - client = AsyncOpenAI( - api_key="your-deepseek-api-key", - base_url="https://api.deepseek.com/v1" - ) +Open **Observe** and select `deepseek-app`. The model span has provider `deepseek`, the model name, and token counts when returned by the API. Prompt and output text are hidden with the configuration above; set the corresponding options to `False` only when you intend to capture that content. Reasoning and tool-argument aliases follow the output privacy policy. To also hide invocation parameters and exception details, use `hide_llm_invocation_parameters=True` and `hide_error_details=True`. Tool definitions remain visible; do not put secrets in tool names, descriptions, or schemas. - response = await client.chat.completions.create( - model="deepseek-chat", - messages=[{"role": "user", "content": "Hello!"}] - ) - print(response.choices[0].message.content) +The instrumentor wraps OpenAI SDK requests process-wide, including clients configured for other providers. Provider attribution uses the API host, not the requested model name; custom proxies retain the `openai` fallback. -asyncio.run(main()) -``` +## Streaming and Responses -### JSON Mode +Use the same instrumented client for Chat Completions streaming. Keep the stream in a context manager so early exits also close the HTTP response and finish its span. ```python -from openai import OpenAI - -client = OpenAI( - api_key="your-deepseek-api-key", - base_url="https://api.deepseek.com/v1" -) - -response = client.chat.completions.create( - model="deepseek-chat", - messages=[ - {"role": "system", "content": "Output valid JSON only."}, - {"role": "user", "content": "List 3 programming languages with their main use cases"} - ], - response_format={"type": "json_object"} -) - -import json -data = json.loads(response.choices[0].message.content) -print(data) +with client.chat.completions.create( + model=os.environ["DEEPSEEK_MODEL"], + messages=[{"role": "user", "content": "Explain prompt caching in one sentence."}], + stream=True, + stream_options={"include_usage": True}, +) as stream: + for chunk in stream: + if chunk.choices and chunk.choices[0].delta.content: + print(chunk.choices[0].delta.content, end="") ``` -## Configuration Options +Run this block inside the client's `with` block above. For async code, use `AsyncOpenAI`, await `create()`, and use `async with` / `async for`. -### TraceConfig +DeepSeek's [Responses API](https://api-docs.deepseek.com/guides/responses_api) uses the same base URL and shared instrumentor: ```python -from fi_instrumentation import TraceConfig -from traceai_deepseek import DeepSeekInstrumentor - -config = TraceConfig( - hide_inputs=False, - hide_outputs=False, -) - -DeepSeekInstrumentor().instrument( - tracer_provider=provider, - config=config +response = client.responses.create( + model=os.environ["DEEPSEEK_MODEL"], + input="Explain prompt caching in one sentence.", ) +print(response.output_text) ``` -## Captured Attributes - -### Common Attributes - -| Attribute | Description | -|-----------|-------------| -| `gen_ai.span.kind` | "LLM" | -| `gen_ai.provider.name` | "deepseek" | -| `gen_ai.request.model` | Model name (deepseek-chat, deepseek-reasoner, etc.) | -| `gen_ai.usage.input_tokens` | Input token count | -| `gen_ai.usage.output_tokens` | Output token count | -| `gen_ai.usage.total_tokens` | Total token count | - -### DeepSeek-Specific Attributes - -| Attribute | Description | -|-----------|-------------| -| `deepseek.response_id` | Unique response ID | -| `deepseek.finish_reason` | Response finish reason (stop, tool_calls, length) | -| `deepseek.prompt_cache_hit_tokens` | Tokens served from cache | -| `deepseek.prompt_cache_miss_tokens` | Tokens not in cache | -| `deepseek.reasoning_tokens` | Tokens used for reasoning (R1 models) | -| `deepseek.reasoning_content` | Full reasoning text (R1 models) | -| `deepseek.tool_calls_count` | Number of tool calls | -| `deepseek.tools_count` | Number of tools provided | - -## Available Models - -| Model | Description | -|-------|-------------| -| `deepseek-chat` | General chat model | -| `deepseek-coder` | Code-focused model | -| `deepseek-reasoner` | Reasoning model (R1) with chain-of-thought | - -## Prompt Caching - -DeepSeek supports prompt caching to reduce costs and latency for repeated prompts. The instrumentation tracks cache hit/miss metrics: - -```python -response = client.chat.completions.create( - model="deepseek-chat", - messages=[ - {"role": "system", "content": "You are a helpful assistant."}, # Cached - {"role": "user", "content": "Hello!"} - ] -) - -# Access cache metrics from usage -print(f"Cache hits: {response.usage.prompt_cache_hit_tokens}") -print(f"Cache misses: {response.usage.prompt_cache_miss_tokens}") -``` +## Migrate from `DeepSeekInstrumentor` -## Important Notes +`DeepSeekInstrumentor` remains available for existing Chat Completions applications, including OpenAI SDK 1.0–1.68. It is deprecated, not removed. It does not instrument the Responses API; upgrade the SDK and switch to `OpenAIInstrumentor` for that API. -1. **OpenAI SDK Required**: DeepSeek uses the OpenAI-compatible API, so you need the `openai` package installed. +Replace the `traceai_deepseek` import and `DeepSeekInstrumentor().instrument(...)` call with the shared setup above. Upgrade both tracing packages before enabling both entry points during migration. With the updated packages, the shared instrumentor owns the request and its tracer provider and privacy configuration apply. An older shared package without DeepSeek extraction keeps the legacy behavior (including duplicate spans) rather than silently losing reasoning and cache fields. Removing either instrumentor leaves the other operational. -2. **Base URL**: Always set `base_url="https://api.deepseek.com/v1"` when creating the client. +The shared route preserves the Chat Completions `deepseek.reasoning_content`, `deepseek.output_reasoning`, `deepseek.prompt_cache_hit_tokens`, `deepseek.prompt_cache_miss_tokens`, `deepseek.reasoning_tokens`, response ID, finish reason, and tool-call aliases when present. Token/cache counters are span attributes, not a separate metrics stream. Span names become `ChatCompletion` / `Response`; `output.value` contains the shared integration's text/tool summary rather than the legacy raw response JSON. Request tools use the shared semantic attributes rather than `deepseek.tools_count`. -3. **API Key**: Get your API key from the [DeepSeek Platform](https://platform.deepseek.com/). +Do not uninstall the legacy package until every application has switched its imports and no longer needs the older SDK floor. -4. **Selective Instrumentation**: The instrumentor only traces calls to DeepSeek's API. Regular OpenAI API calls are not affected. +## Troubleshooting -5. **R1 Models**: When using `deepseek-reasoner`, the `reasoning_content` field contains the model's thought process before the final answer. +- **No spans:** instrument before creating the client, check your Future AGI keys, and flush the provider before a short-lived process exits. +- **Provider is `openai`:** use the exact `api.deepseek.com` API host. A custom proxy is not automatically attributed to DeepSeek. +- **Missing streaming token counts:** request `stream_options={"include_usage": True}` and consume the stream through its final usage chunk. +- **401 or model error:** check your DeepSeek key and the model's availability. See the [DeepSeek API documentation](https://api-docs.deepseek.com/). ## License diff --git a/python/frameworks/deepseek/traceai_deepseek/__init__.py b/python/frameworks/deepseek/traceai_deepseek/__init__.py index 242eb9074..569460d6d 100644 --- a/python/frameworks/deepseek/traceai_deepseek/__init__.py +++ b/python/frameworks/deepseek/traceai_deepseek/__init__.py @@ -1,4 +1,5 @@ import logging +import warnings from importlib import import_module from typing import Any, Collection @@ -55,6 +56,14 @@ def instrumentation_dependencies(self) -> Collection[str]: return _instruments def _instrument(self, **kwargs: Any) -> None: + warnings.warn( + "DeepSeekInstrumentor is a legacy Chat Completions integration. " + "Use traceAI-openai's OpenAIInstrumentor for shared DeepSeek Chat " + "Completions and Responses tracing. When both are enabled, the " + "shared instrumentor's tracer provider and privacy config apply.", + DeprecationWarning, + stacklevel=2, + ) if not (tracer_provider := kwargs.get("tracer_provider")): tracer_provider = trace_api.get_tracer_provider() if not (config := kwargs.get("config")): diff --git a/python/frameworks/deepseek/traceai_deepseek/_stream.py b/python/frameworks/deepseek/traceai_deepseek/_stream.py new file mode 100644 index 000000000..51238ce8f --- /dev/null +++ b/python/frameworks/deepseek/traceai_deepseek/_stream.py @@ -0,0 +1,150 @@ +"""Compatibility proxies: preserve the OpenAI stream's lifetime and interface.""" + +import logging +from typing import Any + +from opentelemetry import trace +from wrapt import ObjectProxy + +from traceai_deepseek._utils import _finish_tracing, _to_dict + +logger = logging.getLogger(__name__) + + +class _StreamState: + def __init__(self, span: Any, extractor: Any, parameters: Any) -> None: + self.span = span + self.extractor = extractor + self.parameters = parameters + self.finished = False + + def process(self, chunk: Any) -> None: + if self.finished: + return + try: + self.extractor.process_chunk(_to_dict(chunk)) + except Exception: + logger.exception("Failed to accumulate DeepSeek response") + + def finish(self, error: Any = None) -> None: + if self.finished: + return + self.finished = True + status = trace.Status(trace.StatusCode.OK) + if error is not None: + self.span.record_exception(error) + status = trace.Status( + trace.StatusCode.ERROR, f"{type(error).__name__}: {error}" + ) + try: + _finish_tracing( + status, + self.span, + self.extractor.get_attributes(), + self.extractor.get_extra_attributes(self.parameters), + ) + except Exception: + logger.exception("Failed to finalize DeepSeek stream") + self.span.finish_tracing(status=status) + + +class _Stream(ObjectProxy): + def __init__(self, stream: Any, state: _StreamState) -> None: + super().__init__(stream) + self._self_state = state + self._self_iterator = iter(stream) + + def __iter__(self): + return self + + def __next__(self): + try: + chunk = next(self._self_iterator) + except StopIteration: + self._self_state.finish() + raise + except BaseException as error: + self._self_state.finish(error) + raise + self._self_state.process(chunk) + return chunk + + def __enter__(self): + if enter := getattr(self.__wrapped__, "__enter__", None): + enter() + return self + + def __exit__(self, exc_type, exc, tb): + try: + if exit := getattr(self.__wrapped__, "__exit__", None): + return exit(exc_type, exc, tb) + self.__wrapped__.response.close() + except BaseException as error: + exc = error + raise + finally: + self._self_state.finish(exc) + + def close(self): + try: + if close := getattr(self.__wrapped__, "close", None): + return close() + return self.__wrapped__.response.close() + except BaseException as error: + self._self_state.finish(error) + raise + finally: + self._self_state.finish() + + +class _AsyncStream(ObjectProxy): + def __init__(self, stream: Any, state: _StreamState) -> None: + super().__init__(stream) + self._self_state = state + self._self_iterator = stream.__aiter__() + + def __aiter__(self): + return self + + async def __anext__(self): + try: + chunk = await self._self_iterator.__anext__() + except StopAsyncIteration: + self._self_state.finish() + raise + except BaseException as error: + self._self_state.finish(error) + raise + self._self_state.process(chunk) + return chunk + + async def __aenter__(self): + if enter := getattr(self.__wrapped__, "__aenter__", None): + await enter() + return self + + async def __aexit__(self, exc_type, exc, tb): + try: + if exit := getattr(self.__wrapped__, "__aexit__", None): + return await exit(exc_type, exc, tb) + await self.__wrapped__.response.aclose() + except BaseException as error: + exc = error + raise + finally: + self._self_state.finish(exc) + + async def close(self): + try: + if close := getattr(self.__wrapped__, "close", None): + return await close() + return await self.__wrapped__.response.aclose() + except BaseException as error: + self._self_state.finish(error) + raise + finally: + self._self_state.finish() + + async def aclose(self): + # Retain the async generator's old close spelling as well as SDK close(). + return await self.close() diff --git a/python/frameworks/deepseek/traceai_deepseek/_utils.py b/python/frameworks/deepseek/traceai_deepseek/_utils.py index 9dbb91aed..4309625c8 100644 --- a/python/frameworks/deepseek/traceai_deepseek/_utils.py +++ b/python/frameworks/deepseek/traceai_deepseek/_utils.py @@ -1,4 +1,6 @@ import logging +import sys +from urllib.parse import urlsplit from typing import Any, Dict, Iterator, Mapping, Optional, Tuple from fi_instrumentation import safe_json_dumps @@ -10,11 +12,19 @@ logger = logging.getLogger(__name__) logger.addHandler(logging.NullHandler()) -# DeepSeek base URL patterns -DEEPSEEK_BASE_URLS = [ - "api.deepseek.com", - "deepseek.com", -] +def shared_openai_is_active() -> bool: + """The shared instrumentor owns requests when both entry points are enabled. + + Do not import an optional package on the legacy OpenAI 1.0 route. Checking + at call time supports either initialization and independent teardown order. + """ + module = sys.modules.get("traceai_openai") + instrumentor = getattr(module, "OpenAIInstrumentor", None) + return bool( + instrumentor + and getattr(instrumentor, "_supports_deepseek_extensions", False) + and instrumentor().is_instrumented_by_opentelemetry + ) def is_deepseek_client(instance: Any) -> bool: @@ -28,8 +38,8 @@ def is_deepseek_client(instance: Any) -> bool: base_url = getattr(client, "base_url", None) if base_url is not None: - base_url_str = str(base_url).lower() - return any(url in base_url_str for url in DEEPSEEK_BASE_URLS) + host = urlsplit(str(base_url)).hostname + return bool(host and host.lower().rstrip(".") == "api.deepseek.com") except Exception: pass return False diff --git a/python/frameworks/deepseek/traceai_deepseek/_wrappers.py b/python/frameworks/deepseek/traceai_deepseek/_wrappers.py index 732bf4f99..b51649d04 100644 --- a/python/frameworks/deepseek/traceai_deepseek/_wrappers.py +++ b/python/frameworks/deepseek/traceai_deepseek/_wrappers.py @@ -17,7 +17,10 @@ _ChatCompletionResponseAttributesExtractor, _StreamingChatCompletionResponseExtractor, ) -from traceai_deepseek._utils import _finish_tracing, _to_dict, is_deepseek_client +from traceai_deepseek._stream import _AsyncStream, _Stream, _StreamState +from traceai_deepseek._utils import ( + _finish_tracing, _to_dict, is_deepseek_client, shared_openai_is_active, +) from traceai_deepseek._with_span import _WithSpan logger = logging.getLogger(__name__) @@ -76,7 +79,7 @@ def __call__( args: Tuple[Any, ...], kwargs: Mapping[str, Any], ) -> Any: - if context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY): + if shared_openai_is_active() or context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY): return wrapped(*args, **kwargs) # Check if this is a DeepSeek client @@ -134,40 +137,10 @@ def __call__( def _handle_streaming_response( self, span: _WithSpan, response: Any, request_parameters: Dict[str, Any] ): - """Handle streaming response.""" - stream_extractor = _StreamingChatCompletionResponseExtractor(config=self._config) - - def streaming_wrapper(): - try: - for chunk in response: - chunk_dict = _to_dict(chunk) - stream_extractor.process_chunk(chunk_dict) - yield chunk - - except Exception as exception: - span.record_exception(exception) - status = trace_api.Status( - status_code=trace_api.StatusCode.ERROR, - description=f"{type(exception).__name__}: {exception}", - ) - span.finish_tracing(status=status) - raise - else: - try: - _finish_tracing( - status=trace_api.Status(status_code=trace_api.StatusCode.OK), - with_span=span, - attributes=stream_extractor.get_attributes(), - extra_attributes=stream_extractor.get_extra_attributes(request_parameters), - ) - except Exception: - logger.exception("Failed to finalize streaming response") - span.finish_tracing() - finally: - if span._span.is_recording(): - span.finish_tracing() - - return streaming_wrapper() + return _Stream(response, _StreamState( + span, _StreamingChatCompletionResponseExtractor(config=self._config), + request_parameters, + )) class _AsyncChatCompletionWrapper(_WithTracer): @@ -185,7 +158,7 @@ async def __call__( args: Tuple[Any, ...], kwargs: Mapping[str, Any], ) -> Any: - if context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY): + if shared_openai_is_active() or context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY): return await wrapped(*args, **kwargs) # Check if this is a DeepSeek client @@ -241,37 +214,7 @@ async def __call__( def _handle_async_streaming_response( self, span: _WithSpan, response: Any, request_parameters: Dict[str, Any] ): - """Handle async streaming response.""" - stream_extractor = _StreamingChatCompletionResponseExtractor(config=self._config) - - async def async_streaming_wrapper(): - try: - async for chunk in response: - chunk_dict = _to_dict(chunk) - stream_extractor.process_chunk(chunk_dict) - yield chunk - - except Exception as exception: - span.record_exception(exception) - status = trace_api.Status( - status_code=trace_api.StatusCode.ERROR, - description=f"{type(exception).__name__}: {exception}", - ) - span.finish_tracing(status=status) - raise - else: - try: - _finish_tracing( - status=trace_api.Status(status_code=trace_api.StatusCode.OK), - with_span=span, - attributes=stream_extractor.get_attributes(), - extra_attributes=stream_extractor.get_extra_attributes(request_parameters), - ) - except Exception: - logger.exception("Failed to finalize async streaming response") - span.finish_tracing() - finally: - if span._span.is_recording(): - span.finish_tracing() - - return async_streaming_wrapper() + return _AsyncStream(response, _StreamState( + span, _StreamingChatCompletionResponseExtractor(config=self._config), + request_parameters, + )) diff --git a/python/frameworks/openai/tests/test_deepseek_route.py b/python/frameworks/openai/tests/test_deepseek_route.py new file mode 100644 index 000000000..b2b2cac8c --- /dev/null +++ b/python/frameworks/openai/tests/test_deepseek_route.py @@ -0,0 +1,412 @@ +"""Real SDK, synthetic provider transport, and localhost OTLP migration contract.""" + +import asyncio +import json +import threading +from dataclasses import fields +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import openai +import pytest +from fi_instrumentation import TraceConfig +from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( + ExportTraceServiceRequest, +) +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor + +DeepSeekInstrumentor = pytest.importorskip("traceai_deepseek").DeepSeekInstrumentor + +try: + from openai._base_client import httpx2 as httpx +except ImportError: + from openai._base_client import httpx + +MODERN = tuple(map(int, openai.__version__.split(".")[:2])) >= (1, 69) +if MODERN: + from traceai_openai import OpenAIInstrumentor + +BASE = dict(id="completion-offline", created=1, model="deepseek-reasoner") +REASONING = "PRIVATE_REASONING_7382" +ANSWER = "PRIVATE_ANSWER_1294" +TOOL_ARGS = '{"secret":"PRIVATE_TOOL_9271"}' +USAGE = dict( + prompt_tokens=5, + completion_tokens=7, + total_tokens=12, + prompt_cache_hit_tokens=3, + prompt_cache_miss_tokens=2, + completion_tokens_details=dict(reasoning_tokens=4), +) +TOOL = dict( + id="call-offline", + type="function", + function=dict(name="lookup", arguments=TOOL_ARGS), +) +MESSAGE = dict( + role="assistant", content=ANSWER, reasoning_content=REASONING, tool_calls=[TOOL] +) +BODY = dict( + BASE, + object="chat.completion", + usage=USAGE, + choices=[dict(index=0, message=MESSAGE, finish_reason="tool_calls")], +) +RESPONSE = dict( + id="resp-offline", + object="response", + created_at=1, + status="completed", + model="deepseek-reasoner", + output=[ + dict( + id="msg-offline", + type="message", + status="completed", + role="assistant", + content=[dict(type="output_text", text=ANSWER, annotations=[])], + ) + ], + usage=dict( + input_tokens=5, + output_tokens=7, + total_tokens=12, + input_tokens_details=dict(cached_tokens=3), + output_tokens_details=dict(reasoning_tokens=4), + ), +) + + +def respond(request): + payload = json.loads(request.content) + if request.url.path.endswith("/responses"): + if payload.get("stream"): + event = dict( + type="response.completed", sequence_number=1, response=RESPONSE + ) + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content="data: " + json.dumps(event) + "\n\n", + ) + return httpx.Response(200, json=RESPONSE) + if payload.get("stream"): + deltas = [ + dict(role="assistant", reasoning_content=REASONING[:10]), + dict( + reasoning_content=REASONING[10:], + content=ANSWER, + tool_calls=[ + dict( + TOOL, + index=0, + function=dict(name="lookup", arguments=TOOL_ARGS[:12]), + ) + ], + ), + dict(tool_calls=[dict(index=0, function=dict(arguments=TOOL_ARGS[12:]))]), + ] + chunks = [ + dict( + BASE, + object="chat.completion.chunk", + usage=None, + choices=[dict(index=0, delta=d, finish_reason=None)], + ) + for d in deltas + ] + chunks.append( + dict( + BASE, + object="chat.completion.chunk", + usage=USAGE, + choices=[dict(index=0, delta={}, finish_reason="tool_calls")], + ) + ) + data = "".join("data: " + json.dumps(c) + "\n\n" for c in chunks) + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=data + "data: [DONE]\n\n", + ) + return httpx.Response(200, json=BODY) + + +@pytest.fixture +def wire(monkeypatch): + for field in fields(TraceConfig): + monkeypatch.delenv(field.metadata["env_var"], raising=False) + captured = [] + + class Receiver(BaseHTTPRequestHandler): + def do_POST(self): + payload = self.rfile.read(int(self.headers["Content-Length"])) + decoded = ExportTraceServiceRequest.FromString(payload) + captured.extend( + span + for resource in decoded.resource_spans + for scope in resource.scope_spans + for span in scope.spans + ) + self.send_response(200) + self.end_headers() + + def log_message(self, *_args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Receiver) + thread = threading.Thread( + target=lambda: server.serve_forever(poll_interval=0.01), daemon=True + ) + thread.start() + provider = TracerProvider() + provider.add_span_processor( + SimpleSpanProcessor( + OTLPSpanExporter( + endpoint=f"http://127.0.0.1:{server.server_port}/v1/traces", timeout=5 + ) + ) + ) + instruments = [] + + def start(route, **privacy): + if route != "legacy" and not MODERN: + pytest.skip("Shared OpenAI instrumentor requires SDK >=1.69") + classes = dict(legacy=[DeepSeekInstrumentor]) + if MODERN: + classes.update( + shared=[OpenAIInstrumentor], + legacy_first=[DeepSeekInstrumentor, OpenAIInstrumentor], + shared_first=[OpenAIInstrumentor, DeepSeekInstrumentor], + ) + for cls in classes[route]: + inst = cls() + inst.instrument(tracer_provider=provider, config=TraceConfig(**privacy)) + instruments.append(inst) + return captured + + yield start + for inst in reversed(instruments): + if inst.is_instrumented_by_opentelemetry: + inst.uninstrument() + provider.shutdown() + server.shutdown() + thread.join() + server.server_close() + + +def attrs(spans): + assert len(spans) == 1 + return { + a.key: getattr(a.value, a.value.WhichOneof("value")) + for a in spans[0].attributes + } + + +def call( + asynchronous=False, + streaming=False, + mode="full", + host="api.deepseek.com", + responses=False, +): + params = dict(model="deepseek-reasoner", stream=streaming) + params.update( + dict(input="PRIVATE_INPUT_7134") + if responses + else dict(messages=[dict(role="user", content="PRIVATE_INPUT_7134")]) + ) + + async def run(): + async with openai.AsyncOpenAI( + api_key="offline-placeholder", + base_url=f"https://{host}/v1", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(respond)), + ) as client: + resource = client.responses if responses else client.chat.completions + response = await resource.create(**params) + if streaming: + if mode == "close": + await response.close() + else: + async with response as entered: + assert entered is response + if mode != "empty": + async for _chunk in response: + if mode == "early": + break + assert response.response.is_closed + return response + + if asynchronous: + return asyncio.run(run()) + with openai.OpenAI( + api_key="offline-placeholder", + base_url=f"https://{host}/v1", + http_client=httpx.Client(transport=httpx.MockTransport(respond)), + ) as client: + resource = client.responses if responses else client.chat.completions + response = resource.create(**params) + if streaming: + if mode == "close": + response.close() + else: + with response as entered: + assert entered is response + if mode != "empty": + for _chunk in response: + if mode == "early": + break + assert response.response.is_closed + return response + + +@pytest.mark.parametrize("route", ["legacy", "shared", "legacy_first", "shared_first"]) +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +def test_chat_migration_parity(wire, route, asynchronous, streaming): + spans = wire(route) + response = call(asynchronous, streaming) + a = attrs(spans) + assert a["gen_ai.provider.name"] == "deepseek" + assert a["gen_ai.request.model"] == "deepseek-reasoner" + assert a["deepseek.prompt_cache_hit_tokens"] == 3 + assert a["deepseek.prompt_cache_miss_tokens"] == 2 + assert a["deepseek.reasoning_tokens"] == 4 + assert a["gen_ai.usage.output_tokens"] == 7 + assert a["deepseek.reasoning_content"] == REASONING + assert a["deepseek.output_reasoning"] == REASONING + assert a["gen_ai.output.messages.0.reasoning_content"] == REASONING + assert a["deepseek.finish_reason"] == "tool_calls" + assert a["deepseek.tool_calls_count"] == 1 + assert a["deepseek.tool_calls.0.function.arguments"] == TOOL_ARGS + if not streaming: + assert response.choices[0].message.reasoning_content == REASONING + + +@pytest.mark.parametrize("route", ["legacy", "shared", "legacy_first", "shared_first"]) +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("mode", ["early", "empty", "close"]) +def test_stream_lifecycle(wire, route, asynchronous, mode): + spans = wire(route) + call(asynchronous, streaming=True, mode=mode) + assert attrs(spans)["gen_ai.provider.name"] == "deepseek" + + +@pytest.mark.parametrize("route", ["shared", "legacy_first", "shared_first"]) +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +def test_responses(wire, route, asynchronous, streaming): + spans = wire(route) + call(asynchronous, streaming, responses=True) + assert attrs(spans)["gen_ai.provider.name"] == "deepseek" + assert attrs(spans)["gen_ai.usage.output_tokens"] == 7 + + +@pytest.mark.parametrize("route", ["legacy", "shared", "legacy_first", "shared_first"]) +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize( + "policy", ["hide_outputs", "hide_output_messages", "hide_output_text"] +) +def test_output_privacy_on_wire(wire, route, asynchronous, streaming, policy): + spans = wire(route, **{policy: True}) + call(asynchronous, streaming) + a = attrs(spans) + assert a["deepseek.prompt_cache_hit_tokens"] == 3 + assert a["deepseek.reasoning_tokens"] == 4 + if policy != "hide_outputs": + a.pop("output.value", None) # Message/text policies do not hide raw output. + for value in [REASONING, ANSWER, "PRIVATE_TOOL_9271"]: + assert value not in json.dumps(a) + + +@pytest.mark.parametrize( + "host", + [ + "api.deepseek.com.evil.example", + "proxy.example/api.deepseek.com", + "notdeepseek.com", + ], +) +def test_legacy_exact_host(wire, host): + spans = wire("legacy") + call(host=host) + assert not spans + + +@pytest.mark.parametrize("route", ["legacy_first", "shared_first"]) +@pytest.mark.parametrize("remove", ["legacy", "shared"]) +def test_independent_teardown(wire, route, remove): + spans = wire(route) + ( + DeepSeekInstrumentor() if remove == "legacy" else OpenAIInstrumentor() + ).uninstrument() + call() + assert attrs(spans)["deepseek.reasoning_content"] == REASONING + + +@pytest.mark.parametrize("route", ["legacy", "shared", "legacy_first", "shared_first"]) +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_api_error_privacy(wire, monkeypatch, route, asynchronous): + spans = wire(route, hide_error_details=True) + + def fail(request): + return httpx.Response( + 400, + json=dict( + error=dict(message="PRIVATE_ERROR_1842", type="invalid_request_error") + ), + ) + + monkeypatch.setattr(__import__(__name__, fromlist=["respond"]), "respond", fail) + with pytest.raises(openai.BadRequestError, match="PRIVATE_ERROR_1842"): + call(asynchronous) + assert attrs(spans)["gen_ai.request.model"] == "deepseek-reasoner" + assert spans[0].status.code == 2 + assert b"PRIVATE_ERROR_1842" not in spans[0].SerializeToString() + + +@pytest.mark.parametrize("route", ["legacy", "shared", "legacy_first", "shared_first"]) +def test_idempotent_instrumentation(wire, route): + spans = wire(route) + DeepSeekInstrumentor().instrument() + call() + assert attrs(spans)["gen_ai.provider.name"] == "deepseek" + # The fixture owns legacy teardown only when it started that instrumentor. + if route == "shared": + DeepSeekInstrumentor().uninstrument() + + +@pytest.mark.parametrize("route", ["legacy_first", "shared_first"]) +def test_shared_configuration_owns_request(wire, route): + spans = wire(route) + DeepSeekInstrumentor().uninstrument() + DeepSeekInstrumentor().instrument(config=TraceConfig(hide_outputs=True)) + call() + assert attrs(spans)["deepseek.reasoning_content"] == REASONING + + +def test_legacy_does_not_defer_to_shared_without_extension_support(wire, monkeypatch): + spans = wire("legacy_first") + monkeypatch.setattr(OpenAIInstrumentor, "_supports_deepseek_extensions", False) + call() + assert len(spans) == 2 # Existing behavior with an older shared package. + legacy = [span for span in spans if span.name == "deepseek.chat.completions"] + assert attrs(legacy)["deepseek.reasoning_content"] == REASONING + + +def test_shared_non_deepseek_has_no_aliases(wire): + spans = wire("shared") + call(host="api.openai.com") + assert attrs(spans)["gen_ai.provider.name"] == "openai" + assert not any(key.startswith("deepseek.") for key in attrs(spans)) + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_explicit_configuration_overrides_environment(wire, monkeypatch, asynchronous): + spans = wire("shared", hide_outputs=False) + monkeypatch.setenv("FI_HIDE_OUTPUTS", "true") + call(asynchronous, streaming=True) + assert attrs(spans)["deepseek.reasoning_content"] == REASONING diff --git a/python/frameworks/openai/traceai_openai/__init__.py b/python/frameworks/openai/traceai_openai/__init__.py index 0747c8745..eba537544 100644 --- a/python/frameworks/openai/traceai_openai/__init__.py +++ b/python/frameworks/openai/traceai_openai/__init__.py @@ -30,6 +30,10 @@ class OpenAIInstrumentor(BaseInstrumentor): # type: ignore An instrumentor for openai """ + # Legacy provider entry points only defer once extraction/privacy parity is + # present, not merely because an older shared package happens to be enabled. + _supports_deepseek_extensions = True + __slots__ = ( "_original_request", "_original_async_request", diff --git a/python/frameworks/openai/traceai_openai/_deepseek.py b/python/frameworks/openai/traceai_openai/_deepseek.py new file mode 100644 index 000000000..fc5de10d0 --- /dev/null +++ b/python/frameworks/openai/traceai_openai/_deepseek.py @@ -0,0 +1,141 @@ +"""DeepSeek response fields on the shared OpenAI request/stream lifecycle.""" + +from typing import Any, Iterator, Tuple + +from fi_instrumentation import TraceConfig +from fi_instrumentation.fi_types import MessageAttributes, SpanAttributes +from opentelemetry.util.types import AttributeValue + +from traceai_openai._response_accumulator import ( + _ChatCompletionAccumulator, + _IndexedAccumulator, + _StringAccumulator, + _ValuesAccumulator, +) +from traceai_openai._utils import _to_dict + + +def _output_alias( + config: TraceConfig, key: str, value: Any +) -> Iterator[Tuple[str, AttributeValue]]: + # Alias keys are not known to older cores: apply the output-text policy + # explicitly, including explicit configuration overrides of environment flags. + masked = config.mask( + f"{SpanAttributes.GEN_AI_OUTPUT_MESSAGES}.0.{MessageAttributes.MESSAGE_CONTENT}", + value, + ) + if masked is not None: + yield key, masked + + +def deepseek_attributes( + response: Any, config: TraceConfig +) -> Iterator[Tuple[str, AttributeValue]]: + data = _to_dict(response) + if response_id := data.get("id"): + yield "deepseek.response_id", response_id + usage = data.get("usage") or {} + for key in ("prompt_cache_hit_tokens", "prompt_cache_miss_tokens"): + if (value := usage.get(key)) is not None: + yield f"deepseek.{key}", value + details = ( + usage.get("completion_tokens_details") + or usage.get("output_tokens_details") + or {} + ) + if (tokens := details.get("reasoning_tokens")) is not None: + yield "deepseek.reasoning_tokens", tokens + for position, choice in enumerate(data.get("choices") or []): + index = choice.get("index", position) + message = choice.get("message") or {} + if reasoning := message.get("reasoning_content"): + yield from _output_alias( + config, + f"{SpanAttributes.GEN_AI_OUTPUT_MESSAGES}.{index}.reasoning_content", + reasoning, + ) + # Retain the existing single-response aliases for the first choice. + if position == 0: + yield from _output_alias( + config, "deepseek.reasoning_content", reasoning + ) + yield from _output_alias(config, "deepseek.output_reasoning", reasoning) + if position != 0: + continue + if finish_reason := choice.get("finish_reason"): + yield "deepseek.finish_reason", finish_reason + if choice.get("logprobs"): + yield "deepseek.has_logprobs", True + tools = message.get("tool_calls") or [] + if tools: + yield "deepseek.tool_calls_count", len(tools) + for i, tool in enumerate(tools[:10]): + for key in ("id", "type"): + if (value := tool.get(key)) is not None: + yield from _output_alias( + config, f"deepseek.tool_calls.{i}.{key}", value + ) + for key in ("name", "arguments"): + if (value := (tool.get("function") or {}).get(key)) is not None: + yield from _output_alias( + config, f"deepseek.tool_calls.{i}.function.{key}", value + ) + + +class _DeepSeekChatAccumulator(_ChatCompletionAccumulator): + """Accumulate extension fields without replacing the SDK stream interface. + + Use the shared indexed/string primitives so tool arguments and reasoning + fragments stay separate across choices. No second list of raw chunks is kept. + """ + + def __init__(self, **kwargs: Any) -> None: + super().__init__(**kwargs) + self._values = _ValuesAccumulator( + choices=_IndexedAccumulator( + lambda: _ValuesAccumulator( + message=_ValuesAccumulator( + content=_StringAccumulator(), + reasoning_content=_StringAccumulator(), + function_call=_ValuesAccumulator( + arguments=_StringAccumulator() + ), + tool_calls=_IndexedAccumulator( + lambda: _ValuesAccumulator( + function=_ValuesAccumulator( + arguments=_StringAccumulator() + ) + ) + ), + ) + ) + ) + ) + + def process_chunk(self, chunk: Any) -> None: + values = _to_dict(chunk) + choices = [] + for choice in values.get("choices") or []: + choice = dict(choice) + choice["message"] = choice.pop("delta", None) or {} + choices.append(choice) + self._values += {**values, "choices": choices} + self._is_null = False + self._cached_result = None + + def get_attributes(self) -> Iterator[Tuple[str, AttributeValue]]: + result = self._result() or {} + choices = result.get("choices") or [] + if choices: + message = choices[0].get("message") or {} + if content := message.get("content"): + yield SpanAttributes.OUTPUT_VALUE, content + elif tools := message.get("tool_calls"): + yield ( + SpanAttributes.OUTPUT_VALUE, + " \n ".join( + f"Function: {tool.get('function', {}).get('name', '')}" + f"({tool.get('function', {}).get('arguments', '')})" + for tool in tools + ), + ) diff --git a/python/frameworks/openai/traceai_openai/_request.py b/python/frameworks/openai/traceai_openai/_request.py index 61f6a9aab..4c931de60 100644 --- a/python/frameworks/openai/traceai_openai/_request.py +++ b/python/frameworks/openai/traceai_openai/_request.py @@ -16,6 +16,7 @@ from opentelemetry.context import _SUPPRESS_INSTRUMENTATION_KEY from opentelemetry.trace import INVALID_SPAN from opentelemetry.util.types import AttributeValue +from traceai_openai._deepseek import _DeepSeekChatAccumulator from traceai_openai._providers import provider_from_host from traceai_openai._request_attributes_extractor import _RequestAttributesExtractor from traceai_openai._response_accumulator import ( @@ -198,6 +199,7 @@ def _finalize_response( with_span: _WithSpan, cast_to: type, request_parameters: Mapping[str, Any], + instance: Any = None, ) -> Any: """ Monkey-patch the response object to trace the stream, or finish tracing if the response is @@ -209,6 +211,13 @@ def _finalize_response( except Exception: logger.exception(f"Failed to parse response of type {type(response)}") + provider = provider_from_host(getattr(getattr(instance, "base_url", None), "host", None)) + extractor = self._response_attributes_extractor + if provider == "deepseek": + # Keep host-specific extraction local to this request: concurrent + # requests through another provider must not inherit DeepSeek aliases. + extractor = _ResponseAttributesExtractor(self._openai, self._config, provider=provider) + if self._is_streaming(response): # Determine if the response is asynchronous if self._is_async_stream(response): @@ -226,6 +235,20 @@ def _finalize_response( else None ) + if provider == "deepseek": + if cast_to is self._openai.types.chat.ChatCompletion: + response_accumulator = _DeepSeekChatAccumulator( + request_parameters=request_parameters, + chat_completion_type=cast_to, + response_attributes_extractor=extractor, + ) + elif cast_to is self._openai.types.responses.response.Response: + response_accumulator = _ResponsesAccumulator( + request_parameters=request_parameters, + chat_completion_type=cast_to, + response_attributes_extractor=extractor, + ) + # Wrap the response with the appropriate stream wrapper wrapped_stream = stream_wrapper( stream=response, @@ -242,7 +265,7 @@ def _finalize_response( has_attributes=_ResponseAttributes( request_parameters=request_parameters, response=response, - response_attributes_extractor=self._response_attributes_extractor, + response_attributes_extractor=extractor, ), ) return response @@ -336,6 +359,7 @@ def __call__( with_span=with_span, cast_to=cast_to, request_parameters=request_parameters, + instance=instance, ) except Exception as e: status = trace_api.Status( @@ -419,6 +443,7 @@ async def __call__( with_span=with_span, cast_to=cast_to, request_parameters=request_parameters, + instance=instance, ) except Exception: logger.exception( diff --git a/python/frameworks/openai/traceai_openai/_response_attributes_extractor.py b/python/frameworks/openai/traceai_openai/_response_attributes_extractor.py index e57fb61d0..290f61c08 100644 --- a/python/frameworks/openai/traceai_openai/_response_attributes_extractor.py +++ b/python/frameworks/openai/traceai_openai/_response_attributes_extractor.py @@ -27,6 +27,7 @@ ) from opentelemetry.util.types import AttributeValue from traceai_openai._attributes._responses_api import _ResponsesApiAttributes +from traceai_openai._deepseek import _output_alias, deepseek_attributes from traceai_openai._utils import _get_openai_version, _get_texts if TYPE_CHECKING: @@ -56,9 +57,12 @@ class _ResponseAttributesExtractor: "_create_embedding_response_type", "_images_response_type", "_responses_type", + "_provider", ) - def __init__(self, openai: ModuleType, config: Optional[TraceConfig] = None) -> None: + def __init__(self, openai: ModuleType, config: Optional[TraceConfig] = None, + provider: Optional[str] = None) -> None: + self._provider = provider self._openai = openai self._config = config if config is not None else TraceConfig() self._chat_completion_type: Type["ChatCompletion"] = ( @@ -78,6 +82,8 @@ def get_attributes_from_response( response: Any, request_parameters: Mapping[str, Any], ) -> Iterator[Tuple[str, AttributeValue]]: + if self._provider == "deepseek": + yield from deepseek_attributes(response, self._config) if isinstance(response, self._chat_completion_type): yield from self._get_attributes_from_chat_completion( completion=response, @@ -134,7 +140,11 @@ def _get_attributes_from_chat_completion( for key, value in self._get_attributes_from_chat_completion_message( message ): - yield f"{SpanAttributes.GEN_AI_OUTPUT_MESSAGES}.{index}.{key}", value + attribute = f"{SpanAttributes.GEN_AI_OUTPUT_MESSAGES}.{index}.{key}" + if self._provider == "deepseek" and key != MessageAttributes.MESSAGE_ROLE: + yield from _output_alias(self._config, attribute, value) + else: + yield attribute, value def _get_attributes_from_completion( self, diff --git a/python/frameworks/openai/traceai_openai/_stream.py b/python/frameworks/openai/traceai_openai/_stream.py index a3a60042e..5eae8d769 100644 --- a/python/frameworks/openai/traceai_openai/_stream.py +++ b/python/frameworks/openai/traceai_openai/_stream.py @@ -74,6 +74,16 @@ def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> Any: finally: _finish_on_exit(self, exc) + def close(self) -> Any: + error = None + try: + return self.__wrapped__.close() + except BaseException as exc: + error = exc + raise + finally: + _finish_on_exit(self, error) + def __iter__(self) -> Iterator[Any]: return self @@ -164,6 +174,16 @@ async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> Any: finally: _finish_on_exit(self, exc) + async def close(self) -> Any: + error = None + try: + return await self.__wrapped__.close() + except BaseException as exc: + error = exc + raise + finally: + _finish_on_exit(self, error) + def __aiter__(self): return self From a14984ae2331b944b698f2f3f0601217aac144e3 Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Mon, 2 Nov 2026 07:02:29 -0800 Subject: [PATCH 2/6] test(deepseek): cover latest OpenAI SDK release Run the shared-route contracts on OpenAI 3.42.1 with source and released core, retaining both SDK floors. Co-Authored-By: Claude Code --- .github/workflows/deepseek-python-tests.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/deepseek-python-tests.yml b/.github/workflows/deepseek-python-tests.yml index 353117ab1..c0670631d 100644 --- a/.github/workflows/deepseek-python-tests.yml +++ b/.github/workflows/deepseek-python-tests.yml @@ -24,8 +24,8 @@ jobs: include: - {python: '3.10', openai: '1.0.0', core: source, shared: false} - {python: '3.10', openai: '1.69.0', core: source, shared: true} - - {python: '3.12', openai: '3.42.0', core: source, shared: true} - - {python: '3.12', openai: '3.42.0', core: '1.2.2', shared: true} + - {python: '3.12', openai: '3.42.1', core: source, shared: true} + - {python: '3.12', openai: '3.42.1', core: '1.2.2', shared: true} steps: - uses: actions/checkout@v4 with: From f748d84991735bfe850c134465a47b9612dabc21 Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Mon, 2 Nov 2026 08:01:41 -0800 Subject: [PATCH 3/6] fix(openai): protect DeepSeek Responses and finalize cancelled streams Mask DeepSeek Responses tool output under the configured text policy, retain incomplete and failed terminal payloads, and keep cancellation finalization idempotent. Exclude vendor errors from raw output and report failed Responses as errors without changing SDK return values. Add real-SDK localhost OTLP contracts across sync, async, stream, and non-stream routes. Co-Authored-By: Claude Code --- .../openai/tests/test_deepseek_route.py | 147 ++++++++++++++++++ .../openai/traceai_openai/_request.py | 10 +- .../traceai_openai/_response_accumulator.py | 38 +++-- .../_response_attributes_extractor.py | 12 +- .../openai/traceai_openai/_span_io_handler.py | 6 +- .../openai/traceai_openai/_stream.py | 16 +- .../openai/traceai_openai/_with_span.py | 2 +- 7 files changed, 215 insertions(+), 16 deletions(-) diff --git a/python/frameworks/openai/tests/test_deepseek_route.py b/python/frameworks/openai/tests/test_deepseek_route.py index b2b2cac8c..5507b17da 100644 --- a/python/frameworks/openai/tests/test_deepseek_route.py +++ b/python/frameworks/openai/tests/test_deepseek_route.py @@ -1,7 +1,9 @@ """Real SDK, synthetic provider transport, and localhost OTLP migration contract.""" import asyncio +import copy import json +import sys import threading from dataclasses import fields from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -410,3 +412,148 @@ def test_explicit_configuration_overrides_environment(wire, monkeypatch, asynchr monkeypatch.setenv("FI_HIDE_OUTPUTS", "true") call(asynchronous, streaming=True) assert attrs(spans)["deepseek.reasoning_content"] == REASONING + + +@pytest.mark.parametrize("route", ["legacy", "shared", "legacy_first", "shared_first"]) +@pytest.mark.parametrize("explicit_close", [False, True]) +def test_cancelled_stream_is_finished_as_error(wire, route, explicit_close): + spans = wire(route, hide_error_details=True) + + async def run(): + reading = asyncio.Event() + cancellations = [] + + class WaitingBody(httpx.AsyncByteStream): + async def __aiter__(self): + reading.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError as error: + cancellations.append(error) + raise + yield b"" + + def respond(request): + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, stream=WaitingBody() + ) + + async with openai.AsyncOpenAI( + api_key="offline-key", + base_url="https://api.deepseek.com", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(respond)), + ) as client: + stream = await client.chat.completions.create( + model="deepseek-reasoner", + messages=[{"role": "user", "content": "hello"}], + stream=True, + ) + + async def consume(): + try: + async for _ in stream: + pass + except asyncio.CancelledError as error: + cancellations.append(error) + raise + finally: + if explicit_close: + await stream.close() + + task = asyncio.create_task(consume()) + await asyncio.wait_for(reading.wait(), timeout=5) + task.cancel("PRIVATE_CANCEL_REASON") + with pytest.raises(asyncio.CancelledError): + await task + # Python 3.10 may lose the cancellation message at the Task boundary. + # Assert identity where the exception passes through instrumentation. + assert len(cancellations) == 2 + assert cancellations[0] is cancellations[1] + assert len(spans) == 1, "cancelled requests must export exactly one span" + assert spans[0].status.code == 2 + assert any(event.name == "exception" for event in spans[0].events) + assert b"PRIVATE_CANCEL_REASON" not in spans[0].SerializeToString() + await stream.close() + assert len(spans) == 1 + assert spans[0].status.code == 2 + + asyncio.run(run()) + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize( + "policy", [None, "hide_outputs", "hide_output_messages", "hide_output_text"] +) +def test_responses_tool_output_privacy(wire, monkeypatch, asynchronous, streaming, policy): + spans = wire("shared", **({policy: True} if policy else {})) + body = copy.deepcopy(RESPONSE) + body["output"] = [ + dict( + type="function_call", id="fc-offline", call_id="call-offline", + name="lookup", arguments=TOOL_ARGS, status="completed", + ) + ] + monkeypatch.setattr(sys.modules[__name__], "RESPONSE", body) + call(asynchronous=asynchronous, streaming=streaming, responses=True) + a = attrs(spans) + assert a["gen_ai.usage.output_tokens"] == 7 + if not policy: + arguments = "gen_ai.output.messages.0.message.tool_calls.0.tool_call.function.arguments" + assert a[arguments] == TOOL_ARGS + return + if policy != "hide_outputs": + a.pop("output.value", None) # Raw output has its own documented hide policy. + assert "PRIVATE_TOOL_9271" not in json.dumps(a) + + +@pytest.mark.parametrize( + "event_type", ["response.completed", "response.incomplete", "response.failed"] +) +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("has_output", [False, True]) +@pytest.mark.parametrize( + "streaming,mode", [(True, "full"), (True, "early"), (False, "full")] +) +def test_responses_terminal_events( + wire, monkeypatch, event_type, asynchronous, has_output, streaming, mode +): + spans = wire("shared", hide_error_details=True) + body = copy.deepcopy(RESPONSE) + body["status"] = event_type.split(".")[1] + if not has_output: + body["output"] = [] + if event_type == "response.incomplete": + body["incomplete_details"] = {"reason": "max_output_tokens"} + if event_type == "response.failed": + body["error"] = {"code": "server_error", "message": "PRIVATE_VENDOR_ERROR"} + + def respond(request): + if not streaming: + return httpx.Response(200, json=body) + event = dict(type=event_type, sequence_number=1, response=body) + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, + content="event: " + event_type + "\ndata: " + json.dumps(event) + "\n\n", + ) + + monkeypatch.setattr(sys.modules[__name__], "respond", respond) + response = call( + asynchronous=asynchronous, streaming=streaming, responses=True, mode=mode + ) + if not streaming: + assert response.status == body["status"] + if event_type == "response.failed": + assert response.error.message == "PRIVATE_VENDOR_ERROR" + a = attrs(spans) + assert a["gen_ai.usage.input_tokens"] == 5 + assert a["gen_ai.usage.output_tokens"] == 7 + assert a["gen_ai.usage.total_tokens"] == 12 + assert a["deepseek.reasoning_tokens"] == 4 + assert a["deepseek.response_id"] == "resp-offline" + if has_output: + if streaming: + assert a["output.value"] == ANSWER + assert a["gen_ai.output.messages.0.message.content"] == ANSWER + assert spans[0].status.code == (2 if event_type == "response.failed" else 1) + assert b"PRIVATE_VENDOR_ERROR" not in spans[0].SerializeToString() diff --git a/python/frameworks/openai/traceai_openai/_request.py b/python/frameworks/openai/traceai_openai/_request.py index c12554b18..99358c7a6 100644 --- a/python/frameworks/openai/traceai_openai/_request.py +++ b/python/frameworks/openai/traceai_openai/_request.py @@ -262,8 +262,16 @@ def _finalize_response( return wrapped_stream else: + failed_response = ( + isinstance(response, self._openai.types.responses.response.Response) + and response.status == "failed" + ) _finish_tracing( - status=trace_api.Status(status_code=trace_api.StatusCode.OK), + status=trace_api.Status( + status_code=( + trace_api.StatusCode.ERROR if failed_response else trace_api.StatusCode.OK + ) + ), with_span=with_span, has_attributes=_ResponseAttributes( request_parameters=request_parameters, diff --git a/python/frameworks/openai/traceai_openai/_response_accumulator.py b/python/frameworks/openai/traceai_openai/_response_accumulator.py index 17417d213..c7d0b1da7 100644 --- a/python/frameworks/openai/traceai_openai/_response_accumulator.py +++ b/python/frameworks/openai/traceai_openai/_response_accumulator.py @@ -19,6 +19,7 @@ from fi_instrumentation import safe_json_dumps from fi_instrumentation.fi_types import FiMimeTypeValues, SpanAttributes +from opentelemetry import trace as trace_api from opentelemetry.util.types import AttributeValue from traceai_openai._utils import _as_output_attributes, _to_dict, _ValueAndType, _io_value_and_type @@ -26,7 +27,6 @@ from openai.types import Completion from openai.types.chat import ChatCompletion from openai.types.responses.response import Response - from openai.types.responses.response_completed_event import ResponseCompletedEvent __all__ = ( "_CompletionAccumulator", @@ -64,19 +64,29 @@ def __init__( self._response_attributes_extractor = response_attributes_extractor self._is_null = True self._cached_result: Optional[Dict[str, Any]] = None - self._values: Optional["ResponseCompletedEvent"] = None + self._values: Optional["Response"] = None def process_chunk(self, chunk: Any) -> None: - if type(chunk).__name__ == "ResponseCompletedEvent": + if getattr(chunk, "type", None) in ( + "response.completed", + "response.incomplete", + "response.failed", + ): self._is_null = False self._cached_result = None - self._values = chunk + self._values = chunk.response def _result(self) -> Any: if self._is_null: return None - if self._values: - return self._values.response + return self._values + + def get_status(self) -> Optional[trace_api.Status]: + if self._values is not None and self._values.status == "failed": + # Vendor error messages can contain private input. Status alone is + # sufficient here; do not copy error details into output or events. + return trace_api.Status(status_code=trace_api.StatusCode.ERROR) + return None def get_attributes(self) -> Iterator[Tuple[str, AttributeValue]]: if not (result := self._result()): @@ -97,10 +107,12 @@ def get_attributes(self) -> Iterator[Tuple[str, AttributeValue]]: if output_content: yield SpanAttributes.OUTPUT_VALUE, output_content else: - # Fallback to full response if we can't extract content - yield from _as_output_attributes( - _io_value_and_type(result), - ) + # Preserve non-text output, but never treat vendor error details as + # assistant output (output.value has a separate privacy policy). + output = { + key: value for key, value in _to_dict(result).items() if key != "error" + } + yield from _as_output_attributes(_io_value_and_type(output)) def get_extra_attributes(self) -> Iterator[Tuple[str, AttributeValue]]: if not (result := self._result()): @@ -224,6 +236,9 @@ def get_attributes(self) -> Iterator[Tuple[str, AttributeValue]]: output_value = safe_json_dumps(messages) yield SpanAttributes.OUTPUT_VALUE, output_value + def get_status(self) -> Optional[trace_api.Status]: + return None + def get_extra_attributes(self) -> Iterator[Tuple[str, AttributeValue]]: if not (result := self._result()): return @@ -297,6 +312,9 @@ def get_attributes(self) -> Iterator[Tuple[str, AttributeValue]]: _ValueAndType(json_string, FiMimeTypeValues.JSON), ) + def get_status(self) -> Optional[trace_api.Status]: + return None + def get_extra_attributes(self) -> Iterator[Tuple[str, AttributeValue]]: if not (result := self._result()): return diff --git a/python/frameworks/openai/traceai_openai/_response_attributes_extractor.py b/python/frameworks/openai/traceai_openai/_response_attributes_extractor.py index 290f61c08..39331b6b5 100644 --- a/python/frameworks/openai/traceai_openai/_response_attributes_extractor.py +++ b/python/frameworks/openai/traceai_openai/_response_attributes_extractor.py @@ -117,7 +117,17 @@ def _get_attributes_from_responses_response( response: Response, request_parameters: Mapping[str, Any], ) -> Iterator[Tuple[str, AttributeValue]]: - yield from _ResponsesApiAttributes._get_attributes_from_response(response) + for key, value in _ResponsesApiAttributes._get_attributes_from_response(response): + if ( + self._provider == "deepseek" + and key.startswith(f"{SpanAttributes.GEN_AI_OUTPUT_MESSAGES}.") + and not key.endswith(f".{MessageAttributes.MESSAGE_ROLE}") + ): + # Tool arguments and reasoning are output text too; older cores + # only recognize the standard message.content privacy key. + yield from _output_alias(self._config, key, value) + else: + yield key, value def _get_attributes_from_chat_completion( self, diff --git a/python/frameworks/openai/traceai_openai/_span_io_handler.py b/python/frameworks/openai/traceai_openai/_span_io_handler.py index 028d2faa5..ab4c25920 100644 --- a/python/frameworks/openai/traceai_openai/_span_io_handler.py +++ b/python/frameworks/openai/traceai_openai/_span_io_handler.py @@ -98,7 +98,11 @@ def _json_output_for(output_data: Any) -> Optional[str]: return None if isinstance(output_data, Response): # Lead with the answer text (the SDK's `output_text`) so previews show it. - body = json.loads(output_data.model_dump_json(exclude_none=True)) + # Errors are not assistant output. Keep vendor details out of the raw + # output fallback, whose privacy policy is independent of error masking. + body = json.loads( + output_data.model_dump_json(exclude_none=True, exclude={"error"}) + ) output_text = getattr(output_data, "output_text", None) if output_text: body = {"output_text": output_text, **body} diff --git a/python/frameworks/openai/traceai_openai/_stream.py b/python/frameworks/openai/traceai_openai/_stream.py index 5eae8d769..40d8c7932 100644 --- a/python/frameworks/openai/traceai_openai/_stream.py +++ b/python/frameworks/openai/traceai_openai/_stream.py @@ -20,6 +20,8 @@ class _ResponseAccumulator(Protocol): def process_chunk(self, chunk: Any) -> None: ... + def get_status(self) -> Optional[trace_api.Status]: ... + def get_attributes(self) -> Iterator[Tuple[str, AttributeValue]]: ... def get_extra_attributes(self) -> Iterator[Tuple[str, AttributeValue]]: ... @@ -95,7 +97,7 @@ def __next__(self) -> Any: status = trace_api.Status(status_code=trace_api.StatusCode.OK) self._finish_tracing(status=status) raise - except Exception as exception: + except BaseException as exception: if not self._self_is_finished: status = trace_api.Status( status_code=trace_api.StatusCode.ERROR, @@ -125,6 +127,11 @@ def _finish_tracing( self, status: Optional[trace_api.Status] = None, ) -> None: + if ( + self._self_response_accumulator is not None + and (status is None or status.status_code != trace_api.StatusCode.ERROR) + ): + status = self._self_response_accumulator.get_status() or status _finish_tracing( status=status, with_span=self._self_with_span, @@ -195,7 +202,7 @@ async def __anext__(self): status = trace_api.Status(status_code=trace_api.StatusCode.OK) self._finish_tracing(status=status) raise - except Exception as exception: + except BaseException as exception: if not self._self_is_finished: status = trace_api.Status( status_code=trace_api.StatusCode.ERROR, @@ -225,6 +232,11 @@ def _finish_tracing( self, status: Optional[trace_api.Status] = None, ) -> None: + if ( + self._self_response_accumulator is not None + and (status is None or status.status_code != trace_api.StatusCode.ERROR) + ): + status = self._self_response_accumulator.get_status() or status _finish_tracing( status=status, with_span=self._self_with_span, diff --git a/python/frameworks/openai/traceai_openai/_with_span.py b/python/frameworks/openai/traceai_openai/_with_span.py index 5f5419ad0..923a31af8 100644 --- a/python/frameworks/openai/traceai_openai/_with_span.py +++ b/python/frameworks/openai/traceai_openai/_with_span.py @@ -49,7 +49,7 @@ def __init__( def is_finished(self) -> bool: return self._is_finished - def record_exception(self, exception: Exception) -> None: + def record_exception(self, exception: BaseException) -> None: if self._is_finished: return try: From e7a981701a4f49f4b50cbd15f8cf4098434bec06 Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Mon, 2 Nov 2026 08:48:36 -0800 Subject: [PATCH 4/6] fix(deepseek): finalize cancelled requests and separate response reasoning Finish async request spans when cancellation occurs before create returns, retaining exception identity and configured error privacy. Extract typed and dictionary reasoning content separately from assistant output and retain reasoning summaries. Co-Authored-By: Claude Code --- .../deepseek/traceai_deepseek/_with_span.py | 2 +- .../deepseek/traceai_deepseek/_wrappers.py | 3 +- .../openai/tests/test_deepseek_route.py | 107 ++++++++++++++++++ .../_attributes/_responses_api.py | 25 ++-- .../openai/traceai_openai/_request.py | 3 +- .../traceai_openai/_response_accumulator.py | 9 +- 6 files changed, 136 insertions(+), 13 deletions(-) diff --git a/python/frameworks/deepseek/traceai_deepseek/_with_span.py b/python/frameworks/deepseek/traceai_deepseek/_with_span.py index dd0d014dd..d470d9de6 100644 --- a/python/frameworks/deepseek/traceai_deepseek/_with_span.py +++ b/python/frameworks/deepseek/traceai_deepseek/_with_span.py @@ -18,7 +18,7 @@ def __init__( self._context_attributes = context_attributes or {} self._extra_attributes = extra_attributes or {} - def record_exception(self, exception: Exception) -> None: + def record_exception(self, exception: BaseException) -> None: """Record an exception on the span.""" if self._span.is_recording(): self._span.record_exception(exception) diff --git a/python/frameworks/deepseek/traceai_deepseek/_wrappers.py b/python/frameworks/deepseek/traceai_deepseek/_wrappers.py index b51649d04..34942cbff 100644 --- a/python/frameworks/deepseek/traceai_deepseek/_wrappers.py +++ b/python/frameworks/deepseek/traceai_deepseek/_wrappers.py @@ -184,7 +184,8 @@ async def __call__( ) as span: try: response = await wrapped(*args, **kwargs) - except Exception as exception: + # Cancellation can arrive before a stream/response takes ownership. + except BaseException as exception: span.record_exception(exception) status = trace_api.Status( status_code=trace_api.StatusCode.ERROR, diff --git a/python/frameworks/openai/tests/test_deepseek_route.py b/python/frameworks/openai/tests/test_deepseek_route.py index 5507b17da..2c786aed0 100644 --- a/python/frameworks/openai/tests/test_deepseek_route.py +++ b/python/frameworks/openai/tests/test_deepseek_route.py @@ -557,3 +557,110 @@ def respond(request): assert a["gen_ai.output.messages.0.message.content"] == ANSWER assert spans[0].status.code == (2 if event_type == "response.failed" else 1) assert b"PRIVATE_VENDOR_ERROR" not in spans[0].SerializeToString() + + +@pytest.mark.parametrize("route", ["legacy", "shared", "legacy_first", "shared_first"]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("responses", [False, True]) +@pytest.mark.parametrize("hide_error_details", [False, True]) +def test_cancelled_create_is_finished_as_error( + wire, route, streaming, responses, hide_error_details +): + spans = wire(route, hide_error_details=hide_error_details) + if responses and route == "legacy": + pytest.skip("Legacy instrumentor supports Chat Completions only") + + async def run(): + waiting = asyncio.Event() + cancellations = [] + + async def respond(request): + waiting.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError as error: + cancellations.append(error) + raise + + async with openai.AsyncOpenAI( + api_key="offline-key", + base_url="https://api.deepseek.com", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(respond)), + ) as client: + async def create(): + resource = client.responses if responses else client.chat.completions + params = dict(model="deepseek-reasoner", stream=streaming) + params.update( + dict(input="hello") if responses + else dict(messages=[dict(role="user", content="hello")]) + ) + try: + await resource.create(**params) + except asyncio.CancelledError as error: + cancellations.append(error) + raise + + task = asyncio.create_task(create()) + await asyncio.wait_for(waiting.wait(), timeout=5) + task.cancel("PRIVATE_CREATE_CANCEL") + with pytest.raises(asyncio.CancelledError): + await task + # Compare before the Task boundary, which can discard details on 3.10. + assert len(cancellations) == 2 + assert cancellations[0] is cancellations[1] + assert len(spans) == 1, "cancelled create must finish before client close" + assert spans[0].status.code == 2 + assert sum(event.name == "exception" for event in spans[0].events) == 1 + exported = spans[0].SerializeToString() + assert (b"PRIVATE_CREATE_CANCEL" not in exported) == hide_error_details + assert len(spans) == 1 + assert spans[0].status.code == 2 + + asyncio.run(run()) + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("has_summary", [False, True]) +@pytest.mark.parametrize( + "policy", [None, "hide_outputs", "hide_output_messages", "hide_output_text"] +) +def test_responses_plaintext_reasoning( + wire, monkeypatch, asynchronous, streaming, has_summary, policy +): + spans = wire("shared", **({policy: True} if policy else {})) + body = copy.deepcopy(RESPONSE) + summary = "PRIVATE_SUMMARY_7382" + body["output"].insert(0, dict( + id="rs-offline", type="reasoning", status="completed", + summary=[dict(type="summary_text", text=summary)] if has_summary else [], + content=[dict(type="reasoning_text", text=text) for text in (REASONING, "more")], + )) + monkeypatch.setattr(sys.modules[__name__], "RESPONSE", body) + response = call(asynchronous=asynchronous, streaming=streaming, responses=True) + if not streaming: + # Instrumentation must not strip reasoning from the SDK object. + assert response.model_dump()["output"][0]["content"][0]["text"] == REASONING + a = attrs(spans) + assert a["gen_ai.usage.output_tokens"] == 7 + assert spans[0].status.code == 1 + if policy != "hide_outputs": + if streaming: + assert a["output.value"] == ANSWER + else: + assert json.loads(a["output.value"])["output_text"] == ANSWER + if policy: + if policy != "hide_outputs": + a.pop("output.value", None) # Raw output has an independent policy. + encoded = json.dumps(a) + assert REASONING not in encoded + assert summary not in encoded + assert ANSWER not in encoded + else: + prefix = "gen_ai.output.messages.0.message.contents." + offset = int(has_summary) + if has_summary: + assert a[f"{prefix}0.message_content.text"] == summary + assert a[f"{prefix}{offset}.message_content.text"] == REASONING + assert a[f"{prefix}{offset + 1}.message_content.text"] == "more" + assert a["gen_ai.output.messages.1.message.content"] == ANSWER diff --git a/python/frameworks/openai/traceai_openai/_attributes/_responses_api.py b/python/frameworks/openai/traceai_openai/_attributes/_responses_api.py index f92d6ec07..0080d5fcb 100644 --- a/python/frameworks/openai/traceai_openai/_attributes/_responses_api.py +++ b/python/frameworks/openai/traceai_openai/_attributes/_responses_api.py @@ -1,7 +1,7 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, Any, Callable, Iterable, Iterator, Tuple, Union +from typing import TYPE_CHECKING, Any, Callable, Iterable, Iterator, Mapping, Tuple, Union from opentelemetry.util.types import AttributeValue from typing_extensions import assert_never @@ -595,12 +595,23 @@ def _get_attributes_from_response_reasoning_item( obj: responses.response_reasoning_item.ResponseReasoningItem, prefix: str = "", ) -> Iterator[Tuple[str, AttributeValue]]: - if isinstance(obj.summary, Iterable): - for i, item in enumerate(obj.summary): - yield from cls._get_attributes_from_response_reasoning_item_summary( - item, - f"{prefix}{MessageAttributes.MESSAGE_CONTENTS}.{i}.", - ) + summary = list(obj.summary or []) + for i, item in enumerate(summary): + yield from cls._get_attributes_from_response_reasoning_item_summary( + item, + f"{prefix}{MessageAttributes.MESSAGE_CONTENTS}.{i}.", + ) + # DeepSeek returns plaintext reasoning in content, not summary. Older + # SDKs retain these extra items as dictionaries; newer SDKs type them. + for i, item in enumerate(getattr(obj, "content", None) or [], start=len(summary)): + if isinstance(item, Mapping): + kind, text = item.get("type"), item.get("text") + else: + kind, text = getattr(item, "type", None), getattr(item, "text", None) + if kind == "reasoning_text" and isinstance(text, str): + inner_prefix = f"{prefix}{MessageAttributes.MESSAGE_CONTENTS}.{i}." + yield f"{inner_prefix}{MessageContentAttributes.MESSAGE_CONTENT_TYPE}", "text" + yield f"{inner_prefix}{MessageContentAttributes.MESSAGE_CONTENT_TEXT}", text @classmethod @stop_on_exception diff --git a/python/frameworks/openai/traceai_openai/_request.py b/python/frameworks/openai/traceai_openai/_request.py index 99358c7a6..81b7c912e 100644 --- a/python/frameworks/openai/traceai_openai/_request.py +++ b/python/frameworks/openai/traceai_openai/_request.py @@ -438,7 +438,8 @@ async def __call__( add_io_to_span_attributes( with_span, None, response, is_streaming=self._is_streaming(response) ) - except Exception as exception: + # Cancellation can arrive before a stream/response takes ownership. + except BaseException as exception: with_span.record_exception(exception) status = trace_api.Status( status_code=trace_api.StatusCode.ERROR, diff --git a/python/frameworks/openai/traceai_openai/_response_accumulator.py b/python/frameworks/openai/traceai_openai/_response_accumulator.py index c7d0b1da7..f5930f7f3 100644 --- a/python/frameworks/openai/traceai_openai/_response_accumulator.py +++ b/python/frameworks/openai/traceai_openai/_response_accumulator.py @@ -97,9 +97,12 @@ def get_attributes(self) -> Iterator[Tuple[str, AttributeValue]]: # For Responses API, extract output items output_items = [] for item in result.output: - if hasattr(item, "content") and item.content: - for content in item.content: - if hasattr(content, "text"): + if ( + getattr(item, "type", None) == "message" + and getattr(item, "role", None) == "assistant" + ): + for content in item.content or []: + if getattr(content, "type", None) == "output_text": output_items.append(content.text) if output_items: output_content = " ".join(output_items) From e8f61b63de4cb711fc16876fcee016931e82b925 Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Mon, 2 Nov 2026 13:17:31 -0800 Subject: [PATCH 5/6] fix(openai): trace raw DeepSeek streams through parsed lifecycle Defer raw response span finalization until parsed streaming iteration or close; retain headers and capture stream errors with the configured privacy policy. Cover sync and async Chat/Responses at SDK floor and latest. Co-Authored-By: Claude Code --- .../tests/test_deepseek_raw_response.py | 177 ++++++++++++++++++ .../openai/traceai_openai/_request.py | 175 ++++++++++++++++- 2 files changed, 347 insertions(+), 5 deletions(-) create mode 100644 python/frameworks/openai/tests/test_deepseek_raw_response.py diff --git a/python/frameworks/openai/tests/test_deepseek_raw_response.py b/python/frameworks/openai/tests/test_deepseek_raw_response.py new file mode 100644 index 000000000..ccddb4f1e --- /dev/null +++ b/python/frameworks/openai/tests/test_deepseek_raw_response.py @@ -0,0 +1,177 @@ +"""Independent SDK-interface probes, synthetic keys and loopback OTLP only.""" + +import asyncio +import importlib.util +import json +from pathlib import Path +import sys + +import openai +import pytest + +source = Path(__file__).with_name("test_deepseek_route.py") +spec = importlib.util.spec_from_file_location("round3_route", source) +route = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = route +spec.loader.exec_module(route) +wire, httpx = route.wire, route.httpx + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("responses", [False, True]) +def test_raw_response_interface(wire, asynchronous, streaming, responses): + spans = wire("shared") + params = dict(model="deepseek-reasoner", stream=streaming) + params.update( + dict(input="hello") + if responses + else dict(messages=[dict(role="user", content="hello")]) + ) + + async def run(): + async with openai.AsyncOpenAI( + api_key="synthetic-key", + base_url="https://api.deepseek.com", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(route.respond)), + ) as client: + resource = client.responses if responses else client.chat.completions + raw = await resource.with_raw_response.create(**params) + if streaming: + assert not spans # The raw headers arrive before stream completion. + result = raw.parse() + if streaming: + assert result is raw.parse() + if streaming: + async with result: + async for _ in result: + pass + return result + + if asynchronous: + result = asyncio.run(run()) + else: + with openai.OpenAI( + api_key="synthetic-key", + base_url="https://api.deepseek.com", + http_client=httpx.Client(transport=httpx.MockTransport(route.respond)), + ) as client: + resource = client.responses if responses else client.chat.completions + raw = resource.with_raw_response.create(**params) + if streaming: + assert not spans + result = raw.parse() + if streaming: + assert result is raw.parse() + with result: + for _ in result: + pass + a = route.attrs(spans) + assert a["gen_ai.usage.output_tokens"] == 7 + assert a["deepseek.response_id"] == ( + "resp-offline" if responses else "completion-offline" + ) + if not responses: + assert a["deepseek.reasoning_content"] == route.REASONING + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("route_name", ["shared", "legacy_first", "shared_first"]) +def test_raw_stream_failure_status(wire, asynchronous, route_name): + spans = wire(route_name, hide_error_details=True) + + def respond(request): + return httpx.Response( + 200, + headers={ + "content-type": "text/event-stream", + "x-request-id": "req-synthetic", + }, + content="data: " + + json.dumps( + {"error": {"message": "PRIVATE_STREAM_ERROR", "type": "server_error"}} + ) + + "\n\n", + ) + + params = dict( + model="deepseek-reasoner", + messages=[dict(role="user", content="hello")], + stream=True, + ) + + async def run(): + async with openai.AsyncOpenAI( + api_key="synthetic-key", + base_url="https://api.deepseek.com", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(respond)), + ) as client: + raw = await client.chat.completions.with_raw_response.create(**params) + assert raw.headers["x-request-id"] == "req-synthetic" + assert not spans + async with raw.parse() as stream: + async for _ in stream: + pass + + with pytest.raises(openai.APIError, match="PRIVATE_STREAM_ERROR"): + if asynchronous: + asyncio.run(run()) + else: + with openai.OpenAI( + api_key="synthetic-key", + base_url="https://api.deepseek.com", + http_client=httpx.Client(transport=httpx.MockTransport(respond)), + ) as client: + raw = client.chat.completions.with_raw_response.create(**params) + assert raw.headers["x-request-id"] == "req-synthetic" + assert not spans + with raw.parse() as stream: + for _ in stream: + pass + assert len(spans) == 1 + assert spans[0].status.code == 2 + assert any(e.name == "exception" for e in spans[0].events) + assert b"PRIVATE_STREAM_ERROR" not in spans[0].SerializeToString() + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("parse_first", [False, True]) +def test_raw_stream_close_before_exhaustion(wire, asynchronous, parse_first): + spans = wire("shared") + params = dict( + model="deepseek-reasoner", + messages=[dict(role="user", content="hello")], + stream=True, + ) + + async def run(): + async with openai.AsyncOpenAI( + api_key="synthetic-key", + base_url="https://api.deepseek.com", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(route.respond)), + ) as client: + raw = await client.chat.completions.with_raw_response.create(**params) + assert not spans + if parse_first: + stream = raw.parse() + await stream.__anext__() + await raw.aclose() + await raw.aclose() + + if asynchronous: + asyncio.run(run()) + else: + with openai.OpenAI( + api_key="synthetic-key", + base_url="https://api.deepseek.com", + http_client=httpx.Client(transport=httpx.MockTransport(route.respond)), + ) as client: + raw = client.chat.completions.with_raw_response.create(**params) + assert not spans + if parse_first: + stream = raw.parse() + next(stream) + raw.close() + raw.close() + assert len(spans) == 1 + assert spans[0].status.code == 1 diff --git a/python/frameworks/openai/traceai_openai/_request.py b/python/frameworks/openai/traceai_openai/_request.py index 81b7c912e..9586efa01 100644 --- a/python/frameworks/openai/traceai_openai/_request.py +++ b/python/frameworks/openai/traceai_openai/_request.py @@ -1,3 +1,4 @@ +import inspect import logging from abc import ABC from contextlib import contextmanager @@ -36,6 +37,7 @@ ) from traceai_openai._with_span import _WithSpan from typing_extensions import TypeAlias +from wrapt import ObjectProxy __all__ = ( "_Request", @@ -117,7 +119,7 @@ def responses_accumulator(request_parameters: _RequestParameters) -> Any: chat_completion_type=openai.types.responses.response.Response, response_attributes_extractor=self._response_attributes_extractor, ) - + self._response_accumulator_factories: Mapping[ type, Callable[[_RequestParameters], _ResponseAccumulator] ] = { @@ -192,6 +194,15 @@ def _get_extra_attributes_from_request( def _is_streaming(self, response: Any) -> bool: return isinstance(response, self._stream_types) + def _is_raw_stream(self, response: Any) -> bool: + return bool( + ( + getattr(response, "_stream", False) + or getattr(response, "_is_sse_stream", False) + ) + and callable(getattr(response, "parse", None)) + ) + def _is_async_stream(self, response: Any) -> bool: """Check if the response is an asynchronous stream.""" return hasattr(response, "__aiter__") and callable(response.__aiter__) @@ -208,6 +219,21 @@ def _finalize_response( Monkey-patch the response object to trace the stream, or finish tracing if the response is not a stream. """ + # A raw streamed response exposes headers now but yields its parsed stream + # later. Keep the raw interface and transfer span ownership to that stream + # when the application calls parse(), rather than ending the span here. + if self._is_raw_stream(response): + return _RawStreamResponse( + response=response, + finalize=lambda parsed: self._finalize_response( + response=parsed, + with_span=with_span, + cast_to=cast_to, + request_parameters=request_parameters, + instance=instance, + ), + with_span=with_span, + ) if hasattr(response, "parse") and callable(response.parse): try: response.parse() @@ -304,6 +330,144 @@ def response_to_dict(self, response: Any) -> Any: return str(response) +class _RawStreamResponse(ObjectProxy): # type: ignore + """Preserve the SDK's raw headers while tracing its lazily parsed stream.""" + + __slots__ = ("_self_finalize", "_self_with_span", "_self_parsed") + + def __init__( + self, response: Any, finalize: Callable[[Any], Any], with_span: _WithSpan + ) -> None: + super().__init__(response) + self._self_finalize = finalize + self._self_with_span = with_span + self._self_parsed = None + + def _self_error(self, error: BaseException) -> None: + if not self._self_with_span.is_finished: + self._self_with_span.record_exception(error) + self._self_with_span.finish_tracing( + status=trace_api.Status( + status_code=trace_api.StatusCode.ERROR, + description=f"{type(error).__name__}: {error}", + ) + ) + + def _self_wrap(self, parsed: Any) -> Any: + if self._self_parsed is not None and parsed is self._self_parsed.__wrapped__: + return self._self_parsed + wrapped = self._self_finalize(parsed) + if isinstance(wrapped, (_Stream, _AsyncStream)): + self._self_parsed = wrapped + return wrapped + + def parse(self, *args: Any, **kwargs: Any) -> Any: + try: + parsed = self.__wrapped__.parse(*args, **kwargs) + except BaseException as error: + self._self_error(error) + raise + if inspect.isawaitable(parsed): + + async def await_parsed() -> Any: + try: + return self._self_wrap(await parsed) + except BaseException as error: + self._self_error(error) + raise + + return await_parsed() + try: + return self._self_wrap(parsed) + except BaseException as error: + self._self_error(error) + raise + + def close(self) -> Any: + # For async raw responses, callers should await the returned close. + if isinstance(self._self_parsed, _AsyncStream) or inspect.iscoroutinefunction( + getattr(self.__wrapped__, "close", None) + ): + return self.aclose() + try: + if self._self_parsed is not None: + self._self_parsed.close() + close = getattr(self.__wrapped__, "close", None) + if close is None: + close = self.__wrapped__.http_response.close + return close() + except BaseException as error: + self._self_error(error) + raise + finally: + if not self._self_with_span.is_finished: + self._self_with_span.finish_tracing( + status=trace_api.Status(status_code=trace_api.StatusCode.OK) + ) + + async def aclose(self) -> None: + try: + if self._self_parsed is not None: + closed = self._self_parsed.close() + if inspect.isawaitable(closed): + await closed + close = getattr(self.__wrapped__, "aclose", None) + if close is None: + close = getattr(self.__wrapped__.http_response, "aclose", None) + if close is None: + close = ( + getattr(self.__wrapped__, "close", None) + or self.__wrapped__.http_response.close + ) + closed = close() + if inspect.isawaitable(closed): + await closed + except BaseException as error: + self._self_error(error) + raise + finally: + if not self._self_with_span.is_finished: + self._self_with_span.finish_tracing( + status=trace_api.Status(status_code=trace_api.StatusCode.OK) + ) + + def __enter__(self) -> "_RawStreamResponse": + enter = getattr(self.__wrapped__, "__enter__", None) + if enter is not None: + enter() + return self + + def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> Any: + try: + exit_raw = getattr(self.__wrapped__, "__exit__", None) + return exit_raw(exc_type, exc, tb) if exit_raw is not None else None + except BaseException as close_error: + exc = close_error + raise + finally: + if exc is not None: + self._self_error(exc) + self.close() + + async def __aenter__(self) -> "_RawStreamResponse": + enter = getattr(self.__wrapped__, "__aenter__", None) + if enter is not None: + await enter() + return self + + async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> Any: + try: + exit_raw = getattr(self.__wrapped__, "__aexit__", None) + return await exit_raw(exc_type, exc, tb) if exit_raw is not None else None + except BaseException as close_error: + exc = close_error + raise + finally: + if exc is not None: + self._self_error(exc) + await self.aclose() + + class _Request(_WithTracer, _WithOpenAI): def __call__( self, @@ -358,7 +522,7 @@ def __call__( ) # Add output data to span attributes after getting response - if not self._is_streaming(response): + if not self._is_streaming(response) and not self._is_raw_stream(response): add_io_to_span_attributes( with_span, None, @@ -435,9 +599,10 @@ async def __call__( self._raw_embeddings_json(response), ) - add_io_to_span_attributes( - with_span, None, response, is_streaming=self._is_streaming(response) - ) + if not self._is_raw_stream(response): + add_io_to_span_attributes( + with_span, None, response, is_streaming=self._is_streaming(response) + ) # Cancellation can arrive before a stream/response takes ownership. except BaseException as exception: with_span.record_exception(exception) From 7bcfdbf058b18dcbdca36e73123addd173d247ef Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Mon, 2 Nov 2026 13:46:41 -0800 Subject: [PATCH 6/6] fix(openai): honor DeepSeek overrides and await raw responses Co-Authored-By: Claude Code --- .../tests/test_deepseek_explicit_raw.py | 78 +++++++++++++++++++ .../openai/traceai_openai/_request.py | 46 ++++++----- 2 files changed, 104 insertions(+), 20 deletions(-) create mode 100644 python/frameworks/openai/tests/test_deepseek_explicit_raw.py diff --git a/python/frameworks/openai/tests/test_deepseek_explicit_raw.py b/python/frameworks/openai/tests/test_deepseek_explicit_raw.py new file mode 100644 index 000000000..b8d6c7173 --- /dev/null +++ b/python/frameworks/openai/tests/test_deepseek_explicit_raw.py @@ -0,0 +1,78 @@ +"""Provider override and async raw-response parsing with real SDK MockTransport.""" + +import asyncio + +import openai +import pytest +from traceai_openai import set_client_provider + +from .test_deepseek_route import REASONING, attrs, httpx, respond, wire + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +def test_explicit_provider_enables_deepseek_extraction(wire, asynchronous, streaming): + spans = wire("shared") + params = dict(model="deepseek-reasoner", messages=[{"role": "user", "content": "hello"}], stream=streaming) + + async def run(): + async with openai.AsyncOpenAI( + api_key="synthetic-key", base_url="https://proxy.example/v1", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(respond)), + ) as client: + set_client_provider(client, provider="deepseek") + result = await client.chat.completions.create(**params) + if streaming: + async with result: + async for _ in result: + pass + + if asynchronous: + asyncio.run(run()) + else: + with openai.OpenAI( + api_key="synthetic-key", base_url="https://proxy.example/v1", + http_client=httpx.Client(transport=httpx.MockTransport(respond)), + ) as client: + set_client_provider(client, provider="deepseek") + result = client.chat.completions.create(**params) + if streaming: + with result: + for _ in result: + pass + a = attrs(spans) + assert a["gen_ai.provider.name"] == "deepseek" + assert a["deepseek.reasoning_content"] == REASONING + assert a["deepseek.prompt_cache_hit_tokens"] == 3 + + +@pytest.mark.parametrize("raw_mode", ["with_streaming_response", "with_raw_response"]) +@pytest.mark.parametrize("streaming", [False, True]) +def test_async_raw_response_parses_before_export(wire, raw_mode, streaming): + spans = wire("shared") + params = dict(model="deepseek-reasoner", messages=[{"role": "user", "content": "hello"}], stream=streaming) + + async def run(): + async with openai.AsyncOpenAI( + api_key="synthetic-key", base_url="https://api.deepseek.com", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(respond)), + ) as client: + if raw_mode == "with_streaming_response": + async with client.chat.completions.with_streaming_response.create(**params) as raw: + assert not spans + result = await raw.parse() + if streaming: + async for _ in result: + pass + else: + raw = await client.chat.completions.with_raw_response.create(**params) + result = raw.parse() + if streaming: + async for _ in result: + pass + assert result is not None + + asyncio.run(run()) + a = attrs(spans) + assert a["deepseek.reasoning_content"] == REASONING + assert a["gen_ai.usage.output_tokens"] == 7 diff --git a/python/frameworks/openai/traceai_openai/_request.py b/python/frameworks/openai/traceai_openai/_request.py index 9586efa01..be83c32ed 100644 --- a/python/frameworks/openai/traceai_openai/_request.py +++ b/python/frameworks/openai/traceai_openai/_request.py @@ -143,15 +143,18 @@ def _get_span_kind(self, cast_to: type) -> str: else FiSpanKindValues.LLM.value ) + def _provider_for_instance(self, instance: Any) -> str: + # Explicit client attribution wins; never guess the vendor of a local server. + provider = _get_client_provider(instance) + if provider is not None: + return provider + base_url = getattr(instance, "base_url", None) + return provider_from_host(getattr(base_url, "host", None)) + def _get_attributes_from_instance( self, instance: Any ) -> Iterator[Tuple[str, AttributeValue]]: - # Explicit client attribution wins; never guess the vendor of a local server. - base_url = getattr(instance, "base_url", None) - provider = _get_client_provider(instance) - if provider is None: - provider = provider_from_host(getattr(base_url, "host", None)) - yield SpanAttributes.GEN_AI_PROVIDER_NAME, provider + yield SpanAttributes.GEN_AI_PROVIDER_NAME, self._provider_for_instance(instance) def _get_attributes_from_request( self, @@ -195,12 +198,14 @@ def _is_streaming(self, response: Any) -> bool: return isinstance(response, self._stream_types) def _is_raw_stream(self, response: Any) -> bool: + parse = getattr(response, "parse", None) return bool( - ( + callable(parse) + and ( getattr(response, "_stream", False) or getattr(response, "_is_sse_stream", False) + or inspect.iscoroutinefunction(parse) ) - and callable(getattr(response, "parse", None)) ) def _is_async_stream(self, response: Any) -> bool: @@ -219,9 +224,9 @@ def _finalize_response( Monkey-patch the response object to trace the stream, or finish tracing if the response is not a stream. """ - # A raw streamed response exposes headers now but yields its parsed stream - # later. Keep the raw interface and transfer span ownership to that stream - # when the application calls parse(), rather than ending the span here. + # Raw streams and asynchronously parsed raw responses expose headers before + # their contents are ready. Preserve the raw interface and transfer span + # ownership when the application parses the response. if self._is_raw_stream(response): return _RawStreamResponse( response=response, @@ -240,10 +245,10 @@ def _finalize_response( except Exception: logger.exception(f"Failed to parse response of type {type(response)}") - provider = provider_from_host(getattr(getattr(instance, "base_url", None), "host", None)) + provider = self._provider_for_instance(instance) extractor = self._response_attributes_extractor if provider == "deepseek": - # Keep host-specific extraction local to this request: concurrent + # Keep provider-specific extraction local to this request: concurrent # requests through another provider must not inherit DeepSeek aliases. extractor = _ResponseAttributesExtractor(self._openai, self._config, provider=provider) @@ -331,9 +336,9 @@ def response_to_dict(self, response: Any) -> Any: class _RawStreamResponse(ObjectProxy): # type: ignore - """Preserve the SDK's raw headers while tracing its lazily parsed stream.""" + """Preserve raw headers while tracing the SDK's lazily parsed response.""" - __slots__ = ("_self_finalize", "_self_with_span", "_self_parsed") + __slots__ = ("_self_finalize", "_self_with_span", "_self_parsed", "_self_parsed_raw") def __init__( self, response: Any, finalize: Callable[[Any], Any], with_span: _WithSpan @@ -342,6 +347,7 @@ def __init__( self._self_finalize = finalize self._self_with_span = with_span self._self_parsed = None + self._self_parsed_raw = None def _self_error(self, error: BaseException) -> None: if not self._self_with_span.is_finished: @@ -354,11 +360,11 @@ def _self_error(self, error: BaseException) -> None: ) def _self_wrap(self, parsed: Any) -> Any: - if self._self_parsed is not None and parsed is self._self_parsed.__wrapped__: + if self._self_parsed is not None and parsed is self._self_parsed_raw: return self._self_parsed wrapped = self._self_finalize(parsed) - if isinstance(wrapped, (_Stream, _AsyncStream)): - self._self_parsed = wrapped + self._self_parsed_raw = parsed + self._self_parsed = wrapped return wrapped def parse(self, *args: Any, **kwargs: Any) -> Any: @@ -390,7 +396,7 @@ def close(self) -> Any: ): return self.aclose() try: - if self._self_parsed is not None: + if isinstance(self._self_parsed, (_Stream, _AsyncStream)): self._self_parsed.close() close = getattr(self.__wrapped__, "close", None) if close is None: @@ -407,7 +413,7 @@ def close(self) -> Any: async def aclose(self) -> None: try: - if self._self_parsed is not None: + if isinstance(self._self_parsed, (_Stream, _AsyncStream)): closed = self._self_parsed.close() if inspect.isawaitable(closed): await closed