from __future__ import annotations """监督协调节点:汇总Validator和Reviewer的意见,根据校验结果与重试次数决定replan/finalize/terminate去向。""" from typing import Any, Literal from app.graph.state import TravelState SupervisorRoute = Literal[ "replan", "finalize", "finalize_with_risks", "terminate", ] def _unique_messages( values: list[str], ) -> list[str]: """去除空字符串和重复修改意见。""" result: list[str] = [] for value in values: normalized = value.strip() if normalized and normalized not in result: result.append(normalized) return result def collect_revision_feedback( state: TravelState, ) -> list[str]: """收集Validator与Reviewer提出的修改意见。""" feedback = list( state.get("revision_feedback", []) ) validation = state.get( "plan_validation_result" ) if validation is not None: feedback.extend( validation.revision_feedback ) for issue in validation.issues: if ( issue.severity == "error" and issue.message ): feedback.append(issue.message) review = state.get("trip_review_result") if review is not None: feedback.extend( review.revision_feedback ) for issue in review.issues: if ( issue.severity == "blocking" and issue.message ): feedback.append( issue.suggestion or issue.message ) return _unique_messages(feedback) def make_supervisor_node( *, max_planning_attempts: int = 3, ): """创建多角色工作流协调节点。 max_planning_attempts=3表示: 1次初始规划 + 最多2次修改。 """ if max_planning_attempts < 1: raise ValueError( "max_planning_attempts必须大于等于1。" ) def supervisor_node( state: TravelState, ) -> dict[str, Any]: """监督节点:基于校验与评审结果及当前尝试次数,决策工作流的下一步——继续重规划、定稿(含风险)或终止。""" system_errors = state.get( "errors", [], ) if system_errors: return { "supervisor_decision": "terminate", "supervisor_reason": ( "工作流存在系统级错误," "不能继续重新规划。" ), "workflow_status": "failed", } validation = state.get( "plan_validation_result" ) review = state.get( "trip_review_result" ) if validation is None or review is None: return { "supervisor_decision": "terminate", "supervisor_reason": ( "Supervisor缺少Validator或" "Reviewer的结构化结果。" ), "workflow_status": "failed", "errors": [ "Supervisor输入状态不完整。" ], } validation_passed = validation.is_valid review_passed = bool( state.get("review_passed", False) ) planning_attempts = state.get( "planning_attempts", 0, ) if validation_passed and review_passed: return { "supervisor_decision": "finalize", "supervisor_reason": ( "确定性校验和独立审查均已通过。" ), "workflow_status": "approved", "revision_feedback": [], "final_answer": ( "Supervisor批准当前方案," "进入最终输出阶段。" ), } feedback = collect_revision_feedback( state ) if not feedback: feedback = [ "重新检查方案与用户需求、" "确定性校验结果和审查意见," "修复所有阻塞问题。" ] if planning_attempts < max_planning_attempts: next_attempt = planning_attempts + 1 return { "supervisor_decision": "replan", "supervisor_reason": ( "当前方案未通过,且仍有" "剩余重新规划次数。" ), "workflow_status": "running", "revision_feedback": feedback, "validation_passed": False, "review_passed": False, "final_answer": ( "Supervisor要求重新规划:" f"即将执行第{next_attempt}次规划。" ), } return { "supervisor_decision": ( "finalize_with_risks" ), "supervisor_reason": ( "方案仍存在问题,但已经达到" f"最大规划次数" f"{max_planning_attempts}次。" ), "workflow_status": ( "completed_with_risks" ), "revision_feedback": feedback, "final_answer": ( "已达到最大重新规划次数," "系统将保留风险说明并输出" "当前最佳方案。" ), } return supervisor_node def route_after_supervisor( state: TravelState, ) -> SupervisorRoute: """根据Supervisor决策选择后续路径。""" decision = state.get( "supervisor_decision", "terminate", ) if decision in { "replan", "finalize", "finalize_with_risks", "terminate", }: return decision return "terminate"