chendandan_homework/scripts/test_reviewer_graph.py

from __future__ import annotations

"""测试Trip Reviewer阶段子图：验证行程独立评审与修改意见生成。"""

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,
            "review_attempts": 0,
            "revision_feedback": [],
        },
        config={
            "recursion_limit": 50,
        },
    )

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

    console.print(result["final_answer"])

    validation = result.get(
        "plan_validation_result"
    )

    review = result.get(
        "trip_review_result"
    )

    console.print(
        {
            "Planner执行次数": result.get(
                "planning_attempts"
            ),
            "Reviewer执行次数": result.get(
                "review_attempts"
            ),
            "确定性校验通过": (
                validation.is_valid
                if validation
                else None
            ),
            "Reviewer原始结论": (
                review.approved
                if review
                else None
            ),
            "系统最终审查结论": result.get(
                "review_passed"
            ),
        }
    )

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

    console.rule("[bold]审查总结")
    console.print(review.summary)

    issue_table = Table(
        title="Reviewer发现的问题"
    )

    issue_table.add_column("级别")
    issue_table.add_column("分类")
    issue_table.add_column("日期")
    issue_table.add_column("活动")
    issue_table.add_column("问题")
    issue_table.add_column("建议")

    for issue in review.issues:
        issue_table.add_row(
            issue.severity,
            issue.category,
            (
                str(issue.day_index)
                if issue.day_index
                else ""
            ),
            issue.activity_name or "",
            issue.message,
            issue.suggestion or "",
        )

    console.print(issue_table)

    if review.strengths:
        console.rule("[bold green]方案优点")

        for strength in review.strengths:
            console.print(f"- {strength}")

    if result.get("revision_feedback"):
        console.rule(
            "[bold yellow]下一轮修改要求"
        )

        for feedback in result[
            "revision_feedback"
        ]:
            console.print(f"- {feedback}")

    if review.recommendations:
        console.rule("[bold]非阻塞建议")

        for recommendation in (
            review.recommendations
        ):
            console.print(
                f"- {recommendation}"
            )


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