mirror of
https://github.com/open-webui/open-webui.git
synced 2026-09-07 08:27:05 +00:00
061f5e3a6d
* perf: stop re-parsing the whole tool-argument buffer on every streamed chunk Converting an OpenAI stream to Anthropic events buffers each tool call's arguments and, to find out when the JSON is complete, parsed the entire buffer again on every chunk. A tool call with large arguments pays that parse thousands of times, and the cost grows with the square of the argument size. The parse now runs only when the buffer could actually be complete. A JSON object can only close on its final brace, so a chunk that does not end there cannot complete it. Arguments that are not an object, or that start with whitespace, keep parsing on every chunk exactly as before. Measured on CPython 3.12 with 130 KB of tool arguments over 7648 chunks: | | before | after | |---|---|---| | parses | 7648 | 1 | | time | 382 ms | 0.82 ms | The block closes on exactly the same chunk as before, verified by replaying randomized fragmentations of objects with braces inside strings, escaped characters, unicode escapes, arrays, bare scalars, leading and trailing whitespace and a buffer that never completes, against both JSON backends. * refactor: read tool['arguments'] directly in the JSON completion guard Restores the pre-existing comment above the guard to its original wording and drops the `buffered` local, so the guard and the parse call both read `tool['arguments']`, the name the rest of the file already uses for that buffer. Behaviour is unchanged: same three conditions in the same order, same short-circuit result. * perf: strip whitespace in the tool-argument completion guard The character guard only looked at the first and last byte of the buffer, so a chunk that ended in a space still triggered a full parse and a tool argument with leading whitespace fell back to parsing on every chunk. Stripping first collapses both cases to a single parse at the end of the stream. Measured on a streamed tool call, parses and wall time for the whole stream, orjson on the left of the slash and stdlib json on the right: | argument shape | before | after | |---|---|---| | 20 KB string, char-by-char deltas | 3678 parses, 56 / 28 ms | 1 parse, 3.1 / 3.1 ms | | 200 KB, 20-char deltas | 1473 parses, 176 / 63 ms | 1 parse, 3.1 / 2.8 ms | | 8 KB prose, leading whitespace | 715 parses, 5.5 / 2.5 ms | 1 parse, 0.18 ms | | 8 KB of spaces inside a value | 713 parses, 6.0 / 2.9 ms | 1 parse, 0.83 / 0.72 ms | The strip costs about 20 ns per delta on arguments that have no whitespace at either end, which is where the old form was already optimal: a 20 KB compact argument goes from 191 to 216 us over 1786 deltas. Soundness is unchanged, the guard can still only skip a parse that would have failed: 2660892 buffers (exhaustive to length 6 over a JSON-lexical alphabet, every prefix of 26 named cases with a trailing byte appended, and every codepoint below U+3000 after a complete document) with zero cases where a parse would have succeeded.
981 lines
42 KiB
Python
981 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'):
|
|
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:
|
|
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()
|