Pydantic-AI 中 Agent 工具调用流程控制的深度解析

背景与核心问题

在 Pydantic-AI 框架中，Agent 的工具调用机制是一个关键功能。开发者经常需要精细控制 Agent 在执行工具后的行为流程，比如在某些工具执行后立即终止流程，或者对工具返回结果进行预处理后再继续执行。

当前解决方案分析

Pydantic-AI 目前提供了 Agent.iter 方法来实现对 Agent 执行流程的细粒度控制。通过这个方法，开发者可以手动遍历 Agent 的执行图，并在每个节点做出决策：

async with agent.iter(...) as run:
    node = run.next_node
    while not isinstance(node, End_):
        if Agent_.is_call_tools_node(node):
            # 处理工具调用
            ...
        node = await run.next(node)

这种方法虽然灵活，但在处理复杂场景时（如并行工具调用、重试机制等）会变得相当复杂。

技术实现细节

在底层实现上，Pydantic-AI 的 Agent 执行流程由多个节点组成：

• 工具调用节点（CallToolsNode）
• 结束节点（End_）
• 其他处理节点

每个节点的执行都会产生新的消息，这些消息可以通过 node.request 和 node.model_response 获取，并最终被添加到 message_history 中。

最佳实践建议

对于需要精细控制工具调用后流程的场景，推荐以下模式：

1. 简单控制：使用 Agent.run 方法获取完整执行结果
2. 中等复杂度：使用 Agent.iter 遍历执行图
3. 高级场景：结合自定义工具实现和流程控制

例如，对于有副作用的工具调用：

@agent.tool_plain
async def send_document(document_id: str) -> str:
    document = await get_document(document_id)
    if not document:
        raise ModelRetry(f"Invalid Document ID: {document_id}")
    
    await send_document_to_user(document)  # 副作用操作
    return f"Sent: {document.name}"  # 保留在历史记录中

这种模式确保了：

• 副作用立即执行
• 返回信息保留在对话历史中
• 错误情况自动触发重试

未来发展方向

虽然当前方案已经提供了足够的灵活性，但在以下方面仍有优化空间：

1. 更简洁的流程控制API
2. 内置的并行工具调用管理
3. 更智能的错误处理和重试机制

开发者社区可以继续探索这些方向，为 Pydantic-AI 贡献更强大的流程控制能力。