diff --git a/src/agents/extensions/models/any_llm_model.py b/src/agents/extensions/models/any_llm_model.py index b72f182a1b..aa217cb2a1 100644 --- a/src/agents/extensions/models/any_llm_model.py +++ b/src/agents/extensions/models/any_llm_model.py @@ -53,6 +53,7 @@ from ...tracing.span_data import GenerationSpanData from ...tracing.spans import Span from ...usage import Usage +from ...util._error_tracing import model_span_errors, record_model_error_on_span from ...util._json import _to_dump_compatible try: @@ -361,7 +362,14 @@ async def _get_response_via_responses( conversation_id: str | None, prompt: ResponsePromptParam | None, ) -> ModelResponse: - with response_span(disabled=tracing.is_disabled()) as span_response: + with ( + response_span(disabled=tracing.is_disabled()) as span_response, + model_span_errors( + span_response, + message="Error getting response", + trace_include_sensitive_data=tracing.include_data(), + ), + ): response = await self._fetch_responses_response( system_instructions=system_instructions, input=input, @@ -425,7 +433,14 @@ async def _stream_response_via_responses( conversation_id: str | None, prompt: ResponsePromptParam | None, ) -> AsyncGenerator[ResponseStreamEvent, None]: - with response_span(disabled=tracing.is_disabled()) as span_response: + with ( + response_span(disabled=tracing.is_disabled()) as span_response, + model_span_errors( + span_response, + message="Error streaming response", + trace_include_sensitive_data=tracing.include_data(), + ), + ): stream = await self._fetch_responses_response( system_instructions=system_instructions, input=input, @@ -472,6 +487,17 @@ async def _stream_response_via_responses( if tracing.include_data() and final_response: span_response.span_data.response = final_response span_response.span_data.input = input + if terminal_failure_error is not None: + # The failure is already known here. A consumer that stops at + # this event closes the generator, which raises GeneratorExit + # at the yield below and skips the raise after the loop, so + # recording later would miss it entirely. + record_model_error_on_span( + span_response, + message="Error streaming response", + error=terminal_failure_error, + trace_include_sensitive_data=tracing.include_data(), + ) yield chunk except asyncio.CancelledError: close_stream_in_background = True @@ -506,15 +532,22 @@ async def _get_response_via_chat( tracing: ModelTracing, prompt: ResponsePromptParam | None, ) -> ModelResponse: - with generation_span( - model=str(self.model), - model_config=model_config_for_trace( - model_settings, - base_url=self.base_url or "", - extra_config={"provider": self._provider_name, "model_impl": "any-llm"}, + with ( + generation_span( + model=str(self.model), + model_config=model_config_for_trace( + model_settings, + base_url=self.base_url or "", + extra_config={"provider": self._provider_name, "model_impl": "any-llm"}, + ), + disabled=tracing.is_disabled(), + ) as span_generation, + model_span_errors( + span_generation, + message="Error getting response", + trace_include_sensitive_data=tracing.include_data(), ), - disabled=tracing.is_disabled(), - ) as span_generation: + ): response = await self._fetch_chat_response( system_instructions=system_instructions, input=input, @@ -608,15 +641,22 @@ async def _stream_response_via_chat( tracing: ModelTracing, prompt: ResponsePromptParam | None, ) -> AsyncGenerator[TResponseStreamEvent, None]: - with generation_span( - model=str(self.model), - model_config=model_config_for_trace( - model_settings, - base_url=self.base_url or "", - extra_config={"provider": self._provider_name, "model_impl": "any-llm"}, + with ( + generation_span( + model=str(self.model), + model_config=model_config_for_trace( + model_settings, + base_url=self.base_url or "", + extra_config={"provider": self._provider_name, "model_impl": "any-llm"}, + ), + disabled=tracing.is_disabled(), + ) as span_generation, + model_span_errors( + span_generation, + message="Error streaming response", + trace_include_sensitive_data=tracing.include_data(), ), - disabled=tracing.is_disabled(), - ) as span_generation: + ): response, stream = await self._fetch_chat_response( system_instructions=system_instructions, input=input, diff --git a/src/agents/extensions/models/litellm_model.py b/src/agents/extensions/models/litellm_model.py index d8430ccc69..185b661031 100644 --- a/src/agents/extensions/models/litellm_model.py +++ b/src/agents/extensions/models/litellm_model.py @@ -59,6 +59,7 @@ from ...tracing.span_data import GenerationSpanData from ...tracing.spans import Span from ...usage import Usage, _cache_write_tokens, _make_input_tokens_details +from ...util._error_tracing import model_span_errors from ...util._json import _to_dump_compatible @@ -214,15 +215,22 @@ async def get_response( conversation_id: str | None = None, # unused prompt: Any | None = None, ) -> ModelResponse: - with generation_span( - model=str(self.model), - model_config=model_config_for_trace( - model_settings, - base_url=self.base_url or "", - extra_config={"model_impl": "litellm"}, + with ( + generation_span( + model=str(self.model), + model_config=model_config_for_trace( + model_settings, + base_url=self.base_url or "", + extra_config={"model_impl": "litellm"}, + ), + disabled=tracing.is_disabled(), + ) as span_generation, + model_span_errors( + span_generation, + message="Error getting response", + trace_include_sensitive_data=tracing.include_data(), ), - disabled=tracing.is_disabled(), - ) as span_generation: + ): response = await self._fetch_response( system_instructions, input, @@ -377,15 +385,22 @@ async def stream_response( conversation_id: str | None = None, # unused prompt: Any | None = None, ) -> AsyncIterator[TResponseStreamEvent]: - with generation_span( - model=str(self.model), - model_config=model_config_for_trace( - model_settings, - base_url=self.base_url or "", - extra_config={"model_impl": "litellm"}, + with ( + generation_span( + model=str(self.model), + model_config=model_config_for_trace( + model_settings, + base_url=self.base_url or "", + extra_config={"model_impl": "litellm"}, + ), + disabled=tracing.is_disabled(), + ) as span_generation, + model_span_errors( + span_generation, + message="Error streaming response", + trace_include_sensitive_data=tracing.include_data(), ), - disabled=tracing.is_disabled(), - ) as span_generation: + ): response, stream = await self._fetch_response( system_instructions, input, diff --git a/src/agents/models/openai_chatcompletions.py b/src/agents/models/openai_chatcompletions.py index b7f9d8e00a..0ac0a2690c 100644 --- a/src/agents/models/openai_chatcompletions.py +++ b/src/agents/models/openai_chatcompletions.py @@ -32,6 +32,7 @@ from ..tracing.span_data import GenerationSpanData from ..tracing.spans import Span from ..usage import Usage +from ..util._error_tracing import model_span_errors from ..util._json import _to_dump_compatible from ._openai_retry import get_openai_retry_advice from ._retry_runtime import should_disable_provider_managed_retries @@ -206,11 +207,18 @@ async def get_response( ) self._handle_unsupported_prompt(prompt) - with generation_span( - model=str(self.model), - model_config=model_config_for_trace(model_settings, base_url=self._client.base_url), - disabled=tracing.is_disabled(), - ) as span_generation: + with ( + generation_span( + model=str(self.model), + model_config=model_config_for_trace(model_settings, base_url=self._client.base_url), + disabled=tracing.is_disabled(), + ) as span_generation, + model_span_errors( + span_generation, + message="Error getting response", + trace_include_sensitive_data=tracing.include_data(), + ), + ): response = await self._fetch_response( system_instructions, input, @@ -340,11 +348,18 @@ async def stream_response( ) self._handle_unsupported_prompt(prompt) - with generation_span( - model=str(self.model), - model_config=model_config_for_trace(model_settings, base_url=self._client.base_url), - disabled=tracing.is_disabled(), - ) as span_generation: + with ( + generation_span( + model=str(self.model), + model_config=model_config_for_trace(model_settings, base_url=self._client.base_url), + disabled=tracing.is_disabled(), + ) as span_generation, + model_span_errors( + span_generation, + message="Error streaming response", + trace_include_sensitive_data=tracing.include_data(), + ), + ): response, stream = await self._fetch_response( system_instructions, input, diff --git a/src/agents/models/openai_responses.py b/src/agents/models/openai_responses.py index cafb168f4b..d3dd46f8f7 100644 --- a/src/agents/models/openai_responses.py +++ b/src/agents/models/openai_responses.py @@ -75,6 +75,7 @@ ) from ..tracing import SpanError, response_span from ..usage import Usage, _response_usage_to_usage, model_usage_to_span_usage +from ..util._error_tracing import record_model_error_on_span from ..util._json import _to_dump_compatible from ..version import __version__ from ._openai_retry import get_openai_retry_advice @@ -593,6 +594,17 @@ async def stream_response( "response.error", }: yielded_terminal_event = True + if terminal_failure_error is not None: + # A consumer that stops at this event closes the + # generator, which raises GeneratorExit at the yield + # below and skips the raise after the loop, so the + # span has to be annotated here or not at all. + record_model_error_on_span( + span_response, + message="Error streaming response", + error=terminal_failure_error, + trace_include_sensitive_data=tracing.include_data(), + ) yield chunk except asyncio.CancelledError: close_stream_in_background = True diff --git a/src/agents/util/_error_tracing.py b/src/agents/util/_error_tracing.py index 7f714482a5..27eae75ec2 100644 --- a/src/agents/util/_error_tracing.py +++ b/src/agents/util/_error_tracing.py @@ -1,3 +1,5 @@ +import contextlib +from collections.abc import Iterator from typing import Any from .. import _debug @@ -29,3 +31,82 @@ def attach_error_to_current_span(error: SpanError) -> None: logger.warning("No active span; trace error was not attached") else: logger.warning("No span to add error %s to", error) + + +def _model_error_text(error: Exception, *, trace_include_sensitive_data: bool) -> str: + """Render the span text for a failed model call. + + The exception is only stringified when its text will actually be exported, so a + provider exception with a side-effecting `__str__` is not invoked just to have + its output thrown away by redaction. + """ + if not trace_include_sensitive_data: + return REDACTED_TRACE_ERROR_MESSAGE + try: + return str(error) + except Exception: + logger.warning( + "Could not stringify %s for the model span; recording the type only", + type(error).__name__, + ) + return f"Unrenderable {type(error).__name__}" + + +def record_model_error_on_span( + span: Span[Any], + *, + message: str, + error: Exception, + trace_include_sensitive_data: bool, +) -> None: + """Record an already-known model failure on its span. + + Streaming providers learn about a terminal failure before they raise it, and a + consumer that stops at that terminal event closes the generator, so the raise + never happens. Recording at the point of knowledge keeps the span accurate in + that case. Best-effort: never raises, so annotating a span cannot change what + the caller sees. + """ + try: + attach_error_to_span( + span, + SpanError( + message=message, + data={ + "error": _model_error_text( + error, + trace_include_sensitive_data=trace_include_sensitive_data, + ) + }, + ), + ) + except Exception: + logger.warning("Could not record the model error on the span", exc_info=True) + + +@contextlib.contextmanager +def model_span_errors( + span: Span[Any], + *, + message: str, + trace_include_sensitive_data: bool, +) -> Iterator[None]: + """Record a failing model call on the span it happened in, then re-raise. + + `Span.__exit__` finishes a span without attaching an exception, so a provider + that does not annotate its own span exports a failed model call that is + indistinguishable from a successful one. + + Recording is best-effort on purpose: the exception the caller sees is always the + one the provider raised, never one produced while annotating the span. + """ + try: + yield + except Exception as error: + record_model_error_on_span( + span, + message=message, + error=error, + trace_include_sensitive_data=trace_include_sensitive_data, + ) + raise diff --git a/tests/test_provider_span_errors.py b/tests/test_provider_span_errors.py new file mode 100644 index 0000000000..2ea1765363 --- /dev/null +++ b/tests/test_provider_span_errors.py @@ -0,0 +1,430 @@ +"""Every model provider must record a failed model call on its own span. + +`Span.__exit__` finishes a span without attaching an exception, so a provider that +does not annotate its span exports a failed model call that is indistinguishable +from a successful one. `OpenAIResponsesModel` has always annotated its span; these +tests pin the same behavior for the other providers. +""" + +from __future__ import annotations + +from typing import Any + +import pytest +from openai import AsyncOpenAI + +from agents import ModelSettings, ModelTracing, OpenAIChatCompletionsModel, trace + +from .testing_processor import fetch_ordered_spans + + +class _Boom(Exception): + pass + + +def _span_error(span_filter: str) -> dict[str, Any] | None: + for span in fetch_ordered_spans(): + if span.span_data.type == span_filter and span.error is not None: + return dict(span.error) + return None + + +async def _drain(agen: Any) -> None: + async for _ in agen: + pass + + +def _chatcompletions_model() -> OpenAIChatCompletionsModel: + return OpenAIChatCompletionsModel( + model="gpt-4", openai_client=AsyncOpenAI(api_key="test", base_url="http://localhost:1") + ) + + +def _call_kwargs() -> dict[str, Any]: + return { + "system_instructions": None, + "input": "hi", + "model_settings": ModelSettings(), + "tools": [], + "output_schema": None, + "handoffs": [], + "tracing": ModelTracing.ENABLED, + "previous_response_id": None, + "conversation_id": None, + "prompt": None, + } + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_chatcompletions_get_response_records_span_error(monkeypatch) -> None: + model = _chatcompletions_model() + + async def boom(*args: Any, **kwargs: Any) -> Any: + raise _Boom("upstream exploded") + + monkeypatch.setattr(model, "_fetch_response", boom) + with trace(workflow_name="test"): + with pytest.raises(_Boom): + await model.get_response(**_call_kwargs()) + + error = _span_error("generation") + assert error is not None, "generation span carried no error" + assert error["message"] == "Error getting response" + assert "upstream exploded" in error["data"]["error"] + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_chatcompletions_stream_response_records_span_error(monkeypatch) -> None: + model = _chatcompletions_model() + + async def boom(*args: Any, **kwargs: Any) -> Any: + raise _Boom("stream exploded") + + monkeypatch.setattr(model, "_fetch_response", boom) + with trace(workflow_name="test"): + with pytest.raises(_Boom): + await _drain(model.stream_response(**_call_kwargs())) + + error = _span_error("generation") + assert error is not None, "generation span carried no error" + assert error["message"] == "Error streaming response" + assert "stream exploded" in error["data"]["error"] + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_chatcompletions_span_error_is_redacted_without_sensitive_data(monkeypatch) -> None: + """With tracing data disabled the exception text must not reach the span.""" + model = _chatcompletions_model() + + async def boom(*args: Any, **kwargs: Any) -> Any: + raise _Boom("secret-connection-string") + + monkeypatch.setattr(model, "_fetch_response", boom) + kwargs = _call_kwargs() + kwargs["tracing"] = ModelTracing.ENABLED_WITHOUT_DATA + with trace(workflow_name="test"): + with pytest.raises(_Boom): + await model.get_response(**kwargs) + + error = _span_error("generation") + assert error is not None + assert "secret-connection-string" not in error["data"]["error"] + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_litellm_get_response_records_span_error(monkeypatch) -> None: + pytest.importorskip("litellm") + from agents.extensions.models.litellm_model import LitellmModel + + model = LitellmModel(model="gpt-4", api_key="test") + + async def boom(*args: Any, **kwargs: Any) -> Any: + raise _Boom("litellm exploded") + + monkeypatch.setattr(model, "_fetch_response", boom) + with trace(workflow_name="test"): + with pytest.raises(_Boom): + await model.get_response(**_call_kwargs()) + + error = _span_error("generation") + assert error is not None, "generation span carried no error" + assert error["message"] == "Error getting response" + assert "litellm exploded" in error["data"]["error"] + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_litellm_stream_response_records_span_error(monkeypatch) -> None: + pytest.importorskip("litellm") + from agents.extensions.models.litellm_model import LitellmModel + + model = LitellmModel(model="gpt-4", api_key="test") + + async def boom(*args: Any, **kwargs: Any) -> Any: + raise _Boom("litellm stream exploded") + + monkeypatch.setattr(model, "_fetch_response", boom) + with trace(workflow_name="test"): + with pytest.raises(_Boom): + await _drain(model.stream_response(**_call_kwargs())) + + error = _span_error("generation") + assert error is not None, "generation span carried no error" + assert error["message"] == "Error streaming response" + assert "litellm stream exploded" in error["data"]["error"] + + +def _any_llm_model() -> Any: + from agents.extensions.models.any_llm_model import AnyLLMModel + + return AnyLLMModel(model="openai/gpt-4", api_key="test") + + +_ANY_LLM_BASE_KWARGS: dict[str, Any] = { + "system_instructions": None, + "input": "hi", + "model_settings": ModelSettings(), + "tools": [], + "output_schema": None, + "handoffs": [], + "tracing": ModelTracing.ENABLED, + "prompt": None, +} + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("method", "fetch", "span_type", "message", "streaming"), + [ + ( + "_get_response_via_responses", + "_fetch_responses_response", + "response", + "Error getting response", + False, + ), + ( + "_stream_response_via_responses", + "_fetch_responses_response", + "response", + "Error streaming response", + True, + ), + ( + "_get_response_via_chat", + "_fetch_chat_response", + "generation", + "Error getting response", + False, + ), + ( + "_stream_response_via_chat", + "_fetch_chat_response", + "generation", + "Error streaming response", + True, + ), + ], +) +async def test_any_llm_records_span_error( + monkeypatch, method: str, fetch: str, span_type: str, message: str, streaming: bool +) -> None: + pytest.importorskip("any_llm") + model = _any_llm_model() + + async def boom(*args: Any, **kwargs: Any) -> Any: + raise _Boom("any_llm exploded") + + monkeypatch.setattr(model, fetch, boom) + kwargs = dict(_ANY_LLM_BASE_KWARGS) + if "via_responses" in method: + kwargs.update({"previous_response_id": None, "conversation_id": None}) + + with trace(workflow_name="test"): + with pytest.raises(_Boom): + if streaming: + await _drain(getattr(model, method)(**kwargs)) + else: + await getattr(model, method)(**kwargs) + + error = _span_error(span_type) + assert error is not None, f"{span_type} span carried no error" + assert error["message"] == message + assert "any_llm exploded" in error["data"]["error"] + + +class _SideEffectingStr(Exception): + """A provider exception whose `__str__` must not be called speculatively.""" + + def __init__(self) -> None: + super().__init__() + self.str_calls = 0 + + def __str__(self) -> str: + self.str_calls += 1 + return "sensitive detail" + + +class _BrokenStr(Exception): + def __str__(self) -> str: + raise ValueError("__str__ exploded") + + +def test_redacted_tracing_does_not_stringify_the_exception() -> None: + """`ENABLED_WITHOUT_DATA` must not evaluate `str(error)` just to discard it.""" + from agents.tracing import generation_span + from agents.util._error_tracing import REDACTED_TRACE_ERROR_MESSAGE, model_span_errors + + original = _SideEffectingStr() + with trace(workflow_name="test"): + with generation_span() as span: + with pytest.raises(_SideEffectingStr) as exc_info: + with model_span_errors( + span, + message="Error getting response", + trace_include_sensitive_data=False, + ): + raise original + + assert exc_info.value is original + assert original.str_calls == 0 + error = _span_error("generation") + assert error is not None + assert error["data"]["error"] == REDACTED_TRACE_ERROR_MESSAGE + + +def test_sensitive_tracing_stringifies_once() -> None: + from agents.tracing import generation_span + from agents.util._error_tracing import model_span_errors + + original = _SideEffectingStr() + with trace(workflow_name="test"): + with generation_span() as span: + with pytest.raises(_SideEffectingStr): + with model_span_errors( + span, + message="Error getting response", + trace_include_sensitive_data=True, + ): + raise original + + assert original.str_calls == 1 + error = _span_error("generation") + assert error is not None + assert error["data"]["error"] == "sensitive detail" + + +@pytest.mark.parametrize("include_sensitive_data", [True, False]) +def test_broken_str_preserves_the_provider_exception(include_sensitive_data: bool) -> None: + """A broken `__str__` must not replace the provider failure the caller sees.""" + from agents.tracing import generation_span + from agents.util._error_tracing import model_span_errors + + original = _BrokenStr() + with trace(workflow_name="test"): + with generation_span() as span: + with pytest.raises(_BrokenStr) as exc_info: + with model_span_errors( + span, + message="Error getting response", + trace_include_sensitive_data=include_sensitive_data, + ): + raise original + + assert exc_info.value is original + assert _span_error("generation") is not None + + +def test_failing_span_recording_preserves_the_provider_exception( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """If recording itself raises, the caller still sees the provider's exception.""" + from agents.tracing import generation_span + from agents.util import _error_tracing + from agents.util._error_tracing import model_span_errors + + def explode(*_args: Any, **_kwargs: Any) -> None: + raise RuntimeError("span backend is down") + + monkeypatch.setattr(_error_tracing, "attach_error_to_span", explode) + + original = _Boom("provider failed") + with trace(workflow_name="test"): + with generation_span() as span: + with pytest.raises(_Boom) as exc_info: + with model_span_errors( + span, + message="Error getting response", + trace_include_sensitive_data=True, + ): + raise original + + assert exc_info.value is original + + +class _TerminalFailureEvent: + """A terminal `response.failed` event with no response payload attached.""" + + type = "response.failed" + response = None + + +class _SingleEventStream: + def __init__(self) -> None: + self._sent = False + + def __aiter__(self) -> _SingleEventStream: + return self + + async def __anext__(self) -> _TerminalFailureEvent: + if self._sent: + raise StopAsyncIteration + self._sent = True + return _TerminalFailureEvent() + + async def aclose(self) -> None: + return None + + +async def _stop_at_terminal_event(agen: Any) -> None: + """Consume the terminal event and close the generator, as a raw consumer would.""" + first = await agen.__anext__() + assert getattr(first, "type", None) == "response.failed" + await agen.aclose() + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_openai_responses_records_terminal_failure_when_consumer_stops(monkeypatch) -> None: + """Closing the stream at a terminal failure must still mark the span. + + The failure is known when the terminal event is yielded, but `aclose()` raises + `GeneratorExit` at that yield, which skips the `raise terminal_failure_error` + after the loop. `GeneratorExit` is a `BaseException`, so nothing downstream + records it either and the span exports as if the call had succeeded. + """ + from agents import OpenAIResponsesModel + + model = OpenAIResponsesModel( + model="gpt-4", openai_client=AsyncOpenAI(api_key="test", base_url="http://localhost:1") + ) + + async def fake_fetch(*args: Any, **kwargs: Any) -> Any: + return _SingleEventStream() + + monkeypatch.setattr(model, "_fetch_response", fake_fetch) + + with trace(workflow_name="test"): + await _stop_at_terminal_event(model.stream_response(**_call_kwargs())) + + error = _span_error("response") + assert error is not None, "response span carried no error" + assert error["message"] == "Error streaming response" + assert "response.failed" in error["data"]["error"] + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +async def test_any_llm_responses_records_terminal_failure_when_consumer_stops(monkeypatch) -> None: + pytest.importorskip("any_llm") + from agents.extensions.models.any_llm_model import AnyLLMModel + + model = AnyLLMModel(model="openai/gpt-4") + + async def fake_fetch(*args: Any, **kwargs: Any) -> Any: + return _SingleEventStream() + + monkeypatch.setattr(model, "_fetch_responses_response", fake_fetch) + + with trace(workflow_name="test"): + await _stop_at_terminal_event(model._stream_response_via_responses(**_call_kwargs())) + + error = _span_error("response") + assert error is not None, "response span carried no error" + assert error["message"] == "Error streaming response" + assert "response.failed" in error["data"]["error"]