chendandan_homework/app/graph/validation_nodes.py

from __future__ import annotations
"""方案校验节点：调用PlanValidationService对行程做确定性规则校验，验证航班/酒店ID真实性与预算合理性。"""

from typing import Any, Literal

from app.graph.state import TravelState
from app.services.plan_validation_service import (
    PlanValidationService,
)


def route_after_planning(
    state: TravelState,
) -> Literal[
    "validate_plan",
    "end",
]:
    """Planner成功后进入确定性校验。"""

    if state.get("errors"):
        return "end"

    if not state.get("itinerary_plan"):
        return "end"

    return "validate_plan"


def make_plan_validation_node(
    service: PlanValidationService
    | None = None,
):
    """创建确定性行程校验节点。"""

    validation_service = (
        service
        or PlanValidationService()
    )

    async def plan_validation_node(
        state: TravelState,
    ) -> dict[str, Any]:
        """方案校验节点：用确定性规则校验行程方案，将校验结果和预算估算写入state。"""

        try:
            result = validation_service.validate(
                request=state["travel_request"],
                plan=state["itinerary_plan"],
                selection=state[
                    "candidate_selection_result"
                ],
                map_result=state[
                    "map_research_result"
                ],
                route_result=state[
                    "route_evaluation_result"
                ],
            )

            warnings = [
                issue.message
                for issue in result.issues
                if issue.severity == "warning"
            ]

            if result.is_valid:
                message = (
                    "确定性校验通过："
                    f"发现{result.warning_count}项"
                    "非阻塞警告；"
                    "等待Trip Reviewer审查。"
                )
            else:
                message = (
                    "确定性校验未通过："
                    f"发现{result.error_count}项错误、"
                    f"{result.warning_count}项警告；"
                    "需要Planner修改方案。"
                )

            # 注意：
            # 业务校验失败不能写入全局errors，
            # 否则LangGraph会立即终止，
            # 后续无法进入重新规划循环。
            return {
                "plan_validation_result": result,
                "validation_passed": (
                    result.is_valid
                ),
                "validation_warnings": warnings,
                "revision_feedback": (
                    result.revision_feedback
                ),
                "final_answer": message,
            }

        except Exception as exc:
            message = (
                "确定性校验节点执行失败："
                f"{type(exc).__name__}: {exc}"
            )

            existing_errors = list(
                state.get("errors", [])
            )

            existing_errors.append(message)

            return {
                "errors": existing_errors,
                "validation_passed": False,
                "final_answer": message,
            }

    return plan_validation_node