All guides
OpenAI

事件与项目

Checked 09/15/2026View original
English originalChinese translation

AI translation, not an official translation. Refer to the original for technical details.

On this page

完整的文档索引，请参阅 llms.txt。在页面 URL 后附加 .md 即可获取文档页面的 Markdown 版本。

事件用于报告智能体工作时发生的情况。项目是可供后续检索的已保存消息和工具调用。使用事件可实时更新应用程序，使用项目可展示已保存的历史记录。

您的应用程序发送输入事件以提交消息、取消轮次或返回工具结果。智能体发送事件以报告输出内容及会话变更。有关发送输入的详细信息，请参阅运行并继续会话。

消费流

在发送工作之前先订阅，以便您的应用程序能接收到轮次的早期事件。传入您的 API 客户端、会话 ID 和事件处理器：

流式传输会话事件

// Pass your saved session ID to this helper.
async function streamSession(client, sessionId, handleEvent) {
  const events = await client.beta.agents.sessions.events.stream(sessionId);
  try {
    for await (const event of events) {
      await handleEvent(event);
      switch (event.type) {
        case "agent.session.idle":
          continue;
        case "error":
          throw new Error(event.error.message);
        case "agent.session.failed":
        case "agent.session.environment.failed":
          throw new Error(`Agent lifecycle failure: ${event.type}`);
        case "agent.session.turn.failed":
          if (event.turn.subagent_id === null) {
            throw new Error(
              `${event.type}: ${event.turn.error?.message ?? ""}`
            );
          }
          break;
        case "agent.session.turn.cancelled":
          if (event.turn.subagent_id === null) {
            throw new Error("The agent turn was cancelled");
          }
          break;
        case "agent.session.turn.completed":
          if (event.turn.subagent_id === null) return;
          break;
      }
    }
    throw new Error(
      "Stream closed before a turn ended. Retrieve the saved state."
    );
  } finally {
    events.controller.abort();
  }
}

# Pass your saved session ID to this helper.
def stream_session(client: OpenAI, session_id: str, handle_event):
    with client.beta.agents.sessions.events.stream(session_id) as events:
        for event in events:
            handle_event(event)
            match event.type:
                case "agent.session.idle":
                    continue
                case "error":
                    raise RuntimeError(event.error.message)
                case "agent.session.failed" | "agent.session.environment.failed":
                    raise RuntimeError(f"Agent lifecycle failure: {event.type}")
                case "agent.session.turn.failed":
                    if event.turn.subagent_id is None:
                        detail = event.turn.error.message if event.turn.error else ""
                        raise RuntimeError(f"{event.type}: {detail}")
                case "agent.session.turn.cancelled":
                    if event.turn.subagent_id is None:
                        raise RuntimeError("The agent turn was cancelled")
                case "agent.session.turn.completed":
                    if event.turn.subagent_id is None:
                        return
    raise RuntimeError("Stream closed before a turn ended. Retrieve the saved state.")

// Pass your saved session ID to this helper.
func streamSession(ctx context.Context, client *openai.Client, sessionID string, handleEvent func(openai.AgentSessionEventUnion)) error {
	events := client.Beta.Agents.Sessions.Events.StreamStreaming(ctx, sessionID)
	defer events.Close()
	for events.Next() {
		event := events.Current()
		handleEvent(event)
		switch event.Type {
		case "agent.session.idle":
			continue
		case "error":
			return fmt.Errorf("agent error: %s", event.RawJSON())
		case "agent.session.failed", "agent.session.environment.failed":
			return fmt.Errorf("agent lifecycle failure: %s", event.RawJSON())
		case "agent.session.turn.failed", "agent.session.turn.cancelled":
			if event.Turn.SubagentID == "" {
				return fmt.Errorf("agent turn did not complete: %s", event.RawJSON())
			}
		case "agent.session.turn.completed":
			if event.Turn.SubagentID == "" {
				return nil
			}
		}
	}
	if err := events.Err(); err != nil {
		return err
	}
	return fmt.Errorf("stream closed before a turn ended; retrieve the saved state")
}

// Pass your saved session ID to this helper.
public static void streamSession(
    OpenAIClient client, String sessionId, Consumer<AgentSessionEvent> handleEvent) {
  try (StreamResponse<AgentSessionEvent> events =
      client.beta().agents().sessions().events().streamStreaming(sessionId)) {
    var iterator = events.stream().iterator();
    while (iterator.hasNext()) {
      var event = iterator.next();
      handleEvent.accept(event);
      if (event.idle().isPresent()) {
        continue;
      }
      if (event.error().isPresent()) {
        throw new IllegalStateException("Agent error: " + event);
      }
      if (event.failed().isPresent() || event.environmentFailed().isPresent()) {
        throw new IllegalStateException("Agent lifecycle failure: " + event);
      }
      if (event.turnFailed().filter(e -> e.turn().subagentId().isEmpty()).isPresent()
          || event.turnCancelled().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
        throw new IllegalStateException("Agent turn did not complete: " + event);
      }
      if (event.turnCompleted().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
        return;
      }
    }
    throw new IllegalStateException(
        "Stream closed before a turn ended. Retrieve the saved state.");
  }
}

# Pass your saved session ID to this helper.
def stream_session(client, session_id, &handle_event)
  events = client.beta.agents.sessions.events.stream_streaming(session_id)
  begin
    events.each do |event|
      handle_event.call(event)
      case event.type.to_s
      when "agent.session.idle"
        next
      when "error"
        raise event.error.message
      when "agent.session.failed", "agent.session.environment.failed"
        raise "Agent lifecycle failure: #{event.type}"
      when "agent.session.turn.failed"
        raise "#{event.type}: #{event.turn.error&.message}" if event.turn.subagent_id.nil?
      when "agent.session.turn.cancelled"
        raise "The agent turn was cancelled" if event.turn.subagent_id.nil?
      when "agent.session.turn.completed"
        return nil if event.turn.subagent_id.nil?
      end
    end
    raise "Stream closed before a turn ended. Retrieve the saved state."
  ensure
    events.close
  end
end

curl -N \
  "https://api.openai.com/v1/agents/sessions/$session_id/events?stream=true" \
  -H "OpenAI-Beta: agents=v1" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Accept: text/event-stream"

该辅助函数将每个事件传递给您的处理器，然后检查常见事件类型。它在遇到 agent.session.idle 时继续，并在根轮次完成时返回。若根轮次失败或被取消、会话或环境失败，或者到达 error 事件，它将抛出错误。子智能体轮次事件不会结束流。您的处理器负责决定如何展示输出；调用方负责处理辅助函数抛出的错误。若流在轮次结束前关闭，辅助函数将抛出错误。请参阅恢复断开的流。

此版本接受一条消息，并在打开流后提交：

发送消息并进行流式传输

// Pass your saved session ID and message to this helper.
async function sendAndStream(client, sessionId, text, handleEvent) {
  const events = await client.beta.agents.sessions.events.stream(sessionId);
  try {
    await client.beta.agents.sessions.events.create(sessionId, {
      events: [
        {
          type: "agent.session.input.message",
          input: [{ role: "user", content: [{ type: "input_text", text }] }],
        },
      ],
    });
    for await (const event of events) {
      await handleEvent(event);
      switch (event.type) {
        case "agent.session.idle":
          continue;
        case "error":
          throw new Error(event.error.message);
        case "agent.session.failed":
        case "agent.session.environment.failed":
          throw new Error(`Agent lifecycle failure: ${event.type}`);
        case "agent.session.turn.failed":
          if (event.turn.subagent_id === null) {
            throw new Error(
              `${event.type}: ${event.turn.error?.message ?? ""}`
            );
          }
          break;
        case "agent.session.turn.cancelled":
          if (event.turn.subagent_id === null) {
            throw new Error("The agent turn was cancelled");
          }
          break;
        case "agent.session.turn.completed":
          if (event.turn.subagent_id === null) return;
          break;
      }
    }
    throw new Error(
      "Stream closed before a turn ended. Retrieve the saved state."
    );
  } finally {
    events.controller.abort();
  }
}

# Pass your saved session ID and message to this helper.
def send_and_stream(client: OpenAI, session_id: str, text, handle_event):
    with client.beta.agents.sessions.events.stream(session_id) as events:
        client.beta.agents.sessions.events.create(
            session_id,
            events=[
                {
                    "type": "agent.session.input.message",
                    "input": [
                        {
                            "role": "user",
                            "content": [{"type": "input_text", "text": text}],
                        }
                    ],
                }
            ],
        )
        for event in events:
            handle_event(event)
            match event.type:
                case "agent.session.idle":
                    continue
                case "error":
                    raise RuntimeError(event.error.message)
                case "agent.session.failed" | "agent.session.environment.failed":
                    raise RuntimeError(f"Agent lifecycle failure: {event.type}")
                case "agent.session.turn.failed":
                    if event.turn.subagent_id is None:
                        detail = event.turn.error.message if event.turn.error else ""
                        raise RuntimeError(f"{event.type}: {detail}")
                case "agent.session.turn.cancelled":
                    if event.turn.subagent_id is None:
                        raise RuntimeError("The agent turn was cancelled")
                case "agent.session.turn.completed":
                    if event.turn.subagent_id is None:
                        return
    raise RuntimeError("Stream closed before a turn ended. Retrieve the saved state.")

// Pass your saved session ID and message to this helper.
func sendAndStream(ctx context.Context, client *openai.Client, sessionID string, text string, handleEvent func(openai.AgentSessionEventUnion)) error {
	events := client.Beta.Agents.Sessions.Events.StreamStreaming(ctx, sessionID)
	defer events.Close()
	if err := events.Err(); err != nil {
		return err
	}
	err := client.Beta.Agents.Sessions.Events.New(ctx,
		sessionID,
		openai.BetaAgentSessionEventNewParams{
			Events: []openai.AgentSessionInputParamUnion{
				{
					OfParamAgentSessionInputMessage: &openai.AgentSessionInputParamAgentSessionInputMessage{
						Input: []openai.AgentSessionInputMessageParam{
							{
								Content: []openai.InputContentParamUnion{
									{
										OfParamInputText: &openai.InputContentParamInputText{
											Text: text,
										},
									},
								},
							},
						},
					},
				},
			},
		})
	if err != nil {
		return err
	}
	for events.Next() {
		event := events.Current()
		handleEvent(event)
		switch event.Type {
		case "agent.session.idle":
			continue
		case "error":
			return fmt.Errorf("agent error: %s", event.RawJSON())
		case "agent.session.failed", "agent.session.environment.failed":
			return fmt.Errorf("agent lifecycle failure: %s", event.RawJSON())
		case "agent.session.turn.failed", "agent.session.turn.cancelled":
			if event.Turn.SubagentID == "" {
				return fmt.Errorf("agent turn did not complete: %s", event.RawJSON())
			}
		case "agent.session.turn.completed":
			if event.Turn.SubagentID == "" {
				return nil
			}
		}
	}
	if err := events.Err(); err != nil {
		return err
	}
	return fmt.Errorf("stream closed before a turn ended; retrieve the saved state")
}

// Pass your saved session ID and message to this helper.
public static void sendAndStream(
    OpenAIClient client, String sessionId, String text, Consumer<AgentSessionEvent> handleEvent) {
  try (StreamResponse<AgentSessionEvent> events =
      client.beta().agents().sessions().events().streamStreaming(sessionId)) {
    client
        .beta()
        .agents()
        .sessions()
        .events()
        .create(
            EventCreateParams.builder()
                .sessionId(sessionId)
                .addEvent(
                    AgentSessionInputParam.AgentSessionInputMessage.builder()
                        .addInput(
                            AgentSessionInputMessageParam.builder()
                                .addInputTextContent(text)
                                .build())
                        .build())
                .build());
    var iterator = events.stream().iterator();
    while (iterator.hasNext()) {
      var event = iterator.next();
      handleEvent.accept(event);
      if (event.idle().isPresent()) {
        continue;
      }
      if (event.error().isPresent()) {
        throw new IllegalStateException("Agent error: " + event);
      }
      if (event.failed().isPresent() || event.environmentFailed().isPresent()) {
        throw new IllegalStateException("Agent lifecycle failure: " + event);
      }
      if (event.turnFailed().filter(e -> e.turn().subagentId().isEmpty()).isPresent()
          || event.turnCancelled().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
        throw new IllegalStateException("Agent turn did not complete: " + event);
      }
      if (event.turnCompleted().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
        return;
      }
    }
    throw new IllegalStateException(
        "Stream closed before a turn ended. Retrieve the saved state.");
  }
}

# Pass your saved session ID and message to this helper.
def send_and_stream(client, session_id, text, &handle_event)
  events = client.beta.agents.sessions.events.stream_streaming(session_id)
  begin
    client.beta.agents.sessions.events.create(
      session_id,
      events: [
        {
          type: "agent.session.input.message",
          input: [
            {
              role: "user",
              content: [
                {
                  type: "input_text",
                  text: text
                }
              ]
            }
          ]
        }
      ]
    )
    events.each do |event|
      handle_event.call(event)
      case event.type.to_s
      when "agent.session.idle"
        next
      when "error"
        raise event.error.message
      when "agent.session.failed", "agent.session.environment.failed"
        raise "Agent lifecycle failure: #{event.type}"
      when "agent.session.turn.failed"
        raise "#{event.type}: #{event.turn.error&.message}" if event.turn.subagent_id.nil?
      when "agent.session.turn.cancelled"
        raise "The agent turn was cancelled" if event.turn.subagent_id.nil?
      when "agent.session.turn.completed"
        return nil if event.turn.subagent_id.nil?
      end
    end
    raise "Stream closed before a turn ended. Retrieve the saved state."
  ensure
    events.close
  end
end

处理更新

使用事件的 type 来决定应用程序应执行的操作：

  • 显示文本： 将 agent.session.turn.output_text.delta 追加到相关内容部分。当 agent.session.turn.output_text.done 到达时，用其完整文本替换该部分。增量内容可能不存在。
  • 跟踪工作： 会话、轮次和项目事件报告进度。检查 agent.session.turn.completed、agent.session.turn.failed 或 agent.session.turn.cancelled 以确定轮次的结果。
  • 提供所需输入： 收到 agent.session.requires_action 时，检索会话并检查 required_actions。您的代码可能需要返回函数结果或连接环境。

空闲的会话或已关闭的流本身并不代表成功。已完成的轮次也不保证每个工具都执行成功。请检查智能体的输出。

使用 item_id、output_index 和 content_index 将文本更新关联到同一内容部分。例如，以下缩写事件更新同一部分：

{
  "type": "agent.session.turn.output_text.delta",
  "item_id": "msg_789",
  "output_index": 0,
  "content_index": 0,
  "delta": "Acme competes"
}

{
  "type": "agent.session.turn.output_text.done",
  "item_id": "msg_789",
  "output_index": 0,
  "content_index": 0,
  "text": "Acme competes on price and distribution."
}

每个事件都有其自己的 event_id。共享的 item_id 用于标识已保存的项目，其中包含消息的内容、状态和阶段。请参阅检索已保存的工作。

有关所有事件类型和字段，请参阅流式事件参考。这些流式事件与 Webhook 不同。有关子智能体活动和命令归因，请参阅观察委派。

获取项目和轮次

使用应用程序会话状态中的会话 ID 来检索已保存的工作：

列表端点每次返回一页结果。使用 SDK 分页辅助函数或 after 游标检索更多结果。单页内容可能不包含某个轮次的所有项目。使用 order: "asc" 以从最旧到最新的顺序读取项目。

如何恢复断开的流

流不会重播错过的事件。要恢复应用程序的视图：

  1. 打开新流并缓冲传入的事件。
  2. 在流保持连接的同时，检索会话及其已保存的项目。
  3. 以项目 ID 为键，从这些项目中恢复本地状态。
  4. 使用 item_id 应用已缓冲的项目更新。对于在已检索历史记录中已达到最终状态的项目，丢弃其更新。
  5. 恢复处理实时事件。

output_text.done 事件可将临时文本缓冲区替换为完整文本。已保存的项目可帮助您恢复已完成的工作，但无法恢复错过的每个中间事件。