Skip to Content

流式响应

使用当前 Agent SDK 的异步生成器逐条消费响应事件。

基础流式调用

from isa_agent_sdk import query, ISAAgentOptions
 
options = ISAAgentOptions(model="gpt-4o-mini")
 
async for message in query("解释量子计算", options=options):
    if message.is_text:
        print(message.content, end="", flush=True)
    elif message.is_tool_use:
        print(f"\n[使用工具：{message.tool_name}]")
    elif message.is_complete:
        print("\n[完成]")

在 FastAPI 中转发 SSE

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from isa_agent_sdk import query
 
app = FastAPI()
 
@app.post("/chat/stream")
async def chat_stream(message: str):
    async def events():
        async for item in query(message):
            if item.is_text:
                yield f"data: {item.content}\n\n"
        yield "data: [DONE]\n\n"
 
    return StreamingResponse(events(), media_type="text/event-stream")

同步场景可以使用 query_sync；只需要最终文本时使用 ask 或 ask_sync。更多事件字段见消息。

如果更喜欢面向对象的封装，也可以使用当前 SDK 导出的 Agent(...).stream()；本页使用 query 以展示完整事件流。

Home