在本文中，我们将探讨如何使用 PydanticAI和LLM（如DeepSeek R1或V3）构建一个简单的航班预订助手。该助手通过多个 AI 智能体的协作，帮助用户根据其出发地和目的地寻找航班，并根据用户的偏好选择座位。我们将从需求描述开始，逐步分析需求、设计方案，并最终实现具体的代码。

需求描述

本文的目标是开发一个交互式系统，满足以下需求：

• 询问用户出发地和目的地：系统需要从用户那里获取航班的出发地和目的地信息。
• 查找合适的航班：利用 AI 智能体根据用户输入查找匹配的航班。
• 询问用户座位偏好：系统需要向用户询问其座位选择，例如具体行号和座位字母。
• 提取并验证座位偏好：利用另一个 AI 智能体解析并验证用户的座位选择，确保符合飞机座位布局的约束。
• 输出结果：最终输出航班信息和座位选择。

分析

为了实现上述需求，我们需要将任务分解为以下几个部分：

1. 航班搜索：
   需要一个智能体，能够理解用户的出发地和目的地请求。
   该智能体需要调用工具（例如 API 或网页抓取）来查找航班信息。
   应处理可能的失败情况，例如找不到合适的航班。
2. 座位选择：
   需要另一个智能体，能够解析用户的座位偏好请求。
   应验证座位选择的有效性，例如行号是否在合理范围内，座位字母是否合法。
   提供座位相关信息，例如哪些座位是靠窗座位，哪些行有额外腿部空间。
3. 任务协调：
   需要一个机制来协调这两个智能体，按顺序处理用户输入。
   确保系统能够处理多次尝试（例如，当用户输入无效时）和 AI 模型的使用限制（如请求次数限制）。

方案设计

我们将使用 pydantic_ai 库来构建 AI 智能体，并结合 Pydantic 模型来定义数据结构。系统将包含以下组件：

1. 智能体设计：
   flight_search_agent：负责处理航班搜索任务，利用 flight_search 工具查找航班。
   seat_preference_agent：负责解析和验证用户的座位偏好。
   每个智能体都使用 OpenAI 的 GPT-4o 模型，并通过系统提示（system prompt）明确其职责。
2. 数据模型：
   使用 Pydantic 定义结构化的数据模型，如 FlightDetails（航班信息）、SeatPreference（座位偏好）和 Failed（失败情况）。
   这些模型将用于验证数据并确保输出符合预期格式。
3. 任务协调：
   在异步的 main 函数中协调智能体的工作流程：
   首先调用 find_flight 函数查找航班。
   如果找到航班，则调用 find_seat 函数获取座位偏好。
   使用 Usage 和 UsageLimits 类管理 AI 模型的使用限制，例如请求次数限制。
4. 失败处理：
   如果智能体无法完成任务（例如找不到航班或无法解析座位偏好），系统将提示用户重新尝试。
   对于航班搜索，允许最多三次尝试；对于座位选择，则持续尝试直到成功。
5. 流程图

具体实现

from typing import Literal, Union

from pydantic import BaseModel, Field
from rich.prompt import Prompt

from pydantic_ai import Agent, RunContext
from pydantic_ai.messages import ModelMessage
from pydantic_ai.usage import Usage, UsageLimits


class FlightDetails(BaseModel):
    flight_number: str


class Failed(BaseModel):
    """无法找到满意的选项。"""


flight_search_agent = Agent[None, Union[FlightDetails, Failed]](
    'deepseek:r1',  # 使用 DeepSeek R1 模型
    result_type=Union[FlightDetails, Failed],  # 返回类型为 FlightDetails 或 Failed
    system_prompt=(
        '使用 "flight_search" 工具查找从给定出发地到目的地的航班。'
    ),
)


@flight_search_agent.tool
async def flight_search(
    ctx: RunContext[None], origin: str, destination: str
) -> Union[FlightDetails, None]:
    # 在实际应用中，这里会调用航班查询API或从数据库中检索信息
    return FlightDetails(flight_number='AK456')


usage_limits = UsageLimits(request_limit=15)  # 限制请求次数


async def find_flight(usage: Usage) -> Union[FlightDetails, None]:
    message_history: Union[list[ModelMessage], None] = None
    for _ in range(3):
        prompt = Prompt.ask(
            '您想从哪里出发，到哪里去？',
        )
        result = await flight_search_agent.run(
            prompt,
            message_history=message_history,
            usage=usage,
            usage_limits=usage_limits,
        )
        if isinstance(result.data, FlightDetails):
            return result.data
        else:
            message_history = result.all_messages(
                result_tool_return_content='请再试一次。'
            )
    return None


class SeatPreference(BaseModel):
    row: int = Field(ge=1, le=30)
    seat: Literal['A', 'B', 'C', 'D', 'E', 'F']


seat_preference_agent = Agent[None, Union[SeatPreference, Failed]](
    'deepseek:r1',  # 使用 DeepSeek R1 模型
    result_type=Union[SeatPreference, Failed],  # 返回类型为 SeatPreference 或 Failed
    system_prompt=(
        "提取用户的座位偏好。A 和 F 是靠窗座位。第 1 排是前排，有额外的腿部空间。第 14 排和第 20 排也有额外的腿部空间。"
    ),
)


async def find_seat(usage: Usage) -> SeatPreference:
    message_history: Union[list[ModelMessage], None] = None
    while True:
        answer = Prompt.ask('您希望选择哪个座位？')
        result = await seat_preference_agent.run(
            answer,
            message_history=message_history,
            usage=usage,
            usage_limits=usage_limits,
        )
        if isinstance(result.data, SeatPreference):
            return result.data
        else:
            print('无法理解座位偏好，请重新输入。')
            message_history = result.all_messages()


async def main():
    usage: Usage = Usage()

    opt_flight_details = await find_flight(usage)
    if opt_flight_details is not None:
        print(f'找到航班：{opt_flight_details.flight_number}')
        seat_preference = await find_seat(usage)
        print(f'座位偏好：{seat_preference}')
    else:
        print("未能找到航班。")

小结

在本文中，我们展示了如何使用 PydanticAI框架和大模型构建一个基于多智能体协作的航班预订助手。通过将任务分解为航班搜索和座位选择两个独立的部分，并为每个部分设计专门的智能体，我们创建了一个模块化且易于维护的系统。PydanticAI模型确保数据的结构化和验证，DeepSeek R1则作为大脑完成自然语言的理解和决策。

需要注意的是，本示例是一个简化版本。在实际应用中，你需要：

• 集成真实的航班搜索 API。
• 处理更复杂的用户交互。
• 更健壮地管理错误和异常。

尽管如此，本示例展示了PydanticAI和DeepSeek R1等大模型结合的强大潜力，为最终实现通用型智能化助手提供了良好的起点。