已合并
feat: VA上游报错后调Agent生成降级回复，确保SSE流包含final_answer #394
xiongxing创建于 7月23日
feat: VA上游报错后调Agent生成降级回复，确保SSE流包含final_answer #394
已合并
共 4 个文件变更+178-15
| @@ -375,7 +375,53 @@ class RemoteAgentHandler: | |||
| 375 | }, | 375 | }, |
| 376 | ) | 376 | ) |
| 377 | elif finalized: | 377 | elif finalized: |
| 378 | - pass | 378 | + # VA 上游报错：_finalize_failed 已在 _call_versatile_adapter 内部完成 |
| 379 | + # （标记 FAILED + 清 va_task_id + 入队 FAILED 事件）。 | ||
| 380 | + # 此处调 Agent 生成降级回复，确保 SSE 流包含 final_answer / conversation_end。 | ||
| 381 | + # 防递归：同一个 turn_ctx 只允许一次降级回复，避免 Agent 再次调 VA 失败后无限循环。 | ||
| 382 | + if getattr(turn_ctx, "_degrade_attempted", False): | ||
| 383 | + logger.warning( | ||
| 384 | + f"[RemoteAgentHandler] 降级回复已尝试过，跳过: conv={turn_ctx.conv_id}" | ||
| 385 | + ) | ||
| 386 | + else: | ||
| 387 | + object.__setattr__(turn_ctx, "_degrade_attempted", True) | ||
| 388 | + executor = context.get("executor") | ||
| 389 | + if executor is not None: | ||
| 390 | + query = context.get("query", "") | ||
| 391 | + original_body = context.get("original_body", {}) | ||
| 392 | + step_counter = context.get("step_counter") | ||
| 393 | + logger.info( | ||
| 394 | + f"[RemoteAgentHandler] VA 上游报错，启动 Agent 降级回复: " | ||
| 395 | + f"conv={turn_ctx.conv_id}, task_id={turn_ctx.task_id}, " | ||
| 396 | + f"query={query!r:.60}" | ||
| 397 | + ) | ||
| 398 | + import time as _time | ||
| 399 | + _degrade_start = _time.monotonic() | ||
| 400 | + try: | ||
| 401 | + await executor.run_agent( | ||
| 402 | + turn_ctx, | ||
| 403 | + query=query, | ||
| 404 | + original_body=original_body, | ||
| 405 | + cascade_result={"workflow_result": None, "error": "upstream_failed"}, | ||
| 406 | + run_options={ | ||
| 407 | + "step_counter": step_counter, | ||
| 408 | + "heartbeat_runtime": context.get("heartbeat_runtime"), | ||
| 409 | + }, | ||
| 410 | + ) | ||
| 411 | + _degrade_ms = (_time.monotonic() - _degrade_start) * 1000 | ||
| 412 | + logger.info( | ||
| 413 | + f"[RemoteAgentHandler] Agent 降级回复完成: " | ||
| 414 | + f"conv={turn_ctx.conv_id}, duration={_degrade_ms:.2f}ms" | ||
| 415 | + ) | ||
| 416 | + except Exception as e: | ||
| 417 | + _degrade_ms = (_time.monotonic() - _degrade_start) * 1000 | ||
| 418 | + logger.error( | ||
| 419 | + f"[RemoteAgentHandler] Agent 降级回复失败: " | ||
| 420 | + f"conv={turn_ctx.conv_id}, task_id={turn_ctx.task_id}, " | ||
| 421 | + f"duration={_degrade_ms:.2f}ms, " | ||
| 422 | + f"error_type={type(e).__name__}, error={e}", | ||
| 423 | + exc_info=True, | ||
| 424 | + ) | ||
| 379 | else: | 425 | else: |
| 380 | await self._suspend_task(turn_ctx, va_task_id, context) | 426 | await self._suspend_task(turn_ctx, va_task_id, context) |
| 381 | 427 | ||
| @@ -1880,6 +1926,46 @@ class RemoteAgentHandler: | |||
| 1880 | elif upstream_error is not None: | 1926 | elif upstream_error is not None: |
| 1881 | # VA 续轮也报错：同样落 FAILED + 清空 va_task_id，破解 conv_id 锁死 | 1927 | # VA 续轮也报错：同样落 FAILED + 清空 va_task_id，破解 conv_id 锁死 |
| 1882 | await self._finalize_failed(turn_ctx, upstream_error) | 1928 | await self._finalize_failed(turn_ctx, upstream_error) |
| 1929 | + # 调 Agent 生成降级回复，确保 SSE 流包含 final_answer / conversation_end | ||
| 1930 | + # 防递归：同一个 turn_ctx 只允许一次降级回复 | ||
| 1931 | + if getattr(turn_ctx, "_degrade_attempted", False): | ||
| 1932 | + logger.warning( | ||
| 1933 | + f"[RemoteAgentHandler] 续轮降级回复已尝试过，跳过: conv={conv_id}" | ||
| 1934 | + ) | ||
| 1935 | + else: | ||
| 1936 | + object.__setattr__(turn_ctx, "_degrade_attempted", True) | ||
| 1937 | + executor = (context or {}).get("executor") | ||
| 1938 | + if executor is not None: | ||
| 1939 | + logger.info( | ||
| 1940 | + f"[RemoteAgentHandler] VA 续轮上游报错，启动 Agent 降级回复: " | ||
| 1941 | + f"conv={conv_id}, task_id={task_id}" | ||
| 1942 | + ) | ||
| 1943 | + import time as _time | ||
| 1944 | + _degrade_start = _time.monotonic() | ||
| 1945 | + try: | ||
| 1946 | + await executor.run_agent( | ||
| 1947 | + turn_ctx, | ||
| 1948 | + query="", | ||
| 1949 | + original_body=original_body, | ||
| 1950 | + cascade_result={"workflow_result": None, "error": "upstream_failed"}, | ||
| 1951 | + run_options={ | ||
| 1952 | + "heartbeat_runtime": (context or {}).get("heartbeat_runtime"), | ||
| 1953 | + }, | ||
| 1954 | + ) | ||
| 1955 | + _degrade_ms = (_time.monotonic() - _degrade_start) * 1000 | ||
| 1956 | + logger.info( | ||
| 1957 | + f"[RemoteAgentHandler] 续轮 Agent 降级回复完成: " | ||
| 1958 | + f"conv={conv_id}, duration={_degrade_ms:.2f}ms" | ||
| 1959 | + ) | ||
| 1960 | + except Exception as e: | ||
| 1961 | + _degrade_ms = (_time.monotonic() - _degrade_start) * 1000 | ||
| 1962 | + logger.error( | ||
| 1963 | + f"[RemoteAgentHandler] 续轮 Agent 降级回复失败: " | ||
| 1964 | + f"conv={conv_id}, task_id={task_id}, " | ||
| 1965 | + f"duration={_degrade_ms:.2f}ms, " | ||
| 1966 | + f"error_type={type(e).__name__}, error={e}", | ||
| 1967 | + exc_info=True, | ||
| 1968 | + ) | ||
| 1883 | else: | 1969 | else: |
| 1884 | await self._state_manager.save_input_required( | 1970 | await self._state_manager.save_input_required( |
| 1885 | InputRequiredState( | 1971 | InputRequiredState( |
| @@ -124,7 +124,7 @@ class TaskStateManager: | |||
| 124 | 124 | ||
| 125 | async def finalize_completed(self, task_id: str, call_context: Any = None) -> None: | 125 | async def finalize_completed(self, task_id: str, call_context: Any = None) -> None: |
| 126 | task = await self.get_task(task_id, call_context) | 126 | task = await self.get_task(task_id, call_context) |
| 127 | - if task and task.get("status_state") != "COMPLETED": | 127 | + if task and task.get("status_state") not in ("COMPLETED", "FAILED"): |
| 128 | task["status_state"] = "COMPLETED" | 128 | task["status_state"] = "COMPLETED" |
| 129 | await self.save_task(task_id, task, call_context) | 129 | await self.save_task(task_id, task, call_context) |
| 130 | logger.debug(f"[Executor] Task 标记 COMPLETED：task={task_id}") | 130 | logger.debug(f"[Executor] Task 标记 COMPLETED：task={task_id}") |
| @@ -278,12 +278,67 @@ def test_build_va_message_packs_target_for_workflow_routing(): | |||
| 278 | } | 278 | } |
| 279 | 279 | ||
| 280 | 280 | ||
| 281 | -def test_a2a_message_target_routes_to_va_workflow_adapter(tmp_path): | 281 | +def _load_versatile_runner(): |
| 282 | - va_root = Path(__file__).resolve().parents[3] / "versatile_adapter" | 282 | + """加载 VersatileAdapterRunner，用 append 避免模块重名风险。""" |
| 283 | - if str(va_root) not in sys.path: | 283 | + va_root = str(Path(__file__).resolve().parents[3] / "versatile_adapter") |
| 284 | - sys.path.insert(0, str(va_root)) | 284 | + if va_root not in sys.path: |
| 285 | - | 285 | + sys.path.append(va_root) |
| 286 | from dispatcher.runner import VersatileAdapterRunner | 286 | from dispatcher.runner import VersatileAdapterRunner |
| 287 | + return VersatileAdapterRunner | ||
| 288 | + | ||
| 289 | + | ||
| 290 | +def test_a2a_message_target_routes_to_a2a_gateway_adapter(tmp_path): | ||
| 291 | + """VA_WORKFLOW_ADAPTER_TYPE=a2a_gateway（默认）时，type: a2a_gateway 的 adapter 能被路由匹配。""" | ||
| 292 | + runner_cls = _load_versatile_runner() | ||
| 293 | + | ||
| 294 | + config_path = tmp_path / "versatile_proxy.yaml" | ||
| 295 | + config_path.write_text( | ||
| 296 | + """ | ||
| 297 | +adapters: | ||
| 298 | + - name: default_controller | ||
| 299 | + type: controller | ||
| 300 | + url_template: "http://mock-host/v1/agents/agent-a/conversations/{conversation_id}" | ||
| 301 | + - name: wf_wealth | ||
| 302 | + type: a2a_gateway | ||
| 303 | + a2a_gateway_base: "https://a2a-gateway.example.com" | ||
| 304 | + agent_card_name: "WealthAgent" | ||
| 305 | + url_template: "{a2a_gateway_base}/a2a/{agent_card_name}" | ||
| 306 | + workflow_id: wf_wealth | ||
| 307 | + intent: "理财推荐" | ||
| 308 | +""", | ||
| 309 | + encoding="utf-8", | ||
| 310 | + ) | ||
| 311 | + executor = _make_executor_with_va_stream([]) | ||
| 312 | + request = executor._test_remote_handler._build_va_message( | ||
| 313 | + _VaRequestPayload( | ||
| 314 | + query="查理财", | ||
| 315 | + headers={}, | ||
| 316 | + body={"custom_data": {}}, | ||
| 317 | + params={}, | ||
| 318 | + conv_id=CONV_ID, | ||
| 319 | + target={ | ||
| 320 | + "type": "workflow", | ||
| 321 | + "intent": "理财推荐", | ||
| 322 | + "workflow_id": "wf_wealth", | ||
| 323 | + }, | ||
| 324 | + ) | ||
| 325 | + ) | ||
| 326 | + | ||
| 327 | + data_part = next(p for p in request.message.parts if p.WhichOneof("content") == "data") | ||
| 328 | + target = MessageToDict(data_part.data)["target"] | ||
| 329 | + runner = runner_cls(config_path=config_path) | ||
| 330 | + cfg = runner._match_workflow(target) | ||
| 331 | + | ||
| 332 | + assert cfg is not None | ||
| 333 | + assert cfg.name == "wf_wealth" | ||
| 334 | + | ||
| 335 | + | ||
| 336 | +def test_a2a_message_target_routes_to_workflow_adapter(tmp_path, monkeypatch): | ||
| 337 | + """VA_WORKFLOW_ADAPTER_TYPE=workflow 时，type: workflow 的 adapter 能被路由匹配。""" | ||
| 338 | + runner_cls = _load_versatile_runner() | ||
| 339 | + | ||
| 340 | + # VA_WORKFLOW_ADAPTER_TYPE 默认 a2a_gateway，测试 workflow 模式时改成 workflow | ||
| 341 | + monkeypatch.setenv("VA_WORKFLOW_ADAPTER_TYPE", "workflow") | ||
| 287 | 342 | ||
| 288 | config_path = tmp_path / "versatile_proxy.yaml" | 343 | config_path = tmp_path / "versatile_proxy.yaml" |
| 289 | config_path.write_text( | 344 | config_path.write_text( |
| @@ -318,7 +373,7 @@ adapters: | |||
| 318 | 373 | ||
| 319 | data_part = next(p for p in request.message.parts if p.WhichOneof("content") == "data") | 374 | data_part = next(p for p in request.message.parts if p.WhichOneof("content") == "data") |
| 320 | target = MessageToDict(data_part.data)["target"] | 375 | target = MessageToDict(data_part.data)["target"] |
| 321 | - runner = VersatileAdapterRunner(config_path=config_path) | 376 | + runner = runner_cls(config_path=config_path) |
| 322 | cfg = runner._match_workflow(target) | 377 | cfg = runner._match_workflow(target) |
| 323 | 378 | ||
| 324 | assert cfg is not None | 379 | assert cfg is not None |
| @@ -695,8 +750,14 @@ async def test_va_error_event_enqueues_failed_status_with_message(monkeypatch): | |||
| 695 | va_events = [_va_failed()] | 750 | va_events = [_va_failed()] |
| 696 | executor, _task, _task_store = _make_executor_with_real_task(va_events) | 751 | executor, _task, _task_store = _make_executor_with_real_task(va_events) |
| 697 | 752 | ||
| 753 | + call_count = [0] | ||
| 754 | + | ||
| 698 | async def fake_agent_stream(**kwargs): | 755 | async def fake_agent_stream(**kwargs): |
| 699 | - yield DelegateRequest(intent="查", task_description="查") | 756 | + call_count[0] += 1 |
| 757 | + if call_count[0] == 1: | ||
| 758 | + yield DelegateRequest(intent="查", task_description="查") | ||
| 759 | + else: | ||
| 760 | + yield ConversationEndEvent() | ||
| 700 | 761 | ||
| 701 | monkeypatch.setattr("orchestrator.executor.agent_stream", fake_agent_stream) | 762 | monkeypatch.setattr("orchestrator.executor.agent_stream", fake_agent_stream) |
| 702 | 763 | ||
| @@ -760,8 +821,14 @@ async def test_va_failed_event_with_plain_text_error_is_forwarded(monkeypatch): | |||
| 760 | event.status.message.CopyFrom(message) | 821 | event.status.message.CopyFrom(message) |
| 761 | executor, _task, _task_store = _make_executor_with_real_task([event]) | 822 | executor, _task, _task_store = _make_executor_with_real_task([event]) |
| 762 | 823 | ||
| 824 | + call_count = [0] | ||
| 825 | + | ||
| 763 | async def fake_agent_stream(**kwargs): | 826 | async def fake_agent_stream(**kwargs): |
| 764 | - yield DelegateRequest(intent="查", task_description="查") | 827 | + call_count[0] += 1 |
| 828 | + if call_count[0] == 1: | ||
| 829 | + yield DelegateRequest(intent="查", task_description="查") | ||
| 830 | + else: | ||
| 831 | + yield ConversationEndEvent() | ||
| 765 | 832 | ||
| 766 | monkeypatch.setattr("orchestrator.executor.agent_stream", fake_agent_stream) | 833 | monkeypatch.setattr("orchestrator.executor.agent_stream", fake_agent_stream) |
| 767 | 834 | ||
| @@ -789,8 +856,14 @@ async def test_va_failed_event_without_payload_falls_back_to_generic_message(mon | |||
| 789 | va_events = [_va_failed_event(error_payload=None)] | 856 | va_events = [_va_failed_event(error_payload=None)] |
| 790 | executor, _task, _task_store = _make_executor_with_real_task(va_events) | 857 | executor, _task, _task_store = _make_executor_with_real_task(va_events) |
| 791 | 858 | ||
| 859 | + call_count = [0] | ||
| 860 | + | ||
| 792 | async def fake_agent_stream(**kwargs): | 861 | async def fake_agent_stream(**kwargs): |
| 793 | - yield DelegateRequest(intent="查", task_description="查") | 862 | + call_count[0] += 1 |
| 863 | + if call_count[0] == 1: | ||
| 864 | + yield DelegateRequest(intent="查", task_description="查") | ||
| 865 | + else: | ||
| 866 | + yield ConversationEndEvent() | ||
| 794 | 867 | ||
| 795 | monkeypatch.setattr("orchestrator.executor.agent_stream", fake_agent_stream) | 868 | monkeypatch.setattr("orchestrator.executor.agent_stream", fake_agent_stream) |
| 796 | 869 | ||
| @@ -817,8 +890,8 @@ async def test_va_failed_event_without_payload_falls_back_to_generic_message(mon | |||
| 817 | 890 | ||
| 818 | 891 | ||
| 819 | 892 | ||
| 820 | -async def test_va_failed_event_does_not_trigger_cascade(monkeypatch): | 893 | +async def test_va_failed_event_triggers_downgrade_reply(monkeypatch): |
| 821 | - """VA FAILED 不应被当成成功完成 → 不该触发 cascade 续轮。""" | 894 | + """VA FAILED 现在会触发降级回复（agent_stream 被调用第二次）。""" |
| 822 | va_events = [_va_failed_event({"code": "103104", "message": "错"})] | 895 | va_events = [_va_failed_event({"code": "103104", "message": "错"})] |
| 823 | executor, _task, _task_store = _make_executor_with_real_task(va_events) | 896 | executor, _task, _task_store = _make_executor_with_real_task(va_events) |
| 824 | 897 | ||
| @@ -826,7 +899,10 @@ async def test_va_failed_event_does_not_trigger_cascade(monkeypatch): | |||
| 826 | 899 | ||
| 827 | async def fake_agent_stream(**kwargs): | 900 | async def fake_agent_stream(**kwargs): |
| 828 | call_count[0] += 1 | 901 | call_count[0] += 1 |
| 829 | - yield DelegateRequest(intent="查", task_description="查") | 902 | + if call_count[0] == 1: |
| 903 | + yield DelegateRequest(intent="查", task_description="查") | ||
| 904 | + else: | ||
| 905 | + yield ConversationEndEvent() | ||
| 830 | 906 | ||
| 831 | monkeypatch.setattr("orchestrator.executor.agent_stream", fake_agent_stream) | 907 | monkeypatch.setattr("orchestrator.executor.agent_stream", fake_agent_stream) |
| 832 | 908 | ||
| @@ -838,4 +914,4 @@ async def test_va_failed_event_does_not_trigger_cascade(monkeypatch): | |||
| 838 | cascade_result=None, | 914 | cascade_result=None, |
| 839 | ) | 915 | ) |
| 840 | 916 | ||
| 841 | - assert call_count[0] == 1, "VA FAILED 路径不应触发 cascade（agent_stream 不应被调用第二次）" | 917 | + assert call_count[0] == 2, "VA FAILED 现在会触发降级回复（agent_stream 被调用第二次）" |
| @@ -309,6 +309,7 @@ class A2aVersatileExecutor(AgentExecutor): | |||
| 309 | return { | 309 | return { |
| 310 | "trace_id": input_data.get("trace_id", ""), | 310 | "trace_id": input_data.get("trace_id", ""), |
| 311 | "conv_id": conv_id, | 311 | "conv_id": conv_id, |
| 312 | + "agent_id": input_data.get("agent_id", ""), | ||
| 312 | } | 313 | } |
| 313 | 314 | ||
| 314 | 315 | ||