流式传输
Agents SDK 可以逐步输出来自模型和其他执行步骤的结果。流式传输让你的 UI 保持响应,无需等待整个最终执行结果后再更新用户。
启用流式传输
Section titled “启用流式传输”向 Runner.run() 传入 { stream: true } 选项以获取流对象,而不是完整的执行结果:
import { Agent, run } from '@openai/agents';
const agent = new Agent({ name: 'Storyteller', instructions: 'You are a storyteller. You will be given a topic and you will tell a story about it.',});
const result = await run(agent, 'Tell me a story about a cat.', { stream: true,});启用流式传输后,返回的 stream 实现了 AsyncIterable 接口。每个产出的事件都是一个对象,描述运行期间发生的事。该流会产出三种事件类型之一,分别对应智能体执行的不同部分。大多数应用只关心模型的文本,因此流提供了辅助方法。
获取文本输出
Section titled “获取文本输出”调用 stream.toTextStream() 以获取已发出的文本流。当 compatibleWithNodeStreams 为 true 时,返回值是标准的 Node.js Readable。我们可以将其直接管道到 process.stdout 或其他目标。
import { Agent, run } from '@openai/agents';
const agent = new Agent({ name: 'Storyteller', instructions: 'You are a storyteller. You will be given a topic and you will tell a story about it.',});
const result = await run(agent, 'Tell me a story about a cat.', { stream: true,});
result .toTextStream({ compatibleWithNodeStreams: true, }) .pipe(process.stdout);当运行及所有挂起的回调完成后,stream.completed 这个 promise 会被解析。如果你需要确保没有更多输出,请务必等待它。
监听所有事件
Section titled “监听所有事件”你可以使用 for await 循环在事件到达时逐个检查。实用信息包括底层模型事件、任何智能体切换以及 SDK 特定的运行信息:
import { Agent, run } from '@openai/agents';
const agent = new Agent({ name: 'Storyteller', instructions: 'You are a storyteller. You will be given a topic and you will tell a story about it.',});
const result = await run(agent, 'Tell me a story about a cat.', { stream: true,});
for await (const event of result) { // these are the raw events from the model if (event.type === 'raw_model_stream_event') { console.log(`${event.type} %o`, event.data); } // agent updated events if (event.type === 'agent_updated_stream_event') { console.log(`${event.type} %s`, event.agent.name); } // Agent SDK specific events if (event.type === 'run_item_stream_event') { console.log(`${event.type} %o`, event.item); }}请参阅流式示例,其中包含一个完整脚本,同时打印纯文本流和原始事件流。
该流会产出三种不同的事件类型:
raw_model_stream_event
Section titled “raw_model_stream_event”type RunRawModelStreamEvent = { type: 'raw_model_stream_event'; data: ResponseStreamEvent;};示例:
{ "type": "raw_model_stream_event", "data": { "type": "output_text_delta", "delta": "Hello" }}run_item_stream_event
Section titled “run_item_stream_event”type RunItemStreamEvent = { type: 'run_item_stream_event'; name: RunItemStreamEventName; item: RunItem;};示例交接负载:
{ "type": "run_item_stream_event", "name": "handoff_occurred", "item": { "type": "handoff_call", "id": "h1", "status": "completed", "name": "transfer_to_refund_agent" }}agent_updated_stream_event
Section titled “agent_updated_stream_event”type RunAgentUpdatedStreamEvent = { type: 'agent_updated_stream_event'; agent: Agent<any, any>;};示例:
{ "type": "agent_updated_stream_event", "agent": { "name": "Refund Agent" }}流式传输中的人工干预
Section titled “流式传输中的人工干预”流式传输兼容会暂停执行的交接(例如工具需要审批时)。流对象上的 interruption 字段会暴露中断信息,你可以通过对每个中断调用 state.approve() 或 state.reject() 来继续执行。再次以 { stream: true } 执行即可恢复流式输出。
import { Agent, run } from '@openai/agents';
const agent = new Agent({ name: 'Storyteller', instructions: 'You are a storyteller. You will be given a topic and you will tell a story about it.',});
let stream = await run( agent, 'What is the weather in San Francisco and Oakland?', { stream: true },);stream.toTextStream({ compatibleWithNodeStreams: true }).pipe(process.stdout);await stream.completed;
while (stream.interruptions?.length) { console.log( 'Human-in-the-loop: approval required for the following tool calls:', ); const state = stream.state; for (const interruption of stream.interruptions) { const approved = confirm( `Agent ${interruption.agent.name} would like to use the tool ${interruption.rawItem.name} with "${interruption.rawItem.arguments}". Do you approve?`, ); if (approved) { state.approve(interruption); } else { state.reject(interruption); } }
// Resume execution with streaming output stream = await run(agent, state, { stream: true }); const textStream = stream.toTextStream({ compatibleWithNodeStreams: true }); textStream.pipe(process.stdout); await stream.completed;}一个更完整、可与用户交互的示例见
human-in-the-loop-stream.ts。
- 退出前请记得等待
stream.completed,以确保所有输出已刷新。 - 初始的
{ stream: true }选项只适用于提供它的那次调用。如果你使用RunState重新运行,必须再次指定该选项。 - 如果你的应用只关心文本结果,优先使用
toTextStream(),以避免处理单个事件对象。
借助流式传输和事件系统,你可以将智能体集成到聊天界面、终端应用或任何用户受益于增量更新的场景中。