diff --git a/litellm/litellm_core_utils/prompt_templates/factory.py b/litellm/litellm_core_utils/prompt_templates/factory.py index c907ed32b95..16d8e93cbfc 100644 --- a/litellm/litellm_core_utils/prompt_templates/factory.py +++ b/litellm/litellm_core_utils/prompt_templates/factory.py @@ -2018,6 +2018,223 @@ def anthropic_process_openai_file_message( ) +def _sanitize_empty_text_content( + message: AllMessageValues, +) -> AllMessageValues: + """ + Case C: Sanitize empty text content + - Replace empty or whitespace-only text content with a placeholder message. + + Returns: + The message with sanitized content if needed, otherwise the original message + """ + if message.get("role") in ["user", "assistant"]: + content = message.get("content") + if isinstance(content, str): + if not content or not content.strip(): + message = dict(message) # Make a copy + message["content"] = "[System: Empty message content sanitised to satisfy protocol]" + verbose_logger.debug( + f"_sanitize_empty_text_content: Replaced empty text content in {message.get('role')} message" + ) + return message + + +def _add_missing_tool_results( + current_message: AllMessageValues, + messages: List[AllMessageValues], + current_index: int, +) -> List[AllMessageValues]: + """ + Case A: Missing tool_result for tool_use (orphaned tool calls) + - If an assistant message has tool_calls but no corresponding tool result follows, + add a dummy tool result message indicating the user did not provide the result. + + Returns: + A list containing the assistant message followed by any dummy tool results needed + """ + result_messages: List[AllMessageValues] = [] + tool_calls = current_message.get("tool_calls") + + if not tool_calls or len(tool_calls) == 0: + return [current_message] + + # Collect all tool_call_ids from this assistant message + expected_tool_call_ids = set() + for tool_call in tool_calls: + tool_call_id = None + if isinstance(tool_call, dict): + tool_call_id = tool_call.get("id") + else: + tool_call_id = getattr(tool_call, "id", None) + if tool_call_id: + expected_tool_call_ids.add(tool_call_id) + + found_tool_call_ids = set() + j = current_index + 1 + + while j < len(messages): + next_msg = messages[j] + next_role = next_msg.get("role") + + if next_role == "assistant": + break + + if next_role in ["tool", "function"]: + tool_call_id = next_msg.get("tool_call_id") + if tool_call_id: + found_tool_call_ids.add(tool_call_id) + + j += 1 + + # Find missing tool results + missing_tool_call_ids = expected_tool_call_ids - found_tool_call_ids + + if missing_tool_call_ids: + verbose_logger.debug( + f"_add_missing_tool_results: Found {len(missing_tool_call_ids)} orphaned tool calls. Adding dummy tool results." + ) + + result_messages.append(current_message) + + for tool_call_id in missing_tool_call_ids: + tool_name = "unknown_tool" + for tool_call in tool_calls: + tc_id = None + if isinstance(tool_call, dict): + tc_id = tool_call.get("id") + else: + tc_id = getattr(tool_call, "id", None) + + if tc_id == tool_call_id: + if isinstance(tool_call, dict): + function = tool_call.get("function", {}) + if isinstance(function, dict): + tool_name = function.get("name", "unknown_tool") + else: + tool_name = getattr(function, "name", "unknown_tool") + else: + function = getattr(tool_call, "function", None) + if function: + tool_name = getattr(function, "name", "unknown_tool") + break + + dummy_tool_result: ChatCompletionToolMessage = { + "role": "tool", + "tool_call_id": tool_call_id, + "content": f"[System: Tool execution skipped/interrupted by user. No result provided for tool '{tool_name}'.]", + } + result_messages.append(dummy_tool_result) + + return result_messages + + return [current_message] + + +def _is_orphaned_tool_result( + current_message: AllMessageValues, + sanitized_messages: List[AllMessageValues], +) -> bool: + """ + Case B: Orphaned tool_result (unexpected result) + - Check if a tool message references a tool_call_id that doesn't exist in the previous + assistant message. + + Returns: + True if this is an orphaned tool result that should be removed, False otherwise + """ + if current_message.get("role") not in ["tool", "function"]: + return False + + tool_call_id = current_message.get("tool_call_id") + + if not tool_call_id: + return False + + # Look back to find the most recent assistant message with tool_calls + found_matching_tool_call = False + + for j in range(len(sanitized_messages) - 1, -1, -1): + prev_msg = sanitized_messages[j] + if prev_msg.get("role") == "assistant": + tool_calls = prev_msg.get("tool_calls") + if tool_calls: + for tool_call in tool_calls: + tc_id = None + if isinstance(tool_call, dict): + tc_id = tool_call.get("id") + else: + tc_id = getattr(tool_call, "id", None) + + if tc_id == tool_call_id: + found_matching_tool_call = True + break + + break + + if not found_matching_tool_call: + verbose_logger.debug( + f"_is_orphaned_tool_result: Found orphaned tool result with tool_call_id={tool_call_id}" + ) + return True + + return False + + +def sanitize_messages_for_tool_calling( + messages: List[AllMessageValues], +) -> List[AllMessageValues]: + """ + Sanitize messages for tool calling to handle common issues when modify_params=True: + + Case A: Missing tool_result for tool_use (orphaned tool calls) + - If an assistant message has tool_calls but no corresponding tool result follows, + add a dummy tool result message indicating the user did not provide the result. + + Case B: Orphaned tool_result (unexpected result) + - If a tool message references a tool_call_id that doesn't exist in the previous + assistant message, remove that tool message. + + Case C: Empty text content + - Replace empty or whitespace-only text content with a placeholder message. + + This function operates on OpenAI format messages before they are converted to + provider-specific formats. + """ + if not litellm.modify_params: + return messages + + sanitized_messages: List[AllMessageValues] = [] + i = 0 + + while i < len(messages): + current_message = messages[i] + + # Case C: Sanitize empty text content + current_message = _sanitize_empty_text_content(current_message) + + # Case A: Check if assistant message has tool_calls without following tool results + if current_message.get("role") == "assistant": + result_messages = _add_missing_tool_results(current_message, messages, i) + + # If dummy tool results were added, extend sanitized_messages and continue + if len(result_messages) > 1: + sanitized_messages.extend(result_messages) + i += 1 + continue + + # Case B: Check for orphaned tool results + if _is_orphaned_tool_result(current_message, sanitized_messages): + i += 1 + continue # Skip this orphaned tool result + + # Add the message to sanitized list + sanitized_messages.append(current_message) + i += 1 + + return sanitized_messages + + def anthropic_messages_pt( # noqa: PLR0915 messages: List[AllMessageValues], model: str, @@ -2037,6 +2254,9 @@ def anthropic_messages_pt( # noqa: PLR0915 5. System messages are a separate param to the Messages API 6. Ensure we only accept role, content. (message.name is not supported) """ + # Sanitize messages for tool calling issues when modify_params=True + messages = sanitize_messages_for_tool_calling(messages) + # add role=tool support to allow function call result/error submission user_message_types = {"user", "tool", "function"} # reformat messages to ensure user/assistant are alternating, if there's either 2 consecutive 'user' messages or 2 consecutive 'assistant' message, merge them. diff --git a/tests/test_litellm/llms/anthropic/test_message_sanitization.py b/tests/test_litellm/llms/anthropic/test_message_sanitization.py new file mode 100644 index 00000000000..489ef527b48 --- /dev/null +++ b/tests/test_litellm/llms/anthropic/test_message_sanitization.py @@ -0,0 +1,380 @@ +""" +Test message sanitization for Anthropic API when modify_params=True + +Tests three cases: +A. Missing tool_result for tool_use (orphaned tool calls) +B. Orphaned tool_result without matching tool_use +C. Empty text content +""" + +import pytest +import sys +import os + +# Add the parent directory to the path so we can import litellm +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../.."))) + +import litellm +from litellm.litellm_core_utils.prompt_templates.factory import ( + sanitize_messages_for_tool_calling, + anthropic_messages_pt, +) + + +class TestMessageSanitization: + """Test message sanitization for tool calling scenarios""" + + def setup_method(self): + """Setup for each test""" + # Save original modify_params value + self.original_modify_params = litellm.modify_params + litellm.modify_params = True + + def teardown_method(self): + """Cleanup after each test""" + # Restore original modify_params value + litellm.modify_params = self.original_modify_params + + def test_case_a_orphaned_tool_call_single(self): + """ + Test Case A: Assistant message with tool_calls but no tool result + Should add a dummy tool result message + """ + messages = [ + { + "role": "user", + "content": "What is the weather in Nashik?" + }, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "toolu_01Kus2cC3ydjBW7UK4GJqBP4", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"location": "Nashik, India"}' + } + } + ] + } + ] + + sanitized = sanitize_messages_for_tool_calling(messages) + + # Should have 3 messages: user, assistant, and dummy tool result + assert len(sanitized) == 3 + assert sanitized[0]["role"] == "user" + assert sanitized[1]["role"] == "assistant" + assert sanitized[2]["role"] == "tool" + assert sanitized[2]["tool_call_id"] == "toolu_01Kus2cC3ydjBW7UK4GJqBP4" + assert "skipped" in sanitized[2]["content"].lower() or "interrupted" in sanitized[2]["content"].lower() + assert "get_weather" in sanitized[2]["content"] + + def test_case_a_orphaned_tool_call_multiple(self): + """ + Test Case A: Assistant message with multiple tool_calls, some missing results + """ + messages = [ + { + "role": "user", + "content": "Get weather for Nashik and Mumbai" + }, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"location": "Nashik"}' + } + }, + { + "id": "call_2", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"location": "Mumbai"}' + } + } + ] + }, + { + "role": "tool", + "tool_call_id": "call_1", + "content": "Weather in Nashik: 25°C" + } + ] + + sanitized = sanitize_messages_for_tool_calling(messages) + + # Should have 4 messages: user, assistant, tool result for call_1, dummy for call_2 + assert len(sanitized) == 4 + assert sanitized[0]["role"] == "user" + assert sanitized[1]["role"] == "assistant" + assert sanitized[2]["tool_call_id"] == "call_2" # Dummy added first + assert sanitized[3]["tool_call_id"] == "call_1" # Original tool result + + def test_case_b_orphaned_tool_result(self): + """ + Test Case B: Tool result without matching tool_call in previous assistant message + Should remove the orphaned tool result + """ + messages = [ + { + "role": "user", + "content": "Hello" + }, + { + "role": "assistant", + "content": "Hi there!" + }, + { + "role": "tool", + "tool_call_id": "nonexistent_id", + "content": "Some result" + } + ] + + sanitized = sanitize_messages_for_tool_calling(messages) + + # Should have only 2 messages, orphaned tool result removed + assert len(sanitized) == 2 + assert sanitized[0]["role"] == "user" + assert sanitized[1]["role"] == "assistant" + + def test_case_b_valid_tool_result_preserved(self): + """ + Test Case B: Valid tool result with matching tool_call should be preserved + """ + messages = [ + { + "role": "user", + "content": "What's the weather?" + }, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_123", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"location": "Boston"}' + } + } + ] + }, + { + "role": "tool", + "tool_call_id": "call_123", + "content": "Weather: 20°C" + } + ] + + sanitized = sanitize_messages_for_tool_calling(messages) + + # All messages should be preserved + assert len(sanitized) == 3 + assert sanitized[2]["role"] == "tool" + assert sanitized[2]["tool_call_id"] == "call_123" + + def test_case_c_empty_text_content_user(self): + """ + Test Case C: Empty text content in user message + Should replace with placeholder + """ + messages = [ + { + "role": "user", + "content": "" + }, + { + "role": "assistant", + "content": "Hello!" + } + ] + + sanitized = sanitize_messages_for_tool_calling(messages) + + assert len(sanitized) == 2 + assert sanitized[0]["role"] == "user" + assert sanitized[0]["content"] == "[System: Empty message content sanitised to satisfy protocol]" + + def test_case_c_whitespace_only_content(self): + """ + Test Case C: Whitespace-only content + Should replace with placeholder + """ + messages = [ + { + "role": "user", + "content": " \n \t " + }, + { + "role": "assistant", + "content": " " + } + ] + + sanitized = sanitize_messages_for_tool_calling(messages) + + assert len(sanitized) == 2 + assert sanitized[0]["content"] == "[System: Empty message content sanitised to satisfy protocol]" + assert sanitized[1]["content"] == "[System: Empty message content sanitised to satisfy protocol]" + + def test_case_c_valid_content_preserved(self): + """ + Test Case C: Valid non-empty content should be preserved + """ + messages = [ + { + "role": "user", + "content": "Hello" + }, + { + "role": "assistant", + "content": "Hi there!" + } + ] + + sanitized = sanitize_messages_for_tool_calling(messages) + + assert len(sanitized) == 2 + assert sanitized[0]["content"] == "Hello" + assert sanitized[1]["content"] == "Hi there!" + + def test_combined_cases(self): + """ + Test combination of multiple cases + """ + messages = [ + { + "role": "user", + "content": "Get weather" + }, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"location": "NYC"}' + } + } + ] + }, + # Missing tool result for call_1 + { + "role": "user", + "content": "" # Empty content + }, + { + "role": "assistant", + "content": "Response" + }, + { + "role": "tool", + "tool_call_id": "orphaned_id", # Orphaned tool result + "content": "Some data" + } + ] + + sanitized = sanitize_messages_for_tool_calling(messages) + + # Should have: user, assistant, dummy tool result, user (sanitized), assistant + # Orphaned tool result should be removed + assert len(sanitized) == 5 + assert sanitized[0]["role"] == "user" + assert sanitized[1]["role"] == "assistant" + assert sanitized[2]["role"] == "tool" + assert sanitized[2]["tool_call_id"] == "call_1" # Dummy added + assert sanitized[3]["role"] == "user" + assert sanitized[3]["content"] == "[System: Empty message content sanitised to satisfy protocol]" + assert sanitized[4]["role"] == "assistant" + + def test_modify_params_false_no_sanitization(self): + """ + Test that sanitization is skipped when modify_params=False + """ + litellm.modify_params = False + + messages = [ + { + "role": "user", + "content": "" + }, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{}' + } + } + ] + } + ] + + sanitized = sanitize_messages_for_tool_calling(messages) + + # Messages should be unchanged + assert len(sanitized) == 2 + assert sanitized[0]["content"] == "" + assert len(sanitized[1].get("tool_calls", [])) == 1 + + def test_anthropic_messages_pt_integration(self): + """ + Test that sanitization is integrated into anthropic_messages_pt + """ + litellm.modify_params = True + + messages = [ + { + "role": "user", + "content": "What is the weather in Nashik?" + }, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "toolu_01Kus2cC3ydjBW7UK4GJqBP4", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"location": "Nashik, India"}' + } + } + ] + } + ] + + # This should not raise an error and should add dummy tool result + result = anthropic_messages_pt( + messages=messages, + model="claude-sonnet-4-5", + llm_provider="anthropic" + ) + + # Should have at least 2 messages (user and assistant) + # The tool result will be merged into user content + assert len(result) >= 2 + assert result[0]["role"] == "user" + assert result[1]["role"] == "assistant" + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])