第 4 章：回调与流式响应

在 YouTube 观看

关于本课

本课程的视频是当时的一个快照。Strands 一直在积极开发中，所以这一页里的代码反映的是最新的写法，但视频里讲的概念依然成立。有疑问的时候，以代码为准。

本课代码：samples/04-callbacks-streaming

Callback 决定用户能看到什么

Callback 控制 Agent 的输出怎么呈现给用户。默认的 handler 把文本流式写到 stdout，放在终端 demo 里没问题，但对 UI、API，或者后台默默跑着的 sub-agent 就不合适了。你可以换成自定义 handler、彻底关掉输出，或者在异步服务里用 stream_async()。

自定义 Callback Handler

callback handler 就是一个接收 **kwargs 的函数。每个 agent event 都会触发它：文本分片、tool call、以及完整消息。下面这个 handler 忽略 stream 分片，只打印完整的 assistant 消息：

from strands import Agent, tool

@tool
def calculator(a: float, b: float, operation: str = "add") -> str:
    """Perform a math operation on two numbers.

    Args:
        a: First number
        b: Second number
        operation: One of "add", "subtract", "multiply", "divide", "power"
    """
    if operation == "add":
        result = a + b
    elif operation == "subtract":
        result = a - b
    elif operation == "multiply":
        result = a * b
    elif operation == "divide":
        result = a / b if b != 0 else "Error: division by zero"
    elif operation == "power":
        result = a ** b
    else:
        result = f"Unknown operation: {operation}"
    return str(result)

def buffered_handler(**kwargs):
    # Only show complete messages, not individual streaming chunks
    if "message" in kwargs and kwargs["message"].get("role") == "assistant":
        content = kwargs["message"].get("content", [])
        for block in content:
            if "text" in block:
                print(block["text"])

agent = Agent(tools=[calculator], callback_handler=buffered_handler)
agent("What is 2 to the power of 16, minus 1?")

📂 callbacks_streaming.py

kwargs 里几个常用的事件 key：

• data：一段流式文本分片
• current_tool_use：Model 正在调用的 Tool，包含它的输入
• message：一轮结束后产生的完整消息

静默模式

把 callback_handler=None 设上，就能压掉所有输出。Agent 照常运行并返回结果，你可以拿去做程序化处理：

agent = Agent(tools=[calculator], callback_handler=None)
result = agent("What is 42 * 42?")
print(f"Captured result: {result}")

多 Agent 系统里那些后台跑的 sub-agent 必须这么干。只有 orchestrator 才应该往用户那边推流。

异步 Streaming（FastAPI）

异步服务用 agent.stream_async()，这是个 async generator，事件一发生就往外 yield。下面用它撑起一个 FastAPI 的 streaming endpoint：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from strands import Agent, tool

@tool
def calculator(a: float, b: float, operation: str = "add") -> str:
    """Perform a math operation on two numbers.

    Args:
        a: First number
        b: Second number
        operation: One of "add", "subtract", "multiply", "divide", "power"
    """
    if operation == "add":
        result = a + b
    elif operation == "subtract":
        result = a - b
    elif operation == "multiply":
        result = a * b
    elif operation == "divide":
        result = a / b if b != 0 else "Error: division by zero"
    elif operation == "power":
        result = a ** b
    else:
        result = f"Unknown operation: {operation}"
    return str(result)

app = FastAPI()


class PromptRequest(BaseModel):
    prompt: str


@app.post("/stream")
async def stream_response(request: PromptRequest):
    async def generate():
        agent = Agent(tools=[calculator], callback_handler=None)
        async for event in agent.stream_async(request.prompt):
            if "data" in event:
                yield event["data"]

    return StreamingResponse(generate(), media_type="text/plain")

📂 fastapi_streaming.py · async_streaming.py

先 pip install fastapi uvicorn，再 uvicorn fastapi_streaming:app --reload 跑起来，然后往 /stream 发 POST。

参考资源

• 📖 Callback Handlers
• 📖 Tools Overview