From 89a384776eac14b459098122e14bb114e7309ede Mon Sep 17 00:00:00 2001 From: sj Date: Thu, 20 Aug 2026 10:44:19 +0800 Subject: [PATCH] fix: expose Router upstream errors --- gfa_core/mas/agents/router_runtime.py | 132 +++++++++++++- .../test_router_analysis_handoff_boundary.py | 167 ++++++++++++++++++ 2 files changed, 292 insertions(+), 7 deletions(-) diff --git a/gfa_core/mas/agents/router_runtime.py b/gfa_core/mas/agents/router_runtime.py index a2a5c948e..97487a3ea 100644 --- a/gfa_core/mas/agents/router_runtime.py +++ b/gfa_core/mas/agents/router_runtime.py @@ -1,5 +1,7 @@ +import asyncio import json import hashlib +import logging import os import re from collections import Counter @@ -18,6 +20,63 @@ from gfa_core.mas.workflow_catalog import ( from gfa_core.mas.workflow_models import GoalLedger +logger = logging.getLogger(__name__) + + +_ROUTER_ERROR_TEXT_LIMIT = 1000 +_ROUTER_CREDENTIAL_PATTERN = re.compile( + r"(?i)\b(api[_-]?key|authorization|(?:access|refresh)[_-]?token|token|secret)" + r"\b\s*[:=]\s*[^,;}\]\r\n]+" +) + + +def _sanitize_router_error_text(value: Any) -> str: + text = str(value or "").strip() + if not text: + return "" + return _ROUTER_CREDENTIAL_PATTERN.sub(r"\1=", text)[ + :_ROUTER_ERROR_TEXT_LIMIT + ] + + +class RouterModelCallError(RuntimeError): + """Allow-listed details from a failed Router model request.""" + + def __init__(self, cause: Exception, *, attempts: int, model: str) -> None: + self.status_code = getattr(cause, "status_code", None) + self.attempts = attempts + self.model = _sanitize_router_error_text(model) or "unknown" + self.request_id = _sanitize_router_error_text( + getattr(cause, "request_id", "") + ) + + body = getattr(cause, "body", None) + error: Mapping[str, Any] = {} + if isinstance(body, Mapping): + nested_error = body.get("error") + error = nested_error if isinstance(nested_error, Mapping) else body + self.error_code = _sanitize_router_error_text(error.get("code")) + self.upstream_message = _sanitize_router_error_text(error.get("message")) + if not self.upstream_message: + self.upstream_message = _sanitize_router_error_text(str(cause)) + if not self.upstream_message: + self.upstream_message = type(cause).__name__ + + super().__init__(self.user_message()) + + def user_message(self) -> str: + details = [] + if self.status_code is not None: + details.append(f"HTTP {self.status_code}") + details.append(f"模型 {self.model}") + details.append(f"已尝试 {self.attempts} 次") + if self.error_code: + details.append(f"错误码 {self.error_code}") + if self.request_id: + details.append(f"Request ID {self.request_id}") + return f"Router 调用 DeepSeek 失败（{'，'.join(details)}）：{self.upstream_message}" + + InteractionKind = Literal[ "supplement", "correction", @@ -400,6 +459,7 @@ class RouterRuntime: ) -> None: self.llm_manager = llm_manager self.tools = tools + self._last_model_call_error: RouterModelCallError | None = None if max_iterations is None: raw_budget = os.environ.get("PLANTAI_MAIN_AGENT_MAX_TOOL_CALLS", "10") try: @@ -416,6 +476,54 @@ class RouterRuntime: else: self.max_iterations = min(max(0, max_iterations), 5) + async def _call_router_model( + self, + messages: list[LLMMessage], + **kwargs: Any, + ) -> Any: + """Call the Router model with bounded retries for transient failures.""" + max_attempts = 3 + # This OpenAI-compatible upstream can emit a transient 402 from its + # billing gateway even when the same credential succeeds immediately + # afterwards. Keep retries bounded so a real balance problem still + # fails quickly and remains visible in the logs. + retryable_statuses = {402, 408, 409, 429, 500, 502, 503, 504} + for attempt in range(1, max_attempts + 1): + try: + return await self.llm_manager.acall( + messages, + provider="deepseek", + allow_fallback=False, + **kwargs, + ) + except Exception as exc: + status = getattr(exc, "status_code", None) + retryable = status is None or status in retryable_statuses + logger.warning( + "Router model call failed: attempt=%s/%s exception=%s status=%s retryable=%s", + attempt, + max_attempts, + type(exc).__name__, + status, + retryable, + ) + if not retryable or attempt >= max_attempts: + failure = RouterModelCallError( + exc, + attempts=attempt, + model=str( + getattr( + self.llm_manager, + "_default_model", + "deepseek-v4-flash", + ) + or "deepseek-v4-flash" + ), + ) + logger.error("Router model unavailable: %s", failure.user_message()) + raise failure from exc + await asyncio.sleep(0.5 * (2 ** (attempt - 1))) + @staticmethod def _action_for_route( decision: RouteDecision, @@ -603,14 +711,15 @@ class RouterRuntime: calls = 0 for attempt in range(min(2, self.max_iterations)): try: - response = await self.llm_manager.acall( + response = await self._call_router_model( route_messages, - provider="deepseek", - allow_fallback=False, tools=[], tool_choice="none", temperature=0, ) + except RouterModelCallError as exc: + self._last_model_call_error = exc + return None, calls + 1, [] except Exception: return None, calls + 1, [] calls += 1 @@ -651,6 +760,7 @@ class RouterRuntime: job_id: str, goal_ledger: GoalLedger | None = None, ) -> RouterOutcome: + self._last_model_call_error = None ui_action = self._config_dialog_ui_action(query) if ui_action is not None: return RouterOutcome( @@ -732,7 +842,11 @@ class RouterRuntime: if not decision_trace: return RouterOutcome( status="agent_unavailable", - message="The Router model is currently unavailable.", + message=( + self._last_model_call_error.user_message() + if self._last_model_call_error is not None + else "The Router model is currently unavailable." + ), ) if decision_calls < 2: return RouterOutcome( @@ -865,16 +979,20 @@ class RouterRuntime: ] force_final = not remaining_needs or recognized_intent is not None try: - response = await self.llm_manager.acall( + response = await self._call_router_model( messages, - provider="deepseek", - allow_fallback=False, tools=[] if force_final else available_definitions, tool_choice=( "none" if force_final else "required" ), temperature=0, ) + except RouterModelCallError as exc: + return RouterOutcome( + status="agent_unavailable", + message=exc.user_message(), + tool_trace=tool_trace, + ) except Exception: return RouterOutcome( status="agent_unavailable", diff --git a/tests/unit/test_router_analysis_handoff_boundary.py b/tests/unit/test_router_analysis_handoff_boundary.py index e47e83eda..cdbd39c0a 100644 --- a/tests/unit/test_router_analysis_handoff_boundary.py +++ b/tests/unit/test_router_analysis_handoff_boundary.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +from types import SimpleNamespace from typing import Any import pytest @@ -99,3 +100,169 @@ def test_new_analysis_decision_discards_main_agent_workflow_and_clarification() assert normalized.workflow_id is None assert normalized.clarification_question is None assert normalized.analysis_request_kind == "execution" + + +def test_route_decision_retries_transient_model_failures( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class TransientPaymentGatewayFailure(RuntimeError): + status_code = 402 + + class TransientLLM: + calls = 0 + + async def acall(self, *_: Any, **__: Any) -> Any: + self.calls += 1 + if self.calls < 3: + raise TransientPaymentGatewayFailure("temporary billing gateway failure") + return SimpleNamespace( + content=( + '{"goal":"new_analysis","needs":["analysis_intent"],' + '"subject":"执行数据准备","analysis_request_kind":"execution"}' + ) + ) + + async def no_sleep(_: float) -> None: + return None + + llm = TransientLLM() + runtime = RouterRuntime(llm, _PassThroughTools(), max_iterations=5) # type: ignore[arg-type] + monkeypatch.setattr(asyncio, "sleep", no_sleep) + + decision, calls, trace = asyncio.run( + runtime._decide_route( + query="执行数据准备", + history=[], + session_id="retry-session", + job_id="job-retry", + goal_ledger=None, + ) + ) + + assert llm.calls == 3 + assert calls == 1 + assert trace == [] + assert decision is not None + assert decision.goal == "new_analysis" + + +def test_route_decision_does_not_retry_authentication_failure() -> None: + class AuthenticationFailure(RuntimeError): + status_code = 401 + + class UnauthorizedLLM: + calls = 0 + + async def acall(self, *_: Any, **__: Any) -> Any: + self.calls += 1 + raise AuthenticationFailure("unauthorized") + + llm = UnauthorizedLLM() + runtime = RouterRuntime(llm, _PassThroughTools(), max_iterations=5) # type: ignore[arg-type] + + decision, calls, trace = asyncio.run( + runtime._decide_route( + query="执行数据准备", + history=[], + session_id="auth-session", + job_id="job-auth", + goal_ledger=None, + ) + ) + + assert llm.calls == 1 + assert decision is None + assert calls == 1 + assert trace == [] + + +def test_route_failure_exposes_sanitized_upstream_reason( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class StructuredPaymentGatewayFailure(RuntimeError): + status_code = 402 + request_id = "req-router-402" + body = { + "error": { + "code": "request_policy_rejected", + "message": ( + "the concrete upstream reason; " + "Authorization: Bearer must-not-leak" + ), + "api_key": "must-not-leak", + } + } + + class FailingLLM: + _default_model = "deepseek-v4-flash" + calls = 0 + + async def acall(self, *_: Any, **__: Any) -> Any: + self.calls += 1 + raise StructuredPaymentGatewayFailure("generic SDK error") + + async def no_sleep(_: float) -> None: + return None + + llm = FailingLLM() + runtime = RouterRuntime(llm, _PassThroughTools(), max_iterations=5) # type: ignore[arg-type] + monkeypatch.setattr(asyncio, "sleep", no_sleep) + + outcome = asyncio.run( + runtime.run( + query="执行数据准备", + history=[], + session_id="failure-detail-session", + job_id="job-failure-detail", + goal_ledger=None, + ) + ) + + assert llm.calls == 3 + assert outcome.status == "agent_unavailable" + assert "HTTP 402" in outcome.message + assert "request_policy_rejected" in outcome.message + assert "the concrete upstream reason" in outcome.message + assert "deepseek-v4-flash" in outcome.message + assert "3 次" in outcome.message + assert "req-router-402" in outcome.message + assert "must-not-leak" not in outcome.message + + +def test_route_failure_reads_flat_openai_error_body( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class FlatOpenAIError(RuntimeError): + status_code = 402 + body = { + "code": "invalid_request_error", + "message": "Insufficient Balance", + "type": "unknown_error", + } + + class FailingLLM: + _default_model = "deepseek-v4-flash" + + async def acall(self, *_: Any, **__: Any) -> Any: + raise FlatOpenAIError("serialized SDK fallback") + + async def no_sleep(_: float) -> None: + return None + + runtime = RouterRuntime(FailingLLM(), _PassThroughTools(), max_iterations=5) # type: ignore[arg-type] + monkeypatch.setattr(asyncio, "sleep", no_sleep) + + outcome = asyncio.run( + runtime.run( + query="执行数据准备", + history=[], + session_id="flat-openai-error-session", + job_id="job-flat-openai-error", + goal_ledger=None, + ) + ) + + assert outcome.status == "agent_unavailable" + assert "错误码 invalid_request_error" in outcome.message + assert outcome.message.endswith("Insufficient Balance") + assert "serialized SDK fallback" not in outcome.message