ai-python-streaming-tools 属于 Vercel Labs 的 AI SDK for Python，用于在工具执行需要展示进度的场景下使用异步生成器（async-generator）形式的工具。流式工具会产出多个值，但智能体在下一轮模型调用时仍需要一个最终的单一工具结果，返回类型决定了 SDK 如何合并这些产出值。
主要能力
文本块（Text Chunks）
当产出的字符串需要拼接为最终工具结果时，使用 ai.StreamingTextTool。示例：
@ai.tool async def draft_reply(topic: str) -> ai.StreamingTextTool: yield "Checking records for " yield topic
调用方将每次产出视为 ai.events.PartialToolCallResult，模型之后看到的是一个字符串，例如 "Checking records for {topic}"。
先进度后结果（Progress Then Result）
当最后一个产出值即为工具结果时，使用 ai.StreamingStatusTool[T]。示例：
@ai.tool async def ask_mothership(question: str) -> ai.StreamingStatusTool[str]: yield "connecting" yield "transmitting" yield f"The mothership says: {question} is under review."
进度值会流式传给调用方，而模型只看到最后产出的那个值。
子智能体（Subagents）
当工具需要运行另一个智能体并流式转发其事件时，使用 ai.SubAgentTool。示例：
@ai.tool async def research(topic: str) -> ai.SubAgentTool: researcher = ai.Agent() messages = [ ai.system_message("Research briefly."), ai.user_message(topic), ] async with researcher.run(model, messages) as stream: async for event in stream: yield event
父级调用方通过 PartialToolCallResult.value 接收嵌套事件：
async with agent.run(model, messages) as stream: async for event in stream: if isinstance(event, ai.events.PartialToolCallResult): if isinstance(event.value, ai.events.TextDelta): yield event.value.chunk
SubAgentTool 会把带类型的 MessageBundle 存为工具结果。父级模型看到的是嵌套智能体的最终助手文本，而不是原始 bundle。
保存历史记录时，需要保留带类型的消息数据：
data = message.model_dump(mode="json") message = ai.messages.Message.model_validate(data)
不要将 MessageBundle 字符串化，也不要丢弃 result_kind。
自定义聚合（Custom Aggregation）
优先使用上述别名。如需自定义聚合，可使用 @ai.tool(aggregator=...) 或 Annotated 返回类型，但不要同时使用两者。
from collections.abc import AsyncGenerator from typing import Annotated
JoinedLines = Annotated[ AsyncGenerator[str], ai.agents.Aggregate(ai.agents.ConcatAggregator, delim="\n"), ]
@ai.tool async def outline(topic: str) -> JoinedLines: yield f"# {topic}" yield "- first point"
自定义聚合器需实现 ai.events.Aggregator。
规则
- 流式工具必须是异步生成器。
- 每个流式工具都需要一个聚合器，通常来自返回类型别名。
- 通过 ai.events.PartialToolCallResult 消费实时输出。
- 最终聚合值会作为普通工具结果发送回模型。
- 在自定义智能体循环中，需保持 ToolRunner 事件持续流动，否则部分工具输出无法到达调用方。
适用场景与限制
该模块适用于需要在工具执行过程中向用户或调用方展示进度或中间结果的 Python AI 应用，例如研究、长篇生成、嵌套智能体调用等。使用时应遵循上述规则，尤其是必须使用异步生成器、必须提供聚合器，以及在自定义循环中保持事件流动。关于收费、价格、支持平台等具体信息，资料中暂未提供。