diff --git a/.sampo/changesets/responses-terminal-stop-reason.md b/.sampo/changesets/responses-terminal-stop-reason.md new file mode 100644 index 000000000..cddddf2ef --- /dev/null +++ b/.sampo/changesets/responses-terminal-stop-reason.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: patch +--- + +Only terminal Responses API statuses become `$ai_stop_reason`: a queued or in-progress background run no longer records a lifecycle state as its stop reason, and an incomplete run is named by what cut it short (`incomplete_details.reason`, e.g. `max_output_tokens`). Streaming runs that end incomplete or failed now carry a stop reason too, and the LangChain callback reads stop reasons from `response_metadata` as well, covering Responses API and Anthropic runs that previously recorded none. diff --git a/posthog/ai/langchain/callbacks.py b/posthog/ai/langchain/callbacks.py index 91fdece52..49832a05d 100644 --- a/posthog/ai/langchain/callbacks.py +++ b/posthog/ai/langchain/callbacks.py @@ -48,6 +48,7 @@ _extract_cache_creation_ttl_breakdown, finalize_ai_content, get_model_params, + _responses_stop_reason, with_privacy_mode, ) from posthog.client import Client @@ -709,14 +710,11 @@ def _capture_generation( finalize_ai_content(completions, self._ph_client), ) - # Extract stop reason from generation info + # Extract the stop reason from the generation and its metadata if output.generations and output.generations[-1]: - last_gen = output.generations[-1][-1] - gen_info = getattr(last_gen, "generation_info", None) - if isinstance(gen_info, dict): - finish_reason = gen_info.get("finish_reason") - if finish_reason is not None: - event_properties["$ai_stop_reason"] = finish_reason + stop_reason = _extract_stop_reason(output.generations[-1][-1]) + if stop_reason is not None: + event_properties["$ai_stop_reason"] = stop_reason _capture_ai_event( self._ph_client, @@ -738,6 +736,37 @@ def _log_debug_event( ) +def _extract_stop_reason(generation: Any) -> Optional[str]: + """ + Providers spread the stop reason across `generation_info` and + `response_metadata` under two spellings, so read every source in priority + order. The Responses API reports no finish_reason at all: an incomplete + run is named by what cut it short, and only terminal statuses count. + """ + + def as_dict(value: Any) -> dict: + return value if isinstance(value, dict) else {} + + message = as_dict( + getattr(getattr(generation, "message", None), "response_metadata", None) + ) + info = as_dict(getattr(generation, "generation_info", None)) + nested = as_dict(info.get("response_metadata")) + + for source, key in ( + (message, "finish_reason"), + (message, "stop_reason"), + (info, "finish_reason"), + (nested, "stop_reason"), + (nested, "finish_reason"), + (info, "stop_reason"), + ): + if source.get(key) is not None: + return str(source[key]) + + return _responses_stop_reason(message) or _responses_stop_reason(nested) + + def _extract_raw_response(last_response): """Extract the response from the last response of the LLM call.""" # We return the text of the response if not empty diff --git a/posthog/ai/openai/_streaming.py b/posthog/ai/openai/_streaming.py index 3f2f38e74..23b6f5968 100644 --- a/posthog/ai/openai/_streaming.py +++ b/posthog/ai/openai/_streaming.py @@ -4,7 +4,7 @@ from typing import Any, Dict, List, Optional from ..types import StreamingEventData, TokenUsage -from ..utils import merge_usage_stats +from ..utils import merge_usage_stats, _responses_stop_reason from .openai_converter import ( accumulate_openai_tool_calls, extract_openai_content_from_chunk, @@ -35,10 +35,12 @@ def process_chunk(self, chunk: Any) -> None: if content is not None: self.output.extend(content) - if getattr(chunk, "type", None) == "response.completed" and response: - status = getattr(response, "status", None) - if status is not None: - self.stop_reason = status + # A stream can end on response.completed, response.incomplete, or + # response.failed; any terminal response names the stop reason. + if response: + stop_reason = _responses_stop_reason(response) + if stop_reason is not None: + self.stop_reason = stop_reason @dataclass diff --git a/posthog/ai/openai/openai_converter.py b/posthog/ai/openai/openai_converter.py index ecd8c8c90..3f3bc869c 100644 --- a/posthog/ai/openai/openai_converter.py +++ b/posthog/ai/openai/openai_converter.py @@ -17,7 +17,7 @@ FormattedTextContent, TokenUsage, ) -from posthog.ai.utils import serialize_raw_usage +from posthog.ai.utils import _responses_stop_reason, serialize_raw_usage def _item_attr(item: Any, name: str, default: Any = None) -> Any: @@ -445,7 +445,7 @@ def extract_openai_stop_reason(response: Any) -> Optional[str]: return getattr(response.choices[0], "finish_reason", None) # Responses API if hasattr(response, "status"): - return getattr(response, "status", None) + return _responses_stop_reason(response) return None diff --git a/posthog/ai/utils.py b/posthog/ai/utils.py index c456c0b72..a83ace709 100644 --- a/posthog/ai/utils.py +++ b/posthog/ai/utils.py @@ -303,6 +303,35 @@ def format_response(response, provider: str): return [] +# A Responses API run is only finished on these statuses; `queued` and +# `in_progress` are lifecycle states a background run passes through. +_TERMINAL_RESPONSE_STATUSES = frozenset( + {"completed", "failed", "cancelled", "incomplete"} +) + + +def _read_response_field(source: Any, key: str) -> Any: + return source.get(key) if isinstance(source, dict) else getattr(source, key, None) + + +def _responses_stop_reason(response: Any) -> Optional[str]: + """ + Map a Responses API outcome to a `$ai_stop_reason`: an incomplete run is + named by what cut it short (`incomplete_details.reason`, e.g. + `max_output_tokens`), the other terminal statuses stand for themselves, + and a non-terminal status yields None. Accepts an SDK response object or + a LangChain `response_metadata` dict. + """ + status = _read_response_field(response, "status") + if not isinstance(status, str) or status not in _TERMINAL_RESPONSE_STATUSES: + return None + details = _read_response_field(response, "incomplete_details") + reason = _read_response_field(details, "reason") + if status == "incomplete" and isinstance(reason, str) and reason: + return reason + return status + + def extract_stop_reason(response: Any, provider: str) -> Optional[str]: """Extract stop reason from response based on provider.""" if provider == "openai": diff --git a/posthog/test/ai/langchain/test_callbacks.py b/posthog/test/ai/langchain/test_callbacks.py index 11f021099..43325ac05 100644 --- a/posthog/test/ai/langchain/test_callbacks.py +++ b/posthog/test/ai/langchain/test_callbacks.py @@ -2839,3 +2839,58 @@ def test_ai_lane_client_routes_through_capture_ai(mock_client): events = [c[1]["event"] for c in mock_client.capture_ai.call_args_list] assert "$ai_generation" in events assert "$ai_trace" in events + + +@pytest.mark.parametrize( + "generation_info,response_metadata,expected", + [ + # generation_info finish_reason keeps priority + ({"finish_reason": "stop"}, {"status": "completed"}, "stop"), + # Responses API: a terminal status carries no finish_reason at all + (None, {"status": "completed", "incomplete_details": None}, "completed"), + # ... an incomplete run is named by what cut it short + ( + None, + { + "status": "incomplete", + "incomplete_details": {"reason": "max_output_tokens"}, + }, + "max_output_tokens", + ), + # a queued background run has no stop reason yet + (None, {"status": "queued", "incomplete_details": None}, None), + # Anthropic reports through response_metadata.stop_reason + (None, {"stop_reason": "end_turn"}, "end_turn"), + ], +) +def test_stop_reason_resolution( + mock_client, generation_info, response_metadata, expected +): + from langchain_core.outputs import ChatGeneration, LLMResult + + cb = CallbackHandler(mock_client) + run_id = uuid.uuid4() + cb._set_llm_metadata( + serialized={}, + run_id=run_id, + messages=[{"role": "user", "content": "test"}], + metadata={"ls_provider": "openai", "ls_model_name": "gpt-4o"}, + ) + response = LLMResult( + generations=[ + [ + ChatGeneration( + message=AIMessage( + content="Response", response_metadata=response_metadata + ), + generation_info=generation_info, + ) + ] + ], + llm_output={}, + ) + + cb._pop_run_and_capture_generation(run_id, None, response) + + props = mock_client.capture.call_args.kwargs["properties"] + assert props.get("$ai_stop_reason") == expected diff --git a/posthog/test/ai/openai/test_openai_converter.py b/posthog/test/ai/openai/test_openai_converter.py index 1c3668f68..e23ecac87 100644 --- a/posthog/test/ai/openai/test_openai_converter.py +++ b/posthog/test/ai/openai/test_openai_converter.py @@ -1,3 +1,5 @@ +import types + import pytest try: @@ -7,7 +9,11 @@ except ImportError: OPENAI_AVAILABLE = False -from posthog.ai.openai.openai_converter import format_openai_input +from posthog.ai.openai._streaming import _ResponsesStreamState +from posthog.ai.openai.openai_converter import ( + extract_openai_stop_reason, + format_openai_input, +) from posthog.test.ai.utils import make_response_usage pytestmark = pytest.mark.skipif(not OPENAI_AVAILABLE, reason="openai not available") @@ -198,3 +204,56 @@ def _chunk(delta_kwargs): refusal_block = next(b for b in content if b["type"] == "refusal") assert refusal_block["refusal"] == "I can't help with that" + + +def _response(status, incomplete_reason=None, **extra): + details = ( + types.SimpleNamespace(reason=incomplete_reason) if incomplete_reason else None + ) + return types.SimpleNamespace(status=status, incomplete_details=details, **extra) + + +@pytest.mark.parametrize( + "status,incomplete_reason,expected", + [ + # Terminal statuses stand for themselves + ("completed", None, "completed"), + ("failed", None, "failed"), + ("cancelled", None, "cancelled"), + # An incomplete run is named by what cut it short + ("incomplete", "max_output_tokens", "max_output_tokens"), + ("incomplete", None, "incomplete"), + # Lifecycle states of a background run are not stop reasons + ("queued", None, None), + ("in_progress", None, None), + ], +) +def test_extract_stop_reason_maps_responses_statuses( + status, incomplete_reason, expected +): + assert extract_openai_stop_reason(_response(status, incomplete_reason)) == expected + + +@pytest.mark.parametrize( + "status,incomplete_reason,expected", + [ + ("completed", None, "completed"), + ("incomplete", "max_output_tokens", "max_output_tokens"), + ("failed", None, "failed"), + ("in_progress", None, None), + ], +) +def test_responses_stream_records_every_terminal_stop_reason( + status, incomplete_reason, expected +): + state = _ResponsesStreamState() + state.process_chunk( + types.SimpleNamespace( + type=f"response.{status}", + response=_response( + status, incomplete_reason, model="gpt-4o", usage=None, output=[] + ), + ) + ) + + assert state.stop_reason == expected