(ctx context.Context, w http.ResponseWriter, resp responsesResponse)
| 506 | } |
| 507 | |
| 508 | func (s *Server) sendResponsesStream(ctx context.Context, w http.ResponseWriter, resp responsesResponse) { |
| 509 | w.Header().Set("Content-Type", "text/event-stream") |
| 510 | w.Header().Set("Cache-Control", "no-cache") |
| 511 | w.Header().Set("Connection", "keep-alive") |
| 512 | w.WriteHeader(http.StatusOK) |
| 513 | |
| 514 | flusher, ok := w.(http.Flusher) |
| 515 | if !ok { |
| 516 | s.logger.Error(ctx, "responseWriter does not support flushing", |
| 517 | slog.F("response_id", resp.ID), |
| 518 | ) |
| 519 | return |
| 520 | } |
| 521 | |
| 522 | writeChunk := func(data string) bool { |
| 523 | if _, err := fmt.Fprintf(w, "%s", data); err != nil { |
| 524 | s.logger.Error(ctx, "failed to write OpenAI responses stream chunk", |
| 525 | slog.F("response_id", resp.ID), |
| 526 | slog.Error(err), |
| 527 | slog.F("error_type", "write_error"), |
| 528 | slog.F("likely_cause", "network_error"), |
| 529 | ) |
| 530 | return false |
| 531 | } |
| 532 | flusher.Flush() |
| 533 | return true |
| 534 | } |
| 535 | |
| 536 | deltaChunk := map[string]interface{}{ |
| 537 | "id": resp.ID, |
| 538 | "object": "response.output_text.delta", |
| 539 | "created": resp.Created, |
| 540 | "model": resp.Model, |
| 541 | "output_index": 0, |
| 542 | "content_index": 0, |
| 543 | "delta": resp.Output[0].Content[0].Text, |
| 544 | } |
| 545 | deltaBytes, _ := json.Marshal(deltaChunk) |
| 546 | if !writeChunk(fmt.Sprintf("data: %s\n\n", deltaBytes)) { |
| 547 | return |
| 548 | } |
| 549 | |
| 550 | finalChunk := map[string]interface{}{ |
| 551 | "id": resp.ID, |
| 552 | "object": "response.completed", |
| 553 | "created": resp.Created, |
| 554 | "model": resp.Model, |
| 555 | "response": map[string]interface{}{ |
| 556 | "id": resp.ID, |
| 557 | "object": resp.Object, |
| 558 | "created": resp.Created, |
| 559 | "model": resp.Model, |
| 560 | "output": resp.Output, |
| 561 | "usage": resp.Usage, |
| 562 | }, |
| 563 | } |
| 564 | finalBytes, _ := json.Marshal(finalChunk) |
| 565 | if !writeChunk(fmt.Sprintf("data: %s\n\n", finalBytes)) { |
no test coverage detected