refac
This commit is contained in:
@@ -630,6 +630,13 @@ else:
|
|||||||
CHAT_RESPONSE_MAX_TOOL_CALL_RETRIES = 30
|
CHAT_RESPONSE_MAX_TOOL_CALL_RETRIES = 30
|
||||||
|
|
||||||
|
|
||||||
|
# WARNING: Experimental. Only enable if your upstream Responses API endpoint
|
||||||
|
# supports stateful sessions (i.e. server-side response storage with
|
||||||
|
# previous_response_id anchoring). Most proxies and third-party endpoints
|
||||||
|
# are stateless and will break if this is enabled.
|
||||||
|
ENABLE_RESPONSES_API_STATEFUL = os.environ.get('ENABLE_RESPONSES_API_STATEFUL', 'False').lower() == 'true'
|
||||||
|
|
||||||
|
|
||||||
CHAT_STREAM_RESPONSE_CHUNK_MAX_BUFFER_SIZE = os.environ.get('CHAT_STREAM_RESPONSE_CHUNK_MAX_BUFFER_SIZE', '')
|
CHAT_STREAM_RESPONSE_CHUNK_MAX_BUFFER_SIZE = os.environ.get('CHAT_STREAM_RESPONSE_CHUNK_MAX_BUFFER_SIZE', '')
|
||||||
|
|
||||||
if CHAT_STREAM_RESPONSE_CHUNK_MAX_BUFFER_SIZE == '':
|
if CHAT_STREAM_RESPONSE_CHUNK_MAX_BUFFER_SIZE == '':
|
||||||
|
|||||||
@@ -779,6 +779,31 @@ def convert_to_azure_payload(url, payload: dict, api_version: str):
|
|||||||
return url, payload
|
return url, payload
|
||||||
|
|
||||||
|
|
||||||
|
# Fields accepted by the Responses API for each input item type.
|
||||||
|
RESPONSES_ALLOWED_FIELDS: dict[str, set[str]] = {
|
||||||
|
'message': {'type', 'role', 'content'},
|
||||||
|
'function_call': {'type', 'call_id', 'name', 'arguments', 'id'},
|
||||||
|
'function_call_output': {'type', 'call_id', 'output'},
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _normalize_stored_item(item: dict) -> dict:
|
||||||
|
"""Strip local-only fields from a stored output item before replaying it.
|
||||||
|
|
||||||
|
Open WebUI stores extra bookkeeping fields (``id``, ``status``,
|
||||||
|
``started_at``, ``ended_at``, ``duration``, ``_tag_type``,
|
||||||
|
``attributes``, ``summary``, etc.) that the Responses API does
|
||||||
|
not accept. This helper returns a copy containing only the
|
||||||
|
fields the API understands.
|
||||||
|
"""
|
||||||
|
item_type = item.get('type', '')
|
||||||
|
allowed = RESPONSES_ALLOWED_FIELDS.get(item_type)
|
||||||
|
if allowed is None:
|
||||||
|
# Unknown type — pass through as-is (e.g. reasoning, extension items).
|
||||||
|
return item
|
||||||
|
return {k: v for k, v in item.items() if k in allowed}
|
||||||
|
|
||||||
|
|
||||||
def convert_to_responses_payload(payload: dict) -> dict:
|
def convert_to_responses_payload(payload: dict) -> dict:
|
||||||
"""
|
"""
|
||||||
Convert Chat Completions payload to Responses API format.
|
Convert Chat Completions payload to Responses API format.
|
||||||
@@ -798,7 +823,7 @@ def convert_to_responses_payload(payload: dict) -> dict:
|
|||||||
# Check for stored output items (from previous Responses API turn)
|
# Check for stored output items (from previous Responses API turn)
|
||||||
stored_output = msg.get('output')
|
stored_output = msg.get('output')
|
||||||
if stored_output and isinstance(stored_output, list):
|
if stored_output and isinstance(stored_output, list):
|
||||||
input_items.extend(stored_output)
|
input_items.extend(_normalize_stored_item(item) for item in stored_output)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
if role == 'system':
|
if role == 'system':
|
||||||
@@ -861,6 +886,12 @@ def convert_to_responses_payload(payload: dict) -> dict:
|
|||||||
|
|
||||||
responses_payload = {**payload, 'input': input_items}
|
responses_payload = {**payload, 'input': input_items}
|
||||||
|
|
||||||
|
# Forward previous_response_id when the middleware has set it
|
||||||
|
# (only used when ENABLE_RESPONSES_API_STATEFUL is enabled).
|
||||||
|
previous_response_id = responses_payload.pop('previous_response_id', None)
|
||||||
|
if previous_response_id:
|
||||||
|
responses_payload['previous_response_id'] = previous_response_id
|
||||||
|
|
||||||
if system_content:
|
if system_content:
|
||||||
responses_payload['instructions'] = system_content
|
responses_payload['instructions'] = system_content
|
||||||
|
|
||||||
|
|||||||
@@ -136,6 +136,7 @@ from open_webui.env import (
|
|||||||
ENABLE_FORWARD_USER_INFO_HEADERS,
|
ENABLE_FORWARD_USER_INFO_HEADERS,
|
||||||
FORWARD_SESSION_INFO_HEADER_CHAT_ID,
|
FORWARD_SESSION_INFO_HEADER_CHAT_ID,
|
||||||
FORWARD_SESSION_INFO_HEADER_MESSAGE_ID,
|
FORWARD_SESSION_INFO_HEADER_MESSAGE_ID,
|
||||||
|
ENABLE_RESPONSES_API_STATEFUL,
|
||||||
)
|
)
|
||||||
from open_webui.utils.headers import include_user_info_headers
|
from open_webui.utils.headers import include_user_info_headers
|
||||||
from open_webui.constants import TASKS
|
from open_webui.constants import TASKS
|
||||||
@@ -850,7 +851,11 @@ def handle_responses_streaming_event(
|
|||||||
if item.get('type') == 'reasoning' and item.get('status') != 'completed':
|
if item.get('type') == 'reasoning' and item.get('status') != 'completed':
|
||||||
item['status'] = 'completed'
|
item['status'] = 'completed'
|
||||||
|
|
||||||
return new_output, {'usage': response_data.get('usage'), 'done': True}
|
return new_output, {
|
||||||
|
'usage': response_data.get('usage'),
|
||||||
|
'done': True,
|
||||||
|
'response_id': response_data.get('id'),
|
||||||
|
}
|
||||||
|
|
||||||
elif event_type == 'response.in_progress':
|
elif event_type == 'response.in_progress':
|
||||||
# State Machine Event: In Progress
|
# State Machine Event: In Progress
|
||||||
@@ -3393,6 +3398,7 @@ async def streaming_chat_response_handler(response, ctx):
|
|||||||
|
|
||||||
usage = None
|
usage = None
|
||||||
prior_output = []
|
prior_output = []
|
||||||
|
last_response_id = None
|
||||||
|
|
||||||
def full_output():
|
def full_output():
|
||||||
return prior_output + output if prior_output else output
|
return prior_output + output if prior_output else output
|
||||||
@@ -3431,6 +3437,7 @@ async def streaming_chat_response_handler(response, ctx):
|
|||||||
nonlocal usage
|
nonlocal usage
|
||||||
nonlocal output
|
nonlocal output
|
||||||
nonlocal prior_output
|
nonlocal prior_output
|
||||||
|
nonlocal last_response_id
|
||||||
|
|
||||||
response_tool_calls = []
|
response_tool_calls = []
|
||||||
|
|
||||||
@@ -3519,6 +3526,10 @@ async def streaming_chat_response_handler(response, ctx):
|
|||||||
# calls. The outer middleware manages the
|
# calls. The outer middleware manages the
|
||||||
# actual completion signal.
|
# actual completion signal.
|
||||||
if response_metadata:
|
if response_metadata:
|
||||||
|
if ENABLE_RESPONSES_API_STATEFUL:
|
||||||
|
response_id = response_metadata.pop('response_id', None)
|
||||||
|
if response_id:
|
||||||
|
last_response_id = response_id
|
||||||
processed_data.update(response_metadata)
|
processed_data.update(response_metadata)
|
||||||
processed_data.pop('done', None)
|
processed_data.pop('done', None)
|
||||||
|
|
||||||
@@ -3927,10 +3938,12 @@ async def streaming_chat_response_handler(response, ctx):
|
|||||||
|
|
||||||
# Responses API path: extract function_call items from output
|
# Responses API path: extract function_call items from output
|
||||||
if not response_tool_calls and output:
|
if not response_tool_calls and output:
|
||||||
# Collect call_ids that already have results
|
# Collect call_ids that already have results,
|
||||||
|
# including those from prior_output so we don't
|
||||||
|
# re-process tool calls from a previous turn.
|
||||||
handled_call_ids = {
|
handled_call_ids = {
|
||||||
item.get('call_id')
|
item.get('call_id')
|
||||||
for item in output
|
for item in (prior_output + output)
|
||||||
if item.get('type') == 'function_call_output'
|
if item.get('type') == 'function_call_output'
|
||||||
}
|
}
|
||||||
responses_api_tool_calls = []
|
responses_api_tool_calls = []
|
||||||
@@ -4249,11 +4262,20 @@ async def streaming_chat_response_handler(response, ctx):
|
|||||||
**form_data,
|
**form_data,
|
||||||
'model': model_id,
|
'model': model_id,
|
||||||
'stream': True,
|
'stream': True,
|
||||||
'messages': [
|
}
|
||||||
|
|
||||||
|
if ENABLE_RESPONSES_API_STATEFUL and last_response_id:
|
||||||
|
system_message = get_system_message(form_data['messages'])
|
||||||
|
new_form_data['messages'] = (
|
||||||
|
([system_message] if system_message else [])
|
||||||
|
+ convert_output_to_messages(output, raw=True)
|
||||||
|
)
|
||||||
|
new_form_data['previous_response_id'] = last_response_id
|
||||||
|
else:
|
||||||
|
new_form_data['messages'] = [
|
||||||
*form_data['messages'],
|
*form_data['messages'],
|
||||||
*convert_output_to_messages(output, raw=True),
|
*convert_output_to_messages(output, raw=True),
|
||||||
],
|
]
|
||||||
}
|
|
||||||
|
|
||||||
res = await generate_chat_completion(
|
res = await generate_chat_completion(
|
||||||
request,
|
request,
|
||||||
@@ -4270,6 +4292,20 @@ async def streaming_chat_response_handler(response, ctx):
|
|||||||
# ensures the UI shows tool history during
|
# ensures the UI shows tool history during
|
||||||
# streaming.
|
# streaming.
|
||||||
prior_output = list(output)
|
prior_output = list(output)
|
||||||
|
# Trim the trailing empty placeholder message
|
||||||
|
# so it doesn't persist as a ghost item once
|
||||||
|
# the new stream produces real content.
|
||||||
|
if (
|
||||||
|
prior_output
|
||||||
|
and prior_output[-1].get('type') == 'message'
|
||||||
|
and prior_output[-1].get('status') == 'in_progress'
|
||||||
|
):
|
||||||
|
msg_parts = prior_output[-1].get('content', [])
|
||||||
|
if (
|
||||||
|
not msg_parts
|
||||||
|
or (len(msg_parts) == 1 and not msg_parts[0].get('text', '').strip())
|
||||||
|
):
|
||||||
|
prior_output.pop()
|
||||||
output = []
|
output = []
|
||||||
await stream_body_handler(res, new_form_data)
|
await stream_body_handler(res, new_form_data)
|
||||||
output[:0] = prior_output
|
output[:0] = prior_output
|
||||||
|
|||||||
Reference in New Issue
Block a user