chendandan_homework/app/graph/supervisor_nodes.py

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"