fix(prompt_templates): do not mutate messages in-place in function_call_prompt

litellm/litellm_core_utils/prompt_templates/factory.py

@@ -5148,25 +5148,45 @@ def _bedrock_tools_pt(tools: list, model: str | None = None) -> list[BedrockTool
     return tool_block_list
 
 
-# Function call template
-def function_call_prompt(messages: list, functions: list):
-    function_prompt = """Produce JSON OUTPUT ONLY! Adhere to this format {"name": "function_name", "arguments":{"argument_name": "argument_value"}} The following functions are available to you:"""
-    for function in functions:
-        function_prompt += f"""\n{function}\n"""
-
-    function_added_to_prompt = False
-    for message in messages:
-        if "system" in message["role"]:
-            if isinstance(message["content"], str):
-                message["content"] += f""" {function_prompt}"""
-            else:
-                message["content"].append({"type": "text", "text": f""" {function_prompt}"""})
-            function_added_to_prompt = True
+def _append_function_prompt_to_message(
+    message: dict[str, object],  # mutable-ok: OpenAI message dict contract
+    function_prompt: str,
+) -> dict[str, object]:  # mutable-ok: OpenAI message dict contract
+    role: Final = message.get("role")
+    if not isinstance(role, str) or "system" not in role:
+        return message
+    content: Final = message.get("content")
+    if isinstance(content, str):
+        return {  # mutable-ok: system message dict contract
+            **message,
+            "content": f"{content} {function_prompt}",
+        }
+    existing_items: Final[Sequence[object]] = (  # pyright: ignore[reportUnknownVariableType]  # list narrowing
+        content if isinstance(content, list) else ()
+    )
+    return {  # mutable-ok: system message dict contract
+        **message,
+        "content": [  # mutable-ok: content block list contract
+            *existing_items,
+            {"type": "text", "text": f" {function_prompt}"},  # mutable-ok: text block dict
+        ],
+    }
 
-    if function_added_to_prompt is False:
-        messages.append({"role": "system", "content": f"""{function_prompt}"""})
 
-    return messages
+# Function call template
+def function_call_prompt(
+    messages: list[dict[str, object]],  # mutable-ok: public signature accepts list
+    functions: list[dict[str, object]],  # mutable-ok: public signature accepts list
+) -> list[dict[str, object]]:  # mutable-ok: public signature returns list
+    header: Final = 'Produce JSON OUTPUT ONLY! Adhere to this format {"name": "function_name", "arguments":{"argument_name": "argument_value"}} The following functions are available to you:'
+    function_prompt: Final = header + "".join(f"\n{function}\n" for function in functions)
+    has_system: Final = any(isinstance(m.get("role"), str) and "system" in str(m.get("role")) for m in messages)
+    if has_system:
+        return [_append_function_prompt_to_message(m, function_prompt) for m in messages]  # mutable-ok: OpenAI messages
+    return [  # mutable-ok: OpenAI messages list contract
+        *messages,
+        {"role": "system", "content": function_prompt},  # mutable-ok: system message dict contract
+    ]
 
 
 def response_schema_prompt(model: str, response_schema: dict) -> str:

tests/test_litellm/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py

@@ -3,6 +3,9 @@
 import logging
 import os
 import re
+from collections.abc import Sequence
+from copy import deepcopy
+from functools import reduce
 from typing import Final
 from unittest.mock import MagicMock, patch
 
@@ -16,17 +19,28 @@
     BedrockImageProcessor,
     _bedrock_converse_messages_pt,
     _bedrock_tools_pt,
-    _rename_duplicate_bedrock_document_names,
     _convert_to_bedrock_tool_call_invoke,
     _convert_to_bedrock_tool_call_result,
+    _rename_duplicate_bedrock_document_names,
     anthropic_messages_pt,
     convert_to_anthropic_tool_result,
     convert_to_gemini_tool_call_result,
+    function_call_prompt,
     make_valid_bedrock_tool_name,
     ollama_pt,
     sanitize_messages_for_tool_calling,
 )
-from litellm.types.llms.openai import ChatCompletionToolMessage
+from litellm.types.llms.openai import (
+    AllMessageValues,
+    ChatCompletionAssistantMessage,
+    ChatCompletionAssistantToolCall,
+    ChatCompletionSystemMessage,
+    ChatCompletionTextObject,
+    ChatCompletionToolCallFunctionChunk,
+    ChatCompletionToolMessage,
+    ChatCompletionUserMessage,
+)
+from litellm.utils import validate_and_fix_openai_messages
 
 
 def _get_gemini_function_response_inline_data_parts(result):
@@ -3821,3 +3835,110 @@ def test_convert_to_anthropic_tool_invoke_keeps_paired_server_tool_use():
         },
         server_result,
     ]
+
+
+def _build_turn_pair(turn: int) -> tuple[ChatCompletionAssistantMessage, ChatCompletionToolMessage]:
+    tool_call_chunk: Final[ChatCompletionToolCallFunctionChunk] = {
+        "name": "read_file",
+        "arguments": json.dumps({"path": f"file_{turn}"}),
+    }
+    tool_call: Final[ChatCompletionAssistantToolCall] = {
+        "id": f"call_{turn}",
+        "type": "function",
+        "function": tool_call_chunk,
+    }
+    assistant_msg: Final[ChatCompletionAssistantMessage] = {
+        "role": "assistant",
+        "content": None,
+        "tool_calls": [tool_call],  # mutable-ok: OpenAI tool_calls list contract
+    }
+    tool_msg: Final[ChatCompletionToolMessage] = {
+        "role": "tool",
+        "tool_call_id": f"call_{turn}",
+        "content": f"File {turn} contents",
+    }
+    return (assistant_msg, tool_msg)
+
+
+_SYS_STR: Final[ChatCompletionSystemMessage] = {"role": "system", "content": "You are helpful."}
+_SYS_TEXT_PART: Final[ChatCompletionTextObject] = {"type": "text", "text": "You are helpful."}
+_SYS_LIST: Final[ChatCompletionSystemMessage] = {
+    "role": "system",
+    "content": [_SYS_TEXT_PART],  # mutable-ok: OpenAI content list contract
+}
+_SYS_CACHE_PART: Final[ChatCompletionTextObject] = {
+    "type": "text",
+    "text": "Use tools.",
+    "cache_control": {"type": "ephemeral"},
+}
+_SYS_MULTI: Final[ChatCompletionSystemMessage] = {
+    "role": "system",
+    "content": [_SYS_CACHE_PART],  # mutable-ok: OpenAI content list contract
+}
+
+
+@pytest.mark.parametrize(
+    "system_messages",
+    (
+        (),
+        (_SYS_STR,),
+        (_SYS_LIST,),
+        (_SYS_STR, _SYS_MULTI),
+    ),
+)
+@pytest.mark.parametrize("validate_messages", (False, True))
+def test_function_call_prompt_does_not_inflate_multi_turn_history(
+    system_messages: Sequence[ChatCompletionSystemMessage], validate_messages: bool
+) -> None:
+    user_msg: Final[ChatCompletionUserMessage] = {"role": "user", "content": "Read the files."}
+    functions: Final = [  # mutable-ok: OpenAI function schema list contract
+        {  # mutable-ok: OpenAI function schema dict contract
+            "name": "read_file",
+            "description": "Read the contents of a file.",
+            "parameters": {"type": "object", "properties": {"path": {"type": "string"}}},  # mutable-ok: JSON schema dict
+        }
+    ]
+    original_functions: Final = deepcopy(functions)
+
+    def _run_turn(history: Sequence[AllMessageValues], turn: int) -> Sequence[AllMessageValues]:
+        turn_messages: Final = [  # mutable-ok: OpenAI messages list contract
+            *history,
+            *_build_turn_pair(turn),
+        ]
+        original_messages: Final = deepcopy(turn_messages)
+        prepared_messages: Final = (
+            validate_and_fix_openai_messages(turn_messages) if validate_messages else turn_messages
+        )
+        original_prepared_messages: Final = deepcopy(prepared_messages)
+
+        result: Final = function_call_prompt(prepared_messages, functions)
+
+        assert turn_messages == original_messages
+        assert prepared_messages == original_prepared_messages
+        assert functions == original_functions
+        assert result is not prepared_messages
+        assert result == function_call_prompt(prepared_messages, functions)
+        assert tuple(m for m in result if m["role"] != "system") == tuple(
+            m for m in prepared_messages if m["role"] != "system"
+        )
+        result_system_messages: Final = tuple(m for m in result if m["role"] == "system")
+        assert len(result_system_messages) == max(1, len(system_messages))
+        for index, message in enumerate(result_system_messages):
+            assert json.dumps(message["content"]).count("Produce JSON OUTPUT ONLY!") == 1
+            assert json.dumps(message["content"]).count("Read the contents of a file.") == 1
+            if system_messages:
+                original_content = system_messages[index]["content"]
+                if isinstance(original_content, list):
+                    assert message["content"][:-1] == original_content
+                    assert message["content"] is not prepared_messages[index]["content"]
+                else:
+                    assert message["content"].startswith(original_content + " ")
+        return turn_messages
+
+    initial_history: Final = [  # mutable-ok: OpenAI messages list contract
+        *deepcopy(system_messages),
+        user_msg,
+    ]
+    reduce(_run_turn, range(4), initial_history)
+
+