Cap Anthropic cache_control breakpoints at 4

src/harbor/llms/utils.py

@@ -5,18 +5,33 @@
 from litellm import Message
 
 
+# Anthropic (and Bedrock) reject a request that carries more than this many
+# ``cache_control`` breakpoints with:
+#   "A maximum of 4 blocks with cache_control may be provided. Found N."
+_MAX_ANTHROPIC_CACHE_BLOCKS = 4
+
+
 def add_anthropic_caching(
     messages: list[dict[str, Any] | Message], model_name: str
 ) -> list[dict[str, Any] | Message]:
     """
     Add ephemeral caching to the most recent messages for Anthropic models.
 
+    Anthropic allows at most ``_MAX_ANTHROPIC_CACHE_BLOCKS`` (4)
+    ``cache_control`` breakpoints per request. Multimodal messages (e.g.
+    computer-use ``[text, image]`` observations) carry multiple content
+    items each, so naively tagging every item in the most recent few
+    messages can exceed the limit and 400 the whole request. To stay under
+    the cap we tag content blocks newest-first and stop once 4 blocks have
+    been placed.
+
     Args:
         messages: List of message dictionaries
         model_name: The model name to check if it's an Anthropic model
 
     Returns:
-        List of messages with caching added to the most recent 3 messages
+        List of messages with caching added to the most recent content
+        blocks, never exceeding 4 ``cache_control`` breakpoints total.
     """
     # Only apply caching for Anthropic models
     if not ("anthropic" in model_name.lower() or "claude" in model_name.lower()):
@@ -25,40 +40,51 @@ def add_anthropic_caching(
     # Create a deep copy to avoid modifying the original messages
     cached_messages = copy.deepcopy(messages)
 
-    # Add cache_control to the most recent 3 messages
-    for n in range(len(cached_messages)):
-        if n >= len(cached_messages) - 3:
-            msg = cached_messages[n]
-
-            # Handle both dict and Message-like objects
-            if isinstance(msg, dict):
-                # Ensure content is in the expected format
-                if isinstance(msg.get("content"), str):
-                    msg["content"] = [
-                        {
-                            "type": "text",
-                            "text": msg["content"],
-                            "cache_control": {"type": "ephemeral"},
-                        }
-                    ]
-                elif isinstance(msg.get("content"), list):
-                    # Add cache_control to each content item
-                    for content_item in msg["content"]:
-                        if isinstance(content_item, dict) and "type" in content_item:
-                            content_item["cache_control"] = {"type": "ephemeral"}
-            elif hasattr(msg, "content"):
-                if isinstance(msg.content, str):
-                    msg.content = [  # type: ignore
-                        {
-                            "type": "text",
-                            "text": msg.content,
-                            "cache_control": {"type": "ephemeral"},
-                        }
-                    ]
-                elif isinstance(msg.content, list):
-                    for content_item in msg.content:
-                        if isinstance(content_item, dict) and "type" in content_item:
-                            content_item["cache_control"] = {"type": "ephemeral"}
+    # Walk messages newest-first and tag content blocks until the budget is
+    # exhausted, so the total number of cache_control markers never exceeds
+    # Anthropic's limit regardless of how many content items each message has.
+    remaining = _MAX_ANTHROPIC_CACHE_BLOCKS
+
+    for msg in reversed(cached_messages):
+        if remaining <= 0:
+            break
+
+        # Handle both dict and Message-like objects
+        if isinstance(msg, dict):
+            if isinstance(msg.get("content"), str):
+                msg["content"] = [
+                    {
+                        "type": "text",
+                        "text": msg["content"],
+                        "cache_control": {"type": "ephemeral"},
+                    }
+                ]
+                remaining -= 1
+            elif isinstance(msg.get("content"), list):
+                # Tag content items newest-first, stopping at the budget.
+                for content_item in reversed(msg["content"]):
+                    if remaining <= 0:
+                        break
+                    if isinstance(content_item, dict) and "type" in content_item:
+                        content_item["cache_control"] = {"type": "ephemeral"}
+                        remaining -= 1
+        elif hasattr(msg, "content"):
+            if isinstance(msg.content, str):
+                msg.content = [  # type: ignore
+                    {
+                        "type": "text",
+                        "text": msg.content,
+                        "cache_control": {"type": "ephemeral"},
+                    }
+                ]
+                remaining -= 1
+            elif isinstance(msg.content, list):
+                for content_item in reversed(msg.content):
+                    if remaining <= 0:
+                        break
+                    if isinstance(content_item, dict) and "type" in content_item:
+                        content_item["cache_control"] = {"type": "ephemeral"}
+                        remaining -= 1
 
     return cached_messages
 

tests/unit/llms/test_anthropic_caching.py

@@ -0,0 +1,99 @@
+"""Tests for ``add_anthropic_caching`` cache_control breakpoint capping.
+
+Anthropic (and Bedrock) reject requests carrying more than 4 ``cache_control``
+breakpoints. Multimodal computer-use messages (``[text, image]``) carry
+multiple content items each, so the helper must cap the total at 4 by tagging
+content blocks newest-first.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from litellm import Message
+
+from harbor.llms.utils import add_anthropic_caching
+
+
+def _count_cache_blocks(messages: list[Any]) -> int:
+    total = 0
+    for msg in messages:
+        content = msg["content"] if isinstance(msg, dict) else msg.content
+        if isinstance(content, list):
+            total += sum(
+                1
+                for item in content
+                if isinstance(item, dict) and "cache_control" in item
+            )
+    return total
+
+
+def _multimodal_user_message() -> dict[str, Any]:
+    return {
+        "role": "user",
+        "content": [
+            {"type": "text", "text": "look at this"},
+            {
+                "type": "image_url",
+                "image_url": {"url": "data:image/webp;base64,AAAA"},
+            },
+        ],
+    }
+
+
+def test_non_anthropic_model_is_untouched() -> None:
+    messages = [{"role": "user", "content": "hello"}]
+    result = add_anthropic_caching(messages, "openai/gpt-4o")
+    assert result is messages
+    assert result[0]["content"] == "hello"
+
+
+def test_caps_multimodal_messages_at_four_blocks() -> None:
+    # 3 multimodal messages = 6 content items; the old behaviour tagged all of
+    # them and 400'd. The cap keeps it at exactly 4.
+    messages = [_multimodal_user_message() for _ in range(3)]
+    result = add_anthropic_caching(messages, "anthropic/claude-opus-4-1")
+    assert _count_cache_blocks(result) == 4
+
+
+def test_tags_newest_blocks_first() -> None:
+    messages = [_multimodal_user_message() for _ in range(3)]
+    result = add_anthropic_caching(messages, "bedrock/claude-opus-4-1")
+
+    # The two items of the last two messages are tagged (4 total); the oldest
+    # message keeps no breakpoints.
+    assert _count_cache_blocks([result[0]]) == 0
+    assert _count_cache_blocks([result[1]]) == 2
+    assert _count_cache_blocks([result[2]]) == 2
+
+
+def test_string_content_is_promoted_to_text_block() -> None:
+    messages = [{"role": "user", "content": "hello"}]
+    result = add_anthropic_caching(messages, "claude-3-5-sonnet")
+    assert result[0]["content"] == [
+        {
+            "type": "text",
+            "text": "hello",
+            "cache_control": {"type": "ephemeral"},
+        }
+    ]
+
+
+def test_original_messages_are_not_mutated() -> None:
+    messages = [_multimodal_user_message()]
+    add_anthropic_caching(messages, "anthropic/claude-opus-4-1")
+    for item in messages[0]["content"]:
+        assert "cache_control" not in item
+
+
+def test_handles_litellm_message_objects() -> None:
+    messages = [Message(role="user", content="hello")]
+    result = add_anthropic_caching(messages, "anthropic/claude-opus-4-1")
+    assert _count_cache_blocks(result) == 1
+    assert result[0].content[0]["cache_control"] == {"type": "ephemeral"}
+
+
+def test_total_never_exceeds_four_across_many_messages() -> None:
+    messages = [_multimodal_user_message() for _ in range(10)]
+    result = add_anthropic_caching(messages, "anthropic/claude-opus-4-1")
+    assert _count_cache_blocks(result) == 4