chendandan_homework/04_work/app/graph/planner_nodes.py

from __future__ import annotations
"""行程规划节点：调用ItineraryPlannerAgent根据候选资源和修改意见生成多日行程。"""

from typing import Any, Literal

from app.agents.itinerary_agent import (
    ItineraryPlannerAgent,
)
from app.graph.state import TravelState
from app.schemas.itinerary import ItineraryPlan


def route_after_route_evaluation(
    state: TravelState,
) -> Literal[
    "plan_itinerary",
    "end",
]:
    """路线评估成功后进入行程规划。"""

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

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

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

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

    return "plan_itinerary"


def make_itinerary_planner_node(
    agent: ItineraryPlannerAgent,
):
    """创建行程规划角色节点。"""

    async def itinerary_planner_node(
        state: TravelState,
    ) -> dict[str, Any]:
        """行程规划节点：调用Agent生成ItineraryPlan，必要时合并revision_feedback进行重规划。"""

        attempts = (
            state.get("planning_attempts", 0)
            + 1
        )

        try:
            result = await agent.plan(
                request=state["travel_request"],
                selection=state[
                    "candidate_selection_result"
                ],
                map_result=state[
                    "map_research_result"
                ],
                route_result=state[
                    "route_evaluation_result"
                ],
                revision_feedback=state.get(
                    "revision_feedback",
                    [],
                ),
            )

            structured_response = result.get(
                "structured_response"
            )

            if structured_response is None:
                message = (
                    "行程规划Agent没有返回"
                    "结构化结果。"
                )

                return {
                    "planning_attempts": attempts,
                    "planner_messages": result.get(
                        "messages",
                        [],
                    ),
                    "errors": [message],
                    "final_answer": message,
                }

            if not isinstance(
                structured_response,
                ItineraryPlan,
            ):
                structured_response = (
                    ItineraryPlan.model_validate(
                        structured_response
                    )
                )

            day_count = len(
                structured_response.days
            )

            activity_count = sum(
                len(day.activities)
                for day
                in structured_response.days
            )

            return {
                "itinerary_plan": (
                    structured_response
                ),
                "planner_messages": result.get(
                    "messages",
                    [],
                ),
                "planning_attempts": attempts,
                "final_answer": (
                    "行程规划角色执行完成："
                    f"生成{day_count}天行程，"
                    f"共安排{activity_count}项活动；"
                    "等待预算和规则校验。"
                ),
            }

        except Exception as exc:
            message = (
                "行程规划Agent执行失败："
                f"{type(exc).__name__}: {exc}"
            )

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

            existing_errors.append(message)

            return {
                "planning_attempts": attempts,
                "errors": existing_errors,
                "final_answer": message,
            }

    return itinerary_planner_node