add anthropic_cache_all_ttl

Author: Wh1isper

pydantic_ai_slim/pydantic_ai/models/anthropic.py

@@ -175,6 +175,13 @@ class AnthropicModelSettings(ModelSettings, total=False):
     See https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching for more information.
     """
 
+    anthropic_cache_all_ttl: Literal['5m', '1h']
+    """The TTL for the last message in the conversation when `anthropic_cache_all` is enabled.
+
+    When `anthropic_cache_all` is enabled, the last message content will have `cache_control` set,
+    See https://docs.claude.com/en/docs/build-with-claude/prompt-caching#1-hour-cache-duration for more information.
+    """
+
 
 @dataclass(init=False)
 class AnthropicModel(Model):
@@ -683,19 +690,22 @@ async def _map_message(  # noqa: C901
         # for subsequent requests that reuse the same context
         if anthropic_messages and model_settings.get('anthropic_cache_all'):
             m = anthropic_messages[-1]
+            ttl = model_settings.get('anthropic_cache_all_ttl', '5m')
             content = m['content']
             if isinstance(content, str):
                 # Convert string content to structured format with cache_control
                 # This typically happens with tool responses
                 m['content'] = [  # pragma: no cover
                     BetaTextBlockParam(
-                        text=content, type='text', cache_control=BetaCacheControlEphemeralParam(type='ephemeral')
+                        text=content,
+                        type='text',
+                        cache_control=BetaCacheControlEphemeralParam(type='ephemeral', ttl=ttl),
                     )
                 ]
             else:
                 # For structured content (lists), add cache_control to the last block
                 content = cast(list[BetaContentBlockParam], content)
-                self._add_cache_control_to_last_param(content)
+                self._add_cache_control_to_last_param(content, ttl=ttl)
 
         if instructions := self._get_instructions(messages, model_request_parameters):
             system_prompt_parts.insert(0, instructions)
@@ -713,7 +723,7 @@ async def _map_message(  # noqa: C901
         return system_prompt, anthropic_messages
 
     @staticmethod
-    def _add_cache_control_to_last_param(params: list[BetaContentBlockParam]) -> None:
+    def _add_cache_control_to_last_param(params: list[BetaContentBlockParam], ttl: Literal['5m', '1h'] = '5m') -> None:
         """Add cache control to the last content block param.
 
         See https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching for more information.
@@ -734,7 +744,7 @@ def _add_cache_control_to_last_param(params: list[BetaContentBlockParam]) -> Non
             raise UserError(f'Cache control not supported for param type: {last_param["type"]}')
 
         # Add cache_control to the last param
-        last_param['cache_control'] = BetaCacheControlEphemeralParam(type='ephemeral')
+        last_param['cache_control'] = BetaCacheControlEphemeralParam(type='ephemeral', ttl=ttl)
 
     @staticmethod
     async def _map_user_prompt(

tests/models/test_anthropic.py

@@ -314,7 +314,11 @@ async def test_cache_point_adds_cache_control(allow_model_requests: None):
             {
                 'role': 'user',
                 'content': [
-                    {'text': 'Some context to cache', 'type': 'text', 'cache_control': {'type': 'ephemeral'}},
+                    {
+                        'text': 'Some context to cache',
+                        'type': 'text',
+                        'cache_control': {'type': 'ephemeral', 'ttl': '5m'},
+                    },
                     {'text': 'Now the question', 'type': 'text'},
                 ],
             }
@@ -339,8 +343,8 @@ async def test_cache_point_multiple_markers(allow_model_requests: None):
 
     assert content == snapshot(
         [
-            {'text': 'First chunk', 'type': 'text', 'cache_control': {'type': 'ephemeral'}},
-            {'text': 'Second chunk', 'type': 'text', 'cache_control': {'type': 'ephemeral'}},
+            {'text': 'First chunk', 'type': 'text', 'cache_control': {'type': 'ephemeral', 'ttl': '5m'}},
+            {'text': 'Second chunk', 'type': 'text', 'cache_control': {'type': 'ephemeral', 'ttl': '5m'}},
             {'text': 'Question', 'type': 'text'},
         ]
     )
@@ -389,7 +393,7 @@ async def test_cache_point_with_image_content(allow_model_requests: None):
             {
                 'source': {'type': 'url', 'url': 'https://example.com/image.jpg'},
                 'type': 'image',
-                'cache_control': {'type': 'ephemeral'},
+                'cache_control': {'type': 'ephemeral', 'ttl': '5m'},
             },
             {'text': 'What is in this image?', 'type': 'text'},
         ]
@@ -573,7 +577,9 @@ async def test_anthropic_cache_all_text_message(allow_model_requests: None):
     content = messages[0]['content']
     assert isinstance(content, list)
     assert len(content) == 1  # pyright: ignore[reportUnknownArgumentType]
-    assert content[0] == snapshot({'type': 'text', 'text': 'test prompt', 'cache_control': {'type': 'ephemeral'}})
+    assert content[0] == snapshot(
+        {'type': 'text', 'text': 'test prompt', 'cache_control': {'type': 'ephemeral', 'ttl': '5m'}}
+    )
 
 
 async def test_anthropic_cache_all_with_multiple_messages(allow_model_requests: None):
@@ -617,7 +623,7 @@ async def test_anthropic_cache_all_with_multiple_messages(allow_model_requests:
     # Second user message should have cache_control
     last_user_content = messages[2]['content']
     assert isinstance(last_user_content, list)
-    assert last_user_content[0]['cache_control'] == {'type': 'ephemeral'}
+    assert last_user_content[0]['cache_control'] == snapshot({'type': 'ephemeral', 'ttl': '5m'})
 
 
 async def test_anthropic_cache_all_combined_with_other_cache_settings(allow_model_requests: None):
@@ -649,17 +655,17 @@ def my_tool(value: str) -> str:  # pragma: no cover
 
     # Check tools have cache_control
     tools = completion_kwargs['tools']
-    assert tools[0]['cache_control'] == {'type': 'ephemeral'}
+    assert tools[0]['cache_control'] == snapshot({'type': 'ephemeral'})
 
     # Check system has cache_control
     system = completion_kwargs['system']
-    assert system[0]['cache_control'] == {'type': 'ephemeral'}
+    assert system[0]['cache_control'] == snapshot({'type': 'ephemeral'})
 
     # Check last message has cache_control
     messages = completion_kwargs['messages']
     last_message_content = messages[-1]['content']
     assert isinstance(last_message_content, list)
-    assert last_message_content[0]['cache_control'] == {'type': 'ephemeral'}
+    assert last_message_content[0]['cache_control'] == snapshot({'type': 'ephemeral', 'ttl': '5m'})
 
 
 async def test_async_request_text_response(allow_model_requests: None):