diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ae412b0..f0945259 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- `SplunkAOTracingProcessor` (OpenAI Agents) now records Responses API LLM spans, the Agents SDK + default, as chat messages: tool calls and tool results keep their own roles instead of one JSON + `user` message, the output is a single assistant message, and workflow/agent spans show the + latest user message and the answer text instead of `"Workflow Step"` and raw `output_text` parts. + Streamed Chat Completions runs (`Runner.run_streamed`), whose LLM spans hold a Responses-style + object, now record the assistant message instead of a JSON dump of the whole response. +- Image and file parts in OpenAI Agents user messages, on both the Responses and Chat Completions + paths, are now recorded as content blocks instead of a JSON dump of the whole message, including + through `SplunkAOLogger(ingestion_hook=...)`. + ## [0.5.0] ### Added diff --git a/src/splunk_ao/handlers/openai_agents/handler.py b/src/splunk_ao/handlers/openai_agents/handler.py index dcfb32db..37bc33da 100644 --- a/src/splunk_ao/handlers/openai_agents/handler.py +++ b/src/splunk_ao/handlers/openai_agents/handler.py @@ -8,7 +8,7 @@ from agents import Span, Trace, TracingProcessor from agents.tracing import ResponseSpanData, get_current_span, get_trace_provider -from galileo_core.schemas.logging.span import LlmMetrics, LlmSpan +from galileo_core.schemas.logging.span import LlmMetrics from galileo_core.schemas.logging.span import Span as SplunkAOSpan from splunk_ao import SplunkAOLogger, splunk_ao_context from splunk_ao.handlers.span_lifecycle import ( @@ -19,6 +19,7 @@ release_handler_steps, ) from splunk_ao.schema.handlers import Node +from splunk_ao.schema.logged import LoggedLlmSpan from splunk_ao.utils import _get_timestamp from splunk_ao.utils.openai_agents import ( SplunkAOCustomSpan, @@ -253,9 +254,12 @@ def _log_node_tree(self, node: Node, state: _OpenAITraceState, reuse_current: bo elif node.node_type in ("llm", "chat"): tools = node.span_params.get("tools") - span = LlmSpan( - input=input, - output=output, + # An explicit None (e.g. a GenerationSpanData that errors before output is set) must not + # reach LoggedLlmSpan, which requires a LoggedMessage, not None. Scoped to this branch only - + # other node types (e.g. a retriever's output=[]) must keep their own falsy-but-valid output. + span = LoggedLlmSpan( + input=input if input is not None else "", + output=output if output is not None else "", name=name or "LLM Span", created_at=start_time_iso, user_metadata=metadata or {}, diff --git a/src/splunk_ao/utils/openai_agents.py b/src/splunk_ao/utils/openai_agents.py index ece9078d..893dffbf 100644 --- a/src/splunk_ao/utils/openai_agents.py +++ b/src/splunk_ao/utils/openai_agents.py @@ -16,18 +16,150 @@ from galileo_core.schemas.logging.span import Span as SplunkAOSpan from splunk_ao.schema.handlers import SPAN_TYPE -from splunk_ao.utils.serialization import serialize_to_str +from splunk_ao.utils.serialization import _convert_langchain_content_block, serialize_to_str _logger = logging.getLogger(__name__) +_TEXT_PART_TYPES = ("text", "input_text", "output_text") +# Responses API tool-call items, by the field holding the call's arguments. +_TOOL_CALL_ITEM_FIELDS = {"function_call": "arguments", "custom_tool_call": "input"} +_TOOL_OUTPUT_ITEM_TYPES = ("function_call_output", "custom_tool_call_output") +# The OTel GenAI message schema has no "developer" role; record it as the equivalent "system". +_ROLE_MAP = {"developer": "system"} + + +def _content_block(part: Any) -> dict[str, Any] | None: + """Convert one Chat Completions or Responses content part into an ingest content block.""" + if isinstance(part, str): + return {"type": "text", "text": part} + if not isinstance(part, dict): + return None + kind = part.get("type") + if kind in _TEXT_PART_TYPES or (kind is None and isinstance(part.get("text"), str)): + return {"type": "text", "text": part.get("text") or ""} + if kind == "refusal": + return {"type": "text", "text": part.get("refusal") or ""} + if kind == "input_image": + if not isinstance(part.get("image_url"), str): + # Only a provider file_id: there is no data or URL to attach. + return {"type": "text", "text": serialize_to_str(part)} + # Responses API carries image_url as a plain string, not {"url": ...}. + part = {"type": "image_url", "image_url": {"url": part["image_url"]}} + elif kind == "input_file": + file_url = part.get("file_url") + if isinstance(file_url, str) and file_url: + return {"type": "data", "modality": "document", "url": file_url} + file_data = part.get("file_data") + if isinstance(file_data, str) and file_data and not file_data.startswith("data:"): + # Raw base64 without a data URI: do not dump the payload as text. + return {"type": "text", "text": serialize_to_str({k: v for k, v in part.items() if k != "file_data"})} + # Responses API carries the file fields on the part itself, not under "file". + part = {"type": "file", "file": part} + return _convert_langchain_content_block(part) + + +def _message_content(content: Any) -> Any: + """Keep string content; convert a list of parts into ingest content blocks.""" + if not isinstance(content, list): + return content + return [block for block in (_content_block(part) for part in content) if block is not None] + + +def _normalize_chat_messages(messages: Any) -> Any: + """Convert multi-part content in Chat Completions messages into ingest content blocks.""" + if not isinstance(messages, list | tuple): + return messages + return [ + {**message, "content": _message_content(message["content"])} + if isinstance(message, dict) and isinstance(message.get("content"), list) + else message + for message in messages + ] + + +def _as_dict(item: Any) -> Any: + return item.model_dump(exclude_none=True) if hasattr(item, "model_dump") else item + + +def _tool_call(item: dict[str, Any]) -> dict[str, Any]: + return { + "id": item.get("call_id"), + "type": "function", + "function": {"name": item.get("name"), "arguments": item.get(_TOOL_CALL_ITEM_FIELDS[item["type"]])}, + } + + +def _normalize_responses_input(items: Any) -> list[dict[str, Any]]: + """Convert Responses API input (a string or item list) into chat messages, in the order sent.""" + if isinstance(items, str): + return [{"role": "user", "content": items}] + messages: list[dict[str, Any]] = [] + for item in map(_as_dict, items or []): + if not isinstance(item, dict): + continue + kind = item.get("type") + if kind in _TOOL_CALL_ITEM_FIELDS: + previous = messages[-1] if messages else None + # Parallel tool calls are one assistant turn that the item list splits up. + if previous and previous["role"] == "assistant" and previous.get("tool_calls") and not previous["content"]: + previous["tool_calls"].append(_tool_call(item)) + else: + messages.append({"role": "assistant", "content": "", "tool_calls": [_tool_call(item)]}) + elif kind in _TOOL_OUTPUT_ITEM_TYPES: + output = item.get("output") + if isinstance(output, str): + content = output + elif isinstance(output, list): + # A tool result can carry multimodal parts (e.g. text + an image); convert them into + # content blocks instead of dumping the whole list, including any base64 payload, as JSON. + content = _message_content(output) + else: + content = serialize_to_str(output) + messages.append({"role": "tool", "tool_call_id": item.get("call_id"), "content": content}) + elif "role" in item: + role = _ROLE_MAP.get(item["role"], item["role"]) + messages.append({"role": role, "content": _message_content(item.get("content") or "")}) + # Items without a role (reasoning, hosted tool calls) are skipped, as in the OTel OpenAI instrumentation. + return messages + + +def _normalize_responses_output(items: Any) -> dict[str, Any] | None: + """Combine Responses API output items into the single assistant message an LLM span records.""" + texts: list[str] = [] + tool_calls: list[dict[str, Any]] = [] + for item in map(_as_dict, items or []): + if not isinstance(item, dict): + continue + if item.get("type") == "message": + content = _message_content(item.get("content") or []) + if isinstance(content, str): + texts.append(content) + continue + texts.extend(block["text"] for block in content if block.get("type") == "text" and block.get("text")) + elif item.get("type") in _TOOL_CALL_ITEM_FIELDS: + tool_calls.append(_tool_call(item)) + if not texts and not tool_calls: + return None + message: dict[str, Any] = {"role": "assistant", "content": "\n".join(texts)} + if tool_calls: + message["tool_calls"] = tool_calls + return message + def _extract_user_input(llm_input: Any) -> str | None: - """Return the last user message content from a GenerationSpanData input list.""" + """Return the text of the latest user message that has any, from an LLM span's message list.""" if not isinstance(llm_input, list): return None for message in reversed(llm_input): - if isinstance(message, dict) and message.get("role") == "user": - return str(message.get("content") or "") + if not isinstance(message, dict) or message.get("role") != "user": + continue + content = message.get("content") + if isinstance(content, str) and content: + return content + if isinstance(content, list): + texts = [b["text"] for b in content if isinstance(b, dict) and isinstance(b.get("text"), str) and b["text"]] + if texts: + return " ".join(texts) return None @@ -287,15 +419,30 @@ def _extract_llm_data(span_data: GenerationSpanData | ResponseSpanData) -> dict[ # GenerationSpanData: input/output are list[dict] — LoggedLlmSpan validators handle conversion. # output is Sequence[Mapping] (a list from the SDK's models, possibly a tuple from a custom model), # but LlmSpanAllowedOutputType only accepts a single dict; take the first choice. + # GenerationSpanData.input/output stay None when tracing.include_data() is false + # (e.g. RunConfig(trace_include_sensitive_data=False)); LoggedLlmSpan requires a message + # list, not None, so an empty list is the valid "no data" representation. + data["input"] = _normalize_chat_messages(data["input"]) if data["input"] is not None else [] if isinstance(data["output"], Sequence) and not isinstance(data["output"], str): choices = list(data["output"]) if len(choices) > 1: _logger.debug("GenerationSpanData.output has %d choices; only the first is recorded", len(choices)) - data["output"] = choices[0] if choices else None + first = choices[0] if choices else None + if isinstance(first, dict) and first.get("object") == "response": + # Streamed Chat Completions record the adapter's Responses-shaped response, not the message. + data["output"] = _normalize_responses_output(first.get("output")) or "" + else: + data["output"] = _normalize_chat_messages([first])[0] if first is not None else None else: - # ResponseSpanData: output is a list of ResponseOutputItem objects — serialize for ingestion. - data["input"] = serialize_to_str(data["input"]) - data["output"] = serialize_to_str(data["output"]) + # ResponseSpanData: input is a string or item list, output a list of ResponseOutputItem objects. + # Both are only set once the model call returns, so they are None when the span starts + # (or stay None if tracing.include_data() is false); an empty list is the valid "no data" + # representation LoggedLlmSpan and the exporter expect, not a literal "null" string. + if data["input"] is None: + data["input"] = [] + else: + data["input"] = _normalize_responses_input(data["input"]) + data["output"] = _normalize_responses_output(data["output"]) or "" if data["temperature"] is not None: try: diff --git a/tests/test_openai_agents.py b/tests/test_openai_agents.py index afc6657d..725a0964 100644 --- a/tests/test_openai_agents.py +++ b/tests/test_openai_agents.py @@ -19,7 +19,7 @@ Runner, set_trace_processors, ) -from agents.tracing import ResponseSpanData, agent_span, generation_span +from agents.tracing import ResponseSpanData, agent_span, generation_span, response_span from opentelemetry.sdk.trace import ReadableSpan from pydantic import BaseModel from pytest import MonkeyPatch, mark @@ -219,6 +219,223 @@ def test_ingestion_hook_root_input_falls_back_to_user_message( logger.terminate() +def test_responses_api_tool_turn_exports_messages_and_workflow_io() -> None: + # Given: a second-turn response span (the SDK's default Responses API path) after one tool call, + # with input and response set after the span starts, as OpenAIResponsesModel does + # When: the trace completes + # Then: the chat span exports user/assistant/tool messages and an assistant reply, and the + # workflow span shows the user prompt as input and the reply as text output + sink = RecordingSink() + logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink) + processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False) + set_trace_processors([processor]) + response = MagicMock() + response.output = [ + {"type": "reasoning", "id": "rs_1", "summary": []}, + {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "Acme Corp."}]}, + ] + response.model = "gpt-5.1" + response.temperature = None + response.usage = None + response.error = None + response.instructions = None + response.model_dump.return_value = {} + try: + with agents_tracing.trace("Agent trace", trace_id="trace_responses_tool_turn") as openai_trace: + with agent_span("SupportAgent", parent=openai_trace) as root_span: + with response_span(parent=root_span) as span: + span.span_data.input = [ + {"role": "user", "content": "Look up customer CUST-123"}, + {"type": "function_call", "call_id": "c1", "name": "get_customer", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "c1", "output": '{"name": "Acme"}'}, + ] + span.span_data.response = response + + chat = next(s for s in sink.spans if (s.attributes or {}).get("gen_ai.operation.name") == "chat") + input_messages = json.loads(chat.attributes["gen_ai.input.messages"]) + assert [m["role"] for m in input_messages] == ["user", "assistant", "tool"] + assert input_messages[1]["parts"][0]["type"] == "tool_call" + assert input_messages[2]["parts"][0]["type"] == "tool_call_response" + output_messages = json.loads(chat.attributes["gen_ai.output.messages"]) + assert output_messages[0]["parts"] == [{"type": "text", "content": "Acme Corp."}] + + workflow = next(s for s in sink.spans if s.name == "invoke_workflow SupportAgent") + assert "Look up customer CUST-123" in workflow.attributes["gen_ai.input.messages"] + workflow_output = json.loads(workflow.attributes["gen_ai.output.messages"]) + assert workflow_output[0]["parts"] == [{"type": "text", "content": "Acme Corp."}] + finally: + set_trace_processors([]) + logger.terminate() + + +def test_responses_api_tool_output_with_image_exports_content_blocks() -> None: + # Given: a tool result (function_call_output) whose output is a list of parts - text and a + # base64 image, as the Agents SDK produces for multimodal tool results + # When: the trace completes + # Then: the exported chat span's tool message holds content blocks, not a JSON dump of the + # whole list (including the base64 payload as text) - end-to-end, not just at the normalizer + sink = RecordingSink() + logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink) + processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False) + set_trace_processors([processor]) + response = MagicMock() + response.output = [{"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "ok"}]}] + response.model = "gpt-5.1" + response.temperature = None + response.usage = None + response.error = None + response.instructions = None + response.model_dump.return_value = {} + try: + with agents_tracing.trace("Agent trace", trace_id="trace_tool_output_image") as openai_trace: + with agent_span("SupportAgent", parent=openai_trace) as root_span: + with response_span(parent=root_span) as span: + span.span_data.input = [ + {"role": "user", "content": "Show me the chart"}, + {"type": "function_call", "call_id": "c1", "name": "get_chart", "arguments": "{}"}, + { + "type": "function_call_output", + "call_id": "c1", + "output": [ + {"type": "input_text", "text": "Here is the chart."}, + {"type": "input_image", "image_url": "data:image/png;base64,AAAA"}, + ], + }, + ] + span.span_data.response = response + + chat = next(s for s in sink.spans if (s.attributes or {}).get("gen_ai.operation.name") == "chat") + input_messages = json.loads(chat.attributes["gen_ai.input.messages"]) + tool_message = next(m for m in input_messages if m["role"] == "tool") + tool_response_parts = tool_message["parts"][0]["response"] + # Structured text + data parts, not the whole list dumped as one JSON text blob. + assert [p["type"] for p in tool_response_parts] == ["text", "data"] + assert tool_response_parts[1]["modality"] == "image" + assert "input_image" not in str(tool_response_parts[0]) + finally: + set_trace_processors([]) + logger.terminate() + + +@patch("splunk_ao.logger.logger.AgentStreams") +@patch("splunk_ao.logger.logger.Projects") +@patch("splunk_ao.logger.logger.Traces") +def test_ingestion_hook_keeps_multimodal_llm_input( + mock_traces_client: Mock, mock_projects_client: Mock, mock_logstreams_client: Mock +) -> None: + # Given: the ingestion-hook path and a user message with text and an image + # When: the trace is committed + # Then: the LLM span keeps text and image content blocks instead of a JSON dump of the message + setup_mock_traces_client(mock_traces_client) + setup_mock_projects_client(mock_projects_client) + setup_mock_logstreams_client(mock_logstreams_client) + logger = SplunkAOLogger(project="test", agent_stream="test", ingestion_hook=lambda _: None) + processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False) + set_trace_processors([processor]) + try: + with agents_tracing.trace("Agent trace", trace_id="trace_hook_multimodal") as openai_trace: + with agent_span("Agent", parent=openai_trace) as root_span: + with generation_span( + input=[ + { + "role": "user", + "content": [ + {"type": "text", "text": "What is this?"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}, + ], + } + ], + output=[{"role": "assistant", "content": "A cat."}], + model="gpt-4", + parent=root_span, + ): + pass + + llm_span = logger.traces[0].spans[0].spans[0] + assert [block.type for block in llm_span.input[0].content] == ["text", "data"] + assert llm_span.input[0].content[1].modality == "image" + finally: + set_trace_processors([]) + logger.terminate() + + +@patch("splunk_ao.logger.logger.AgentStreams") +@patch("splunk_ao.logger.logger.Projects") +@patch("splunk_ao.logger.logger.Traces") +def test_ingestion_hook_llm_span_with_no_output_does_not_crash_trace( + mock_traces_client: Mock, mock_projects_client: Mock, mock_logstreams_client: Mock +) -> None: + # Given: the ingestion-hook path and a generation span that ends with no output set + # (e.g. the model call errored before GenerationSpanData.output was populated) + # When: the trace is committed + # Then: the trace completes instead of failing - LoggedLlmSpan(output=...) requires a LoggedMessage + # and raises on a bare None reaching it + setup_mock_traces_client(mock_traces_client) + setup_mock_projects_client(mock_projects_client) + setup_mock_logstreams_client(mock_logstreams_client) + logger = SplunkAOLogger(project="test", agent_stream="test", ingestion_hook=lambda _: None) + processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False) + set_trace_processors([processor]) + try: + with agents_tracing.trace("Agent trace", trace_id="trace_hook_no_output") as openai_trace: + with agent_span("Agent", parent=openai_trace) as root_span: + with generation_span(input=[{"role": "user", "content": "Hi"}], model="gpt-4", parent=root_span): + pass + + llm_span = logger.traces[0].spans[0].spans[0] + assert llm_span.output.content == "" + finally: + set_trace_processors([]) + logger.terminate() + + +@patch("splunk_ao.logger.logger.AgentStreams") +@patch("splunk_ao.logger.logger.Projects") +@patch("splunk_ao.logger.logger.Traces") +def test_ingestion_hook_builds_responses_llm_span_with_tool_messages( + mock_traces_client: Mock, mock_projects_client: Mock, mock_logstreams_client: Mock +) -> None: + # Given: the ingestion-hook path and a Responses API span with a tool call and tool result + # (only the Chat Completions path was covered on this path before) + # When: the trace is committed + # Then: the LLM span (LoggedLlmSpan) holds the tool-call/tool-result messages with their own + # roles and the assistant reply text, built without crashing + setup_mock_traces_client(mock_traces_client) + setup_mock_projects_client(mock_projects_client) + setup_mock_logstreams_client(mock_logstreams_client) + logger = SplunkAOLogger(project="test", agent_stream="test", ingestion_hook=lambda _: None) + processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False) + set_trace_processors([processor]) + response = MagicMock() + response.output = [ + {"type": "reasoning", "id": "rs_1", "summary": []}, + {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "Acme Corp."}]}, + ] + response.model = "gpt-5.1" + response.temperature = None + response.usage = None + response.error = None + response.instructions = None + response.model_dump.return_value = {} + try: + with agents_tracing.trace("Agent trace", trace_id="trace_hook_responses_tool_turn") as openai_trace: + with agent_span("SupportAgent", parent=openai_trace) as root_span: + with response_span(parent=root_span) as span: + span.span_data.input = [ + {"role": "user", "content": "Look up customer CUST-123"}, + {"type": "function_call", "call_id": "c1", "name": "get_customer", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "c1", "output": '{"name": "Acme"}'}, + ] + span.span_data.response = response + + llm_span = logger.traces[0].spans[0].spans[0] + assert [m.role.value for m in llm_span.input] == ["user", "assistant", "tool"] + assert llm_span.output.content == "Acme Corp." + finally: + set_trace_processors([]) + logger.terminate() + + def test_openai_agents_public_tracing_lifecycle_exports_and_cleans_state() -> None: sink = RecordingSink() logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink) diff --git a/tests/test_openai_agents_utils.py b/tests/test_openai_agents_utils.py index 1da1224e..1d6e9a4a 100644 --- a/tests/test_openai_agents_utils.py +++ b/tests/test_openai_agents_utils.py @@ -26,6 +26,8 @@ _extract_workflow_data, _map_span_name, _map_span_type, + _normalize_responses_input, + _normalize_responses_output, _parse_usage, ) @@ -422,3 +424,253 @@ def test_returns_none_when_no_user_message(self) -> None: # When: user input is extracted # Then: None is returned assert _extract_user_input([{"role": "system", "content": "System only"}]) is None + + def test_joins_text_blocks_and_skips_image_only_message(self) -> None: + # Given: a user message with text and image blocks, followed by an image-only user message + # When: user input is extracted + # Then: the latest user message with text is used, and only its text is returned + llm_input = [ + { + "role": "user", + "content": [{"type": "text", "text": "Describe"}, {"type": "data", "modality": "image", "url": "u"}], + }, + {"role": "user", "content": [{"type": "data", "modality": "image", "url": "u2"}]}, + ] + assert _extract_user_input(llm_input) == "Describe" + + +class TestNormalizeResponsesInput: + """Test _normalize_responses_input function.""" + + def test_string_input_is_one_user_message(self) -> None: + # Given: a plain string Responses input + # When: it is normalized + # Then: it becomes one user message + assert _normalize_responses_input("Hi") == [{"role": "user", "content": "Hi"}] + + def test_tool_turn_items_keep_roles_and_order(self) -> None: + # Given: a second-turn Responses input with developer, reasoning, parallel calls, results and replies + # When: it is normalized + # Then: items map to system/user/assistant/tool messages in order, and reasoning is skipped + items = [ + {"role": "developer", "content": "Be brief."}, + {"role": "user", "content": "Look up CUST-1"}, + {"type": "reasoning", "id": "rs_1", "summary": []}, + {"type": "function_call", "call_id": "c1", "name": "get_customer", "arguments": "{}"}, + {"type": "function_call", "call_id": "c2", "name": "get_tickets", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "c1", "output": '{"name": "Acme"}'}, + {"type": "function_call_output", "call_id": "c2", "output": "[]"}, + {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "Acme."}]}, + ] + result = _normalize_responses_input(items) + assert [m["role"] for m in result] == ["system", "user", "assistant", "tool", "tool", "assistant"] + assert [c["id"] for c in result[2]["tool_calls"]] == ["c1", "c2"] + assert result[3] == {"role": "tool", "tool_call_id": "c1", "content": '{"name": "Acme"}'} + assert result[5]["content"] == [{"type": "text", "text": "Acme."}] + + def test_multimodal_parts_become_content_blocks(self) -> None: + # Given: a Responses user message with text, a base64 image, and an image known only by file_id + # When: it is normalized + # Then: text stays text, the base64 image becomes a data block, and the file_id stays text without payload + items = [ + { + "role": "user", + "content": [ + {"type": "input_text", "text": "What is this?"}, + {"type": "input_image", "image_url": "data:image/png;base64,AAAA"}, + {"type": "input_image", "file_id": "file-123"}, + ], + } + ] + blocks = _normalize_responses_input(items)[0]["content"] + assert blocks[0] == {"type": "text", "text": "What is this?"} + assert blocks[1] == {"type": "data", "modality": "image", "base64": "AAAA", "mime_type": "image/png"} + assert blocks[2]["type"] == "text" and "file-123" in blocks[2]["text"] + + def test_tool_output_with_image_parts_becomes_content_blocks(self) -> None: + # Given: a function_call_output whose output is a list of parts (text + a base64 image), as the + # Agents SDK produces for tool results that return multimodal content + # When: it is normalized + # Then: the tool message's content is content blocks, not a JSON dump of the whole list + # (including the base64 payload as text) + items = [ + { + "type": "function_call_output", + "call_id": "c1", + "output": [ + {"type": "input_text", "text": "Here is the chart."}, + {"type": "input_image", "image_url": "data:image/png;base64,AAAA"}, + ], + } + ] + result = _normalize_responses_input(items) + assert result == [ + { + "role": "tool", + "tool_call_id": "c1", + "content": [ + {"type": "text", "text": "Here is the chart."}, + {"type": "data", "modality": "image", "base64": "AAAA", "mime_type": "image/png"}, + ], + } + ] + + def test_input_file_with_url_becomes_document_data_block(self) -> None: + # Given: a Responses input_file part that carries a URL, not inline data + # When: it is normalized + # Then: it becomes a document data block with that URL, not a JSON dump of the part + items = [{"role": "user", "content": [{"type": "input_file", "file_url": "https://x/a.pdf"}]}] + blocks = _normalize_responses_input(items)[0]["content"] + assert blocks == [{"type": "data", "modality": "document", "url": "https://x/a.pdf"}] + + def test_input_file_with_raw_base64_does_not_dump_payload_as_text(self) -> None: + # Given: a Responses input_file part with file_data that is raw base64, no "data:" URI prefix + # When: it is normalized + # Then: the payload itself is dropped from the text fallback, not embedded in it + items = [ + {"role": "user", "content": [{"type": "input_file", "filename": "a.pdf", "file_data": "AAAAAAAAAAAAAAAA"}]} + ] + blocks = _normalize_responses_input(items)[0]["content"] + assert len(blocks) == 1 + assert blocks[0]["type"] == "text" + assert "AAAAAAAAAAAAAAAA" not in blocks[0]["text"] + assert "a.pdf" in blocks[0]["text"] + + +class TestNormalizeResponsesOutput: + """Test _normalize_responses_output function.""" + + def test_message_and_tool_call_combine_into_one_assistant_message(self) -> None: + # Given: Responses output with reasoning, a text message, and a function call + # When: it is normalized + # Then: one assistant message holds the text and the tool call + items = [ + {"type": "reasoning", "id": "rs_1", "summary": []}, + {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "Checking."}]}, + {"type": "function_call", "call_id": "c1", "name": "lookup", "arguments": '{"id": 1}'}, + ] + assert _normalize_responses_output(items) == { + "role": "assistant", + "content": "Checking.", + "tool_calls": [{"id": "c1", "type": "function", "function": {"name": "lookup", "arguments": '{"id": 1}'}}], + } + + def test_reasoning_only_output_gives_none(self) -> None: + # Given: Responses output with only a reasoning item + # When: it is normalized + # Then: there is no message to record + assert _normalize_responses_output([{"type": "reasoning", "id": "rs_1", "summary": []}]) is None + + +class TestGenerationMultimodalInput: + """Test multi-part Chat Completions content in _extract_llm_data.""" + + def test_image_part_becomes_content_block(self) -> None: + # Given: GenerationSpanData whose user message has text and an image_url part + # When: LLM data is extracted + # Then: the content is converted to text and data blocks instead of being kept as raw parts + span_data = GenerationSpanData( + input=[ + { + "role": "user", + "content": [ + {"type": "text", "text": "What is this?"}, + {"type": "image_url", "image_url": {"url": "https://example.com/a.png"}}, + ], + } + ], + output=[{"role": "assistant", "content": "A cat."}], + model="gpt-4", + ) + content = _extract_llm_data(span_data)["input"][0]["content"] + assert content == [ + {"type": "text", "text": "What is this?"}, + {"type": "data", "modality": "image", "url": "https://example.com/a.png"}, + ] + + def test_tuple_input_parts_become_content_blocks(self) -> None: + # Given: GenerationSpanData input as a tuple, which its Sequence type allows a custom model to pass + # When: LLM data is extracted + # Then: the multi-part content is still converted to content blocks + span_data = GenerationSpanData( + input=( + { + "role": "user", + "content": [ + {"type": "text", "text": "What is this?"}, + {"type": "image_url", "image_url": {"url": "https://example.com/a.png"}}, + ], + }, + ), + output=[{"role": "assistant", "content": "A cat."}], + model="gpt-4", + ) + assert _extract_llm_data(span_data)["input"][0]["content"] == [ + {"type": "text", "text": "What is this?"}, + {"type": "data", "modality": "image", "url": "https://example.com/a.png"}, + ] + + +class TestGenerationStreamedOutput: + """Test GenerationSpanData output from a streamed Chat Completions run.""" + + # Streamed runs record [Response.model_dump()] instead of [ChatCompletionMessage.model_dump()]. + @staticmethod + def _streamed_output(*items: dict[str, Any]) -> list[dict[str, Any]]: + return [{"id": "__fake_id__", "object": "response", "model": "gpt-4", "output": list(items), "usage": None}] + + def test_streamed_tool_call_response_becomes_assistant_tool_calls(self) -> None: + # Given: streamed GenerationSpanData whose output is a response object holding two function calls + # When: LLM data is extracted + # Then: output is one assistant message with both tool calls, not a JSON dump of the response + span_data = GenerationSpanData( + input=[{"role": "user", "content": "Look up CUST-1"}], + output=self._streamed_output( + {"type": "function_call", "call_id": "c1", "name": "get_customer", "arguments": '{"id": "CUST-1"}'}, + {"type": "function_call", "call_id": "c2", "name": "get_tickets", "arguments": '{"id": "CUST-1"}'}, + ), + model="gpt-4", + ) + assert _extract_llm_data(span_data)["output"] == { + "role": "assistant", + "content": "", + "tool_calls": [ + {"id": "c1", "type": "function", "function": {"name": "get_customer", "arguments": '{"id": "CUST-1"}'}}, + {"id": "c2", "type": "function", "function": {"name": "get_tickets", "arguments": '{"id": "CUST-1"}'}}, + ], + } + + def test_streamed_text_response_becomes_assistant_text(self) -> None: + # Given: streamed GenerationSpanData whose output is a response object holding one output_text message + # When: LLM data is extracted + # Then: output is an assistant message with the answer text + span_data = GenerationSpanData( + input=[{"role": "user", "content": "Hi"}], + output=self._streamed_output( + { + "type": "message", + "id": "__fake_id__", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "Hello!", "annotations": []}], + } + ), + model="gpt-4", + ) + assert _extract_llm_data(span_data)["output"] == {"role": "assistant", "content": "Hello!"} + + def test_streamed_reasoning_only_response_does_not_return_none(self) -> None: + # Given: streamed GenerationSpanData whose output is a response object holding only a + # role-less reasoning item (no "message" or tool-call item for _normalize_responses_output + # to combine into an assistant message) + # When: LLM data is extracted + # Then: output falls back to an empty string, not None - on the ingestion-hook path, None + # reaches LoggedLlmSpan(output=...), which requires a LoggedMessage and rejects None. + # Not the literal "null" string either, which would be truthy and defeat every + # downstream fallback (workflow/trace output would show "null" as the answer). + span_data = GenerationSpanData( + input=[{"role": "user", "content": "Hi"}], + output=self._streamed_output({"type": "reasoning", "id": "__fake_id__", "summary": []}), + model="gpt-4", + ) + assert _extract_llm_data(span_data)["output"] == ""