이벤트와 아이템

이벤트와 아이템 (Events and items)

이벤트는 에이전트가 작업하는 동안 일어나는 일을 보고해요. 아이템(item)은 나중에 검색할 수 있는 저장된 메시지와 도구 호출이에요. 이벤트로 애플리케이션을 실시간으로 갱신하고, 아이템으로 저장된 기록을 표시하세요.

출처: 문서

본문

애플리케이션은 메시지를 제출하고, 턴을 취소하고, 도구 결과를 반환하기 위해 입력 이벤트를 보내요. 에이전트는 출력과 세션 변경을 보고하는 이벤트를 보내요. 입력 전송은 세션 실행 및 이어가기를 참고하세요.

스트림 소비하기 (Consume a stream)

애플리케이션이 턴의 초기 이벤트를 받을 수 있도록 작업을 보내기 전에 구독하세요. 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 ***" \
  -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

업데이트 처리하기 (Handle updates)

이벤트의 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는 메시지의 콘텐츠, 상태, 단계를 포함하는 저장된 아이템을 식별해요. 저장된 작업 검색을 참고하세요.

모든 이벤트 유형과 필드는 스트리밍 이벤트 레퍼런스를 참고하세요. 이 스트림 이벤트는 웹훅과 다르다는 점에 유의하세요. 서브에이전트 활동과 명령 귀속은 위임 관찰을 참고하세요.

아이템과 턴 가져오기 (Fetch items and turns)

애플리케이션의 대화 상태에서 세션 ID를 사용해 저장된 작업을 검색하세요:

세션 아이템: 아이템 나열로 턴에 걸친 루트 에이전트의 메시지와 도구 호출을 검색하세요.
턴: 턴 나열로 세션의 작업을 둘러보세요. 턴 검색으로 ID로 턴을 검색해 상태, 타임스탬프, 사용량, 오류를 검사하세요.
한 턴의 아이템: 루트 에이전트 턴에서는 세션 아이템을 turn_id로 필터링하세요. 각 서브에이전트는 자체 아이템 기록과 턴별 아이템 엔드포인트가 있어요.

나열 엔드포인트는 한 번에 한 페이지를 반환해요. SDK 페이지네이션 헬퍼나 after 커서로 더 많은 결과를 검색하세요. 한 페이지가 턴의 모든 아이템을 포함하지 못할 수 있어요. order: "asc"를 사용해 아이템을 오래된 것부터 새로운 것 순으로 읽으세요.

끊긴 스트림 복구하기 (How to recover a disconnected stream)

스트림은 놓친 이벤트를 재생하지 않아요. 애플리케이션의 뷰를 복원하려면:

새 스트림을 열고 들어오는 이벤트를 버퍼링하세요.
스트림이 연결된 동안 세션과 저장된 아이템을 검색하세요.
그 아이템들로 항목 ID를 키로 로컬 상태를 복원하세요.
item_id로 버퍼링된 아이템 업데이트를 적용하세요. 검색된 기록에서 이미 최종 상태에 도달한 아이템의 업데이트는 버리세요.
실시간 이벤트 처리를 재개하세요.

output_text.done 이벤트는 임시 텍스트 버퍼를 완전한 텍스트로 바꿀 수 있어요. 저장된 아이템은 완료된 작업을 복구하게 해 주지만, 놓친 모든 중간 이벤트는 복구하지 못해요.

더 알아보기 (Learn more)

세션 실행 및 이어가기에서 입력 전송과 턴 취소를 확인하세요.
세션 웹훅으로 이벤트 스트림 없이 상태 변경을 받는 방법을 확인하세요.