scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-05 08:07:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 136

Add server side compaction translation from openai to anthropic

Browse source
This commit is contained in:
Sameer Kankute 2026-02-19 16:44:35 +05:30
parent e00c181f0c
commit a52fc738af
6 changed files with 185 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
docs/my-website/docs/response_api.md
View file

@ -1047,6 +1047,8 @@ For long-running conversations, you can enable **server-side compaction** so tha
Supported on the OpenAI Responses API when using the `openai` or `azure` provider. Pass `context_management` with a compaction entry and `compact_threshold` (token count; minimum 1000). When the context crosses the threshold, the server compacts in-stream and continues. Chain turns with `previous_response_id` or by appending output items to your next input array. See [OpenAI Compaction guide](https://developers.openai.com/api/docs/guides/compaction) for details.
> **Note:** You can use openai `context_management` format with Anthropic models via LiteLLM via responses API. LiteLLM will automatically translate this format for Anthropic and handle context management for you.
For explicit control over when compaction runs, use the standalone compact endpoint (`POST /v1/responses/compact`) instead.
### Python SDK

1
litellm/constants.py
View file

@ -644,6 +644,7 @@ DEFAULT_CHAT_COMPLETION_PARAM_VALUES = {
"prompt_cache_retention": None,
"store": None,
"metadata": None,
"context_management": None,
}
openai_compatible_endpoints: List = [

95
litellm/llms/anthropic/chat/transformation.py
View file

@ -191,6 +191,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
"user",
"web_search_options",
"speed",
"context_management",
]
if "claude-3-7-sonnet" in model or supports_reasoning(
@ -825,6 +826,62 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
return hosted_web_search_tool
@staticmethod
def map_openai_context_management_to_anthropic(
context_management: Union[List[Dict[str, Any]], Dict[str, Any]]
) -> Optional[Dict[str, Any]]:
"""
OpenAI format: [{"type": "compaction", "compact_threshold": 200000}]
Anthropic format: {
"edits": [
{
"type": "compact_20260112",
"trigger": {"type": "input_tokens", "value": 150000}
}
]
}
Args:
context_management: OpenAI or Anthropic context_management parameter
Returns:
Anthropic-formatted context_management dict, or None if invalid
"""
# If already in Anthropic format (dict with 'edits'), pass through
if isinstance(context_management, dict) and "edits" in context_management:
return context_management
# If in OpenAI format (list), transform to Anthropic format
if isinstance(context_management, list):
anthropic_edits = []
for entry in context_management:
if not isinstance(entry, dict):
continue
entry_type = entry.get("type")
if entry_type == "compaction":
anthropic_edit: Dict[str, Any] = {
"type": "compact_20260112"
}
compact_threshold = entry.get("compact_threshold")
# Rewrite to 'trigger' with correct nesting if threshold exists
if compact_threshold is not None and isinstance(compact_threshold, (int, float)):
anthropic_edit["trigger"] = {
"type": "input_tokens",
"value": int(compact_threshold)
}
# Map any other keys by passthrough except handled ones
for k in entry:
if k not in {"type", "compact_threshold"}: # only passthrough other keys
anthropic_edit[k] = entry[k]
anthropic_edits.append(anthropic_edit)
if anthropic_edits:
return {"edits": anthropic_edits}
return None
def map_openai_params( # noqa: PLR0915
self,
non_default_params: dict,
@ -931,9 +988,12 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
)
elif param == "extra_headers":
optional_params["extra_headers"] = value
elif param == "context_management" and isinstance(value, dict):
# Pass through Anthropic-specific context_management parameter
optional_params["context_management"] = value
elif param == "context_management":
# Supports both OpenAI list format and Anthropic dict format
if isinstance(value, (list, dict)):
anthropic_context_management = self.map_openai_context_management_to_anthropic(value)
if anthropic_context_management is not None:
optional_params["context_management"] = anthropic_context_management
elif param == "speed" and isinstance(value, str):
# Pass through Anthropic-specific speed parameter for fast mode
optional_params["speed"] = value
@ -1094,32 +1154,39 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
headers["anthropic-beta"] = f"{existing_beta}, {beta_value}"
def _ensure_context_management_beta_header(
self, headers: dict, context_management: dict
self, headers: dict, context_management: object
) -> None:
"""
Add appropriate beta headers based on context_management edits.
- If any edit has type "compact_20260112", add compact-2026-01-12 header
- For all other edits, add context-management-2025-06-27 header
"""
edits = context_management.get("edits", [])
edits = []
# If anthropic format (dict with "edits" key)
if isinstance(context_management, dict) and "edits" in context_management:
edits = context_management.get("edits", [])
# If OpenAI format: list of context management entries
elif isinstance(context_management, list):
edits = context_management
# Defensive: ignore/fallback if context_management not valid
else:
return
has_compact = False
has_other = False
for edit in edits:
edit_type = edit.get("type", "")
if edit_type == "compact_20260112":
if edit_type == "compact_20260112" or edit_type == "compaction":
has_compact = True
else:
has_other = True
# Add compact header if any compact edits exist
# Add compact header if any compact edits/entries exist
if has_compact:
self._ensure_beta_header(
headers, ANTHROPIC_BETA_HEADER_VALUES.COMPACT_2026_01_12.value
)
# Add context management header if any other edits exist
# Add context management header if any other edits/entries exist
if has_other:
self._ensure_beta_header(
headers, ANTHROPIC_BETA_HEADER_VALUES.CONTEXT_MANAGEMENT_2025_06_27.value

11
litellm/llms/anthropic/experimental_pass_through/messages/transformation.py
View file

@ -164,6 +164,17 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
# Remove system parameter if all content was filtered out
anthropic_messages_optional_request_params.pop("system", None)
# Transform context_management from OpenAI format to Anthropic format if needed
context_management_param = anthropic_messages_optional_request_params.get("context_management")
if context_management_param is not None:
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
transformed_context_management = AnthropicConfig.map_openai_context_management_to_anthropic(
context_management_param
)
if transformed_context_management is not None:
anthropic_messages_optional_request_params["context_management"] = transformed_context_management
####### get required params for all anthropic messages requests ######
verbose_logger.debug(f"TRANSFORMATION DEBUG - Messages: {messages}")
anthropic_messages_request: AnthropicMessagesRequest = AnthropicMessagesRequest(

3
litellm/responses/litellm_completion_transformation/transformation.py
View file

@ -211,6 +211,7 @@ class LiteLLMCompletionResponsesConfig:
"web_search_options": web_search_options,
"response_format": response_format,
"reasoning_effort": reasoning_effort,
"context_management": responses_api_request.get("context_management"),
# litellm specific params
"custom_llm_provider": custom_llm_provider,
"extra_headers": extra_headers,
@ -1349,7 +1350,7 @@ class LiteLLMCompletionResponsesConfig:
result.append(tool) # type: ignore
continue
if tool.get("type") == "function":
fn = tool.get("function") or {}
fn = cast(Dict[str, Any], tool.get("function") or {})
parameters = dict(fn.get("parameters", {}) or {})
if not parameters or "type" not in parameters:
parameters["type"] = "object"

88
tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_transformation.py
View file

@ -2582,6 +2582,94 @@ def test_compaction_block_with_other_content_types():
assert tool_calls[0]["function"]["name"] == "get_weather"
def test_map_openai_context_management_to_anthropic():
"""
Test mapping OpenAI Responses API context_management format to Anthropic format.
"""
config = AnthropicConfig()
# Test OpenAI list format with compaction
openai_format = [{"type": "compaction", "compact_threshold": 200000}]
result = config.map_openai_context_management_to_anthropic(openai_format)
assert result is not None
assert "edits" in result
assert len(result["edits"]) == 1
assert result["edits"][0]["type"] == "compact_20260112"
assert result["edits"][0]["trigger"]["type"] == "input_tokens"
assert result["edits"][0]["trigger"]["value"] == 200000
# Test OpenAI format with instructions
openai_format_with_instructions = [{
"type": "compaction",
"compact_threshold": 150000,
"instructions": "Focus on preserving code snippets"
}]
result = config.map_openai_context_management_to_anthropic(openai_format_with_instructions)
assert result is not None
assert result["edits"][0]["trigger"]["value"] == 150000
assert result["edits"][0]["instructions"] == "Focus on preserving code snippets"
# Test Anthropic format (should pass through)
anthropic_format = {
"edits": [{
"type": "compact_20260112",
"trigger": {"type": "input_tokens", "value": 150000}
}]
}
result = config.map_openai_context_management_to_anthropic(anthropic_format)
assert result == anthropic_format
def test_map_openai_params_with_context_management():
"""
Test that map_openai_params correctly transforms context_management from OpenAI to Anthropic format.
"""
config = AnthropicConfig()
# Test with OpenAI list format
non_default_params = {
"context_management": [{"type": "compaction", "compact_threshold": 200000}]
}
optional_params = {}
result = config.map_openai_params(
non_default_params=non_default_params,
optional_params=optional_params,
model="claude-opus-4-6",
drop_params=False
)
assert "context_management" in result
assert "edits" in result["context_management"]
assert result["context_management"]["edits"][0]["type"] == "compact_20260112"
assert result["context_management"]["edits"][0]["trigger"]["value"] == 200000
# Test with Anthropic dict format (should pass through)
non_default_params_anthropic = {
"context_management": {
"edits": [{
"type": "compact_20260112",
"trigger": {"type": "input_tokens", "value": 150000},
"instructions": "Focus on preserving code"
}]
}
}
optional_params = {}
result = config.map_openai_params(
non_default_params=non_default_params_anthropic,
optional_params=optional_params,
model="claude-opus-4-6",
drop_params=False
)
assert "context_management" in result
assert result["context_management"] == non_default_params_anthropic["context_management"]
def test_compaction_block_empty_list_not_added():
"""
Test that empty compaction_blocks list is not added to provider_specific_fields.