chendandan_homework/scripts/test_validation_graph.py

from __future__ import annotations

"""测试方案校验阶段子图：验证确定性规则校验与校验失败后的路由。"""

import asyncio

from rich.console import Console
from rich.table import Table

from app.graph.planner_builder import (
    build_planner_graph,
)


console = Console()


async def main() -> None:
    graph = await build_planner_graph()

    result = await graph.ainvoke(
        {
            "user_query": (
                "我和妻子计划2026年8月10日"
                "从上海去成都，8月13日返回，"
                "总预算8000元。"
                "喜欢熊猫、自然景观和川菜，"
                "希望行程轻松一些。"
                "酒店每晚不超过600元，"
                "评分不低于4.5，"
                "并且希望靠近地铁。"
                "机票价格和便利性综合考虑。"
            ),
            "errors": [],
            "missing_fields": [],
            "planning_attempts": 0,
            "revision_feedback": [],
        }
    )

    console.rule(
        "[bold]Validator工作流结果"
    )

    console.print(result["final_answer"])

    validation = result.get(
        "plan_validation_result"
    )

    if validation is None:
        console.print(
            {
                "系统错误": result.get("errors"),
            }
        )
        return

    console.print(
        {
            "校验通过": validation.is_valid,
            "错误数": validation.error_count,
            "警告数": validation.warning_count,
            "已知费用": (
                validation.budget.known_total
            ),
            "用户预算": (
                validation.budget.user_budget
            ),
            "预算状态": (
                validation.budget.status
            ),
            "已知预算余额": (
                validation
                .budget
                .remaining_known_budget
            ),
        }
    )

    issue_table = Table(
        title="确定性校验问题"
    )

    issue_table.add_column("级别")
    issue_table.add_column("代码")
    issue_table.add_column("日期序号")
    issue_table.add_column("活动序号")
    issue_table.add_column("问题")

    for issue in validation.issues:
        issue_table.add_row(
            issue.severity,
            issue.code,
            (
                str(issue.day_index)
                if issue.day_index is not None
                else ""
            ),
            (
                str(issue.activity_sequence)
                if issue.activity_sequence
                is not None
                else ""
            ),
            issue.message,
        )

    console.print(issue_table)

    console.rule("[bold]预算覆盖说明")

    console.print(
        validation.budget.coverage_note
    )

    if validation.revision_feedback:
        console.rule(
            "[bold yellow]需要Planner修改"
        )

        for feedback in (
            validation.revision_feedback
        ):
            console.print(f"- {feedback}")


if __name__ == "__main__":
    asyncio.run(main())