mirror of
https://github.com/open-webui/open-webui.git
synced 2026-10-11 03:38:02 +00:00
0fe9ed0d3c
When the provider failed partway through a streaming request to the Anthropic Messages endpoint, the stream still ended like a normally finished answer, so Claude Code and the Anthropic SDKs took the cut-off text as complete. The stream now ends with an Anthropic error event, with the provider's error message if it sent one, so clients raise an error. Successful streams are unchanged. Fixes #31403
1001 lines
43 KiB
Python
1001 lines
43 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 (backends reject an empty tools array)
|
|
if anthropic_payload.get('tools'):
|
|
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 and 'tools' in openai_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'
|
|
error_message = None
|
|
|
|
# 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
|
|
|
|
error = data.get('error')
|
|
if error:
|
|
error_message = (
|
|
error.get('message') if isinstance(error, dict) else error
|
|
) or 'Chat completion stream failed'
|
|
break
|
|
|
|
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')
|
|
|
|
if error_message:
|
|
break
|
|
|
|
except Exception as e:
|
|
log.error(f'Error in Anthropic stream conversion: {e}')
|
|
error_message = 'Chat completion stream failed'
|
|
|
|
# Skip message_stop so a failed stream is not reported as complete.
|
|
if error_message:
|
|
error_event = {'type': 'error', 'error': {'type': 'api_error', 'message': error_message}}
|
|
yield f'event: error\ndata: {JSONCodec.dumps(error_event)}\n\n'.encode()
|
|
return
|
|
|
|
# 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()
|