主动追问¶

用户让 Agent 写报告，却没有说明目标读者。Agent 可以先提问，拿到答案后再继续。用 SDK 注册一个自定义 ask_user 工具即可实现。

提问并等待回答¶

完成模型配置，运行：

python examples/runtime/ask_user.py

"""最小追问：等待回答；输入 /skip 跳过。"""

import asyncio
import os

import bumblehive


async def read_answer(question: str, options: list[str] | None = None) -> dict:
    print(f"\n问题：{question}")
    choices = options or []
    for index, option in enumerate(choices, 1):
        print(f"  {index}. {option}")
    while True:
        answer = (await asyncio.to_thread(
            input, "输入答案或选项编号；/skip 跳过："
        )).strip()
        if answer == "/skip":
            return {"status": "skipped"}
        if not answer:
            continue
        if answer.isdecimal() and 1 <= int(answer) <= len(choices):
            answer = choices[int(answer) - 1]
        return {"status": "answered", "answer": answer}


async def main() -> None:
    config = bumblehive.RuntimeArguments(
        model=os.environ["BUMBLEHIVE_MODEL"],
        api_key=os.environ["BUMBLEHIVE_API_KEY"],
        base_url=os.environ["BUMBLEHIVE_BASE_URL"],
        skill_names=[],
        tool_names=["ask_user"],
    )
    async with bumblehive.from_config(config) as runtime:
        @runtime.tools.tool(
            description=(
                "缺少无法从上下文确定的关键信息时，向用户提出一个简短问题。"
                "可提供建议选项，用户也能自由输入。"
                "answered 包含回答；skipped 表示用户跳过，不要重复提问或当作同意。"
            ),
        )
        async def ask_user(question: str, options: list[str] | None = None) -> dict:
            return await read_answer(question, options)

        result = await runtime.run(
            "帮我写一段介绍 Python 的开场白。请先用 ask_user 询问目标读者，"
            "收到回答后再写；如果用户跳过，就写一个通用版本。"
        )
        print(result.error.message if result.error else result.final_content)


if __name__ == "__main__":
    asyncio.run(main())

工具描述告诉模型什么时候提问；read_answer() 接收用户输入；工具把答案返回给模型。本例一直等待，直到用户回答或输入 /skip。

运行效果¶

问题：这段 Python 开场白面向哪些读者？
  1. 没有编程经验的初学者
  2. 有其他语言经验的开发者
输入答案或选项编号；/skip 跳过：1
…Agent 为初学者写出开场白…

问题和选项由模型生成，实际措辞可能不同。输入编号会选择对应答案，也可以直接输入任何文字。不传 options 时只显示输入提示。输入 /skip 返回 skipped，Agent 按任务要求继续或说明缺少什么信息，不把跳过当成同意。

用户没有立即回答怎么办¶

不希望一直等待时，可以设置等待时长。这里复用工具的等待与后台执行，不需要另写一套超时机制。

从仓库根目录运行以下进阶示例。它会导入上面的 read_answer() 和公共示例辅助代码：

python -m examples.runtime.ask_user_timeout

"""从仓库根目录运行：python -m examples.runtime.ask_user_timeout"""

import asyncio
import os

import bumblehive
from examples.runtime.ask_user import read_answer
from examples.runtime.background_support import BackgroundTools, Conversation


WAIT_SECONDS = 60


async def main() -> None:
    config = bumblehive.RuntimeArguments(
        model=os.environ["BUMBLEHIVE_MODEL"],
        api_key=os.environ["BUMBLEHIVE_API_KEY"],
        base_url=os.environ["BUMBLEHIVE_BASE_URL"],
        skill_names=[],
        tool_names=["ask_user"],
    )
    conversation = Conversation()

    def on_answer(task_id: str, value: dict) -> None:
        # 超时后跳过只结束等待，不再给 Agent 发送消息。
        if value["status"] != "skipped":
            conversation.notify(task_id, value)

    background = BackgroundTools(on_answer)
    async with bumblehive.from_config(config) as runtime:
        @runtime.tools.tool(description=(
            "缺少关键信息时询问用户，可提供建议选项，也允许自由输入。"
            f"等待 {WAIT_SECONDS} 秒。answered 包含答案；skipped 表示跳过，不要重复问。"
            "pending 表示还未回答，晚到答案会带原问题作为消息送达。"
            "可以继续独立工作；必须依赖答案时说明阻碍并结束本轮。"
            "pending 和 skipped 都不代表同意，不要猜答案或重复提问。"
        ))
        async def ask_user(question: str, options: list[str] | None = None) -> dict:
            async def read() -> dict:
                return {"question": question, **await read_answer(question, options)}

            return await background.submit(read(), wait_seconds=WAIT_SECONDS)

        await conversation.run(
            runtime,
            "帮我写介绍 Python 的开场白，先用 ask_user 询问目标读者。"
            "等待时可以先列出通用提纲；收到答案再写正文。"
            "如果跳过，就只给通用提纲。不要重复提问。",
            background,
        )


if __name__ == "__main__":
    asyncio.run(main())

修改 WAIT_SECONDS 调整等待时间，例如改为 5 方便体验。这里的等待时长和返回状态都是示例定义的，不是 SDK 内置配置。

用户操作	Agent 收到什么
等待期间回答	工具直接返回 `answered` 和答案
等待期间跳过	工具直接返回 `skipped`
等待超时	工具返回 `pending`，输入仍然有效
超时后回答	原问题和答案作为后续消息送达
超时后跳过	结束提问，不再发送消息

例如先不回答，让等待超时：Agent 可以先给通用提纲。随后输入“初学者”，它会结合回答继续。运行中收到答案用 steer()；本轮已经结束则用原历史开始下一轮，具体代码集中在后台工具示例。

pending 表示尚未回答。如果剩下的工作必须依赖答案，Agent 应说明阻碍并结束本轮；应用继续等待用户，不要猜测答案或反复提问。

这个终端示例使用一个输入口，适合一次回答一个问题。input() 在线程中等待，取消协程不能终止终端读取；退出前输入答案或 /skip。接入界面时，把 read_answer() 换成等待该问题提交事件的异步函数即可。