in databao/executors/lighthouse/graph.py [0:0]
def _set_message_cache_breakpoint(config: LLMConfig, message: BaseMessage) -> BaseMessage:
"""Enable prompt caching for this message (for Anthropic models).
If you have a list of messages, set a breakpoint only on the last message to automatically
cache all previous messages.
See https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching
> Prompt caching references the entire prompt - tools, system, and messages (in that order) up to and including
the block designated with cache_control.
"""
if not ExecuteSubmit._is_anthropic_model(config):
return message
new_content: list[dict[str, Any] | str]
match message.content:
case str() | dict():
new_content = [ExecuteSubmit._set_anthropic_cache_breakpoint(message.content)]
case list():
# Set checkpoint only for the last message
new_content = message.content.copy()
new_content[-1] = ExecuteSubmit._set_anthropic_cache_breakpoint(new_content[-1])
return message.model_copy(update={"content": new_content})