Я интегрирую API Google Vertex AI GenateContentStream с Spring Webflux, чтобы транслировать выходные кусочки из запроса анализа файлов в качестве серверных событий (SSE). < /p>
Идея: < /p>
Загрузка файла Via /v3 /wire chind proaint < /p>
Загрузите файл Via /v3 /wire wire-точки < /p>
file via /v3 /wire wire /witd-pread < /p>
. /> Отправить его в вершину AI с помощью приглашения < /p>
< /li>
plire -кусочкам клиенту, как только вершина ai pend). (Почтальон, браузер) получает все куски только после того, как поток полностью закончен-вместо того, чтобы получить их один за другим в реальном времени. < /P>
Вот мой метод обслуживания: < /p>
public Flux analyzeFileWithPromptStream(Path filePath, String mimeType, DocumentType documentType, String prompt) {
try {
byte[] fileBytes = Files.readAllBytes(filePath);
ByteString content = ByteString.copyFrom(fileBytes);
Part filePart = Part.newBuilder()
.setInlineData(Blob.newBuilder()
.setMimeType(mimeType)
.setData(content)
.build())
.build();
String customPrompt = getCustomPromptForDocumentType(documentType);
String promptText = (prompt != null && !prompt.isEmpty())
? prompt
: """
Analyze this document thoroughly...
""" + (customPrompt.isEmpty() ? "" : ("\n" + customPrompt));
Part promptPart = Part.newBuilder().setText(promptText).build();
Content contentRequest = Content.newBuilder()
.setRole("user")
.addParts(promptPart)
.addParts(filePart)
.build();
// generativeModel.generateContentStream(
// contentRequest)
// .stream()
// .forEach(System.out::println);
// return Flux.interval(Duration.ofMillis(500))
// .map(i -> "Chunk " + i + "\n")
// .take(10);
return Flux.create(sink -> {
try {
ResponseStream stream =
generativeModel.generateContentStream(contentRequest);
for (GenerateContentResponse response : stream) {
for (Candidate candidate : response.getCandidatesList()) {
for (Part part : candidate.getContent().getPartsList()) {
String chunk = part.getText()
.replaceAll("```json", "")
.replaceAll("```", "")
.trim();
if (!chunk.isEmpty()) {
log.info("Streaming chunk: {}", chunk);
sink.next(chunk);
}
}
}
}
sink.complete();
} catch (Exception e) {
log.error("Stream error", e);
sink.error(e);
}
});
} catch (IOException e) {
log.error("Error reading file: {}", e.getMessage(), e);
return Flux.error(new RuntimeException("Error reading file", e));
}
}
< /code>
Мой контроллер: < /p>
public Flux performOcrStream(
@RequestPart("file") MultipartFile file,
@RequestPart("request") OcrReqDto request,
@RequestPart(value = "prompt", required = false) String prompt) {
if (file == null || file.isEmpty()) {
return Flux.error(new IllegalArgumentException("File is required"));
}
if (request == null || request.getDocumentType() == null) {
return Flux.error(new IllegalArgumentException("Document type is required"));
}
try {
Path tempFile = Files.createTempFile("gemini-", file.getOriginalFilename());
file.transferTo(tempFile.toFile());
String mimeType = file.getContentType();
return Flux.using(
() -> tempFile,
path -> geminiVertexService.analyzeFileWithPromptStream(path, mimeType, request.getDocumentType(), prompt)
.map(chunk -> ServerSentEvent.builder(chunk)
.event("message")
.build()),
path -> {
try {
Files.deleteIfExists(path);
log.info("Temporary file deleted: {}", path);
} catch (IOException e) {
log.warn("Failed to delete temporary file: {}", path, e);
}
}
).onErrorResume(ex -> {
return Flux.just(ServerSentEvent.builder("ERROR: " + ex.getMessage())
.event("error")
.build());
});
} catch (IOException e) {
return Flux.just(ServerSentEvent.builder("Failed to process file: " + e.getMessage())
.event("error")
.build());
}
}
< /code>
Что я попробовал: < /p>
Проверьте документацию Google в Link google Docs < /p>
Используемый mediaType.text_event_stream_value < /p>
Пробое было явно с использованием .doonnext (...). Для окончательного завершения перед тем, как показать данные < /p>
Вопрос:
Почему Spring Webflux не отправляет кусочки SSE в клиенту, как только они получены из Google Vertex AI?>
Подробнее здесь: https://stackoverflow.com/questions/797 ... at-once-in
