chendandan_homework/scripts/test_final_graph.py

from __future__ import annotations

"""测试完整多角色工作流端到端流程：从需求输入到最终旅行方案输出。"""

import asyncio

from rich.console import Console

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月14日返回，"
                "总预算5000元。"
                "喜欢熊猫、自然景观和川菜，"
                "希望行程轻松一些。"
                "酒店每晚不超过300元，"
                "评分不低于3，"
                "并且希望靠近地铁。"
                "机票价格和便利性综合考虑。"
            ),
            "errors": [],
            "missing_fields": [],
            "planning_attempts": 0,
            "review_attempts": 0,
            "revision_feedback": [],
            "workflow_status": "running",
        },
        config={
            "recursion_limit": 80,
        },
    )

    console.rule(
        "[bold]最终多角色工作流"
    )

    console.print(
        {
            "工作流状态": result.get(
                "workflow_status"
            ),
            "Supervisor决策": result.get(
                "supervisor_decision"
            ),
            "Planner次数": result.get(
                "planning_attempts"
            ),
            "Reviewer次数": result.get(
                "review_attempts"
            ),
            "校验通过": result.get(
                "validation_passed"
            ),
            "审查通过": result.get(
                "review_passed"
            ),
            "系统错误": result.get(
                "errors"
            ),
        }
    )

    console.rule("[bold]最终旅行方案")

    console.print(
        result.get(
            "final_answer",
            "没有最终回答。",
        )
    )


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