Pydantic-AI 中实时获取AI交互中间消息的技术方案

在基于Pydantic-AI框架开发AI应用时，开发者经常需要获取AI模型与工具交互过程中的中间消息。这些消息对于构建流畅的用户体验至关重要，比如在工具调用前向用户展示AI的意图说明。

核心需求场景

当使用Pydantic-AI的Agent工具时，标准的运行流程会直接返回最终结果。但在实际业务中，我们往往需要：

1. 在工具调用前展示AI的决策过程
2. 实时反馈AI的思考步骤
3. 记录完整的交互历史用于审计或调试

解决方案实现

Pydantic-AI提供了两种主要方式来处理中间消息：

1. 使用迭代器模式处理消息流

通过agent.iter()方法可以逐步处理交互过程中的每个节点：

async def main():
    history = None
    while True:
        user_input = input("用户输入>")
        async with agent.iter(user_input, message_history=history) as r:
            async for node in r:
                if agent.is_call_tools_node(node):
                    for part in node.model_response.parts:
                        if part.part_kind == "text":
                            print("AI回复>", part.content)
            history = r.result.all_messages()

这种方法特别适合需要实时交互的场景，开发者可以在每个节点处理时执行自定义逻辑。

2. 自定义Agent实现

对于需要更精细控制的场景，可以继承基础Agent类实现自定义逻辑：

class CustomAgent(Agent):
    async def process_node(self, node):
        # 自定义节点处理逻辑
        if self.is_text_response(node):
            await self.stream_to_ui(node.content)
        await super().process_node(node)

消息结构解析

Pydantic-AI的消息交互遵循标准结构，主要包含以下几种类型：

1. 用户请求消息：包含用户原始输入
2. 模型响应消息：包含AI的文本回复和工具调用
3. 工具返回消息：记录工具执行结果
4. 最终响应消息：包含处理完成的最终输出

每种消息都包含时间戳、内容类型等元数据，便于开发者追踪完整的交互流程。

最佳实践建议

1. 对于简单场景，直接使用iter()方法即可满足需求
2. 复杂交互系统建议实现自定义Agent类
3. 生产环境应持久化存储完整消息历史
4. 注意处理消息中的多种内容类型（文本、工具调用等）

通过合理利用这些技术，开发者可以构建出交互更加自然、透明的AI应用系统。