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

fix(anthropic): reduce reasoning fallback complexity

Browse source
This commit is contained in:
jesco-absolut 2026-08-01 16:54:28 -04:00
parent 0d7217ce58
commit e3fb8b1f1f
1 changed files with 25 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py
View file

@ -1426,6 +1426,28 @@ class LiteLLMAnthropicMessagesAdapter:
return "text", TextBlock(type="text", text="")
@staticmethod
def _streaming_reasoning_fields(choice: StreamingChoices) -> tuple[str, str]:
reasoning_content = ""
reasoning_signature = ""
thinking_blocks = getattr(choice.delta, "thinking_blocks", None) or []
for thinking_block in thinking_blocks:
if thinking_block["type"] == "thinking":
thinking = thinking_block.get("thinking") or ""
signature = thinking_block.get("signature") or ""
assert isinstance(thinking, str)
assert isinstance(signature, str)
reasoning_content += thinking
reasoning_signature += signature
if reasoning_content or reasoning_signature:
return reasoning_content, reasoning_signature
fallback = getattr(choice.delta, "reasoning_content", None)
return fallback or "", ""
def _translate_streaming_openai_chunk_to_anthropic(
self, choices: List[Union[OpenAIStreamingChoice, StreamingChoices]]
) -> Tuple[
@ -1450,26 +1472,9 @@ class LiteLLMAnthropicMessagesAdapter:
if tool.function is not None and tool.function.arguments is not None:
partial_json = (partial_json or "") + tool.function.arguments
elif isinstance(choice, StreamingChoices):
thinking_blocks = getattr(choice.delta, "thinking_blocks", None) or []
choice_reasoning_content = ""
choice_reasoning_signature = ""
if len(thinking_blocks) > 0:
for thinking_block in thinking_blocks:
if thinking_block["type"] == "thinking":
thinking = thinking_block.get("thinking") or ""
signature = thinking_block.get("signature") or ""
assert isinstance(thinking, str)
assert isinstance(signature, str)
choice_reasoning_content += thinking
choice_reasoning_signature += signature
if choice_reasoning_content or choice_reasoning_signature:
reasoning_content += choice_reasoning_content
reasoning_signature += choice_reasoning_signature
elif hasattr(choice.delta, "reasoning_content"):
if choice.delta.reasoning_content is not None:
reasoning_content += choice.delta.reasoning_content
choice_reasoning_content, choice_reasoning_signature = self._streaming_reasoning_fields(choice)
reasoning_content += choice_reasoning_content
reasoning_signature += choice_reasoning_signature
if partial_json is not None:
return "input_json_delta", ContentJsonBlockDelta(type="input_json_delta", partial_json=partial_json)