Add prompt caching

Author: roaga

requirements-constraints.txt

@@ -99,7 +99,7 @@ chromadb==0.4.14
 google-cloud-storage==2.*
 google-cloud-aiplatform==1.*
 google-cloud-secret-manager==2.*
-anthropic[vertex]==0.34.2
+anthropic[vertex]==0.41.0
 langfuse @ git+https://github.com/jennmueng/langfuse-python.git@9d9350de1e4e84fa548fe84f82c1b826be17956e
 watchdog
 stumpy==1.13.0

requirements.txt

@@ -21,7 +21,7 @@ amqp==5.3.1
     # via kombu
 annotated-types==0.7.0
     # via pydantic
-anthropic==0.34.2
+anthropic==0.41.0
     # via -r requirements-constraints.txt
 anyio==4.7.0
     # via
@@ -268,7 +268,7 @@ httpcore==1.0.7
     # via httpx
 httptools==0.6.4
     # via uvicorn
-httpx==0.27.2
+httpx==0.28.1
     # via
     #   -r requirements-constraints.txt
     #   anthropic
@@ -700,7 +700,6 @@ sniffio==1.3.1
     # via
     #   anthropic
     #   anyio
-    #   httpx
     #   openai
 sqlalchemy==2.0.25
     # via
@@ -730,7 +729,6 @@ threadpoolctl==3.2.0
     #   scikit-learn
 tokenizers==0.15.2
     # via
-    #   anthropic
     #   chromadb
     #   transformers
 torch==2.2.0

src/seer/automation/agent/client.py

@@ -7,6 +7,7 @@
 import anthropic
 from anthropic import NOT_GIVEN
 from anthropic.types import (
+    CacheControlEphemeralParam,
     MessageParam,
     TextBlockParam,
     ToolParam,
@@ -542,10 +543,22 @@ def _prep_message_and_tools(
         message_dicts = [cls.to_message_param(message) for message in messages] if messages else []
         if prompt:
             message_dicts.append(cls.to_message_param(Message(role="user", content=prompt)))
+        # Set caching breakpoints for the last message and the 3rd last message
+        if len(message_dicts) > 0 and message_dicts[-1]["content"]:
+            message_dicts[-1]["content"][0]["cache_control"] = CacheControlEphemeralParam(
+                type="ephemeral"
+            )
+        if len(message_dicts) >= 4 and message_dicts[-4]["content"]:
+            message_dicts[-4]["content"][0]["cache_control"] = CacheControlEphemeralParam(
+                type="ephemeral"
+            )
 
         tool_dicts = (
             [cls.to_tool_dict(tool) for tool in tools] if tools and len(tools) > 0 else None
         )
+        # set caching breakpoint at end of tools
+        if tool_dicts:
+            tool_dicts[-1]["cache_control"] = CacheControlEphemeralParam(type="ephemeral")
 
         return message_dicts, tool_dicts, system_prompt