mirror of
https://github.com/open-webui/open-webui.git
synced 2026-09-17 23:52:29 +00:00
75b1836322
* fix: drop thinking blocks when converting Anthropic Messages requests to Chat Completions Claude Code and other Anthropic SDK clients pointed at /api/v1/messages send the assistant's earlier thinking blocks back with every follow-up request. Since 0.11.0 those blocks were copied into the OpenAI assistant message as content parts of type thinking, a part type Chat Completions does not define. Strict OpenAI-compatible servers such as NVIDIA Dynamo reject the whole request with 400 "data did not match any variant of untagged enum ChatCompletionRequestAssistantMessageContent", so a conversation with a reasoning model died on its second turn. The error reports its position at the very end of the body, which made the request look cut off; it was complete. Thinking and redacted_thinking blocks are now skipped in the conversion, which is what happened before 0.11.0. An assistant turn that held only thinking blocks is kept as an empty assistant message so the turn order survives. Native Anthropic and LiteLLM connections are unaffected because they receive the request untouched, and thinking blocks in responses are still produced. Fixes #29799 * fix: keep signed thinking blocks when converting Anthropic Messages requests Dropping every thinking block also removed the signed ones. Those are the blocks a gateway such as LiteLLM forwards to Anthropic, which needs the signed thinking block of the previous assistant turn when a tool-use turn continues with extended thinking. Only the unsigned blocks are the problem: Open WebUI creates them itself from reasoning_content, and strict Chat Completions backends reject them because thinking is not a content part type they know. Unsigned thinking blocks are now dropped while signed thinking and redacted_thinking blocks are kept, the same rule LiteLLM applies before forwarding to Anthropic and the rule the native chat path already uses for Anthropic reasoning details. Backends that never emit a signature, such as NVIDIA Dynamo and vLLM, keep receiving requests without thinking parts, so the original failure stays fixed.
983 lines
42 KiB
Python
983 lines
42 KiB
Python
import logging
|
|
|
|
import aiohttp
|
|
from open_webui.env import (
|
|
AIOHTTP_CLIENT_SESSION_SSL,
|
|
AIOHTTP_CLIENT_TIMEOUT_MODEL_LIST,
|
|
ENABLE_FORWARD_USER_INFO_HEADERS,
|
|
)
|
|
from open_webui.models.users import UserModel
|
|
from open_webui.utils.headers import include_user_info_headers
|
|
from open_webui.utils.json_codec import JSONCodec
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
ANTHROPIC_VERSION = '2023-06-01'
|
|
|
|
ANTHROPIC_CONVERTED_REQUEST_PARAMS = {
|
|
'model',
|
|
'messages',
|
|
'system',
|
|
'max_tokens',
|
|
'temperature',
|
|
'top_p',
|
|
'top_k',
|
|
'stop_sequences',
|
|
'stream',
|
|
'metadata',
|
|
'service_tier',
|
|
'tools',
|
|
'tool_choice',
|
|
'reasoning_effort',
|
|
}
|
|
|
|
|
|
def is_anthropic_url(url: str) -> bool:
|
|
"""Check if the URL is an Anthropic API endpoint."""
|
|
return 'api.anthropic.com' in url
|
|
|
|
|
|
async def get_anthropic_models(url: str, key: str, user: UserModel = None) -> dict:
|
|
"""
|
|
Fetch models from Anthropic's /v1/models endpoint with pagination.
|
|
Normalizes the response to OpenAI format.
|
|
"""
|
|
timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT_MODEL_LIST)
|
|
all_models = []
|
|
after_id = None
|
|
|
|
try:
|
|
async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session:
|
|
headers = {
|
|
'x-api-key': key,
|
|
'anthropic-version': ANTHROPIC_VERSION,
|
|
}
|
|
|
|
if ENABLE_FORWARD_USER_INFO_HEADERS and user:
|
|
headers = include_user_info_headers(headers, user)
|
|
|
|
while True:
|
|
params = {'limit': 1000}
|
|
if after_id:
|
|
params['after_id'] = after_id
|
|
|
|
async with session.get(
|
|
f'{url}/models',
|
|
headers=headers,
|
|
params=params,
|
|
ssl=AIOHTTP_CLIENT_SESSION_SSL,
|
|
) as response:
|
|
if response.status != 200:
|
|
error_detail = f'HTTP Error: {response.status}'
|
|
try:
|
|
res = await response.json()
|
|
if 'error' in res:
|
|
error_detail = f'External Error: {res["error"]}'
|
|
except Exception:
|
|
pass
|
|
return {'object': 'list', 'data': [], 'error': error_detail}
|
|
|
|
data = await response.json()
|
|
|
|
for model in data.get('data', []):
|
|
all_models.append(
|
|
{
|
|
'id': model.get('id'),
|
|
'object': 'model',
|
|
'created': 0,
|
|
'owned_by': 'anthropic',
|
|
'name': model.get('display_name', model.get('id')),
|
|
}
|
|
)
|
|
|
|
if not data.get('has_more', False):
|
|
break
|
|
after_id = data.get('last_id')
|
|
|
|
except Exception as e:
|
|
log.error(f'Anthropic connection error: {e}')
|
|
return None
|
|
|
|
return {'object': 'list', 'data': all_models}
|
|
|
|
|
|
##############################
|
|
#
|
|
# Anthropic Messages API Conversion Utilities
|
|
#
|
|
##############################
|
|
|
|
|
|
def _copy_cache_control(source: dict, target: dict) -> dict:
|
|
if isinstance(source, dict) and 'cache_control' in source:
|
|
target['cache_control'] = source['cache_control']
|
|
return target
|
|
|
|
|
|
def _has_cache_control(blocks: list) -> bool:
|
|
return any(isinstance(block, dict) and 'cache_control' in block for block in blocks)
|
|
|
|
|
|
def _finalize_openai_content(blocks: list) -> str | list:
|
|
if not blocks:
|
|
return ''
|
|
|
|
if len(blocks) == 1 and blocks[0].get('type') == 'text' and not _has_cache_control(blocks):
|
|
return blocks[0].get('text', '')
|
|
|
|
return blocks
|
|
|
|
|
|
def is_anthropic_messages_passthrough(url: str, api_config: dict | None = None) -> bool:
|
|
api_config = api_config or {}
|
|
provider = str(api_config.get('provider', '')).lower()
|
|
|
|
return is_anthropic_url(url or '') or provider == 'litellm'
|
|
|
|
|
|
def convert_anthropic_to_openai_payload(
|
|
anthropic_payload: dict, passthrough_params: list[str] | str | None = None
|
|
) -> dict:
|
|
"""
|
|
Convert an Anthropic Messages API request to OpenAI Chat Completions format.
|
|
|
|
Anthropic format:
|
|
{model, messages: [{role, content}], system, max_tokens, ...}
|
|
OpenAI format:
|
|
{model, messages: [{role, content}], max_tokens, ...}
|
|
"""
|
|
openai_payload = {}
|
|
|
|
# Model
|
|
openai_payload['model'] = anthropic_payload.get('model', '')
|
|
|
|
# Build messages list
|
|
messages = []
|
|
|
|
# System prompt (Anthropic has it as top-level, OpenAI as a system message)
|
|
system = anthropic_payload.get('system')
|
|
if system:
|
|
if isinstance(system, str):
|
|
messages.append({'role': 'system', 'content': system})
|
|
elif isinstance(system, list):
|
|
openai_content = []
|
|
for block in system:
|
|
if isinstance(block, dict) and block.get('type') == 'text':
|
|
openai_content.append(
|
|
_copy_cache_control(
|
|
block,
|
|
{
|
|
'type': 'text',
|
|
'text': block.get('text', ''),
|
|
},
|
|
)
|
|
)
|
|
elif isinstance(block, str):
|
|
openai_content.append({'type': 'text', 'text': block})
|
|
messages.append({'role': 'system', 'content': _finalize_openai_content(openai_content)})
|
|
|
|
# Convert messages
|
|
for msg in anthropic_payload.get('messages', []):
|
|
role = msg.get('role', 'user')
|
|
content = msg.get('content')
|
|
|
|
if isinstance(content, str):
|
|
messages.append({'role': role, 'content': content})
|
|
elif isinstance(content, list):
|
|
# Convert Anthropic content blocks to OpenAI format
|
|
openai_content = []
|
|
tool_calls = []
|
|
|
|
for block in content:
|
|
block_type = block.get('type', 'text')
|
|
|
|
if block_type == 'text':
|
|
openai_content.append(
|
|
_copy_cache_control(
|
|
block,
|
|
{
|
|
'type': 'text',
|
|
'text': block.get('text', ''),
|
|
},
|
|
)
|
|
)
|
|
elif block_type in ('thinking', 'redacted_thinking'):
|
|
# Unsigned thinking cannot be replayed upstream
|
|
if block_type == 'redacted_thinking' or block.get('signature'):
|
|
openai_content.append(_copy_cache_control(block, dict(block)))
|
|
elif block_type == 'image':
|
|
source = block.get('source', {})
|
|
if source.get('type') == 'base64':
|
|
media_type = source.get('media_type', 'image/png')
|
|
data = source.get('data', '')
|
|
openai_content.append(
|
|
_copy_cache_control(
|
|
block,
|
|
{
|
|
'type': 'image_url',
|
|
'image_url': {
|
|
'url': f'data:{media_type};base64,{data}',
|
|
},
|
|
},
|
|
)
|
|
)
|
|
elif source.get('type') == 'url':
|
|
openai_content.append(
|
|
_copy_cache_control(
|
|
block,
|
|
{
|
|
'type': 'image_url',
|
|
'image_url': {'url': source.get('url', '')},
|
|
},
|
|
)
|
|
)
|
|
elif block_type == 'tool_use':
|
|
tool_calls.append(
|
|
{
|
|
'id': block.get('id', ''),
|
|
'type': 'function',
|
|
'function': {
|
|
'name': block.get('name', ''),
|
|
'arguments': (
|
|
JSONCodec.dumps(block.get('input', {}))
|
|
if isinstance(block.get('input'), dict)
|
|
else str(block.get('input', '{}'))
|
|
),
|
|
},
|
|
}
|
|
)
|
|
elif block_type == 'tool_result':
|
|
# Tool results become separate tool messages in OpenAI format
|
|
tool_result_content = block.get('content', '')
|
|
tool_content: str | list = ''
|
|
|
|
if isinstance(tool_result_content, str):
|
|
tool_content = tool_result_content
|
|
elif isinstance(tool_result_content, list):
|
|
# Build a multimodal content array to preserve
|
|
# images and other non-text content types.
|
|
converted_parts = []
|
|
for content_block in tool_result_content:
|
|
if not isinstance(content_block, dict):
|
|
continue
|
|
content_type = content_block.get('type', 'text')
|
|
|
|
if content_type == 'text':
|
|
converted_parts.append(
|
|
_copy_cache_control(
|
|
content_block,
|
|
{
|
|
'type': 'text',
|
|
'text': content_block.get('text', ''),
|
|
},
|
|
)
|
|
)
|
|
elif content_type == 'image':
|
|
source = content_block.get('source', {})
|
|
if source.get('type') == 'base64':
|
|
media_type = source.get('media_type', 'image/png')
|
|
data = source.get('data', '')
|
|
converted_parts.append(
|
|
_copy_cache_control(
|
|
content_block,
|
|
{
|
|
'type': 'image_url',
|
|
'image_url': {
|
|
'url': f'data:{media_type};base64,{data}',
|
|
},
|
|
},
|
|
)
|
|
)
|
|
elif source.get('type') == 'url':
|
|
converted_parts.append(
|
|
_copy_cache_control(
|
|
content_block,
|
|
{
|
|
'type': 'image_url',
|
|
'image_url': {
|
|
'url': source.get('url', ''),
|
|
},
|
|
},
|
|
)
|
|
)
|
|
elif content_type == 'document':
|
|
# Documents have no direct OpenAI equivalent;
|
|
# convert to a text representation.
|
|
document_source = content_block.get('source', {})
|
|
document_title = content_block.get('title', 'Document')
|
|
document_context = content_block.get('context', '')
|
|
document_text = f'[Document: {document_title}]'
|
|
if document_context:
|
|
document_text += f'\n{document_context}'
|
|
if document_source.get('type') == 'text' and document_source.get('data'):
|
|
document_text += f'\n{document_source["data"]}'
|
|
converted_parts.append({'type': 'text', 'text': document_text})
|
|
elif content_type == 'search_result':
|
|
# Convert search results to a text
|
|
# representation with source attribution.
|
|
search_title = content_block.get('title', '')
|
|
search_url = content_block.get('source', '')
|
|
search_content_blocks = content_block.get('content', [])
|
|
search_texts = []
|
|
for search_block in search_content_blocks:
|
|
if isinstance(search_block, dict) and search_block.get('type') == 'text':
|
|
search_texts.append(search_block.get('text', ''))
|
|
search_body = '\n'.join(search_texts)
|
|
search_text = f'[Search Result: {search_title}]'
|
|
if search_url:
|
|
search_text += f'\nSource: {search_url}'
|
|
if search_body:
|
|
search_text += f'\n{search_body}'
|
|
converted_parts.append({'type': 'text', 'text': search_text})
|
|
|
|
# Flatten to string when only text parts are present
|
|
if all(part.get('type') == 'text' for part in converted_parts) and not _has_cache_control(
|
|
converted_parts
|
|
):
|
|
tool_content = '\n'.join(part.get('text', '') for part in converted_parts)
|
|
elif converted_parts:
|
|
tool_content = converted_parts
|
|
else:
|
|
tool_content = ''
|
|
|
|
# Propagate error status if present
|
|
if block.get('is_error'):
|
|
if isinstance(tool_content, str):
|
|
tool_content = f'Error: {tool_content}'
|
|
elif isinstance(tool_content, list):
|
|
tool_content.insert(
|
|
0,
|
|
{
|
|
'type': 'text',
|
|
'text': 'Error: ',
|
|
},
|
|
)
|
|
|
|
messages.append(
|
|
{
|
|
'role': 'tool',
|
|
'tool_call_id': block.get('tool_use_id', ''),
|
|
'content': tool_content,
|
|
}
|
|
)
|
|
|
|
# Build the message
|
|
if tool_calls:
|
|
# Assistant message with tool calls
|
|
msg_dict = {'role': role}
|
|
if openai_content:
|
|
msg_dict['content'] = _finalize_openai_content(openai_content)
|
|
else:
|
|
msg_dict['content'] = ''
|
|
msg_dict['tool_calls'] = tool_calls
|
|
messages.append(msg_dict)
|
|
elif openai_content or role == 'assistant':
|
|
messages.append({'role': role, 'content': _finalize_openai_content(openai_content)})
|
|
else:
|
|
messages.append({'role': role, 'content': str(content) if content else ''})
|
|
|
|
openai_payload['messages'] = messages
|
|
|
|
# max_tokens
|
|
if 'max_tokens' in anthropic_payload:
|
|
openai_payload['max_tokens'] = anthropic_payload['max_tokens']
|
|
|
|
captured_passthrough_params = {
|
|
param: value for param, value in anthropic_payload.items() if param not in ANTHROPIC_CONVERTED_REQUEST_PARAMS
|
|
}
|
|
if isinstance(passthrough_params, str):
|
|
passthrough_params = passthrough_params.split(',')
|
|
elif not isinstance(passthrough_params, (list, tuple, set)):
|
|
passthrough_params = []
|
|
passthrough_param_names = {str(item).strip() for item in passthrough_params if str(item).strip()}
|
|
if '*' in passthrough_param_names:
|
|
openai_payload.update(captured_passthrough_params)
|
|
else:
|
|
for param in passthrough_param_names:
|
|
if param in captured_passthrough_params:
|
|
openai_payload[param] = captured_passthrough_params[param]
|
|
|
|
output_config = anthropic_payload.get('output_config')
|
|
if isinstance(output_config, dict):
|
|
if 'effort' in output_config and 'reasoning_effort' not in anthropic_payload:
|
|
openai_payload['reasoning_effort'] = output_config['effort']
|
|
|
|
format_config = output_config.get('format')
|
|
if isinstance(format_config, dict):
|
|
format_type = format_config.get('type')
|
|
if format_type == 'json_schema':
|
|
json_schema = {
|
|
'name': format_config.get('name', 'response_schema'),
|
|
'schema': format_config.get('schema', {}),
|
|
}
|
|
if 'description' in format_config:
|
|
json_schema['description'] = format_config['description']
|
|
if 'strict' in format_config:
|
|
json_schema['strict'] = format_config['strict']
|
|
openai_payload['response_format'] = {
|
|
'type': 'json_schema',
|
|
'json_schema': json_schema,
|
|
}
|
|
elif format_type == 'json_object':
|
|
openai_payload['response_format'] = {'type': format_type}
|
|
|
|
if 'reasoning_effort' in anthropic_payload:
|
|
openai_payload['reasoning_effort'] = anthropic_payload['reasoning_effort']
|
|
|
|
# Common parameters
|
|
for param in ('temperature', 'top_p', 'top_k', 'stop_sequences', 'stream', 'metadata', 'service_tier'):
|
|
if param in anthropic_payload:
|
|
if param == 'stop_sequences':
|
|
openai_payload['stop'] = anthropic_payload[param]
|
|
else:
|
|
openai_payload[param] = anthropic_payload[param]
|
|
|
|
# Tools conversion: Anthropic → OpenAI
|
|
if 'tools' in anthropic_payload:
|
|
openai_tools = []
|
|
for tool in anthropic_payload['tools']:
|
|
openai_tools.append(
|
|
_copy_cache_control(
|
|
tool,
|
|
{
|
|
'type': 'function',
|
|
'function': {
|
|
'name': tool.get('name', ''),
|
|
'description': tool.get('description', ''),
|
|
'parameters': tool.get('input_schema', {}),
|
|
},
|
|
},
|
|
)
|
|
)
|
|
openai_payload['tools'] = openai_tools
|
|
|
|
# tool_choice
|
|
if 'tool_choice' in anthropic_payload:
|
|
tool_choice = anthropic_payload['tool_choice']
|
|
if isinstance(tool_choice, dict):
|
|
tool_choice_type = tool_choice.get('type', 'auto')
|
|
if tool_choice_type == 'auto':
|
|
openai_payload['tool_choice'] = 'auto'
|
|
elif tool_choice_type == 'any':
|
|
openai_payload['tool_choice'] = 'required'
|
|
elif tool_choice_type == 'tool':
|
|
openai_payload['tool_choice'] = {
|
|
'type': 'function',
|
|
'function': {'name': tool_choice.get('name', '')},
|
|
}
|
|
|
|
return openai_payload
|
|
|
|
|
|
def convert_openai_to_anthropic_response(
|
|
openai_response: dict, model: str = '', input_tokens: int | None = None
|
|
) -> dict:
|
|
"""
|
|
Convert a non-streaming OpenAI Chat Completions response to Anthropic Messages format.
|
|
"""
|
|
import uuid as _uuid
|
|
|
|
choice = {}
|
|
if openai_response.get('choices'):
|
|
choice = openai_response['choices'][0]
|
|
|
|
message = choice.get('message', {})
|
|
finish_reason = choice.get('finish_reason', 'stop')
|
|
|
|
# Map finish_reason to stop_reason
|
|
stop_reason_map = {
|
|
'stop': 'end_turn',
|
|
'length': 'max_tokens',
|
|
'tool_calls': 'tool_use',
|
|
'content_filter': 'end_turn',
|
|
}
|
|
stop_reason = stop_reason_map.get(finish_reason, 'end_turn')
|
|
|
|
# Build content blocks
|
|
content = []
|
|
message_thinking = message.get('thinking')
|
|
thinking_blocks = message.get('thinking_blocks') or []
|
|
if not thinking_blocks and isinstance(message_thinking, dict):
|
|
thinking_blocks = message_thinking.get('blocks') or []
|
|
|
|
has_thinking = False
|
|
for block in thinking_blocks:
|
|
if not isinstance(block, dict):
|
|
continue
|
|
|
|
if block.get('type') == 'redacted_thinking':
|
|
content.append({k: v for k, v in block.items() if k in {'type', 'data'}})
|
|
has_thinking = True
|
|
continue
|
|
|
|
thinking = block.get('thinking') or block.get('content') or block.get('text')
|
|
if not thinking:
|
|
continue
|
|
|
|
thinking_block = {'type': 'thinking', 'thinking': thinking}
|
|
if block.get('signature'):
|
|
thinking_block['signature'] = block['signature']
|
|
content.append(thinking_block)
|
|
has_thinking = True
|
|
|
|
reasoning_content = message.get('reasoning_content') or message.get('reasoning')
|
|
if not reasoning_content and isinstance(message_thinking, str):
|
|
reasoning_content = message_thinking
|
|
if reasoning_content and not has_thinking:
|
|
content.append({'type': 'thinking', 'thinking': reasoning_content})
|
|
|
|
message_content = message.get('content')
|
|
if message_content:
|
|
content.append({'type': 'text', 'text': message_content})
|
|
|
|
# Tool calls -> tool_use blocks
|
|
tool_calls = message.get('tool_calls') or []
|
|
for tool_call in tool_calls:
|
|
function = tool_call.get('function', {})
|
|
try:
|
|
tool_input = JSONCodec.loads(function.get('arguments', '{}'))
|
|
except (JSONCodec.JSONDecodeError, TypeError):
|
|
tool_input = {}
|
|
content.append(
|
|
{
|
|
'type': 'tool_use',
|
|
'id': tool_call.get('id', f'toolu_{_uuid.uuid4().hex[:24]}'),
|
|
'name': function.get('name', ''),
|
|
'input': tool_input,
|
|
}
|
|
)
|
|
|
|
# Usage
|
|
openai_usage = openai_response.get('usage') or {}
|
|
cache_creation = openai_usage.get('cache_creation_input_tokens')
|
|
cache_read = openai_usage.get('cache_read_input_tokens')
|
|
prompt_details = openai_usage.get('prompt_tokens_details')
|
|
if cache_read is None and isinstance(prompt_details, dict):
|
|
cache_read = prompt_details.get('cached_tokens')
|
|
|
|
usage_input = openai_usage.get('input_tokens')
|
|
if usage_input is None:
|
|
prompt_tokens = openai_usage.get('prompt_tokens')
|
|
if prompt_tokens is not None:
|
|
usage_input = max(prompt_tokens - (cache_creation or 0) - (cache_read or 0), 0)
|
|
|
|
usage_output = openai_usage.get('output_tokens')
|
|
if usage_output is None:
|
|
usage_output = openai_usage.get('completion_tokens')
|
|
|
|
usage = {
|
|
'input_tokens': usage_input if usage_input is not None else (input_tokens if input_tokens is not None else 0),
|
|
'output_tokens': usage_output if usage_output is not None else 0,
|
|
}
|
|
if cache_creation is not None:
|
|
usage['cache_creation_input_tokens'] = cache_creation
|
|
if cache_read is not None:
|
|
usage['cache_read_input_tokens'] = cache_read
|
|
if isinstance(openai_usage.get('output_tokens_details'), dict):
|
|
usage['output_tokens_details'] = openai_usage['output_tokens_details']
|
|
if isinstance(openai_usage.get('server_tool_use'), dict):
|
|
usage['server_tool_use'] = openai_usage['server_tool_use']
|
|
if openai_usage.get('service_tier') is not None:
|
|
usage['service_tier'] = openai_usage['service_tier']
|
|
|
|
return {
|
|
'id': openai_response.get('id', f'msg_{_uuid.uuid4().hex[:24]}'),
|
|
'type': 'message',
|
|
'role': 'assistant',
|
|
'content': content,
|
|
'model': model or openai_response.get('model', ''),
|
|
'stop_reason': stop_reason,
|
|
'stop_sequence': None,
|
|
'usage': usage,
|
|
}
|
|
|
|
|
|
async def openai_stream_to_anthropic_stream(openai_stream_generator, model: str = '', input_tokens: int | None = None):
|
|
"""
|
|
Convert an OpenAI SSE streaming response to Anthropic Messages SSE format.
|
|
|
|
OpenAI sends: data: {"choices": [{"delta": {"content": "..."}}]}
|
|
Anthropic sends: event: content_block_delta\\ndata: {"type": "content_block_delta", ...}
|
|
|
|
Handles text content, tool calls, and mixed content with proper
|
|
multi-block indexing as required by Anthropic's streaming protocol.
|
|
|
|
Tool calls are tracked by their unique id (not OpenAI index) so that
|
|
parallel calls sharing the same index get distinct Anthropic tool_use
|
|
blocks. Each block follows the Anthropic lifecycle: start -> delta -> stop.
|
|
"""
|
|
import uuid as _uuid
|
|
|
|
message_id = f'msg_{_uuid.uuid4().hex[:24]}'
|
|
output_tokens = 0
|
|
cache_creation_input_tokens = None
|
|
cache_read_input_tokens = None
|
|
output_tokens_details = None
|
|
server_tool_use = None
|
|
service_tier = None
|
|
stop_reason = 'end_turn'
|
|
|
|
# Track content blocks with a running index.
|
|
# Each text block or tool_use block gets its own index.
|
|
current_block_index = 0
|
|
thinking_block_open = False
|
|
text_block_open = False
|
|
|
|
# Accumulated state for each tool call, keyed by tool call id.
|
|
# Parallel calls that share the same OpenAI index get distinct entries.
|
|
# Each entry: {id, name, arguments, block_index, started, stopped}
|
|
tracked_tool_calls = {}
|
|
# Map OpenAI tool call index -> tool call id for routing
|
|
# argument-only deltas (deltas that carry arguments but no id).
|
|
index_to_tool_id = {}
|
|
# Whether any tool call block has been emitted (suppresses further text)
|
|
has_tool_calls = False
|
|
|
|
# Emit message_start
|
|
message_start = {
|
|
'type': 'message_start',
|
|
'message': {
|
|
'id': message_id,
|
|
'type': 'message',
|
|
'role': 'assistant',
|
|
'content': [],
|
|
'model': model,
|
|
'stop_reason': None,
|
|
'stop_sequence': None,
|
|
'usage': {'input_tokens': input_tokens or 0, 'output_tokens': 0},
|
|
},
|
|
}
|
|
yield f'event: message_start\ndata: {JSONCodec.dumps(message_start)}\n\n'.encode()
|
|
|
|
try:
|
|
async for chunk in openai_stream_generator:
|
|
if isinstance(chunk, bytes):
|
|
chunk = chunk.decode('utf-8', errors='ignore')
|
|
|
|
for line in chunk.strip().split('\n'):
|
|
line = line.strip()
|
|
|
|
if not line or not line.startswith('data:'):
|
|
continue
|
|
|
|
data_string = line[5:].strip()
|
|
if data_string == '[DONE]':
|
|
continue
|
|
if data_string == '{}':
|
|
continue
|
|
|
|
try:
|
|
data = JSONCodec.loads(data_string)
|
|
except (JSONCodec.JSONDecodeError, TypeError):
|
|
continue
|
|
|
|
usage_data = data.get('usage')
|
|
if isinstance(usage_data, dict):
|
|
cache_creation = usage_data.get('cache_creation_input_tokens')
|
|
cache_read = usage_data.get('cache_read_input_tokens')
|
|
prompt_details = usage_data.get('prompt_tokens_details')
|
|
if cache_read is None and isinstance(prompt_details, dict):
|
|
cache_read = prompt_details.get('cached_tokens')
|
|
|
|
usage_input = usage_data.get('input_tokens')
|
|
if usage_input is None:
|
|
prompt_tokens = usage_data.get('prompt_tokens')
|
|
if prompt_tokens is not None:
|
|
usage_input = max(prompt_tokens - (cache_creation or 0) - (cache_read or 0), 0)
|
|
|
|
usage_output = usage_data.get('output_tokens')
|
|
if usage_output is None:
|
|
usage_output = usage_data.get('completion_tokens')
|
|
|
|
if usage_input is not None:
|
|
input_tokens = usage_input
|
|
if usage_output is not None:
|
|
output_tokens = usage_output
|
|
if cache_creation is not None:
|
|
cache_creation_input_tokens = cache_creation
|
|
if cache_read is not None:
|
|
cache_read_input_tokens = cache_read
|
|
if isinstance(usage_data.get('output_tokens_details'), dict):
|
|
output_tokens_details = usage_data['output_tokens_details']
|
|
if isinstance(usage_data.get('server_tool_use'), dict):
|
|
server_tool_use = usage_data['server_tool_use']
|
|
if usage_data.get('service_tier') is not None:
|
|
service_tier = usage_data['service_tier']
|
|
|
|
choices = data.get('choices', [])
|
|
if not choices:
|
|
continue
|
|
|
|
delta = choices[0].get('delta', {})
|
|
finish_reason = choices[0].get('finish_reason')
|
|
message = choices[0].get('message') or {}
|
|
|
|
reasoning_content = (
|
|
delta.get('reasoning_content')
|
|
or delta.get('reasoning')
|
|
or delta.get('thinking')
|
|
or message.get('reasoning_content')
|
|
or message.get('reasoning')
|
|
)
|
|
if not reasoning_content:
|
|
thinking_blocks = delta.get('thinking_blocks') or message.get('thinking_blocks') or []
|
|
for block in thinking_blocks:
|
|
if isinstance(block, dict):
|
|
reasoning_content = block.get('thinking') or block.get('content') or block.get('text')
|
|
if reasoning_content:
|
|
break
|
|
|
|
if reasoning_content and not text_block_open and not has_tool_calls:
|
|
if not thinking_block_open:
|
|
block_start = {
|
|
'type': 'content_block_start',
|
|
'index': current_block_index,
|
|
'content_block': {'type': 'thinking', 'thinking': ''},
|
|
}
|
|
yield f'event: content_block_start\ndata: {JSONCodec.dumps(block_start)}\n\n'.encode()
|
|
thinking_block_open = True
|
|
|
|
block_delta = {
|
|
'type': 'content_block_delta',
|
|
'index': current_block_index,
|
|
'delta': {'type': 'thinking_delta', 'thinking': reasoning_content},
|
|
}
|
|
yield f'event: content_block_delta\ndata: {JSONCodec.dumps(block_delta)}\n\n'.encode()
|
|
|
|
# --- Handle text content ---
|
|
# Anthropic expects text blocks before tool blocks, so skip
|
|
# text deltas once any tool call has started.
|
|
content = delta.get('content')
|
|
if content and not has_tool_calls:
|
|
if thinking_block_open:
|
|
block_stop = {
|
|
'type': 'content_block_stop',
|
|
'index': current_block_index,
|
|
}
|
|
yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode()
|
|
thinking_block_open = False
|
|
current_block_index += 1
|
|
|
|
if not text_block_open:
|
|
block_start = {
|
|
'type': 'content_block_start',
|
|
'index': current_block_index,
|
|
'content_block': {'type': 'text', 'text': ''},
|
|
}
|
|
yield f'event: content_block_start\ndata: {JSONCodec.dumps(block_start)}\n\n'.encode()
|
|
text_block_open = True
|
|
|
|
block_delta = {
|
|
'type': 'content_block_delta',
|
|
'index': current_block_index,
|
|
'delta': {'type': 'text_delta', 'text': content},
|
|
}
|
|
yield f'event: content_block_delta\ndata: {JSONCodec.dumps(block_delta)}\n\n'.encode()
|
|
|
|
# --- Handle tool calls ---
|
|
# Some providers put tool_calls on the final message object
|
|
# instead of the delta; fall back to that when needed.
|
|
tool_calls = delta.get('tool_calls') or []
|
|
if not tool_calls and message.get('tool_calls'):
|
|
tool_calls = message['tool_calls']
|
|
|
|
if tool_calls:
|
|
# Close text block if one is open (text comes before tools)
|
|
if thinking_block_open:
|
|
block_stop = {
|
|
'type': 'content_block_stop',
|
|
'index': current_block_index,
|
|
}
|
|
yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode()
|
|
thinking_block_open = False
|
|
current_block_index += 1
|
|
|
|
if text_block_open:
|
|
block_stop = {
|
|
'type': 'content_block_stop',
|
|
'index': current_block_index,
|
|
}
|
|
yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode()
|
|
text_block_open = False
|
|
current_block_index += 1
|
|
|
|
for tool_call in tool_calls:
|
|
tool_call_index = tool_call.get('index', 0)
|
|
tool_call_id = tool_call.get('id', '')
|
|
tool_call_name = (tool_call.get('function') or {}).get('name', '')
|
|
arguments_chunk = (tool_call.get('function') or {}).get('arguments', '')
|
|
|
|
# Resolve which tracked tool call this delta belongs to.
|
|
# A delta with an id starts or identifies a specific tool.
|
|
# A delta without an id carries arguments for the most
|
|
# recent tool at this OpenAI index.
|
|
if tool_call_id:
|
|
if tool_call_id not in tracked_tool_calls:
|
|
tracked_tool_calls[tool_call_id] = {
|
|
'id': tool_call_id,
|
|
'name': tool_call_name,
|
|
'arguments': '',
|
|
'block_index': -1,
|
|
'started': False,
|
|
'stopped': False,
|
|
}
|
|
index_to_tool_id[tool_call_index] = tool_call_id
|
|
tool = tracked_tool_calls[tool_call_id]
|
|
elif tool_call_index in index_to_tool_id:
|
|
tool = tracked_tool_calls[index_to_tool_id[tool_call_index]]
|
|
else:
|
|
# First delta for this index with no id; create a
|
|
# provisional entry with a generated fallback id.
|
|
fallback_id = f'toolu_{_uuid.uuid4().hex[:24]}'
|
|
tracked_tool_calls[fallback_id] = {
|
|
'id': fallback_id,
|
|
'name': tool_call_name,
|
|
'arguments': '',
|
|
'block_index': -1,
|
|
'started': False,
|
|
'stopped': False,
|
|
}
|
|
index_to_tool_id[tool_call_index] = fallback_id
|
|
tool = tracked_tool_calls[fallback_id]
|
|
|
|
# Update name if provided on a later delta
|
|
if tool_call_name and not tool['name']:
|
|
tool['name'] = tool_call_name
|
|
|
|
# Emit content_block_start once we have a name
|
|
if not tool['started'] and tool['name']:
|
|
tool['block_index'] = current_block_index
|
|
tool['started'] = True
|
|
has_tool_calls = True
|
|
|
|
block_start = {
|
|
'type': 'content_block_start',
|
|
'index': current_block_index,
|
|
'content_block': {
|
|
'type': 'tool_use',
|
|
'id': tool['id'],
|
|
'name': tool['name'],
|
|
'input': {},
|
|
},
|
|
}
|
|
yield f'event: content_block_start\ndata: {JSONCodec.dumps(block_start)}\n\n'.encode()
|
|
current_block_index += 1
|
|
|
|
# Buffer arguments and emit as input_json_delta
|
|
if arguments_chunk:
|
|
tool['arguments'] += arguments_chunk
|
|
|
|
if tool['started'] and not tool['stopped']:
|
|
block_delta = {
|
|
'type': 'content_block_delta',
|
|
'index': tool['block_index'],
|
|
'delta': {
|
|
'type': 'input_json_delta',
|
|
'partial_json': arguments_chunk,
|
|
},
|
|
}
|
|
yield f'event: content_block_delta\ndata: {JSONCodec.dumps(block_delta)}\n\n'.encode()
|
|
|
|
# Close the block once arguments form complete JSON
|
|
if (
|
|
tool['started']
|
|
and not tool['stopped']
|
|
and (tool['arguments'].rstrip()[-1:] == '}' or tool['arguments'].lstrip()[:1] != '{')
|
|
):
|
|
try:
|
|
JSONCodec.loads(tool['arguments'])
|
|
tool['stopped'] = True
|
|
block_stop = {
|
|
'type': 'content_block_stop',
|
|
'index': tool['block_index'],
|
|
}
|
|
yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode()
|
|
except (JSONCodec.JSONDecodeError, ValueError):
|
|
pass
|
|
|
|
# --- Handle finish reason ---
|
|
if finish_reason is not None:
|
|
stop_reason_map = {
|
|
'stop': 'end_turn',
|
|
'length': 'max_tokens',
|
|
'tool_calls': 'tool_use',
|
|
}
|
|
stop_reason = stop_reason_map.get(finish_reason, 'end_turn')
|
|
|
|
except Exception as e:
|
|
log.error(f'Error in Anthropic stream conversion: {e}')
|
|
|
|
# Close any open thinking block
|
|
if thinking_block_open:
|
|
block_stop = {'type': 'content_block_stop', 'index': current_block_index}
|
|
yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode()
|
|
current_block_index += 1
|
|
|
|
# Flush any tools that buffered arguments but never emitted a block
|
|
for tool in tracked_tool_calls.values():
|
|
if not tool['started'] and tool['name']:
|
|
tool['block_index'] = current_block_index
|
|
tool['started'] = True
|
|
|
|
block_start = {
|
|
'type': 'content_block_start',
|
|
'index': current_block_index,
|
|
'content_block': {
|
|
'type': 'tool_use',
|
|
'id': tool['id'],
|
|
'name': tool['name'],
|
|
'input': {},
|
|
},
|
|
}
|
|
yield f'event: content_block_start\ndata: {JSONCodec.dumps(block_start)}\n\n'.encode()
|
|
current_block_index += 1
|
|
|
|
if tool['arguments']:
|
|
block_delta = {
|
|
'type': 'content_block_delta',
|
|
'index': tool['block_index'],
|
|
'delta': {
|
|
'type': 'input_json_delta',
|
|
'partial_json': tool['arguments'],
|
|
},
|
|
}
|
|
yield f'event: content_block_delta\ndata: {JSONCodec.dumps(block_delta)}\n\n'.encode()
|
|
|
|
# Close any open text block
|
|
if text_block_open:
|
|
block_stop = {'type': 'content_block_stop', 'index': current_block_index}
|
|
yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode()
|
|
|
|
# Close any tool call blocks that are still open
|
|
for tool in tracked_tool_calls.values():
|
|
if tool['started'] and not tool['stopped']:
|
|
block_stop = {'type': 'content_block_stop', 'index': tool['block_index']}
|
|
yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode()
|
|
|
|
# Emit message_delta with stop reason
|
|
usage = {'output_tokens': output_tokens}
|
|
if input_tokens is not None:
|
|
usage['input_tokens'] = input_tokens
|
|
if cache_creation_input_tokens is not None:
|
|
usage['cache_creation_input_tokens'] = cache_creation_input_tokens
|
|
if cache_read_input_tokens is not None:
|
|
usage['cache_read_input_tokens'] = cache_read_input_tokens
|
|
if output_tokens_details is not None:
|
|
usage['output_tokens_details'] = output_tokens_details
|
|
if server_tool_use is not None:
|
|
usage['server_tool_use'] = server_tool_use
|
|
if service_tier is not None:
|
|
usage['service_tier'] = service_tier
|
|
|
|
message_delta = {
|
|
'type': 'message_delta',
|
|
'delta': {
|
|
'stop_reason': stop_reason,
|
|
'stop_sequence': None,
|
|
},
|
|
'usage': usage,
|
|
}
|
|
yield f'event: message_delta\ndata: {JSONCodec.dumps(message_delta)}\n\n'.encode()
|
|
|
|
# Emit message_stop
|
|
yield f'event: message_stop\ndata: {JSONCodec.dumps({"type": "message_stop"})}\n\n'.encode()
|