diff --git a/src/ogx/providers/inline/responses/builtin/responses/streaming.py b/src/ogx/providers/inline/responses/builtin/responses/streaming.py index 45ab032829b..e8772c73063 100644 --- a/src/ogx/providers/inline/responses/builtin/responses/streaming.py +++ b/src/ogx/providers/inline/responses/builtin/responses/streaming.py @@ -128,6 +128,8 @@ # Anything else is either a registered function tool (client-side) or a hallucinated name. _SERVER_SIDE_BUILTIN_TOOL_NAMES = frozenset({"web_search", "knowledge_search", "file_search"}) +_MAX_HALLUCINATED_TOOL_RETRIES = 3 + _GUARDRAIL_BATCH_CHARS = 200 # Maps OpenAI Chat Completions error codes to Responses API error codes @@ -483,6 +485,7 @@ async def create_response(self) -> AsyncIterator[OpenAIResponseObjectStream]: chat_tool_choice = processed_tool_choice.model_dump() n_iter = 0 + n_hallucinated_retries = 0 messages = self.ctx.messages.copy() final_status = "completed" incomplete_reason: str | None = None @@ -617,6 +620,7 @@ async def create_response(self) -> AsyncIterator[OpenAIResponseObjectStream]: non_function_tool_calls, approvals, next_turn_messages, + has_hallucinated_retries, ) = self._separate_tool_calls(current_response, messages, completion_result_data.reasoning_content) # add any approval requests required for tool_call in approvals: @@ -697,9 +701,22 @@ async def create_response(self) -> AsyncIterator[OpenAIResponseObjectStream]: ): yield stream_event messages = next_turn_messages - if not function_tool_calls and not non_function_tool_calls: + if not function_tool_calls and not non_function_tool_calls and not has_hallucinated_retries: break + if has_hallucinated_retries: + n_hallucinated_retries += 1 + if n_hallucinated_retries >= _MAX_HALLUCINATED_TOOL_RETRIES: + logger.warning( + "Exiting inference loop; model keeps hallucinating tool names", + retries=n_hallucinated_retries, + ) + final_status = "incomplete" + incomplete_reason = "max_iterations_exceeded" + break + else: + n_hallucinated_retries = 0 + if function_tool_calls: logger.info("Exiting inference loop since there is a function (client-side) tool call") break @@ -775,12 +792,19 @@ async def create_response(self) -> AsyncIterator[OpenAIResponseObjectStream]: def _separate_tool_calls( self, current_response, messages, reasoning_content: str | None = None - ) -> tuple[list, list, list, list]: - """Separate tool calls into function and non-function categories.""" + ) -> tuple[list, list, list, list, bool]: + """Separate tool calls into function and non-function categories. + + Returns (function_tool_calls, non_function_tool_calls, approvals, + next_turn_messages, has_hallucinated_retries). The last flag is True + when the model hallucinated a tool name in a server-only loop and an + error was fed back — the caller should re-enter the inference loop. + """ function_tool_calls = [] non_function_tool_calls = [] approvals = [] next_turn_messages = messages.copy() + has_hallucinated_retries = False for choice in current_response.choices: # Convert response message to input message format for multi-turn. @@ -814,16 +838,40 @@ def _separate_tool_calls( and tool_call.function.name not in _SERVER_SIDE_BUILTIN_TOOL_NAMES and tool_call.function.name not in self.mcp_tool_to_server ): - # The model called a tool name that is neither a registered function tool, - # nor a server-side built-in, nor an MCP tool — it hallucinated a name. - # Return it to the client as a function_call output item rather than - # crashing the server with an unhandled ValueError. - logger.warning( - "Model called unrecognized tool ; treating as a client-side function call.", - name=tool_call.function.name, - ) - function_tool_calls.append(tool_call) - executed_tool_calls.append(tool_call) + # The model hallucinated a tool name — it doesn't match + # any registered function tool, server-side built-in, or + # MCP tool. + has_client_tools = any(t.type == "function" for t in self.ctx.response_tools) + if has_client_tools: + # A client is expected to handle function calls, so + # surface the hallucinated name as a client-side + # function call to avoid a server 500. + logger.warning( + "Model called unrecognized tool; treating as a client-side function call", + name=tool_call.function.name, + ) + function_tool_calls.append(tool_call) + executed_tool_calls.append(tool_call) + else: + # Server-only loop — no client will ever supply a + # result for this call. Feed an error back to the + # model so it can self-correct on the next iteration. + logger.warning( + "Model called unrecognized tool; returning error to model", + name=tool_call.function.name, + ) + available = sorted(self.mcp_tool_to_server.keys()) + next_turn_messages.append( + OpenAIToolMessageParam( + tool_call_id=tool_call.id, + content=( + f"Error: tool '{tool_call.function.name}' is not available. " + f"Available tools are: {', '.join(available)}. " + "Please use one of these tools instead." + ), + ) + ) + has_hallucinated_retries = True else: if self._approval_required(tool_call.function.name): approval_response = self.ctx.approval_response( @@ -855,7 +903,7 @@ def _separate_tool_calls( else: next_turn_messages.pop() - return function_tool_calls, non_function_tool_calls, approvals, next_turn_messages + return function_tool_calls, non_function_tool_calls, approvals, next_turn_messages, has_hallucinated_retries def _accumulate_chunk_usage(self, chunk: OpenAIChatCompletionChunk) -> None: """Accumulate usage from a streaming chunk into the response usage format.""" diff --git a/tests/unit/providers/inline/responses/builtin/responses/test_streaming.py b/tests/unit/providers/inline/responses/builtin/responses/test_streaming.py index c757da6d68a..fcc0b34b83a 100644 --- a/tests/unit/providers/inline/responses/builtin/responses/test_streaming.py +++ b/tests/unit/providers/inline/responses/builtin/responses/test_streaming.py @@ -165,7 +165,7 @@ def test_single_approval_pops_assistant_message(self): response = _make_response(tool_calls) messages = ["system_msg", "user_msg"] - _, _, approvals, result_messages = orch._separate_tool_calls(response, messages) + _, _, approvals, result_messages, _ = orch._separate_tool_calls(response, messages) assert len(approvals) == 1 assert len(result_messages) == 2 @@ -184,7 +184,7 @@ def test_multiple_approvals_pops_once_not_per_tool_call(self): response = _make_response(tool_calls) messages = ["system_msg", "user_msg"] - _, _, approvals, result_messages = orch._separate_tool_calls(response, messages) + _, _, approvals, result_messages, _ = orch._separate_tool_calls(response, messages) assert len(approvals) == 3 assert len(result_messages) == 2, ( @@ -205,7 +205,7 @@ def test_two_approvals_does_not_eat_user_message(self): response = _make_response(tool_calls) messages = ["system_msg", "user_msg"] - _, _, approvals, result_messages = orch._separate_tool_calls(response, messages) + _, _, approvals, result_messages, _ = orch._separate_tool_calls(response, messages) assert len(approvals) == 2 assert "user_msg" in result_messages @@ -227,7 +227,7 @@ def test_all_denied_pops_assistant_message(self): response = _make_response(tool_calls) messages = ["system_msg", "user_msg"] - _, _, approvals, result_messages = orch._separate_tool_calls(response, messages) + _, _, approvals, result_messages, _ = orch._separate_tool_calls(response, messages) assert len(approvals) == 0 assert len(result_messages) == 2 @@ -258,7 +258,7 @@ def side_effect(name, args): response = _make_response([tc_weather, tc_time]) messages = ["system_msg", "user_msg"] - _, non_function, approvals, result_messages = orch._separate_tool_calls(response, messages) + _, non_function, approvals, result_messages, _ = orch._separate_tool_calls(response, messages) assert len(non_function) == 1 assert non_function[0].id == "call_1" @@ -286,7 +286,7 @@ def test_mix_with_two_executed_one_deferred(self): response = _make_response([tc_weather, tc_time, tc_news]) messages = ["system_msg", "user_msg"] - _, non_function, approvals, result_messages = orch._separate_tool_calls(response, messages) + _, non_function, approvals, result_messages, _ = orch._separate_tool_calls(response, messages) assert len(non_function) == 2 assert len(approvals) == 1 @@ -320,7 +320,7 @@ def side_effect(name, args): response = _make_response([tc_weather, tc_time]) messages = ["system_msg", "user_msg"] - _, non_function, approvals, result_messages = orch._separate_tool_calls(response, messages) + _, non_function, approvals, result_messages, _ = orch._separate_tool_calls(response, messages) assert len(non_function) == 1 assert len(approvals) == 0 @@ -353,7 +353,7 @@ def side_effect(name, args): response = _make_response(tool_calls) messages = ["system_msg", "user_msg"] - _, _, _, result_messages = orch._separate_tool_calls(response, messages) + _, _, _, result_messages, _ = orch._separate_tool_calls(response, messages) assert result_messages[0] == "system_msg" assert result_messages[1] == "user_msg" @@ -374,7 +374,7 @@ def test_no_approvals_needed_keeps_full_assistant_message(self): response = _make_response(tool_calls) messages = ["system_msg", "user_msg"] - _, non_function, approvals, result_messages = orch._separate_tool_calls(response, messages) + _, non_function, approvals, result_messages, _ = orch._separate_tool_calls(response, messages) assert len(non_function) == 2 assert len(approvals) == 0 @@ -400,7 +400,7 @@ def test_all_pre_approved_keeps_full_assistant_message(self): response = _make_response(tool_calls) messages = ["system_msg", "user_msg"] - _, non_function, approvals, result_messages = orch._separate_tool_calls(response, messages) + _, non_function, approvals, result_messages, _ = orch._separate_tool_calls(response, messages) assert len(non_function) == 2 assert len(approvals) == 0 diff --git a/tests/unit/providers/responses/builtin/test_openai_responses_params.py b/tests/unit/providers/responses/builtin/test_openai_responses_params.py index 04c9844edd6..629d81e1fc6 100644 --- a/tests/unit/providers/responses/builtin/test_openai_responses_params.py +++ b/tests/unit/providers/responses/builtin/test_openai_responses_params.py @@ -34,6 +34,7 @@ ) from ogx_api.openai_responses import ( OpenAIResponseInputToolFunction, + OpenAIResponseInputToolMCP, OpenAIResponseMessage, OpenAIResponseText, OpenAIResponseTextFormat, @@ -724,6 +725,160 @@ async def fake_stream_hallucinated_tool(): assert result.output[0].name == "lookup_capital_city" +async def test_hallucinated_tool_call_retries_when_no_client_tools(openai_responses_impl, mock_inference_api): + """When the model hallucinates a tool name and no client-side function tools + are configured, the server should feed an error back to the model and let it + retry — not silently exit the inference loop. + """ + model = "meta-llama/Llama-3.1-8B-Instruct" + + async def hallucinated_stream(): + yield ChatCompletionChunk( + id="hall-1", + choices=[ + Choice( + index=0, + delta=ChoiceDelta( + tool_calls=[ + ChoiceDeltaToolCall( + index=0, + id="tc_hall_1", + function=ChoiceDeltaToolCallFunction( + name="services_list", + arguments='{"namespace": "demo-app"}', + ), + type="function", + ) + ] + ), + ), + ], + created=1, + model=model, + object="chat.completion.chunk", + ) + + async def corrected_stream(): + yield ChatCompletionChunk( + id="corrected-1", + choices=[ + Choice( + index=0, + delta=ChoiceDelta(content="I don't have a services_list tool. Let me use pods_list instead."), + finish_reason="stop", + ), + ], + created=1, + model=model, + object="chat.completion.chunk", + ) + + mock_inference_api.openai_chat_completion.side_effect = [ + hallucinated_stream(), + corrected_stream(), + ] + + mcp_tool = OpenAIResponseInputToolMCP(server_label="k8s", server_url="http://k8s-mcp") + + # Patch _process_tools to populate mcp_tool_to_server without connecting + # to a real MCP server. + from ogx.providers.inline.responses.builtin.responses.streaming import StreamingResponseOrchestrator + + async def patched_process_tools(self, output_messages): + self.mcp_tool_to_server = { + "pods_list_in_namespace": mcp_tool, + "pods_get": mcp_tool, + } + return + yield # make this an async generator + + with patch.object(StreamingResponseOrchestrator, "_process_tools", patched_process_tools): + result = await openai_responses_impl.create_openai_response( + CreateResponseRequest( + input="List services in demo-app", + model=model, + tools=[mcp_tool], + ) + ) + + assert result is not None + assert result.status == "completed" + # The model was called twice: once for the hallucinated call, once after + # the error was fed back. + assert mock_inference_api.openai_chat_completion.call_count == 2 + # The retry message should contain the error about the unavailable tool. + second_call_messages = mock_inference_api.openai_chat_completion.call_args_list[1].args[0].messages + tool_error_messages = [m for m in second_call_messages if getattr(m, "role", None) == "tool"] + assert len(tool_error_messages) == 1 + assert "services_list" in tool_error_messages[0].content + assert "pods_list_in_namespace" in tool_error_messages[0].content + + +async def test_hallucinated_tool_call_retries_exhausted(openai_responses_impl, mock_inference_api): + """After _MAX_HALLUCINATED_TOOL_RETRIES consecutive hallucinations the loop + should stop with status 'incomplete' instead of looping forever. + """ + from ogx.providers.inline.responses.builtin.responses.streaming import ( + _MAX_HALLUCINATED_TOOL_RETRIES, + StreamingResponseOrchestrator, + ) + + model = "meta-llama/Llama-3.1-8B-Instruct" + + def make_hallucinated_stream(call_id, tool_name): + async def stream(): + yield ChatCompletionChunk( + id=call_id, + choices=[ + Choice( + index=0, + delta=ChoiceDelta( + tool_calls=[ + ChoiceDeltaToolCall( + index=0, + id=call_id, + function=ChoiceDeltaToolCallFunction( + name=tool_name, + arguments="{}", + ), + type="function", + ) + ] + ), + ), + ], + created=1, + model=model, + object="chat.completion.chunk", + ) + + return stream() + + mock_inference_api.openai_chat_completion.side_effect = [ + make_hallucinated_stream(f"tc_h{i}", f"fake_tool_{i}") for i in range(_MAX_HALLUCINATED_TOOL_RETRIES) + ] + + mcp_tool = OpenAIResponseInputToolMCP(server_label="k8s", server_url="http://k8s-mcp") + + async def patched_process_tools(self, output_messages): + self.mcp_tool_to_server = {"pods_list_in_namespace": mcp_tool} + return + yield + + with patch.object(StreamingResponseOrchestrator, "_process_tools", patched_process_tools): + result = await openai_responses_impl.create_openai_response( + CreateResponseRequest( + input="List services in demo-app", + model=model, + tools=[mcp_tool], + ) + ) + + assert result is not None + assert result.status == "incomplete" + assert mock_inference_api.openai_chat_completion.call_count == _MAX_HALLUCINATED_TOOL_RETRIES + + async def test_create_openai_response_with_stream_options_merges_with_default( openai_responses_impl, mock_inference_api ):