Skip to content
JavaAgentic

Type at least two characters. Try “RAG”, “pgvector” or “tool calling”.

Functional Programming in Java for AI Pipelines

Functional Java refreshed for AI work: streams for document pipelines, Optional for safe metadata access, and CompletableFuture for concurrent model calls — with practical examples.

Beginner4 min readUpdated
On this page

Functional Java is not academic here. Document ingestion, embedding pipelines and concurrent model calls are naturally expressed as streams and futures. This is a targeted refresh aimed at the patterns you will actually write.

Key Takeaways

  • Streams express an ingestion pipeline (load → filter → split → embed) as the steps it is.
  • Optional is the right tool for the frequently-missing fields in model response metadata.
  • CompletableFuture composes asynchronous stages with explicit dependencies, timeouts and fallbacks.
  • Prefer clarity over cleverness — a stream that needs a comment to parse should be a loop.

Streams for document pipelines

An ingestion pipeline is a sequence of transformations. Streams let you write it as one:

List<Document> chunks = sources.stream()
        .filter(source -> source.text() != null && !source.text().isBlank())
        .map(source -> new Document(source.text(), Map.of("id", source.id())))
        .flatMap(doc -> splitter.apply(List.of(doc)).stream())
        .toList();
 
vectorStore.add(chunks);

Read top to bottom, it says exactly what happens: drop empty sources, wrap each in a Document, split each into chunks, collect. The imperative equivalent is three nested loops and a mutable list.

Grouping and counting

Collectors handle the aggregations you need for analytics and cost reporting:

// Count retrieved chunks by source, to see which documents dominate answers.
Map<String, Long> bySource = retrievedChunks.stream()
        .collect(Collectors.groupingBy(
                d -> (String) d.getMetadata().get("sourceId"),
                Collectors.counting()));
 
// Sum token usage across a batch of responses.
long totalTokens = responses.stream()
        .mapToLong(r -> r.getMetadata().getUsage().getTotalTokens())
        .sum();

Optional for missing metadata

Model responses are loosely structured. Usage may be absent, finish reasons vary, tool calls appear only sometimes. Optional makes the absent case impossible to forget:

public long promptTokens(ChatResponse response) {
    return Optional.ofNullable(response.getMetadata().getUsage())
            .map(Usage::getPromptTokens)
            .orElse(0L);
}

CompletableFuture for concurrent and composed calls

When you need several model calls and some depend on others, CompletableFuture wires the dependency graph explicitly.

// Independent calls, run concurrently, combined at the end.
CompletableFuture<String> summary =
        CompletableFuture.supplyAsync(() -> summarise(document), executor);
CompletableFuture<Sentiment> sentiment =
        CompletableFuture.supplyAsync(() -> classify(document), executor);
 
CompletableFuture<Report> report = summary.thenCombine(sentiment,
        (s, sent) -> new Report(s, sent));
 
Report result = report.get(30, TimeUnit.SECONDS);

Timeouts and fallbacks

A model call can hang. Compose the failure handling in:

String answer = CompletableFuture
        .supplyAsync(() -> chatClient.prompt().user(question).call().content(), executor)
        .orTimeout(20, TimeUnit.SECONDS)
        .exceptionally(ex -> {
            log.warn("model call failed or timed out", ex);
            return "Sorry — I could not answer that right now.";
        })
        .join();

Method references

Small readability win, used everywhere:

documents.stream()
        .map(Document::getText)       // instead of d -> d.getText()
        .filter(Objects::nonNull)
        .forEach(this::index);        // instead of t -> this.index(t)

Function composition for prompt building

Functions compose, which is handy for building a prompt through stages:

Function<String, String> redactPii = this::redact;
Function<String, String> truncate = s -> s.length() > 4000 ? s.substring(0, 4000) : s;
Function<String, String> prepare = redactPii.andThen(truncate);
 
String safe = prepare.apply(rawUserInput);

A complete example: a scored retrieval pipeline

Putting the pieces together — filter, score, sort, take:

public List<ScoredChunk> topChunks(String query, List<Document> candidates, int k) {
    float[] queryVector = embeddingModel.embed(query);
 
    return candidates.stream()
            .map(doc -> new ScoredChunk(doc, cosineSimilarity(queryVector, doc.embedding())))
            .filter(scored -> scored.score() >= 0.7)   // drop weak matches
            .sorted(Comparator.comparingDouble(ScoredChunk::score).reversed())
            .limit(k)
            .toList();
}
 
record ScoredChunk(Document doc, double score) {}

When not to use streams

Streams are the wrong tool when you need:

  • Early exit on a side effect — a loop with break is clearer than anyMatch plus state.
  • Index-based logic — streams hide the index; if you need it, loop.
  • Checked exceptions — lambdas cannot throw them cleanly; a loop handles them naturally.

Clarity wins. A stream you have to decode is worse than the loop it replaced.

Next

Frequently Asked Questions

Should I use streams or a for-loop for document processing?
Use streams when the operation is a pipeline of transformations — map, filter, collect — because it reads as the sequence of steps it performs. Use a plain loop when you need early exit, index tracking, or complex control flow. For AI ingestion pipelines (load, split, embed, store) streams usually read better, but do not force them where a loop is clearer.
CompletableFuture or virtual threads for concurrent model calls?
On Java 21, virtual threads with ordinary blocking calls are simpler for most fan-out cases. CompletableFuture still shines when you need to compose asynchronous stages — call model A, then use its result in calls B and C, then combine — with explicit dependency wiring. They also compose with timeouts and fallbacks cleanly.
Why does Optional matter for AI code?
Model response metadata is loosely structured and frequently missing fields — token usage, finish reason, tool calls may or may not be present. Optional forces you to handle the absent case explicitly instead of risking a NullPointerException on the one response that omits a field.

Related tutorials