diff --git a/apps/web/vue3/main.py b/apps/web/vue3/main.py index eec13e569..e8a239136 100644 --- a/apps/web/vue3/main.py +++ b/apps/web/vue3/main.py @@ -62,12 +62,23 @@ DEFAULT_AGENT_LANG = "zh" if PLANTAI_RAG_ROOT.exists() and str(PLANTAI_RAG_ROOT) not in sys.path: sys.path.insert(0, str(PLANTAI_RAG_ROOT)) +_deepseek_api_key_before_rag_import = os.environ.get("DEEPSEEK_API_KEY") +_plantai_new_api_bundle_configured = all( + str(os.environ.get(name) or "").strip() + for name in ("NEW_API_KEY", "NEW_API_BASE_URL", "NEW_API_MODEL") +) try: from src.rag.unified.bbav5_methodology_config import build_bbav5_methodology_config METHODOLOGY_CONFIG_AVAILABLE = True except Exception as e: METHODOLOGY_CONFIG_AVAILABLE = False print(f"⚠️ 方法学配置模块加载失败: {e}") +finally: + if _plantai_new_api_bundle_configured: + if _deepseek_api_key_before_rag_import is None: + os.environ.pop("DEEPSEEK_API_KEY", None) + else: + os.environ["DEEPSEEK_API_KEY"] = _deepseek_api_key_before_rag_import try: from gfa_core.mas.agents.interpretation_evidence import build_evidence_report diff --git a/gfa_core/mas/agents/router_runtime.py b/gfa_core/mas/agents/router_runtime.py index 692351ee7..3bc9e0a99 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 @@ -20,6 +22,64 @@ from gfa_core.mas.workflow_catalog import ( from gfa_core.mas.workflow_models import CurrentRecommendation, GoalLedger +logger = logging.getLogger(__name__) + + +_ROUTER_ERROR_TEXT_LIMIT = 1000 +_ROUTER_MODEL_FAILURE_MESSAGE_PREFIX = "Router 调用 DeepSeek 失败(" +_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", @@ -561,6 +621,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: @@ -577,6 +638,50 @@ 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 + 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, @@ -1126,15 +1231,16 @@ 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, response_format={"type": "json_object"}, ) + except RouterModelCallError as exc: + self._last_model_call_error = exc + return None, calls + 1, [] except Exception: return None, calls + 1, [] calls += 1 @@ -1181,6 +1287,7 @@ class RouterRuntime: goal_ledger: GoalLedger | None = None, language: str = "zh", ) -> RouterOutcome: + self._last_model_call_error = None ui_action = self._config_dialog_ui_action(query) if ui_action is not None: return RouterOutcome( @@ -1276,7 +1383,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( @@ -1522,10 +1633,8 @@ 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" @@ -1537,6 +1646,12 @@ class RouterRuntime: else None ), ) + except RouterModelCallError as exc: + return RouterOutcome( + status="agent_unavailable", + message=exc.user_message(), + tool_trace=tool_trace, + ) except Exception: return RouterOutcome( status="agent_unavailable", @@ -2271,8 +2386,12 @@ class RouterRuntime: role = item.get("role") if role not in {"user", "assistant"}: continue - content = item.get("content", "") - normalized.append(LLMMessage(role=role, content=str(content or ""))) + content = str(item.get("content", "") or "") + if role == "assistant" and content.startswith( + _ROUTER_MODEL_FAILURE_MESSAGE_PREFIX + ): + continue + normalized.append(LLMMessage(role=role, content=content)) route_context = json.dumps( { @@ -2338,8 +2457,13 @@ class RouterRuntime: role = item.get("role") if role not in {"user", "assistant"}: continue + content = str(item.get("content", "") or "") + if role == "assistant" and content.startswith( + _ROUTER_MODEL_FAILURE_MESSAGE_PREFIX + ): + continue normalized.append( - LLMMessage(role=role, content=str(item.get("content", "") or "")) + LLMMessage(role=role, content=content) ) context = f"\nCurrent session_id={session_id!r}; current job_id={job_id!r}." if goal_ledger is not None: diff --git a/tests/unit/test_router_analysis_handoff_boundary.py b/tests/unit/test_router_analysis_handoff_boundary.py index 74aaaf3a1..6129ba6c7 100644 --- a/tests/unit/test_router_analysis_handoff_boundary.py +++ b/tests/unit/test_router_analysis_handoff_boundary.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import json +from types import SimpleNamespace from typing import Any import pytest @@ -269,3 +270,203 @@ def test_router_response_language_does_not_rewrite_business_query() -> None: assert messages[-1].content == query assert "Respond to the user in English." in messages[0].content + + +def test_router_generated_model_failure_is_not_replayed_as_history() -> None: + failure_message = ( + "Router 调用 DeepSeek 失败(HTTP 402,模型 deepseek-v4-flash," + "已尝试 3 次,错误码 invalid_request_error):Insufficient Balance" + ) + history = [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": failure_message}, + {"role": "assistant", "content": "正常历史回复"}, + ] + decision = RouteDecision(goal="general", needs=[], subject="hello") + + route_messages = RouterRuntime._route_messages( + query="hello", + history=history, + session_id="router-history-session", + job_id="job-router-history", + goal_ledger=None, + ) + response_messages = RouterRuntime._initial_messages( + query="hello", + history=history, + session_id="router-history-session", + job_id="job-router-history", + decision=decision, + goal_ledger=None, + ) + + for messages in (route_messages, response_messages): + contents = [message.content for message in messages] + assert failure_message not in contents + assert "正常历史回复" in contents + + +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 diff --git a/tests/unit/test_web_llm_environment_isolation.py b/tests/unit/test_web_llm_environment_isolation.py new file mode 100644 index 000000000..aabbe9619 --- /dev/null +++ b/tests/unit/test_web_llm_environment_isolation.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import os +from pathlib import Path +import subprocess +import sys + + +PROJECT_ROOT = Path(__file__).resolve().parents[2] + + +def test_web_import_keeps_plantai_new_api_bundle_when_rag_import_sets_deepseek_key( + tmp_path: Path, +) -> None: + rag_root = tmp_path / "rag" + module_dir = rag_root / "src" / "rag" / "unified" + module_dir.mkdir(parents=True) + for package_dir in (rag_root / "src", rag_root / "src" / "rag", module_dir): + (package_dir / "__init__.py").write_text("", encoding="utf-8") + (module_dir / "bbav5_methodology_config.py").write_text( + ( + "import os\n" + "os.environ.setdefault('DEEPSEEK_API_KEY', 'external-stale-key')\n" + "def build_bbav5_methodology_config(*args, **kwargs):\n" + " return {}\n" + ), + encoding="utf-8", + ) + + runtime_root = tmp_path / "runtime" + runtime_root.mkdir() + env = os.environ.copy() + for name in ( + "DEEPSEEK_API_KEY", + "deepseek_api_key", + "NEW_API_KEY", + "NEW_API_BASE_URL", + "NEW_API_MODEL", + ): + env.pop(name, None) + env.update( + { + "NEW_API_KEY": "plantai-current-key", + "NEW_API_BASE_URL": "https://api.deepseek.com", + "NEW_API_MODEL": "deepseek-v4-flash", + "PLANTAI_RAG_ROOT": str(rag_root), + "GF_MAS_ROOT": str(tmp_path / "missing-gf-mas"), + "BBAV5_PROJECT_ROOT": str(runtime_root), + "PLANTAI_PROJECT_ROOT": str(runtime_root), + } + ) + + result = subprocess.run( + [ + sys.executable, + "-c", + ( + "import os\n" + "from pathlib import Path\n" + f"path = Path({str(PROJECT_ROOT / 'apps/web/vue3/main.py')!r})\n" + "source = path.read_text(encoding='utf-8')\n" + "marker = 'try:\\n from gfa_core.mas.agents.interpretation_evidence'\n" + "prefix, separator, _ = source.partition(marker)\n" + "assert separator\n" + "namespace = {'__file__': str(path), '__name__': 'web_import_prefix'}\n" + "exec(compile(prefix, str(path), 'exec'), namespace)\n" + "assert 'DEEPSEEK_API_KEY' not in os.environ\n" + ), + ], + cwd=PROJECT_ROOT, + env=env, + capture_output=True, + text=True, + timeout=120, + check=False, + ) + + assert result.returncode == 0, result.stdout + result.stderr