From 738949031b31a9e42388d528b5503accc6c14de9 Mon Sep 17 00:00:00 2001 From: Alessandro <155005371+3clyp50@users.noreply.github.com> Date: Wed, 1 Jul 2026 15:53:20 +0200 Subject: [PATCH] Stabilize LiteLLM provider fallback Fall back to Chat Completions for provider/proxy Responses endpoint failures before any output is emitted. Preserve streamed Chat Completions tool-call deltas as structured LLMResult function-call items, so fallback providers can still drive tools. Document Docker host-gateway addressing for local model servers, where container localhost does not reach host loopback. Verified with: PYTHONPATH="/home/eclypso/a0/agent-zero" conda run -n a0 pytest tests/test_stream_tool_early_stop.py tests/test_responses_architecture.py -q; PYTHONPATH="/home/eclypso/a0/agent-zero" conda run -n a0 python -m py_compile helpers/litellm_transport.py helpers/llm_result.py; git diff --check. --- docs/setup/installation.md | 6 + helpers/litellm_transport.py | 207 +++++++++++++++++++++- helpers/litellm_transport.py.dox.md | 2 + helpers/llm_result.py | 16 +- helpers/llm_result.py.dox.md | 2 +- tests/test_stream_tool_early_stop.py | 251 ++++++++++++++++++++++++++- 6 files changed, 472 insertions(+), 12 deletions(-) diff --git a/docs/setup/installation.md b/docs/setup/installation.md index fc3713c28..8306508aa 100644 --- a/docs/setup/installation.md +++ b/docs/setup/installation.md @@ -546,6 +546,12 @@ Use the naming format required by your selected provider: > [!TIP] > If you see "Invalid model ID," verify the provider and naming format on the provider website, or search the web for " model naming". +#### Local Model Server Addresses From Docker + +When Agent Zero runs in Docker, `localhost` and `127.0.0.1` inside an API base URL mean the Agent Zero container, not your host machine. For a model server running on the host, use `http://host.docker.internal:` when available, or the Docker host gateway address such as `http://172.17.0.1:` on the default Linux bridge. + +If the model server only listens on host loopback, for example `127.0.0.1:`, the container still cannot reach it through the gateway. Configure the local server to listen on a Docker-reachable address such as `0.0.0.0`, and keep that port limited to trusted clients. + #### Context Window & Memory Split - Set the **total context window** (e.g., 100k) first. diff --git a/helpers/litellm_transport.py b/helpers/litellm_transport.py index a66ed63c6..d456e367f 100644 --- a/helpers/litellm_transport.py +++ b/helpers/litellm_transport.py @@ -260,11 +260,17 @@ class LiteLLMTransport: try: if self.policy.mode is TransportMode.CHAT_COMPLETIONS: iterator = completion(**self._chat_request(stream=True)) + parser = ChatCompletionsStreamParser() for chunk in iterator: - parsed = ChatCompletionsTransport.parse(chunk) + parsed = parser.parse(chunk) if _has_chunk_delta(parsed): got_any_chunk = True yield parsed + parsed = parser.flush() + if _has_chunk_delta(parsed): + got_any_chunk = True + yield parsed + self.last_result = self._stream_result_from_chat_parser(parser) else: request = self._responses_request(stream=True) iterator = responses(**request) @@ -295,11 +301,17 @@ class LiteLLMTransport: try: if self.policy.mode is TransportMode.CHAT_COMPLETIONS: iterator = await acompletion(**self._chat_request(stream=True)) + parser = ChatCompletionsStreamParser() async for chunk in iterator: # type: ignore[union-attr] - parsed = ChatCompletionsTransport.parse(chunk) + parsed = parser.parse(chunk) if _has_chunk_delta(parsed): got_any_chunk = True yield parsed + parsed = parser.flush() + if _has_chunk_delta(parsed): + got_any_chunk = True + yield parsed + self.last_result = self._stream_result_from_chat_parser(parser) else: request = self._responses_request(stream=True) iterator = await aresponses(**request) @@ -393,6 +405,7 @@ class LiteLLMTransport: response=parsed["response_delta"], reasoning=parsed["reasoning_delta"], input_items=ResponsesTransport.input_from_messages(self.messages), + output_items=parsed.get("_output_items"), provider_model_key=self.model, capability=self._capability_metadata(), ) @@ -417,6 +430,20 @@ class LiteLLMTransport: return None return self._llm_result_from_response(parser.completed_response, request) + def _stream_result_from_chat_parser( + self, parser: "ChatCompletionsStreamParser" + ) -> LLMResult | None: + output_items = parser.output_items() + if not output_items: + return None + return LLMResult.from_chat( + response=parser.function_calls_text(), + input_items=ResponsesTransport.input_from_messages(self.messages), + output_items=output_items, + provider_model_key=self.model, + capability=self._capability_metadata(), + ) + def _capability_metadata(self) -> dict[str, Any]: return { "mode": self.policy.mode.value, @@ -489,7 +516,149 @@ class ChatCompletionsTransport: reasoning_delta = _get_value(delta, "reasoning_content") or _get_value( message, "reasoning_content" ) or "" - return {"reasoning_delta": reasoning_delta, "response_delta": response_delta} + parsed = {"reasoning_delta": reasoning_delta, "response_delta": response_delta} + if not response_delta: + tool_calls = _as_list(_get_value(message, "tool_calls")) + response_delta = ChatCompletionsTransport.tool_calls_text(tool_calls) + if response_delta: + parsed["response_delta"] = response_delta + parsed["_output_items"] = ChatCompletionsTransport.output_items( + tool_calls + ) + return parsed + + @classmethod + def tool_calls_text(cls, tool_calls: Any) -> str: + calls = [cls.tool_call_object(call) for call in _as_list(tool_calls)] + calls = [call for call in calls if call] + if not calls: + return "" + if len(calls) == 1: + return json.dumps(calls[0]) + return json.dumps( + {"tool_name": "parallel_tool_calls", "tool_args": {"calls": calls}} + ) + + @classmethod + def output_items(cls, tool_calls: Any) -> list[dict[str, Any]]: + items = [] + for index, tool_call in enumerate(_as_list(tool_calls)): + item = cls.function_call_item(tool_call, fallback_index=index) + if item: + items.append(item) + return items + + @classmethod + def function_call_item( + cls, tool_call: Any, *, fallback_index: int = 0 + ) -> dict[str, Any]: + function = _get_value(tool_call, "function") or {} + name = _get_value(function, "name") or _get_value(tool_call, "name") + if not name: + return {} + raw_arguments = _get_value(function, "arguments") + if raw_arguments is None: + raw_arguments = _get_value(tool_call, "arguments") or "{}" + call_id = str(_get_value(tool_call, "id") or f"call_{fallback_index}") + return { + "type": "function_call", + "id": call_id, + "call_id": call_id, + "name": str(name), + "arguments": raw_arguments + if isinstance(raw_arguments, str) + else json.dumps(raw_arguments), + } + + @classmethod + def tool_call_object(cls, tool_call: Any) -> dict[str, Any]: + item = cls.function_call_item(tool_call) + if not item: + return {} + return ResponsesTransport.function_call_object(item) + + +class ChatCompletionsStreamParser: + def __init__(self) -> None: + self.tool_calls: dict[str, dict[str, Any]] = {} + self.order: list[str] = [] + self.emitted = False + + def parse(self, chunk: Any) -> ChatChunk: + parsed = ChatCompletionsTransport.parse(chunk) + choice = _first_choice(chunk) + delta = _get_value(choice, "delta") or {} + self._append_tool_calls(_get_value(delta, "tool_calls")) + self._append_legacy_function_call(_get_value(delta, "function_call")) + + if _get_value(choice, "finish_reason") in {"tool_calls", "function_call"}: + text = self._emit() + if text and not parsed["response_delta"]: + parsed["response_delta"] = text + return parsed + + def flush(self) -> ChatChunk: + return {"reasoning_delta": "", "response_delta": self._emit()} + + def function_calls_text(self) -> str: + return ChatCompletionsTransport.tool_calls_text(self._ordered_tool_calls()) + + def output_items(self) -> list[dict[str, Any]]: + return ChatCompletionsTransport.output_items(self._ordered_tool_calls()) + + def _append_tool_calls(self, tool_calls: Any) -> None: + for fallback_index, tool_call in enumerate(_as_list(tool_calls)): + key = self._tool_call_key(tool_call, fallback_index) + current = self._current_tool_call(key) + if _get_value(tool_call, "id"): + current["id"] = _get_value(tool_call, "id") + if _get_value(tool_call, "type"): + current["type"] = _get_value(tool_call, "type") + self._append_function_delta(current, _get_value(tool_call, "function")) + + def _append_legacy_function_call(self, function_call: Any) -> None: + if not function_call: + return + current = self._current_tool_call("0") + current["type"] = "function" + self._append_function_delta(current, function_call) + + def _append_function_delta(self, tool_call: dict[str, Any], delta: Any) -> None: + if not delta: + return + function = tool_call.setdefault("function", {}) + if _get_value(delta, "name"): + function["name"] = _get_value(delta, "name") + if _get_value(delta, "arguments") is not None: + function["arguments"] = str(function.get("arguments") or "") + str( + _get_value(delta, "arguments") or "" + ) + + def _current_tool_call(self, key: str) -> dict[str, Any]: + if key not in self.tool_calls: + self.tool_calls[key] = {"type": "function", "function": {}} + self.order.append(key) + return self.tool_calls[key] + + def _ordered_tool_calls(self) -> list[dict[str, Any]]: + return [self.tool_calls[key] for key in self.order] + + def _emit(self) -> str: + if self.emitted: + return "" + text = self.function_calls_text() + if text: + self.emitted = True + return text + + @staticmethod + def _tool_call_key(tool_call: Any, fallback_index: int) -> str: + index = _get_value(tool_call, "index") + if index is not None: + return str(index) + if _get_value(tool_call, "id"): + return str(_get_value(tool_call, "id")) + return str(fallback_index) class ResponsesTransport: @@ -1548,6 +1717,8 @@ def _is_responses_not_supported_error(exc: Exception) -> bool: return False if _is_bad_request_error(exc) and _looks_like_responses_request_rejected(text): return True + if _is_server_error(exc) and _looks_like_responses_endpoint(text): + return True if _is_not_found_error(exc) and _looks_like_responses_endpoint_not_found(text): return True if "/v1/responses" in text and any( @@ -1565,6 +1736,9 @@ def _is_responses_not_supported_error(exc: Exception) -> bool: "no 'tools' defined while 'tool_choice' is specified", "tools` must not be an empty array", "tools must not be an empty array", + "not available through this proxy", + "litellm[proxy]", + "no module named 'fastapi'", ) ) @@ -1585,6 +1759,24 @@ def _is_bad_request_error(exc: Exception) -> bool: return "400" in text and "bad request" in text +def _is_server_error(exc: Exception) -> bool: + status_code = _exception_status_code(exc) + if isinstance(status_code, int) and 500 <= status_code < 600: + return True + type_chain = _exception_type_chain(exc).lower() + if "internalservererror" in type_chain: + return True + text = _exception_text(exc).lower() + return any( + marker in text + for marker in ( + "500 internal server error", + "server error '500", + "internalservererror", + ) + ) + + def _looks_like_responses_request_rejected(text: str) -> bool: if "/v1/responses" in text or "responses api" in text: return True @@ -1613,6 +1805,10 @@ def _looks_like_responses_endpoint_not_found(text: str) -> bool: return "detail" in text and "not found" in text +def _looks_like_responses_endpoint(text: str) -> bool: + return "/responses" in text or "path /api/v1/responses" in text + + def _is_responses_state_unsupported_error(exc: Exception) -> bool: text = _exception_text(exc).lower() if any(marker in text for marker in ("429", "too many requests", "rate limit")): @@ -1742,7 +1938,10 @@ def _first_choice(chunk: Any) -> Any: def _get_value(obj: Any, key: str) -> Any: if isinstance(obj, dict): return obj.get(key) - return getattr(obj, key, None) + value = getattr(obj, key, None) + if value is not None: + return value + return _object_to_dict(obj).get(key) def _as_list(value: Any) -> list[Any]: diff --git a/helpers/litellm_transport.py.dox.md b/helpers/litellm_transport.py.dox.md index a3b7bb73f..2bfa0783c 100644 --- a/helpers/litellm_transport.py.dox.md +++ b/helpers/litellm_transport.py.dox.md @@ -28,6 +28,8 @@ - Normalize function tool parameter schemas with an explicit object `properties` field before Responses requests so OpenAI-compatible chat backends reached through LiteLLM can validate them. - Prefer Responses API when configured, but fallback to Chat Completions when the provider does not support Responses. - Fall back to Chat Completions when a Responses request is rejected before any output by an endpoint-specific or shape-specific Bad Request indicating the provider cannot parse Responses payloads. +- Fall back to Chat Completions when a Responses endpoint fails before output with an endpoint-specific server error, proxy path-unavailable error, or LiteLLM proxy-extra import error. +- Preserve Chat Completions tool calls from both non-streaming responses and streaming deltas as canonical `LLMResult` function-call items. - Preserve provider-state metadata when Responses API calls succeed, and fall back to local replay when provider state is unsupported. - Keep prompt-cache markers only for providers that accept them. diff --git a/helpers/llm_result.py b/helpers/llm_result.py index 61e996c95..874edec9b 100644 --- a/helpers/llm_result.py +++ b/helpers/llm_result.py @@ -125,12 +125,13 @@ class LLMResult: response: str, reasoning: str = "", input_items: list[dict[str, Any]] | None = None, + output_items: list[dict[str, Any]] | None = None, provider_model_key: str = "", capability: dict[str, Any] | None = None, ) -> "LLMResult": - output_items = [] - if response: - output_items.append( + items = [ResponseItem.from_any(item) for item in output_items or []] + if response and not items: + items.append( ResponseItem( type="message", data={ @@ -141,7 +142,7 @@ class LLMResult: ) ) if reasoning: - output_items.insert( + items.insert( 0, ResponseItem( type="reasoning", @@ -151,16 +152,19 @@ class LLMResult: }, ), ) - return cls( + result = cls( response=response, reasoning=reasoning, input_items=list(input_items or []), - output_items=output_items, + output_items=items, provider_model_key=provider_model_key, mode="chat_completions", state="off", capability=dict(capability or {}), ) + if not result.response and result.function_calls: + result.response = result.function_calls_text() + return result @property def function_calls(self) -> list[ResponseFunctionCall]: diff --git a/helpers/llm_result.py.dox.md b/helpers/llm_result.py.dox.md index b48cea191..78b472885 100644 --- a/helpers/llm_result.py.dox.md +++ b/helpers/llm_result.py.dox.md @@ -19,7 +19,7 @@ - `LLMResult.metadata()` stores data under `RESPONSE_METADATA_KEY` so history can round-trip provider state. - `from_response(...)` must preserve provider `response_id`, `previous_response_id`, raw output items, usage, and capability metadata. -- `from_chat(...)` must produce an equivalent chat-completions result with `mode="chat_completions"` and `state="off"`. +- `from_chat(...)` must produce an equivalent chat-completions result with `mode="chat_completions"` and `state="off"`, preserving optional function-call output items when the chat transport supplies them. - Function-call output items must preserve `call_id` and optional acknowledged safety checks. - Argument parsing must tolerate JSON strings, dictionaries, and malformed values without throwing. diff --git a/tests/test_stream_tool_early_stop.py b/tests/test_stream_tool_early_stop.py index 2b73c9503..5caff077c 100644 --- a/tests/test_stream_tool_early_stop.py +++ b/tests/test_stream_tool_early_stop.py @@ -62,6 +62,14 @@ class _FailingAsyncChunkStream: self.closed = True +class _DumpOnly: + def __init__(self, **data): + self._data = data + + def model_dump(self): + return dict(self._data) + + def test_extract_json_root_string_returns_canonical_snapshot(): text = ( 'prefix {"tool_name":"response","tool_args":{"text":"brace } inside"}} ' @@ -553,6 +561,62 @@ async def test_unified_call_falls_back_when_litellm_hides_responses_404_url( assert calls == ["responses", "chat"] +@pytest.mark.parametrize( + "responses_error", + [ + "litellm.exceptions.APIError: Path /api/v1/responses is not " + "available through this proxy.", + "MaskedHTTPStatusError: Server error '500 Internal Server Error' " + "for url 'https://api.venice.ai/api/v1/responses'", + "InternalServerError: OpenAIException - '<=' not supported between " + "instances of 'str' and 'int' for url 'http://192.168.200.52:4000/responses'", + "ImportError Missing dependency No module named 'fastapi'. " + "Run `pip install 'litellm[proxy]'`", + ], +) +@pytest.mark.asyncio +async def test_unified_call_falls_back_for_proxy_responses_failures( + monkeypatch, + responses_error, +): + calls: list[str] = [] + + async def fake_aresponses(*args, **kwargs): + calls.append("responses") + raise RuntimeError(responses_error) + + async def fake_acompletion(*args, **kwargs): + calls.append("chat") + assert kwargs["stream"] is True + assert kwargs["drop_params"] is True + return _AsyncChunkStream([_chunk("fallback")]) + + async def fake_rate_limiter(*args, **kwargs): + return None + + monkeypatch.setattr(litellm_transport, "aresponses", fake_aresponses) + monkeypatch.setattr(litellm_transport, "acompletion", fake_acompletion) + monkeypatch.setattr(models, "apply_rate_limiter", fake_rate_limiter) + + wrapper = models.LiteLLMChatWrapper( + model="test-model", + provider="openai", + model_config=None, + ) + + async def response_callback(chunk: str, full: str): + return None + + response, reasoning = await wrapper.unified_call( + messages=[], + response_callback=response_callback, + ) + + assert response == "fallback" + assert reasoning == "" + assert calls == ["responses", "chat"] + + @pytest.mark.asyncio async def test_unified_call_falls_back_when_responses_bad_request_rejects_shape( monkeypatch, @@ -1169,7 +1233,7 @@ def test_cache_control_policy_keeps_native_responses_first(): def test_responses_fallback_does_not_mask_rate_limits(): exc = RuntimeError( "RateLimitError: 429 Too Many Requests for url " - "https://api.openai.com/v1/responses" + "https://provider.example/v1/responses" ) policy = litellm_transport.TransportPolicy( @@ -1215,6 +1279,191 @@ def test_responses_response_parser_extracts_text_reasoning_and_function_calls(): } +def test_chat_completions_response_parser_extracts_tool_calls(): + parsed = litellm_transport.ChatCompletionsTransport.parse( + { + "choices": [ + { + "message": { + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": { + "name": "lookup", + "arguments": '{"q":"a0"}', + }, + } + ] + } + } + ] + } + ) + + assert extract_tools.json_parse_dirty(parsed["response_delta"]) == { + "tool_name": "lookup", + "tool_args": {"q": "a0"}, + } + assert parsed["_output_items"][0]["name"] == "lookup" + + +def test_chat_completions_stream_parser_accumulates_tool_call_arguments(): + parser = litellm_transport.ChatCompletionsStreamParser() + + assert parser.parse( + { + "choices": [ + { + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": { + "name": "lookup", + "arguments": '{"q":', + }, + } + ] + } + } + ] + } + ) == {"reasoning_delta": "", "response_delta": ""} + parsed = parser.parse( + { + "choices": [ + { + "delta": { + "tool_calls": [ + { + "index": 0, + "function": {"arguments": '"a0"}'}, + } + ] + }, + "finish_reason": "tool_calls", + } + ] + } + ) + + assert extract_tools.json_parse_dirty(parsed["response_delta"]) == { + "tool_name": "lookup", + "tool_args": {"q": "a0"}, + } + assert parser.output_items()[0]["name"] == "lookup" + assert parser.flush() == {"reasoning_delta": "", "response_delta": ""} + + +def test_chat_completions_stream_parser_reads_dumped_tool_calls(): + parser = litellm_transport.ChatCompletionsStreamParser() + + assert parser.parse( + _DumpOnly( + choices=[ + _DumpOnly( + delta=_DumpOnly( + tool_calls=[ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": _DumpOnly( + name="lookup", + arguments='{"q":"a0"}', + ), + } + ] + ) + ) + ] + ) + ) == {"reasoning_delta": "", "response_delta": ""} + + parsed = parser.parse( + _DumpOnly(choices=[_DumpOnly(delta=_DumpOnly(), finish_reason="tool_calls")]) + ) + + assert extract_tools.json_parse_dirty(parsed["response_delta"]) == { + "tool_name": "lookup", + "tool_args": {"q": "a0"}, + } + + +@pytest.mark.asyncio +async def test_unified_turn_preserves_chat_streaming_tool_calls(monkeypatch): + async def fake_acompletion(*args, **kwargs): + return _AsyncChunkStream( + [ + { + "choices": [ + { + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": { + "name": "lookup", + "arguments": '{"q":', + }, + } + ] + } + } + ] + }, + { + "choices": [ + { + "delta": { + "tool_calls": [ + { + "index": 0, + "function": {"arguments": '"a0"}'}, + } + ] + }, + "finish_reason": "tool_calls", + } + ] + }, + ] + ) + + async def fake_rate_limiter(*args, **kwargs): + return None + + monkeypatch.setattr(litellm_transport, "acompletion", fake_acompletion) + monkeypatch.setattr(models, "apply_rate_limiter", fake_rate_limiter) + + wrapper = models.LiteLLMChatWrapper( + model="test-model", + provider="openai", + model_config=None, + ) + + async def response_callback(chunk: str, full: str): + return None + + result = await wrapper.unified_turn( + messages=[], + response_callback=response_callback, + a0_api_mode="chat", + ) + + assert extract_tools.json_parse_dirty(result.response) == { + "tool_name": "lookup", + "tool_args": {"q": "a0"}, + } + assert result.function_calls[0].name == "lookup" + assert result.function_calls[0].arguments == {"q": "a0"} + + def test_responses_stream_parser_accumulates_function_call_arguments(): parser = litellm_transport.ResponsesEventParser()