From 840da4a79bb1539b9221652c9939f1d300d8f835 Mon Sep 17 00:00:00 2001 From: "hanzhi.421" Date: Wed, 15 Jul 2026 15:30:00 +0800 Subject: [PATCH 1/3] feat: support Ark responses context management --- pyproject.toml | 2 +- .../models/test_ark_llm_context_management.py | 41 +++++++++++++++++++ tests/test_agent.py | 23 +++++++++++ veadk/models/ark_llm.py | 1 + 4 files changed, 66 insertions(+), 1 deletion(-) create mode 100644 tests/models/test_ark_llm_context_management.py diff --git a/pyproject.toml b/pyproject.toml index ceda4f49..74a90413 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,7 +30,7 @@ dependencies = [ "opentelemetry-exporter-otlp==1.37.0", "opentelemetry-instrumentation-logging>=0.56b0", "wrapt==1.17.2", # For patching built-in functions - "volcengine-python-sdk>=5.0.1", # For Volcengine API + "volcengine-python-sdk>=5.0.36", # For Volcengine API and Ark Responses context management "volcengine>=1.0.193", # For Volcengine sign and AgentKit Runtime API access "agent-pilot-sdk==0.1.2", # Prompt optimization by Volcengine AgentPilot/PromptPilot toolkits "fastmcp>=2.12.3", # For running MCP diff --git a/tests/models/test_ark_llm_context_management.py b/tests/models/test_ark_llm_context_management.py new file mode 100644 index 00000000..7c85debb --- /dev/null +++ b/tests/models/test_ark_llm_context_management.py @@ -0,0 +1,41 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from veadk.models.ark_llm import request_reorganization_by_ark + + +def test_request_reorganization_preserves_context_management(): + context_management = { + "edits": [ + { + "type": "clear_thinking", + "keep": {"type": "thinking_turns", "value": 1}, + }, + { + "type": "clear_tool_uses", + "trigger": {"type": "tool_uses", "value": 30}, + "keep": {"type": "tool_uses", "value": 20}, + }, + ] + } + + request = request_reorganization_by_ark( + { + "model": "openai/test-model", + "input": [], + "context_management": context_management, + } + ) + + assert request["context_management"] == context_management diff --git a/tests/test_agent.py b/tests/test_agent.py index a24c12ce..c6c6c6e4 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -136,6 +136,29 @@ def test_agent_model_creation(mock_lite_llm): assert agent.model == mock_model +@patch("veadk.models.ark_llm.ArkLlm") +def test_agent_passes_context_management_to_responses_model(mock_ark_llm): + context_management = { + "edits": [ + { + "type": "clear_thinking", + "keep": {"type": "thinking_turns", "value": 1}, + } + ] + } + + Agent( + model_name="test_model", + model_provider="ark", + model_api_key="test_key", + model_api_base="test_base", + enable_responses=True, + model_extra_config={"context_management": context_management}, + ) + + assert mock_ark_llm.call_args.kwargs["context_management"] == context_management + + @patch.dict("os.environ", {"MODEL_AGENT_API_KEY": "mock_api_key"}) def test_agent_with_existing_model(): existing_model = LiteLlm(model="test_model") diff --git a/veadk/models/ark_llm.py b/veadk/models/ark_llm.py index 07cba482..246a084e 100644 --- a/veadk/models/ark_llm.py +++ b/veadk/models/ark_llm.py @@ -101,6 +101,7 @@ "tools", "top_p", "max_tool_calls", + "context_management", "expire_at", "extra_headers", "extra_query", From 4eeaa522086ce44fcbd245698865b607cadd65e6 Mon Sep 17 00:00:00 2001 From: "hanzhi.421" Date: Thu, 16 Jul 2026 10:59:19 +0800 Subject: [PATCH 2/3] fix(observability): harden inline image TOS upload --- tests/test_ve_tos.py | 127 ++++++++++++++++++++++++++++ veadk/integrations/ve_tos/ve_tos.py | 18 ++-- veadk/runner.py | 11 ++- 3 files changed, 146 insertions(+), 10 deletions(-) create mode 100644 tests/test_ve_tos.py diff --git a/tests/test_ve_tos.py b/tests/test_ve_tos.py new file mode 100644 index 00000000..6c46e356 --- /dev/null +++ b/tests/test_ve_tos.py @@ -0,0 +1,127 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import sys +from types import SimpleNamespace + +import pytest +from google.genai import types + +from veadk.integrations.ve_tos.ve_tos import VeTOS +from veadk.runner import _upload_image_to_tos + + +class _FakeTosClient: + def __init__(self, **kwargs): + self.init_kwargs = kwargs + self.put_calls = [] + + def put_object(self, **kwargs): + self.put_calls.append(kwargs) + + +@pytest.fixture +def fake_tos(monkeypatch): + module = SimpleNamespace(TosClientV2=_FakeTosClient) + monkeypatch.setitem(sys.modules, "tos", module) + monkeypatch.setenv("VOLCENGINE_ACCESS_KEY", "test-ak") + monkeypatch.setenv("VOLCENGINE_SECRET_KEY", "test-sk") + return module + + +def test_ve_tos_uses_bucket_and_session_token_from_environment(monkeypatch, fake_tos): + monkeypatch.setenv("DATABASE_TOS_BUCKET", "customer-bucket") + monkeypatch.setenv("VOLCENGINE_SESSION_TOKEN", "test-token") + + client = VeTOS() + + assert client.bucket_name == "customer-bucket" + assert client.session_token == "test-token" + assert client._client.init_kwargs["security_token"] == "test-token" + + +def test_ve_tos_explicit_bucket_wins_over_environment(monkeypatch, fake_tos): + monkeypatch.setenv("DATABASE_TOS_BUCKET", "environment-bucket") + + client = VeTOS(bucket_name="explicit-bucket") + + assert client.bucket_name == "explicit-bucket" + + +@pytest.mark.asyncio +async def test_async_upload_bytes_reports_success(monkeypatch, fake_tos): + client = VeTOS(bucket_name="test-bucket") + monkeypatch.setattr(client, "_ensure_client_and_bucket", lambda bucket: True) + + uploaded = await client.async_upload_bytes( + data=b"image-bytes", object_key="app/session/image.png" + ) + + assert uploaded is True + assert client._client.put_calls == [ + { + "bucket": "test-bucket", + "key": "app/session/image.png", + "content": b"image-bytes", + "meta": None, + } + ] + + +@pytest.mark.asyncio +async def test_async_upload_bytes_reports_bucket_failure(monkeypatch, fake_tos): + client = VeTOS(bucket_name="test-bucket") + monkeypatch.setattr(client, "_ensure_client_and_bucket", lambda bucket: False) + + uploaded = await client.async_upload_bytes(data=b"image-bytes") + + assert uploaded is False + + +def _inline_image_part(): + return types.Part( + inline_data=types.Blob( + data=b"image-bytes", + display_name="image.png", + mime_type="image/png", + ) + ) + + +@pytest.mark.asyncio +async def test_inline_image_upload_failure_is_fail_open(monkeypatch, fake_tos): + async def report_failure(self, **kwargs): + return False + + monkeypatch.setattr(VeTOS, "async_upload_bytes", report_failure) + part = _inline_image_part() + + await _upload_image_to_tos(part, "test-app", "test-user", "test-session") + + assert part.inline_data.display_name == "image.png" + assert part.inline_data.data == b"image-bytes" + + +@pytest.mark.asyncio +async def test_inline_image_upload_exception_is_fail_open(monkeypatch, fake_tos): + async def raise_upload_error(self, **kwargs): + raise RuntimeError("simulated TOS outage") + + monkeypatch.setattr(VeTOS, "async_upload_bytes", raise_upload_error) + part = _inline_image_part() + + await _upload_image_to_tos(part, "test-app", "test-user", "test-session") + + assert part.inline_data.display_name == "image.png" + assert part.inline_data.data == b"image-bytes" diff --git a/veadk/integrations/ve_tos/ve_tos.py b/veadk/integrations/ve_tos/ve_tos.py index 446937bd..403c4111 100644 --- a/veadk/integrations/ve_tos/ve_tos.py +++ b/veadk/integrations/ve_tos/ve_tos.py @@ -21,7 +21,6 @@ from veadk.consts import DEFAULT_TOS_BUCKET_NAME from veadk.utils.logger import get_logger -from veadk.utils.misc import getenv if TYPE_CHECKING: pass @@ -38,11 +37,11 @@ def __init__( sk: str = "", session_token: str = "", region: str = "", - bucket_name: str = DEFAULT_TOS_BUCKET_NAME, + bucket_name: str = "", ) -> None: self.ak = ak if ak else os.getenv("VOLCENGINE_ACCESS_KEY", "") self.sk = sk if sk else os.getenv("VOLCENGINE_SECRET_KEY", "") - self.session_token = session_token + self.session_token = session_token or os.getenv("VOLCENGINE_SESSION_TOKEN", "") # get provider from env provider = (os.getenv("CLOUD_PROVIDER") or "").lower() @@ -75,7 +74,7 @@ def __init__( ) self.bucket_name = ( - bucket_name if bucket_name else getenv("", DEFAULT_TOS_BUCKET_NAME) + bucket_name or os.getenv("DATABASE_TOS_BUCKET") or DEFAULT_TOS_BUCKET_NAME ) self._tos_module = None @@ -454,7 +453,7 @@ async def async_upload_bytes( bucket_name: str = "", object_key: str = "", metadata: dict | None = None, - ) -> None: + ) -> bool: """Asynchronously upload byte data to TOS bucket Args: @@ -462,13 +461,16 @@ async def async_upload_bytes( bucket_name: TOS bucket name object_key: Object key, auto-generated if None metadata: Metadata to associate with the object + + Returns: + ``True`` when the object was uploaded, otherwise ``False``. """ bucket_name = self._check_bucket_name(bucket_name) if not object_key: object_key = self._build_object_key_for_bytes() # Use common function to check client and bucket if not self._ensure_client_and_bucket(bucket_name): - return + return False try: # Use asyncio.to_thread to execute blocking TOS operations in thread await asyncio.to_thread( @@ -479,10 +481,10 @@ async def async_upload_bytes( meta=metadata, ) logger.debug(f"Async upload success, object_key: {object_key}") - return + return True except Exception as e: logger.error(f"Async upload failed: {e}") - return + return False def upload_file( self, diff --git a/veadk/runner.py b/veadk/runner.py index 55cce3b4..01c1a5db 100644 --- a/veadk/runner.py +++ b/veadk/runner.py @@ -285,11 +285,18 @@ async def _upload_image_to_tos( filename = os.path.basename(part.inline_data.display_name) object_key = f"{app_name}/{user_id}-{session_id}-{filename}" ve_tos = VeTOS() - tos_url = ve_tos.build_tos_signed_url(object_key=object_key) - await ve_tos.async_upload_bytes( + uploaded = await ve_tos.async_upload_bytes( object_key=object_key, data=part.inline_data.data, ) + if not uploaded: + logger.error( + "TOS upload did not succeed; continuing without a traced " + f"attachment URL | bucket={ve_tos.bucket_name}, " + f"object_key={object_key}" + ) + return + tos_url = ve_tos.build_tos_signed_url(object_key=object_key) part.inline_data.display_name = tos_url except Exception as e: logger.exception( From c4d1e7e31f88a92e43c0b76e3e72a5f5f7b85133 Mon Sep 17 00:00:00 2001 From: "hanzhi.421" Date: Thu, 16 Jul 2026 11:51:03 +0800 Subject: [PATCH 3/3] feat(model): support fallbacks for Ark Responses API --- .../docs/framework/agent/responses-api.en.mdx | 11 ++ .../docs/framework/agent/responses-api.mdx | 11 ++ tests/models/test_ark_llm_fallbacks.py | 138 ++++++++++++++++++ tests/test_agent.py | 17 +++ veadk/agent.py | 39 ++--- veadk/models/ark_llm.py | 115 ++++++++++----- 6 files changed, 278 insertions(+), 53 deletions(-) create mode 100644 tests/models/test_ark_llm_fallbacks.py diff --git a/docs/content/docs/framework/agent/responses-api.en.mdx b/docs/content/docs/framework/agent/responses-api.en.mdx index 080a9d73..353ffcd5 100644 --- a/docs/content/docs/framework/agent/responses-api.en.mdx +++ b/docs/content/docs/framework/agent/responses-api.en.mdx @@ -41,6 +41,17 @@ root_agent = Agent( ) ``` +**Model fallbacks** + +The Responses API also accepts a `model_name` list for resilience. If a request fails, VeADK tries each subsequent model in order until one succeeds or all models fail. + +```python +root_agent = Agent( + enable_responses=True, + model_name=["doubao-seed-1-8-251228", "deepseek-r1-250528"], +) +``` + **Result** ![responses-api](/assets/images/agents/responses_api.png) diff --git a/docs/content/docs/framework/agent/responses-api.mdx b/docs/content/docs/framework/agent/responses-api.mdx index fc104955..94610751 100644 --- a/docs/content/docs/framework/agent/responses-api.mdx +++ b/docs/content/docs/framework/agent/responses-api.mdx @@ -41,6 +41,17 @@ root_agent = Agent( ) ``` +**多模型容错** + +Responses API 同样支持通过 `model_name` 列表配置容错模型。请求失败时会按顺序尝试后续模型，直到成功或所有模型均失败。 + +```python +root_agent = Agent( + enable_responses=True, + model_name=["doubao-seed-1-8-251228", "deepseek-r1-250528"], +) +``` + **效果展示** ![responses-api](/assets/images/agents/responses_api.png) diff --git a/tests/models/test_ark_llm_fallbacks.py b/tests/models/test_ark_llm_fallbacks.py new file mode 100644 index 00000000..dfd2f0bd --- /dev/null +++ b/tests/models/test_ark_llm_fallbacks.py @@ -0,0 +1,138 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import httpx +import pytest +from google.adk.models import LlmResponse +from volcenginesdkarkruntime._exceptions import ArkBadRequestError + +from veadk.models.ark_llm import ArkLlm + + +@pytest.mark.asyncio +async def test_ark_llm_uses_next_model_when_primary_fails(monkeypatch): + model = ArkLlm( + model="openai/primary-model", + fallbacks=["openai/fallback-model"], + ) + calls = [] + + async def fake_generate(self, responses_args, stream=False): + calls.append((responses_args["model"], stream)) + if responses_args["model"] == "openai/primary-model": + raise RuntimeError("primary failed") + yield LlmResponse(model_version=responses_args["model"]) + + monkeypatch.setattr(ArkLlm, "generate_content_via_responses", fake_generate) + + responses = [ + response + async for response in model._generate_content_with_fallbacks( + {"input": []}, stream=True + ) + ] + + assert calls == [ + ("openai/primary-model", True), + ("openai/fallback-model", True), + ] + assert [response.model_version for response in responses] == [ + "openai/fallback-model" + ] + + +@pytest.mark.asyncio +async def test_ark_llm_raises_last_error_when_all_models_fail(monkeypatch): + model = ArkLlm( + model="openai/primary-model", + fallbacks=["openai/fallback-model"], + ) + + async def fake_generate(self, responses_args, stream=False): + if False: + yield + raise RuntimeError(f"{responses_args['model']} failed") + + monkeypatch.setattr(ArkLlm, "generate_content_via_responses", fake_generate) + + with pytest.raises(RuntimeError, match="openai/fallback-model failed"): + async for _ in model._generate_content_with_fallbacks({"input": []}): + pass + + +@pytest.mark.asyncio +async def test_ark_llm_does_not_fallback_after_stream_output(monkeypatch): + model = ArkLlm( + model="openai/primary-model", + fallbacks=["openai/fallback-model"], + ) + calls = [] + + async def fake_generate(self, responses_args, stream=False): + calls.append(responses_args["model"]) + yield LlmResponse(model_version=responses_args["model"], partial=True) + raise RuntimeError("stream interrupted") + + monkeypatch.setattr(ArkLlm, "generate_content_via_responses", fake_generate) + + generator = model._generate_content_with_fallbacks({"input": []}, stream=True) + first_response = await anext(generator) + assert first_response.model_version == "openai/primary-model" + + with pytest.raises(RuntimeError, match="stream interrupted"): + await anext(generator) + + assert calls == ["openai/primary-model"] + + +@pytest.mark.asyncio +async def test_ark_llm_retries_expired_response_before_fallback(monkeypatch): + model = ArkLlm( + model="openai/primary-model", + fallbacks=["openai/fallback-model"], + ) + calls = [] + request = httpx.Request("POST", "https://ark.example/v1/responses") + expired_error = ArkBadRequestError( + "previous response expired", + response=httpx.Response(400, request=request), + body={"code": "InvalidParameter.PreviousResponseNotFound"}, + request_id="request-id", + ) + + async def fake_generate(self, responses_args, stream=False): + calls.append( + (responses_args["model"], responses_args.get("previous_response_id")) + ) + if len(calls) == 1: + raise expired_error + if responses_args["model"] == "openai/primary-model": + raise RuntimeError("primary failed without cache") + yield LlmResponse(model_version=responses_args["model"]) + + monkeypatch.setattr(ArkLlm, "generate_content_via_responses", fake_generate) + + responses = [ + response + async for response in model._generate_content_with_fallbacks( + {"input": [], "previous_response_id": "expired-id"} + ) + ] + + assert calls == [ + ("openai/primary-model", "expired-id"), + ("openai/primary-model", None), + ("openai/fallback-model", None), + ] + assert responses[0].model_version == "openai/fallback-model" diff --git a/tests/test_agent.py b/tests/test_agent.py index c6c6c6e4..5371d7d7 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -159,6 +159,23 @@ def test_agent_passes_context_management_to_responses_model(mock_ark_llm): assert mock_ark_llm.call_args.kwargs["context_management"] == context_management +@patch("veadk.models.ark_llm.ArkLlm") +def test_agent_configures_responses_model_fallbacks(mock_ark_llm): + Agent( + model_name=["primary-model", "fallback-model-1", "fallback-model-2"], + model_provider="openai", + model_api_key="test_key", + model_api_base="test_base", + enable_responses=True, + ) + + assert mock_ark_llm.call_args.kwargs["model"] == "openai/primary-model" + assert mock_ark_llm.call_args.kwargs["fallbacks"] == [ + "openai/fallback-model-1", + "openai/fallback-model-2", + ] + + @patch.dict("os.environ", {"MODEL_AGENT_API_KEY": "mock_api_key"}) def test_agent_with_existing_model(): existing_model = LiteLlm(model="test_model") diff --git a/veadk/agent.py b/veadk/agent.py index 64892399..a59c6b9b 100644 --- a/veadk/agent.py +++ b/veadk/agent.py @@ -248,35 +248,36 @@ def model_post_init(self, __context: Any) -> None: logger.info(f"Model extra config: {self.model_extra_config}") if not self.model: + fallbacks = None + if isinstance(self.model_name, list): + if self.model_name: + model_name = self.model_name[0] + fallbacks = [ + f"{self.model_provider}/{m}" for m in self.model_name[1:] + ] + logger.info( + f"Using primary model: {model_name}, with fallbacks: {self.model_name[1:]}" + ) + else: + model_name = settings.model.name + logger.warning( + f"Empty model_name list provided, using default model from settings: {model_name}" + ) + else: + model_name = self.model_name + if self.enable_responses: from veadk.models.ark_llm import ArkLlm self.model = ArkLlm( - model=f"{self.model_provider}/{self.model_name}", + model=f"{self.model_provider}/{model_name}", api_key=self.model_api_key, api_base=self.model_api_base, + fallbacks=fallbacks, enable_responses_cache=self.enable_responses_cache, **self.model_extra_config, ) else: - fallbacks = None - if isinstance(self.model_name, list): - if self.model_name: - model_name = self.model_name[0] - fallbacks = [ - f"{self.model_provider}/{m}" for m in self.model_name[1:] - ] - logger.info( - f"Using primary model: {model_name}, with fallbacks: {self.model_name[1:]}" - ) - else: - model_name = settings.model.name - logger.warning( - f"Empty model_name list provided, using default model from settings: {model_name}" - ) - else: - model_name = self.model_name - self.model = LiteLlm( model=f"{self.model_provider}/{model_name}", api_key=self.model_api_key, diff --git a/veadk/models/ark_llm.py b/veadk/models/ark_llm.py index 246a084e..deb7ad8d 100644 --- a/veadk/models/ark_llm.py +++ b/veadk/models/ark_llm.py @@ -15,6 +15,7 @@ # adapted from Google ADK models adk-python/blob/main/src/google/adk/models/lite_llm.py at f1f44675e4a86b75e72cfd838efd8a0399f23e24 · google/adk-python import base64 +import copy import json import time from typing import Any, Dict, Union, AsyncGenerator, Tuple, List, Optional, Literal @@ -701,6 +702,7 @@ async def aresponses( class ArkLlm(Gemini): model: str + fallbacks: Optional[List[str]] = None llm_client: ArkLlmClient = Field(default_factory=ArkLlmClient) _additional_args: Dict[str, Any] = None use_interactions_api: bool = True @@ -722,6 +724,7 @@ def __init__(self, **kwargs): self._additional_args.pop("messages", None) self._additional_args.pop("tools", None) self._additional_args.pop("stream", None) + self._additional_args.pop("fallbacks", None) self._additional_args.pop("enable_responses_cache", None) if drop_params is not None: self._additional_args["drop_params"] = drop_params @@ -766,46 +769,90 @@ async def generate_content_async( if generation_params: responses_args.update(generation_params) - try: - async for llm_response in self.generate_content_via_responses( - responses_args.copy(), stream=stream - ): - yield llm_response - except ArkBadRequestError as e: - # Check if it is PreviousResponseNotFound - is_expired = False - if hasattr(e, "body") and isinstance(e.body, dict): - if e.body.get("code") == "InvalidParameter.PreviousResponseNotFound": - is_expired = True - - if is_expired: - logger.warning( - f"Interaction expired (PreviousResponseNotFound). Retrying without previous_response_id. Error: {e}" - ) - # Remove previous_response_id - if "previous_response_id" in responses_args: - del responses_args["previous_response_id"] - # Retry - try: - async for llm_response in self.generate_content_via_responses( - responses_args.copy(), stream=stream - ): - yield llm_response - except Exception: + async for llm_response in self._generate_content_with_fallbacks( + responses_args, stream=stream + ): + yield llm_response + + async def _generate_content_with_fallbacks( + self, responses_args: dict, stream: bool = False + ) -> AsyncGenerator[LlmResponse, None]: + """Try the primary Ark model and configured fallbacks in order. + + A fallback is safe only before an attempt has yielded output. Once a + streaming response has reached the caller, switching models would mix + chunks from two responses, so the original error is propagated. + """ + models = [self.model, *(self.fallbacks or [])] + + for index, model in enumerate(models): + attempt_args = copy.deepcopy(responses_args) + attempt_args["model"] = model + yielded_response = False + + try: + async for llm_response in self.generate_content_via_responses( + attempt_args, stream=stream + ): + yielded_response = True + yield llm_response + return + except Exception as error: + if yielded_response: logger.exception( - "Retry without previous_response_id failed in Ark Responses API" + f"Ark Responses API streaming request failed after model `{model}` emitted output; fallback is unsafe" ) raise - else: - logger.exception("Ark Responses API request failed") - raise - except Exception: - logger.exception("Unexpected error in Ark Responses API generation") - raise + + failure = error + if self._is_previous_response_not_found(error): + logger.warning( + f"Interaction expired for model `{model}` (PreviousResponseNotFound). Retrying without previous_response_id. Error: {error}" + ) + responses_args = copy.deepcopy(responses_args) + responses_args.pop("previous_response_id", None) + retry_args = copy.deepcopy(responses_args) + retry_args["model"] = model + retry_yielded_response = False + try: + async for llm_response in self.generate_content_via_responses( + retry_args, stream=stream + ): + retry_yielded_response = True + yield llm_response + return + except Exception as retry_error: + if retry_yielded_response: + logger.exception( + f"Ark Responses API streaming retry failed after model `{model}` emitted output; fallback is unsafe" + ) + raise + failure = retry_error + + if index + 1 < len(models): + next_model = models[index + 1] + logger.warning( + f"Ark Responses API request with model `{model}` failed. Falling back to `{next_model}`. Error: {failure}" + ) + continue + + logger.error( + f"Ark Responses API request failed for all configured models. Last model: `{model}`. Error: {failure}" + ) + raise failure.with_traceback(failure.__traceback__) + + @staticmethod + def _is_previous_response_not_found(error: Exception) -> bool: + return ( + isinstance(error, ArkBadRequestError) + and isinstance(getattr(error, "body", None), dict) + and error.body.get("code") == "InvalidParameter.PreviousResponseNotFound" + ) async def generate_content_via_responses( self, responses_args: dict, stream: bool = False ): + model = responses_args["model"] responses_args = request_reorganization_by_ark( responses_args, enable_responses_cache=self.enable_responses_cache ) @@ -813,7 +860,7 @@ async def generate_content_via_responses( responses_args["stream"] = True async for part in await self.llm_client.aresponses(**responses_args): llm_response = event_to_generate_content_response( - event=part, is_partial=True, model_version=self.model + event=part, is_partial=True, model_version=model ) if llm_response: yield llm_response