diff --git a/litellm/integrations/deepeval/deepeval.py b/litellm/integrations/deepeval/deepeval.py index 71d149e4014..eafd07ab241 100644 --- a/litellm/integrations/deepeval/deepeval.py +++ b/litellm/integrations/deepeval/deepeval.py @@ -1,5 +1,9 @@ import os -from typing import Final +from collections.abc import Mapping +from typing import Final, Literal + +from pydantic import TypeAdapter +from typing_extensions import ReadOnly, TypedDict from litellm._logging import verbose_logger from litellm._uuid import uuid @@ -16,6 +20,110 @@ from litellm.integrations.deepeval.utils import ( validate_environment, ) +_STR_MAPPING: Final = TypeAdapter(Mapping[str, object]) +_OBJECT_TUPLE: Final = TypeAdapter(tuple[object, ...]) +_REDACTED: Final = "redacted-by-litellm" +_TEXT_PART_TYPES: Final = frozenset({"text", "output_text"}) +_RESPONSES_TOOL_CALL_TYPES: Final = frozenset({"function_call", "custom_tool_call"}) + + +class _ToolFunction(TypedDict): + name: ReadOnly[str] + arguments: ReadOnly[str] + + +class _ToolCall(TypedDict): + id: ReadOnly[str] + type: ReadOnly[Literal["function"]] + function: ReadOnly[_ToolFunction] + + +class _AssistantMessage(TypedDict): + role: ReadOnly[Literal["assistant"]] + content: ReadOnly[str | None] + tool_calls: ReadOnly[tuple[_ToolCall, ...]] + + +def _as_mapping(value: object) -> Mapping[str, object] | None: + return _STR_MAPPING.validate_python(value) if isinstance(value, dict) else None + + +def _as_str(value: object) -> str | None: + return value if isinstance(value, str) else None + + +def _dict_items(value: object) -> tuple[Mapping[str, object], ...]: + items: Final = _OBJECT_TUPLE.validate_python(value) if isinstance(value, list) else () + return tuple(mapping for item in items if (mapping := _as_mapping(item)) is not None) + + +def _part_text(part: Mapping[str, object]) -> str | None: + return _as_str(part.get("text")) if part.get("type") in _TEXT_PART_TYPES else None + + +def _content_text(content: object) -> str: + """A plain string as is, or the joined text parts of a content-part list.""" + if isinstance(content, str): + return content + return "".join(text for part in _dict_items(content) if (text := _part_text(part)) is not None) + + +def _tool_call(call_id: object, name: object, arguments: object, redact: bool) -> _ToolCall: + function: Final[_ToolFunction] = { + "name": _as_str(name) or "", + "arguments": _REDACTED if redact else (_as_str(arguments) or ""), + } + tool_call: Final[_ToolCall] = {"id": _as_str(call_id) or "", "type": "function", "function": function} + return tool_call + + +def _chat_tool_call(tool_call: Mapping[str, object], redact: bool) -> _ToolCall: + function: Final = _as_mapping(tool_call.get("function")) or {} + return _tool_call(tool_call.get("id"), function.get("name"), function.get("arguments"), redact) + + +def _responses_tool_call(item: Mapping[str, object], redact: bool) -> _ToolCall: + arguments_key: Final = "input" if item.get("type") == "custom_tool_call" else "arguments" + call_id: Final = item.get("call_id") or item.get("id") + return _tool_call(call_id, item.get("name"), item.get(arguments_key), redact) + + +def _span_output(text: str, tool_calls: tuple[_ToolCall, ...], redact: bool) -> _AssistantMessage | str: + span_text: Final = _REDACTED if redact and text else text + if not tool_calls: + return span_text + message: Final[_AssistantMessage] = {"role": "assistant", "content": span_text or None, "tool_calls": tool_calls} + return message + + +def _chat_output(choice: Mapping[str, object], redact: bool) -> _AssistantMessage | str | None: + message: Final = _as_mapping(choice.get("message")) or {} + content: Final = message.get("content") + raw_tool_calls: Final = _dict_items(message.get("tool_calls")) + tool_calls: Final = tuple(_chat_tool_call(tool_call, redact) for tool_call in raw_tool_calls) + if content is None and not tool_calls: + return None + return _span_output(_content_text(content), tool_calls, redact) + + +def _responses_output(output_items: tuple[Mapping[str, object], ...], redact: bool) -> _AssistantMessage | str: + messages: Final = tuple(item for item in output_items if item.get("type") == "message") + message_texts: Final = tuple(_content_text(item.get("content")) for item in messages) + tool_items: Final = tuple(item for item in output_items if item.get("type") in _RESPONSES_TOOL_CALL_TYPES) + tool_calls: Final = tuple(_responses_tool_call(item, redact) for item in tool_items) + text: Final = "\n\n".join(message_text for message_text in message_texts if message_text) + return _span_output(text, tool_calls, redact) if text or tool_calls else "NO_OUTPUT" + + +def _success_output(response: object, redact: bool) -> _AssistantMessage | str | None: + """Chat completions read ``choices``; Responses API payloads read ``output``.""" + response_dict: Final = _as_mapping(response) or {} + choices: Final = _dict_items(response_dict.get("choices")) + if choices: + return _chat_output(choices[0], redact) + output_items: Final = _dict_items(response_dict.get("output")) + return _responses_output(output_items, redact) if output_items else "NO_OUTPUT" + # This file includes the custom callbacks for LiteLLM Proxy # Once defined, these can be passed in proxy_config.yaml @@ -101,12 +209,7 @@ class DeepEvalLogger(CustomLogger): # extract usage usage: Final = standard_logging_object.get("response", {}).get("usage", {}) if is_success: - output = ( - standard_logging_object.get("response", {}) - .get("choices", [{}])[0] - .get("message", {}) - .get("content", "NO_OUTPUT") - ) + output = _success_output(standard_logging_object.get("response", {}), redact=self.turn_off_message_logging) else: output = str(standard_logging_object.get("error_string", "")) return BaseApiSpan( diff --git a/litellm/integrations/deepeval/types.py b/litellm/integrations/deepeval/types.py index eb4bb1f9b2a..00b639fa015 100644 --- a/litellm/integrations/deepeval/types.py +++ b/litellm/integrations/deepeval/types.py @@ -1,4 +1,5 @@ # Duplicate -> https://github.com/confident-ai/deepeval/blob/main/deepeval/tracing/api.py +from collections.abc import Mapping from enum import Enum from typing import Any, ClassVar, Literal @@ -35,7 +36,7 @@ class BaseApiSpan(LiteLLMBaseModel): start_time: str = Field(alias="startTime") end_time: str = Field(alias="endTime") input: dict | list | str | None = None - output: dict | list | str | None = None + output: Mapping[str, object] | str | None = None error: str | None = None # llm diff --git a/tests/unit/integrations/test_deepeval.py b/tests/unit/integrations/test_deepeval.py index 6dd3aef0a03..02517b8c870 100644 --- a/tests/unit/integrations/test_deepeval.py +++ b/tests/unit/integrations/test_deepeval.py @@ -1,13 +1,50 @@ +import json import os import unittest +from collections.abc import Mapping from litellm._uuid import uuid from datetime import datetime, timezone +from typing import Final from unittest.mock import MagicMock, patch from litellm.integrations.deepeval.api import Endpoints, HttpMethods from litellm.integrations.deepeval.deepeval import DeepEvalLogger from litellm.integrations.deepeval.types import SpanApiType, TraceSpanApiStatus +_CHAT_TOOL_CALL: Final = { + "id": "call_1", + "type": "function", + "function": {"name": "get_weather", "arguments": '{"city": "Boston"}'}, +} +_RESPONSES_FUNCTION_CALL: Final = { + "type": "function_call", + "id": "fc_1", + "call_id": "call_1", + "name": "get_weather", + "arguments": '{"city": "Boston"}', +} +_EXPECTED_TOOL_CALL: Final = { + "id": "call_1", + "type": "function", + "function": {"name": "get_weather", "arguments": '{"city": "Boston"}'}, +} + + +def _chat_response(content: str | None, tool_calls: tuple[Mapping[str, object], ...]) -> Mapping[str, object]: + return { + "usage": {"prompt_tokens": 12, "completion_tokens": 7}, + "choices": [{"message": {"role": "assistant", "content": content, "tool_calls": list(tool_calls)}}], + } + + +def _responses_message(*texts: str) -> Mapping[str, object]: + return { + "type": "message", + "id": "msg_1", + "role": "assistant", + "content": [{"type": "output_text", "text": text, "annotations": []} for text in texts], + } + class TestDeepEvalLogger(unittest.TestCase): @patch.dict(os.environ, {"CONFIDENT_API_KEY": "test-api-key"}) @@ -94,6 +131,112 @@ class TestDeepEvalLogger(unittest.TestCase): self.assertEqual(llm_span["inputTokenCount"], 10) self.assertEqual(llm_span["outputTokenCount"], 20) + def _sent_span_output(self, response: Mapping[str, object]) -> object: + """Log ``response`` and return the span output exactly as it goes over the wire (JSON).""" + kwargs: Final = { + "input": self.input_str, + "standard_logging_object": { + "id": self.span_id, + "trace_id": self.trace_id, + "model": self.model, + "response": response, + }, + } + self.logger.log_success_event(kwargs, self.mock_response_obj, self.start_time, self.end_time) + body: Final = self.mock_api_instance.send_request.call_args.kwargs["body"] + return json.loads(json.dumps(body))["llmSpans"][0].get("output") + + def test_chat_tool_call_only_records_tool_call(self): + output: Final = self._sent_span_output(_chat_response(content=None, tool_calls=(_CHAT_TOOL_CALL,))) + + self.assertEqual(output, {"role": "assistant", "content": None, "tool_calls": [_EXPECTED_TOOL_CALL]}) + + def test_chat_tool_call_with_empty_string_content_records_tool_call(self): + output: Final = self._sent_span_output(_chat_response(content="", tool_calls=(_CHAT_TOOL_CALL,))) + + self.assertEqual(output, {"role": "assistant", "content": None, "tool_calls": [_EXPECTED_TOOL_CALL]}) + + def test_chat_text_and_tool_call_records_both(self): + output: Final = self._sent_span_output( + _chat_response(content="Checking the weather.", tool_calls=(_CHAT_TOOL_CALL,)) + ) + + self.assertEqual( + output, + {"role": "assistant", "content": "Checking the weather.", "tool_calls": [_EXPECTED_TOOL_CALL]}, + ) + + def test_responses_text_joins_parts_and_separates_messages(self): + output: Final = self._sent_span_output( + { + "usage": {}, + "output": [ + {"type": "reasoning", "id": "rs_1", "summary": [{"type": "summary_text", "text": "thinking"}]}, + _responses_message("Let me check. "), + _responses_message("Hello ", "from responses"), + ], + } + ) + + self.assertEqual(output, "Let me check. \n\nHello from responses") + + def test_responses_function_call_only_records_tool_call(self): + output: Final = self._sent_span_output({"usage": {}, "output": [_RESPONSES_FUNCTION_CALL]}) + + self.assertEqual(output, {"role": "assistant", "content": None, "tool_calls": [_EXPECTED_TOOL_CALL]}) + + def test_responses_text_and_function_call_records_both(self): + output: Final = self._sent_span_output( + {"usage": {}, "output": [_responses_message("Checking the weather."), _RESPONSES_FUNCTION_CALL]} + ) + + self.assertEqual( + output, + {"role": "assistant", "content": "Checking the weather.", "tool_calls": [_EXPECTED_TOOL_CALL]}, + ) + + def test_responses_reasoning_only_records_no_output(self): + output: Final = self._sent_span_output( + {"usage": {}, "output": [{"type": "reasoning", "id": "rs_1", "summary": [{"type": "summary_text", "text": "secret"}]}]} + ) + + self.assertEqual(output, "NO_OUTPUT") + + def test_turn_off_message_logging_redacts_tool_arguments(self): + self.logger.turn_off_message_logging = True + chat_output: Final = self._sent_span_output(_chat_response(content=None, tool_calls=(_CHAT_TOOL_CALL,))) + responses_output: Final = self._sent_span_output({"usage": {}, "output": [_RESPONSES_FUNCTION_CALL]}) + redacted_tool_call: Final = { + "id": "call_1", + "type": "function", + "function": {"name": "get_weather", "arguments": "redacted-by-litellm"}, + } + + self.assertEqual(chat_output, {"role": "assistant", "content": None, "tool_calls": [redacted_tool_call]}) + self.assertEqual(responses_output, {"role": "assistant", "content": None, "tool_calls": [redacted_tool_call]}) + + def test_turn_off_message_logging_redacts_text(self): + self.logger.turn_off_message_logging = True + chat_text: Final = self._sent_span_output(_chat_response(content="secret answer", tool_calls=())) + responses_text: Final = self._sent_span_output({"usage": {}, "output": [_responses_message("secret answer")]}) + chat_mixed: Final = self._sent_span_output( + _chat_response(content="secret answer", tool_calls=(_CHAT_TOOL_CALL,)) + ) + responses_mixed: Final = self._sent_span_output( + {"usage": {}, "output": [_responses_message("secret answer"), _RESPONSES_FUNCTION_CALL]} + ) + + self.assertEqual(chat_text, "redacted-by-litellm") + self.assertEqual(responses_text, "redacted-by-litellm") + for mixed in (chat_mixed, responses_mixed): + self.assertEqual(mixed["content"], "redacted-by-litellm") + self.assertEqual(mixed["tool_calls"][0]["function"]["arguments"], "redacted-by-litellm") + + def test_empty_choices_records_no_output(self): + output: Final = self._sent_span_output({"usage": {}, "choices": []}) + + self.assertEqual(output, "NO_OUTPUT") + def test_log_failure_event(self): error_message = "This is an error." kwargs = {