Pydantic-AI项目中实现人机交互循环的技术方案解析

在基于Pydantic-AI构建智能对话系统时，实现有效的人机交互循环（Human-in-the-loop）是一个关键需求。本文将深入探讨如何在FastAPI应用中结合Pydantic Agent和Pydantic Graph实现这一机制。

核心架构设计

典型的对话系统状态管理采用以下数据结构：

@dataclass(kw_only=True)
class APPState:
    run_record_id: Optional[int] = None
    user_input: Optional[str] = None
    assistant_reply: Optional[str] = None
    messages: List[ModelMessage] = field(default_factory=list)

这种设计实现了对话状态的持久化，使得系统能够在多个HTTP请求之间保持上下文连续性。

两种实现模式对比

1. 直接输入模式

在CLI环境中，可以通过Python内置的input()函数直接获取用户输入。这种模式简单直接，但在Web环境中存在明显局限：

• 无法中断HTTP请求等待用户响应
• 缺乏状态持久化机制
• 不适合异步交互场景

2. 输出类型判别模式

更成熟的方案是将人机交互设计为输出类型判别机制：

class Ask(BaseModel):
    question: str

class Answer(BaseModel):
    answer: str

agent = Agent("gpt-4o", output_type=Union[Ask, Answer])

这种设计实现了：

• 动态决策：由LLM自主决定何时需要人工介入
• 状态保持：通过消息历史记录维持对话上下文
• 流程控制：区分中间询问和最终答复两种输出状态

Web环境实现要点

在FastAPI等Web框架中实现时需注意：

1. 对话状态管理

   • 使用唯一会话ID标识对话流程
   • 数据库存储消息历史记录
   • 支持对话断点续传

2. 执行流程控制

if is_continued_conversation():
    message_history = get_message_history()
    result = agent.run_sync(question_response, message_history)
else:
    result = agent.run_sync(initial_message)

3. 响应处理逻辑

if isinstance(output, Ask):
    # 存储当前状态，等待用户响应
else:
    # 返回最终答案

高级应用场景

对于复杂业务流程，可以结合Pydantic Graph实现更精细的控制：

• 使用持久化机制保存图状态
• 设计专用节点处理人工输入
• 实现多步骤工作流的暂停与恢复

最佳实践建议

1. 合理设计状态数据结构，确保包含必要元信息
2. 实现健壮的错误处理机制
3. 考虑添加超时和会话过期机制
4. 对于敏感操作建议强制人机验证

这种架构设计既保持了LLM的自主决策能力，又确保了关键环节的人工控制，是构建可靠AI应用的理想选择。