Runtime 与运行结果¶

Runtime 组合模型、工具、Skills 与会话。使用 async with 初始化并释放资源，复用同一 Runtime 可以发起多次运行。

常用接口¶

接口	用途
`from_config`	从配置对象、字典或 JSON 路径创建 Runtime。
`BumblehiveRuntime`	发起运行、消费事件流或在控制台展示过程。
`AgentRun`	`run()` 和 `run_console()` 返回的单次运行对象，支持 `await`、`asyncio.create_task()` 和 `steer()`。
`AgentRunStream`	`stream()` 返回的单次事件流，支持 `steer()`、`result()` 和 `aclose()`。
`AgentRunResult`	运行完成后的结果，包含回答、工具使用情况、模型用量和错误。

创建与运行¶

从配置对象、字典或 JSON 路径创建 Runtime。

from_config ¶

from_config(config: ConfigInput = None) -> BumblehiveRuntime

Create a runtime from a JSON file, mapping, or config object.

等待 run() 获取结果；迭代 stream() 消费事件；等待 run_console() 在终端展示过程并获取结果。history 与 session_id 不能同时使用，approval_handler 仅影响当前调用。

BumblehiveRuntime ¶

High-level agent runtime built from BumblehiveConfig.

from_config `classmethod` ¶

from_config(config: ConfigInput = None) -> BumblehiveRuntime

Create a runtime from a JSON file, mapping, config object, or defaults.

initialize_tools `async` ¶

initialize_tools() -> None

Register built-ins and connect the runtime-scoped MCP servers once.

run ¶

run(message: UserMessage, *, config: Mapping[str, Any] | None = None, hooks: HookInput = None, history: MessageHistory | None = None, session_id: str | None = None, approval_handler: ToolApprovalHandler | None = None) -> AgentRun

Create a steerable turn, executed when awaited or scheduled as a Task.

With neither history nor session_id, the turn is stateless. Passing history reads caller-owned in-memory history without updating it. Use history.replace_run_messages(result.messages) to carry a completed run into the next turn. Passing session_id loads, recovers, and persists a managed session. Passing both raises ValueError.

Do not share one MessageHistory across concurrent runs: await turns sequentially, or use separate histories for parallel conversations.

stream ¶

stream(message: UserMessage, *, config: Mapping[str, Any] | None = None, hooks: HookInput = None, history: MessageHistory | None = None, session_id: str | None = None, approval_handler: ToolApprovalHandler | None = None, max_queue_size: int = DEFAULT_STREAM_QUEUE_SIZE) -> AgentRunStream

Stream one turn with stateless, in-memory, or persisted history.

run_console ¶

run_console(message: UserMessage, *, config: Mapping[str, Any] | None = None, hooks: HookInput = None, history: MessageHistory | None = None, session_id: str | None = None, approval_handler: ToolApprovalHandler | None = None, renderer: Any | None = None, max_queue_size: int = DEFAULT_STREAM_QUEUE_SIZE) -> AgentRun

Create a steerable turn that renders its stream when awaited.

Supports the same await, Task, steering, and cancellation behavior as run(). Rendering consumes an internal stream; keyboard input remains the application's responsibility.

delete_session `async` ¶

delete_session(session_id: str) -> bool

Delete a persisted session and evict its cached state.

close `async` ¶

close() -> None

Release resources owned by this runtime.

运行对象¶

运行期间补充或纠正任务，见运行期间追加指令。

AgentRun ¶

Bases: Coroutine[Any, Any, AgentRunResult]

A single-use coroutine with a queue for supplemental user input.

Await directly or schedule with asyncio.create_task. Only one await or Task may drive the run. Call steer on the owning event loop; it queues input without starting execution. Cancel an active run through its Task.

steer ¶

steer(message: UserMessage) -> None

Validate and copy input into this run; raise RuntimeError if closed.

take_pending ¶

take_pending() -> list[Message]

Remove unconsumed input after closure; raise RuntimeError while open.

close ¶

close() -> None

Discard an unscheduled run; cancel active runs through their Task.

AgentRun 支持直接 await 或交给 asyncio.create_task()，只能由一个 await 或 Task 驱动。需要多次读取结果时，等待同一个 Task。

未调度的对象用 run.close() 丢弃；运行中通过 Task 取消并等待清理。示例见本页运行生命周期。

需要检查协程类型时：run() 和 run_console() 是普通方法，inspect.iscoroutinefunction() 返回 False；返回对象实现 Coroutine 协议，但 inspect.iscoroutine(run) 也为 False。

AgentRunStream ¶

Bases: AsyncEventStream[AgentRunResult]

A single-consumer event stream with supplemental input for its run.

Iteration starts execution. Consume fully before calling result and always call aclose when abandoning the stream. Input can close before buffered events have all been consumed.

steer ¶

steer(message: UserMessage) -> None

Validate and copy input into this run on its owning event loop.

take_pending ¶

take_pending() -> list[Message]

Remove unconsumed input after closure; raise RuntimeError while open.

result `async` ¶

result() -> ResultT

Return the completed background run result.

aclose `async` ¶

aclose() -> None

Close input and cancel/await stream execution, including before start.

AgentRunStream 首次迭代时启动，只能消费一次。完整消费后读取 result()，提前退出时调用 aclose()。它支持 async for，不支持直接 await stream。示例见流式输出。

读取结果¶

final_content 是最终回答，usage 是模型用量，stop_reason 说明结束原因。首先检查 error；tools_used 只记录成功执行的工具。

AgentRunResult `dataclass` ¶

Outcome of one tool-capable model run.

运行生命周期¶

Runtime 应在一组相关任务中复用，并通过 async with 关闭 Provider、MCP 和内置命令会话。退出前应完成或取消它上面的运行：

import asyncio

async with bumblehive.from_config(config) as runtime:
    task = asyncio.create_task(runtime.run("分析项目"))
    try:
        result = await task
    finally:
        if not task.done():
            task.cancel()
        await asyncio.gather(task, return_exceptions=True)

run() 在等待或调度后执行，stream() 在首次迭代时执行。创建后不使用的运行对象调用 close()；事件流调用 await aclose()。

steer() 必须在所属事件循环中调用，输入关闭后抛出 RuntimeError。关闭后可以用 take_pending() 取回尚未消费的消息；此操作清空队列，已加入对话的消息不会退回。

Runtime 与运行结果¶

常用接口¶

创建与运行¶

from_config ¶

BumblehiveRuntime ¶

from_config classmethod ¶

initialize_tools async ¶

run ¶

stream ¶

run_console ¶

delete_session async ¶

close async ¶

运行对象¶

AgentRun ¶

steer ¶

take_pending ¶

close ¶

AgentRunStream ¶

steer ¶

take_pending ¶

result async ¶

aclose async ¶

读取结果¶

AgentRunResult dataclass ¶

运行生命周期¶

from_config `classmethod` ¶

initialize_tools `async` ¶

delete_session `async` ¶

close `async` ¶

result `async` ¶

aclose `async` ¶

AgentRunResult `dataclass` ¶