fix: minimize anthropic prompt cache change#9197
Conversation
There was a problem hiding this comment.
Code Review
This pull request introduces prompt caching support for the Anthropic provider by adding cache control parameters and sorting tools to ensure consistent payloads. The review feedback suggests refactoring the duplicated payload preparation logic shared between _query and _query_stream into a single helper method to reduce code duplication and improve maintainability.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| @classmethod | ||
| def _enable_prompt_caching(cls, payloads: dict) -> None: | ||
| payloads["cache_control"] = dict(cls._PROMPT_CACHE_CONTROL) | ||
|
|
||
| @staticmethod | ||
| def _sort_anthropic_tools(tool_list: list[dict]) -> list[dict]: | ||
| return sorted(tool_list, key=lambda tool: str(tool.get("name", ""))) | ||
|
|
||
| @classmethod | ||
| def _apply_explicit_prompt_cache_breakpoints(cls, payloads: dict) -> None: | ||
| tools = payloads.get("tools") | ||
| if isinstance(tools, list) and tools: | ||
| last_tool = tools[-1] | ||
| if isinstance(last_tool, dict) and "cache_control" not in last_tool: | ||
| last_tool["cache_control"] = dict(cls._PROMPT_CACHE_CONTROL) | ||
|
|
||
| system_blocks = payloads.get("system") | ||
| if not isinstance(system_blocks, list) or not system_blocks: | ||
| return | ||
|
|
||
| last_block = system_blocks[-1] | ||
| if isinstance(last_block, dict) and "cache_control" not in last_block: | ||
| last_block["cache_control"] = dict(cls._PROMPT_CACHE_CONTROL) | ||
|
|
There was a problem hiding this comment.
To avoid code duplication and adhere to the repository's general rules, the identical payload preparation logic in _query and _query_stream should be refactored into a shared helper method (e.g., _prepare_query_payload). This helper will handle sorting tools, setting tool choices, configuring max tokens, enabling prompt caching, applying cache breakpoints, applying thinking config, and sanitizing assistant messages.
@classmethod
def _enable_prompt_caching(cls, payloads: dict) -> None:
payloads["cache_control"] = dict(cls._PROMPT_CACHE_CONTROL)
@staticmethod
def _sort_anthropic_tools(tool_list: list[dict]) -> list[dict]:
return sorted(tool_list, key=lambda tool: str(tool.get("name", "")))
@classmethod
def _apply_explicit_prompt_cache_breakpoints(cls, payloads: dict) -> None:
tools = payloads.get("tools")
if isinstance(tools, list) and tools:
last_tool = tools[-1]
if isinstance(last_tool, dict) and "cache_control" not in last_tool:
last_tool["cache_control"] = dict(cls._PROMPT_CACHE_CONTROL)
system_blocks = payloads.get("system")
if not isinstance(system_blocks, list) or not system_blocks:
return
last_block = system_blocks[-1]
if isinstance(last_block, dict) and "cache_control" not in last_block:
last_block["cache_control"] = dict(cls._PROMPT_CACHE_CONTROL)
def _prepare_query_payload(self, payloads: dict, tools: ToolSet | None) -> None:
if tools:
if tool_list := tools.get_func_desc_anthropic_style():
payloads["tools"] = self._sort_anthropic_tools(tool_list)
payloads["tool_choice"] = self._normalize_tool_choice(
payloads.get("tool_choice", "auto")
)
if "max_tokens" not in payloads:
payloads["max_tokens"] = 65536
self._enable_prompt_caching(payloads)
self._apply_explicit_prompt_cache_breakpoints(payloads)
self._apply_thinking_config(payloads)
self._sanitize_assistant_messages(payloads)References
- When implementing similar functionality for different cases, refactor the logic into a shared helper function to avoid code duplication.
| if tools: | ||
| if tool_list := tools.get_func_desc_anthropic_style(): | ||
| payloads["tools"] = tool_list | ||
| payloads["tools"] = self._sort_anthropic_tools(tool_list) | ||
| payloads["tool_choice"] = self._normalize_tool_choice( | ||
| payloads.get("tool_choice", "auto") | ||
| ) |
There was a problem hiding this comment.
Use the newly introduced _prepare_query_payload helper method here as well to simplify the payload preparation in _query_stream.
self._prepare_query_payload(payloads, tools)References
- When implementing similar functionality for different cases, refactor the logic into a shared helper function to avoid code duplication.
| if "max_tokens" not in payloads: | ||
| payloads["max_tokens"] = 65536 | ||
| self._enable_prompt_caching(payloads) | ||
| self._apply_explicit_prompt_cache_breakpoints(payloads) | ||
| self._apply_thinking_config(payloads) | ||
| self._sanitize_assistant_messages(payloads) |
There was a problem hiding this comment.
423b46c to
67339df
Compare
Related to #9196
Changes / 改动
cache_controlto the lastsystemblock in the Anthropic provider request payloadVerification / 验证
python3 -m py_compile astrbot/core/provider/sources/anthropic_source.pyastrbotDocker container after syncing the updated file