diff --git a/backend/open_webui/env.py b/backend/open_webui/env.py index aab88cdfb3..e891f1d39f 100644 --- a/backend/open_webui/env.py +++ b/backend/open_webui/env.py @@ -630,6 +630,13 @@ else: 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', '') if CHAT_STREAM_RESPONSE_CHUNK_MAX_BUFFER_SIZE == '': diff --git a/backend/open_webui/routers/openai.py b/backend/open_webui/routers/openai.py index be452b55d9..3aa121913e 100644 --- a/backend/open_webui/routers/openai.py +++ b/backend/open_webui/routers/openai.py @@ -779,6 +779,31 @@ def convert_to_azure_payload(url, payload: dict, api_version: str): 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: """ 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) stored_output = msg.get('output') 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 if role == 'system': @@ -861,6 +886,12 @@ def convert_to_responses_payload(payload: dict) -> dict: 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: responses_payload['instructions'] = system_content diff --git a/backend/open_webui/utils/middleware.py b/backend/open_webui/utils/middleware.py index ce56f82210..d721781ce4 100644 --- a/backend/open_webui/utils/middleware.py +++ b/backend/open_webui/utils/middleware.py @@ -136,6 +136,7 @@ from open_webui.env import ( ENABLE_FORWARD_USER_INFO_HEADERS, FORWARD_SESSION_INFO_HEADER_CHAT_ID, FORWARD_SESSION_INFO_HEADER_MESSAGE_ID, + ENABLE_RESPONSES_API_STATEFUL, ) from open_webui.utils.headers import include_user_info_headers 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': 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': # State Machine Event: In Progress @@ -3393,6 +3398,7 @@ async def streaming_chat_response_handler(response, ctx): usage = None prior_output = [] + last_response_id = None def full_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 output nonlocal prior_output + nonlocal last_response_id response_tool_calls = [] @@ -3519,6 +3526,10 @@ async def streaming_chat_response_handler(response, ctx): # calls. The outer middleware manages the # actual completion signal. 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.pop('done', None) @@ -3927,10 +3938,12 @@ async def streaming_chat_response_handler(response, ctx): # Responses API path: extract function_call items from 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 = { item.get('call_id') - for item in output + for item in (prior_output + output) if item.get('type') == 'function_call_output' } responses_api_tool_calls = [] @@ -4249,11 +4262,20 @@ async def streaming_chat_response_handler(response, ctx): **form_data, 'model': model_id, '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'], *convert_output_to_messages(output, raw=True), - ], - } + ] res = await generate_chat_completion( request, @@ -4270,6 +4292,20 @@ async def streaming_chat_response_handler(response, ctx): # ensures the UI shows tool history during # streaming. 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 = [] await stream_body_handler(res, new_form_data) output[:0] = prior_output