交互式聊天助手¶
将输入、流式回答和会话保存组合起来，得到一个可以连续对话的终端助手。
运行¶
完成模型配置，运行：
python examples/runtime/chat.py
"""流式聊天，/quit 退出；同一会话可以在下次运行继续。"""
import asyncio
import os
import bumblehive
from bumblehive.observability import MODEL_STREAM_CONTENT_DELTA
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=[],
)
async with bumblehive.from_config(config) as runtime:
while True:
text = (await asyncio.to_thread(input, "你：")).strip()
if text == "/quit":
break
if not text:
continue
stream = runtime.stream(text, session_id="chat-example")
print("Agent：", end="", flush=True)
try:
async for event in stream:
if event.kind == MODEL_STREAM_CONTENT_DELTA:
print(event.payload["delta"], end="", flush=True)
result = await stream.result()
finally:
await stream.aclose()
print()
if result.error:
print(result.error.message)
if __name__ == "__main__":
asyncio.run(main())
体验对话¶
你：记住，我的项目叫 BumbleHive。
Agent：…确认项目名称…
你：我的项目叫什么？
Agent：你的项目叫 BumbleHive。
你：/quit
同一个 session_id 保留对话，stream() 逐段输出回答。重新运行程序会继续 chat-example 会话；修改 ID 可以开始新对话。
接入界面¶
把终端输入换成用户消息事件，把输出换成界面更新即可。下面的适配示例将文本和工具事件转换成 JSON 行：
python examples/runtime/chat_events.py
"""Map SDK events to small UI updates; print JSON lines as a transport example."""
import asyncio
import json
import os
from typing import Any
import bumblehive
from bumblehive.observability import (
MODEL_STREAM_CONTENT_DELTA,
TOOL_CALL_FINISHED,
TOOL_CALL_STARTED,
)
def to_ui_update(event: bumblehive.AgentEvent) -> dict[str, Any] | None:
update: dict[str, Any] = {"run_id": event.run_id, "session_id": event.session_id}
if event.kind == MODEL_STREAM_CONTENT_DELTA:
return {**update, "type": "text_delta", "text": event.payload["delta"]}
if event.kind == TOOL_CALL_STARTED:
call = event.payload["tool_call"]
return {
**update,
"type": "tool_started",
"call_id": call["call_id"],
"name": call["name"],
}
if event.kind == TOOL_CALL_FINISHED:
result = event.payload["tool_result"]
return {
**update,
"type": "tool_finished",
"call_id": result["tool_call_id"],
"ok": event.payload["ok"],
}
return None
def send(update: dict[str, Any]) -> None:
# A real application sends this update through its WebSocket or SSE layer.
print(json.dumps(update, ensure_ascii=False), flush=True)
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=["add"],
)
async with bumblehive.from_config(config) as runtime:
@runtime.tools.tool(name="add", description="计算两个整数的和。")
def add(a: int, b: int) -> int:
return a + b
stream = runtime.stream("请调用 add 计算 21 加 34。", session_id="chat-events-demo")
try:
async for event in stream:
update = to_ui_update(event)
if update is not None:
send(update)
result = await stream.result()
send({
"type": "finished",
"text": result.final_content,
"usage": result.usage,
"error": result.error.message if result.error else None,
})
finally:
await stream.aclose()
if __name__ == "__main__":
asyncio.run(main())
| 更新 | 用途 |
|---|---|
text_delta |
追加回答文本 |
tool_started |
按 run_id 和 call_id 显示工具开始 |
tool_finished |
更新同一个工具调用的结果 |
finished |
显示最终回答或错误 |