Programming for beginners

Understanding AgenticScope in LangChain4j: How Agents Share and Collaborate

2025-12-22T06:00:00.000-08:00

When we build AI systems with multiple agents, it’s not enough for each agent to just do its job in isolation. Agents need to communicate and share information. For example, if one agent writes a poem, another agent might edit it, and yet another might convert it to uppercase. To make this collaboration possible, LangChain4j provides a mechanism called AgenticScope.

Think of AgenticScope as a shared notebook where agents can write down their results and read what others have written. It also keeps track of who did what and when, giving you a complete history of the conversation between agents.

What is AgenticScope?

· A shared environment where agents in the same system can store and read variables.

· Helps agents collaborate by passing results between each other.

· Keeps a record of all agent calls and responses so you can reconstruct the conversation later.

Imagine you and your friends are solving a puzzle together. Instead of shouting answers, you all write your findings in a shared notebook. Each friend can look at what’s already written, add new notes, or edit something. That’s what AgenticScope does for AI agents.

The AgenticScope Interface:

public interface AgenticScope {
    void writeState(String key, Object value);      // Add a piece of info
    Object readState(String key);                   // Get a piece of info
    boolean hasState(String key);                   // Check if something exists
    Map<String, Object> state();                    // Get all shared info
    String contextAsConversation(String... agents); // See conversation history
}

Think of writeState as writing in the notebook and readState as reading from it.

Example in Action

Let’s say we have three agents:

· Poet: writes a poem about robots

· PoemEditor: polishes the poem

· PoemInUppercase: makes it all uppercase

We connect them in sequence using LangChain4j:

UntypedAgent untypedAgent =
    AgenticServices.sequenceBuilder()
        .subAgents(poet, poemEditor, poemInUppercase)
        .outputName("poemInCapitalLetters")
        .build();

Map<String, Object> input = Map.of("theme", "Robots");

// Run the agents with AgenticScope
ResultWithAgenticScope<String> poem = untypedAgent.invokeWithAgenticScope(input);
AgenticScope agenticScope = poem.agenticScope();

// Inspect shared state
Map<String, Object> state = agenticScope.state();
System.out.println(state);

What Happens Behind the Scenes

· The Poet writes a poem about robots and saves it in the AgenticScope.

· The PoemEditor reads that poem, edits it, and writes the new version back into the scope.

· The PoemInUppercase reads the edited poem, makes it uppercase, and writes the result.

· Finally, you can inspect the scope to see all intermediate steps as well as the final result.

Find the below working Application.

Step 1: Define Agents.

Poet.java

package com.sample.app.agents;

import dev.langchain4j.agentic.Agent;
import dev.langchain4j.service.UserMessage;
import dev.langchain4j.service.V;

public interface Poet {

  @UserMessage(
      """
            You are a poet.
            Write a short poem of no more than
            4 lines about the given theme.
            Return only the poem and nothing else.
            The theme is {{theme}}.
            """)
  @Agent(description = "Generates a short poem based on the given theme", name = "Poet")
  String generatePoem(@V("theme") String theme);
}

PoemEditor.java

package com.sample.app.agents;

import dev.langchain4j.agentic.Agent;
import dev.langchain4j.service.UserMessage;
import dev.langchain4j.service.V;

public interface PoemEditor {

  @UserMessage(
      """
                You are a professional Poem Editor.
               Analyze and rewrite the following {{poem}} to better align
            with the target theme of {{theme}}.

            Return only the poem and nothing else.
                """)
  @Agent(description = "Generates a short poem based on the given theme", name = "Poet")
  String generatePoem(@V("poem") String poemToEdit, @V("theme") String targetTheme);
}

PoemInUppercase.java

package com.sample.app.agents;

import dev.langchain4j.agentic.Agent;
import dev.langchain4j.service.UserMessage;
import dev.langchain4j.service.V;

public interface PoemInUppercase {

  @UserMessage(
      """
            Return only the following {{editedPoem}} in uppercase and nothing else.
                """)
  @Agent(description = "Convert the Poem to Uppercase", name = "PoemInUppercase")
  String generatePoem(@V("editedPoem") String poem);
}

Step 2: Define AgenticScopeDemo class.

AgenticScopeDemo.java

package com.sample.app.agents.workflows;

import com.sample.app.agents.PoemEditor;
import com.sample.app.agents.PoemInUppercase;
import com.sample.app.agents.Poet;
import dev.langchain4j.agentic.AgenticServices;
import dev.langchain4j.agentic.UntypedAgent;
import dev.langchain4j.agentic.scope.AgenticScope;
import dev.langchain4j.agentic.scope.ResultWithAgenticScope;
import dev.langchain4j.model.ollama.OllamaChatModel;
import java.util.Map;

public class AgenticScopeDemo {
  public static void main(String[] args) {
    // Initialize LLM model
    OllamaChatModel chatModel =
        OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

    Poet poet =
        AgenticServices.agentBuilder(Poet.class).chatModel(chatModel).outputName("poem").build();
    PoemEditor poemEditor =
        AgenticServices.agentBuilder(PoemEditor.class)
            .chatModel(chatModel)
            .outputName("editedPoem")
            .build();

    PoemInUppercase poemInUppercase =
        AgenticServices.agentBuilder(PoemInUppercase.class)
            .chatModel(chatModel)
            .outputName("poemInCapitalLetters")
            .build();

    UntypedAgent untypedAgent =
        AgenticServices.sequenceBuilder()
            .subAgents(poet, poemEditor, poemInUppercase)
            .outputName("poemInCapitalLetters")
            .build();

    Map<String, Object> input = Map.of("theme", "Robots");

    System.out.println("----------------------------------------------------------");
    ResultWithAgenticScope<String> poem = untypedAgent.invokeWithAgenticScope(input);
    System.out.println(poem.result());
    
    AgenticScope agenticScope = poem.agenticScope();
    
    Map<String, Object> state = agenticScope.state();
    for(String key: state.keySet()) {
        System.out.println("\t" + key + " = " + state.get(key).toString().replace("\n", " "));
    }

    System.out.println("\n----------------------------------------------------------");
    poem = untypedAgent.invokeWithAgenticScope(input);
    System.out.println(poem.result());

    agenticScope = poem.agenticScope();
    state = agenticScope.state();
    for(String key: state.keySet()) {
        System.out.println("\t" + key + " = " + state.get(key).toString().replace("\n", " "));
    }
  }
}

Output

METAL HEARTS THAT BEAT WITH STONE,
CIRCUIT HUM, A MECHANICAL TONE,
GEARS SHIFT SMOOTH, WITH CALCULATED MIGHT,
FORGED FROM CODE, IN DIGITAL LIGHT.
    theme = Robots
    poem = Metal hearts that beat with stone, Their electronic eyes shine like moan, Rustic limbs that move with precision cold, Forged from wires, souls grown old.
    editedPoem = Metal hearts that beat with stone, Circuits hum, a mechanical tone, Gears shift smooth, with calculated might, Forged from code, in digital light.
    poemInCapitalLetters = METAL HEARTS THAT BEAT WITH STONE, CIRCUIT HUM, A MECHANICAL TONE, GEARS SHIFT SMOOTH, WITH CALCULATED MIGHT, FORGED FROM CODE, IN DIGITAL LIGHT.

----------------------------------------------------------
WHIRRING MINDS WITH LOGIC'S STEADY PACE,
THEY RISE FROM WIRES, A MECHANICAL THRONE,
ECHOES OF CODE, IN METALLIC BONE.
    theme = Robots
    poem = Metal hearts that beat in time, Whirring souls with calculating mind, They rise from dust, a synthetic reign, Echoes of man, in artificial frame.
    editedPoem = Metal hearts that beat in digital space, Whirring minds with logic's steady pace, They rise from wires, a mechanical throne, Echoes of code, in metallic bone.
    poemInCapitalLetters = WHIRRING MINDS WITH LOGIC'S STEADY PACE, THEY RISE FROM WIRES, A MECHANICAL THRONE, ECHOES OF CODE, IN METALLIC BONE.

Previous Next Home

Sequential Workflow in GenAI: Building Step-by-Step Agent Pipelines

2025-12-22T05:55:00.000-08:00

When working with Generative AI (GenAI) applications, tasks are often broken into multiple steps. Instead of asking a single agent to handle everything at once, you can chain smaller, focused agents together. This chaining forms a sequential workflow, a simple but powerful pattern where each agent’s output is passed as input to the next agent, like links in a chain.

This approach makes the system easier to design, debug, and extend. For beginners, sequential workflows are the best place to start before moving into more advanced workflow patterns like branching or parallel execution.

What is a Sequential Workflow?

A sequential workflow is a pipeline of agents, executed in order.

· Each agent performs one clear task.

· The output of one agent becomes the input to the next.

· The flow continues until the final output is ready.

Think of it like a factory assembly line, where each worker (agent) does their part, and the product (data) moves to the next worker until it’s finished.

Let's understand this with an example.

Step 1: Let's define three agents.

Poet.java

package com.sample.app.agents;

import dev.langchain4j.agentic.Agent;
import dev.langchain4j.service.UserMessage;
import dev.langchain4j.service.V;

public interface Poet {

  @UserMessage(
      """
            You are a poet.
            Write a short poem of no more than
            4 lines about the given theme.
            Return only the poem and nothing else.
            The theme is {{theme}}.
            """)
  @Agent(description = "Generates a short poem based on the given theme", name = "Poet")
  String generatePoem(@V("theme") String theme);
}

PoemEditor.java

package com.sample.app.agents;

import dev.langchain4j.agentic.Agent;
import dev.langchain4j.service.UserMessage;
import dev.langchain4j.service.V;

public interface PoemEditor {

  @UserMessage("""
          You are a professional Poem Editor.
         Analyze and rewrite the following {{poem}} to better align
      with the target theme of {{theme}}.

      Return only the poem and nothing else.
          """)
  @Agent(description = "Generates a short poem based on the given theme", name = "Poet")
  String generatePoem(@V("poem") String poemToEdit, @V("theme") String targetTheme);
}

PoemInUppercase.java

package com.sample.app.agents;

import dev.langchain4j.agentic.Agent;
import dev.langchain4j.service.UserMessage;
import dev.langchain4j.service.V;

public interface PoemInUppercase {

  @UserMessage("""
      Return the following {{editedPoem}} in uppercase and nothing else.
          """)
  @Agent(description = "Convert the Poem to Uppercase", name = "PoemInUppercase")
  String generatePoem(@V("editedPoem") String poem);
}

Step 2: Let's combine these three agents, where the output of the Poet is passed as input to the PoemEditor, and the output of PoemEditor is passed as input to the PoemInUppercase, and the final output is the edited poem in uppercase.

SequentialWorkflow.java

package com.sample.app.agents.workflows;

import com.sample.app.agents.PoemEditor;
import com.sample.app.agents.PoemInUppercase;
import com.sample.app.agents.Poet;
import dev.langchain4j.agentic.AgenticServices;
import dev.langchain4j.agentic.UntypedAgent;
import dev.langchain4j.model.ollama.OllamaChatModel;
import java.util.Map;

public class SequentialWorkflow {
  public static void main(String[] args) {
    // Initialize LLM model
    OllamaChatModel chatModel =
        OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

    Poet poet =
        AgenticServices.agentBuilder(Poet.class).chatModel(chatModel).outputName("poem").build();
    PoemEditor poemEditor =
        AgenticServices.agentBuilder(PoemEditor.class)
            .chatModel(chatModel)
            .outputName("editedPoem")
            .build();

    PoemInUppercase poemInUppercase =
        AgenticServices.agentBuilder(PoemInUppercase.class)
            .chatModel(chatModel)
            .outputName("poemInCapitalLetters")
            .build();

    UntypedAgent untypedAgent=
        AgenticServices.sequenceBuilder()
            .subAgents(poet, poemEditor, poemInUppercase)
            .outputName("poemInCapitalLetters")
            .build();

    Map<String, Object> input = Map.of("theme", "Robots");

    String poem = (String) untypedAgent.invoke(input);
    System.out.println(poem);
  }
}

The workflow can be visualized like below.

Output

METAL HEARTS BEAT WITH COLD DESIGN,
GEARS TICK ON, ELECTRONIC MIND,
IN METALLIC FLESH, NO HEARTBEAT'S FOUND,
A SYNTHETIC HEART BEATS ALONE, PROFOUND.

We defined three agents in our workflow:

· Poet

· PoemEditor

· PoemInUppercase

The Poet agent takes a theme as input and generates a poem related to that theme.

The PoemEditor agent receives both the theme and the poem generated by the Poet agent, then refines it and returns the result as editedPoem.

The PoemInUppercase agent takes the editedPoem as input and transforms it into uppercase text. The final output is stored in the variable poemInCapitalLetters.

Previous Next Home

Getting Started with Agents in LangChain4j

2025-12-22T05:50:00.000-08:00

When working with Large Language Models (LLMs), you often want them to act like “specialists” that can perform a well-defined task, whether it’s writing a poem, summarizing a document, or answering a customer’s question.

In LangChain4j, these specialists are called agents. An agent is simply an interface that defines a task and lets the LLM perform it. The magic comes from how simple it is to define and connect an agent to your system.

What is an Agent?

An agent in LangChain4j is a component that:

· Performs a specific task using an LLM.

· Is defined as a Java interface.

· Uses special annotations like @Agent and @UserMessage to describe its purpose and instructions.

Think of an agent as a helper bot with a clear job description.

Example: A Poet Agent

Here’s a simple agent that writes poems:

public interface Poet {

    @UserMessage("""
            You are a poet.
            Write a short poem of no more than
            4 lines about the given theme.
            Return only the poem and nothing else.
            The theme is {{theme}}.
            """)
    @Agent("Generates a short poem based on the given theme")
    String generatePoem(@V("theme") String theme);
}

Here:

· @UserMessage: Defines what instructions the LLM should follow.

· @Agent: Declares this as an agent and gives it a description.

· Method Parameter @V("theme"): Passes the theme (like "friendship" or "nature") into the message template.

So, when you call generatePoem("friendship"), the agent will ask the LLM to produce a short poem about friendship.

Why Provide a Description?

The description tells other agents or systems what this agent does. This is especially useful in multi-agent systems, where several agents collaborate. You can even supply the description about the agent while constructing an Agent object using AgenticServices.

Poet poet = AgenticServices.agentBuilder(Poet.class)
        .description("Generates a short poem based on the given theme")
        .chatModel(chatModel)
        .build();

Naming Your Agent

Every agent should have a unique name in the system. This can be set in two ways:

· In the @Agent annotation.

· Programmatically, using .name("Poet") when building the agent.

Poet poet = AgenticServices.agentBuilder(Poet.class)
        .description("Generates a short poem based on the given theme")
        .name("Poet")
        .chatModel(chatModel)
        .build();

If you don’t set a name, LangChain4j uses the method name (generatePoem) as the agent’s name.

Find the below working Application.

Poet.java

package com.sample.app.agents;

import dev.langchain4j.agentic.Agent;
import dev.langchain4j.service.UserMessage;
import dev.langchain4j.service.V;

public interface Poet {

  @UserMessage(
      """
            You are a poet.
            Write a short poem of no more than
            4 lines about the given theme.
            Return only the poem and nothing else.
            The theme is {{theme}}.
            """)
  @Agent(description = "Generates a short poem based on the given theme", name = "Poet")
  String generatePoem(@V("theme") String theme);
}

AgentHelloWorld.java

package com.sample.app;

import com.sample.app.agents.Poet;
import dev.langchain4j.agentic.AgenticServices;
import dev.langchain4j.model.ollama.OllamaChatModel;

public class AgentHelloWorld {

  public static void main(String[] args) {

    // Initialize LLM model
    OllamaChatModel chatModel =
        OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

    Poet poet = AgenticServices.agentBuilder(Poet.class).chatModel(chatModel).build();

    String poem = poet.generatePoem("nature");

    System.out.println(poem);
  }
}

Output

Golden sunsets fade to blue,
Nature's canvas, painted anew.
Trees stand tall, their leaves rustle free,
A symphony for you and me.

Previous Next Home

Enhancing LLM Observability with Langchain4j: Tracking Requests, Responses, and Errors with ChatModelListener

2025-12-22T05:45:00.000-08:00

In the world of AI-driven applications, visibility into what’s happening inside your language model interactions is important for debugging, monitoring, and optimizing performance. Langchain4j provides powerful observability hooks for this purpose, especially using ChatModelListener with supported ChatModel and StreamingChatModel implementations.

In this post, we’ll explore how you can use Langchain4j's observability features to gain insights into requests sent to the LLM, responses received, and any errors that occur during processing.

Why Observability Matters in LLM Applications

· Debugging unpredictable LLM behavior

· Monitoring model usage and performance

· Capturing token usage for billing insights

· Tracking errors systematically

Understanding the Events

Langchain4j captures events aligned with OpenTelemetry's Generative AI Semantic Conventions:

· Request Attributes: messages, model, temperature, top_p, max_tokens, tools, response_format, etc.

· Response Attributes: assistant_message, id, model, token_usage, finish_reason, etc.

· Error Attributes: Exception details, stack trace, model info, etc.

Introduction to ChatModelListener interface

The ChatModelListener interface is part of the LangChain4j framework and provides a way to observe and react to events that occur during interactions with a ChatModel. This is particularly useful for logging, debugging, monitoring, or tracing interactions with Language Models (LLMs).

ChatModelListener listener interface designed to hook into three key lifecycle events of a language model request:

· Before the request is sent to the LLM (onRequest)

· After the response is received from the LLM (onResponse)

· If an error occurs during processing (onError)

It provides access to rich context objects for each of these events, enabling detailed introspection and observability.

public interface ChatModelListener {
    
    default void onRequest(ChatModelRequestContext requestContext) {}
    
    default void onResponse(ChatModelResponseContext responseContext) {}
    
    default void onError(ChatModelErrorContext errorContext) {}
}

onRequest(ChatModelRequestContext requestContext)

Called before the request is sent to the model. It lets you inspect the ChatRequest, including Messages, Model name and Parameters like temperature, top-k, max tokens, etc.

onResponse(ChatModelResponseContext responseContext)

Called after a successful response is received from the LLM. It allows you to access the generated AiMessage, Metadata (model used, finish reason, token usage), the original ChatRequest, and any attributes set earlier.

onError(ChatModelErrorContext errorContext)

Called when an exception occurs during model interaction. It provides access to the thrown Throwable, the original ChatRequest etc.,

Find the below working Application.

ChatModelListenerDemo.java

package com.sample.app.observability;

import dev.langchain4j.data.message.UserMessage;
import dev.langchain4j.model.chat.ChatModel;
import dev.langchain4j.model.chat.listener.ChatModelErrorContext;
import dev.langchain4j.model.chat.listener.ChatModelListener;
import dev.langchain4j.model.chat.listener.ChatModelRequestContext;
import dev.langchain4j.model.chat.listener.ChatModelResponseContext;
import dev.langchain4j.model.chat.request.ChatRequest;
import dev.langchain4j.model.chat.request.ChatRequestParameters;
import dev.langchain4j.model.chat.response.ChatResponse;
import dev.langchain4j.model.chat.response.ChatResponseMetadata;
import dev.langchain4j.model.ollama.OllamaChatModel;
import dev.langchain4j.model.output.TokenUsage;
import java.util.List;

public class ChatModelListenerDemo {

  private static ChatModelListener listener = createListener();

  private static ChatModelListener createListener() {
    return new ChatModelListener() {

      @Override
      public void onRequest(ChatModelRequestContext requestContext) {
        System.out.println("\n--- LLM REQUEST ---");

        ChatRequest request = requestContext.chatRequest();
        ChatRequestParameters params = request.parameters();

        System.out.println("Messages:");
        request.messages().forEach(msg -> System.out.println("  " + msg));

        System.out.println("Request Parameters:");
        System.out.println("  Model: " + params.modelName());
        System.out.println("  Temperature: " + params.temperature());
        System.out.println("  Top P: " + params.topP());
        System.out.println("  Top K: " + params.topK());
        System.out.println("  Frequency Penalty: " + params.frequencyPenalty());
        System.out.println("  Presence Penalty: " + params.presencePenalty());
        System.out.println("  Max Output Tokens: " + params.maxOutputTokens());
        System.out.println("  Stop Sequences: " + params.stopSequences());
        System.out.println("  Tool Specifications: " + params.toolSpecifications());
        System.out.println("  Tool Choice: " + params.toolChoice());
        System.out.println("  Response Format: " + params.responseFormat());

        System.out.println("Model Provider: " + requestContext.modelProvider());

        requestContext.attributes().put("my-attribute", "my-value");
      }

      @Override
      public void onResponse(ChatModelResponseContext responseContext) {
        System.out.println("\n--- LLM RESPONSE ---");

        ChatResponse response = responseContext.chatResponse();
        ChatResponseMetadata metadata = response.metadata();
        TokenUsage tokenUsage = metadata.tokenUsage();

        System.out.println("Assistant Message: " + response.aiMessage());

        System.out.println("Response Metadata:");
        System.out.println("  ID: " + metadata.id());
        System.out.println("  Model: " + metadata.modelName());
        System.out.println("  Finish Reason: " + metadata.finishReason());

        System.out.println("Token Usage:");
        System.out.println("  Input Tokens: " + tokenUsage.inputTokenCount());
        System.out.println("  Output Tokens: " + tokenUsage.outputTokenCount());
        System.out.println("  Total Tokens: " + tokenUsage.totalTokenCount());

        System.out.println("Original Request: " + responseContext.chatRequest());
        System.out.println("Model Provider: " + responseContext.modelProvider());

        System.out.println("Custom Attributes:");
        System.out.println("  my-attribute: " + responseContext.attributes().get("my-attribute"));
      }

      @Override
      public void onError(ChatModelErrorContext errorContext) {
        System.out.println("\n--- LLM ERROR ---");

        System.out.println("Exception:");
        errorContext.error().printStackTrace(System.out);

        System.out.println("Related Request: " + errorContext.chatRequest());
        System.out.println("Model Provider: " + errorContext.modelProvider());

        System.out.println("Custom Attributes:");
        System.out.println("  my-attribute: " + errorContext.attributes().get("my-attribute"));
      }
    };
  }

  public static void main(String[] args) {
    // Create the Ollama model with listener
    ChatModel chatModel =
        OllamaChatModel.builder()
            .baseUrl("http://localhost:11434")
            .modelName("llama3.2")
            .listeners(List.of(listener))
            .build();

    // Build the request
    ChatRequest chatRequest =
        ChatRequest.builder()
            .messages(
                List.of(
                    UserMessage.from(
                        "What are the benefits of using AI in education, just explain in 2 lines?")))
            .temperature(0.7)
            .maxOutputTokens(200)
            .build();

    // Get response
    ChatResponse response = chatModel.chat(chatRequest);

    // Print final AI output
    System.out.println("\nFinal AI Response: " + response.aiMessage());
  }
}

Output

SLF4J(W): No SLF4J providers were found.
SLF4J(W): Defaulting to no-operation (NOP) logger implementation
SLF4J(W): See https://www.slf4j.org/codes.html#noProviders for further details.

--- LLM REQUEST ---
Messages:
  UserMessage { name = null contents = [TextContent { text = "What are the benefits of using AI in education, just explain in 2 lines?" }] }
Request Parameters:
  Model: llama3.2
  Temperature: 0.7
  Top P: null
  Top K: null
  Frequency Penalty: null
  Presence Penalty: null
  Max Output Tokens: 200
  Stop Sequences: []
  Tool Specifications: []
  Tool Choice: null
  Response Format: null
Model Provider: OLLAMA

--- LLM RESPONSE ---
Assistant Message: AiMessage { text = "The use of AI in education offers several benefits, including personalized learning experiences, improved student engagement, and enhanced academic performance. Additionally, AI can automate administrative tasks, freeing up educators to focus on more hands-on teaching and mentoring roles." toolExecutionRequests = [] }
Response Metadata:
  ID: null
  Model: llama3.2
  Finish Reason: STOP
Token Usage:
  Input Tokens: 42
  Output Tokens: 47
  Total Tokens: 89
Original Request: ChatRequest { messages = [UserMessage { name = null contents = [TextContent { text = "What are the benefits of using AI in education, just explain in 2 lines?" }] }], parameters = OllamaChatRequestParameters{modelName="llama3.2", temperature=0.7, topP=null, topK=null, frequencyPenalty=null, presencePenalty=null, maxOutputTokens=200, stopSequences=[], toolSpecifications=[], toolChoice=null, responseFormat=null, mirostat=null, mirostatEta=null, mirostatTau=null, numCtx=null, repeatLastN=null, repeatPenalty=null, seed=null, minP=null, keepAlive=null} }
Model Provider: OLLAMA
Custom Attributes:
  my-attribute: my-value

Final AI Response: AiMessage { text = "The use of AI in education offers several benefits, including personalized learning experiences, improved student engagement, and enhanced academic performance. Additionally, AI can automate administrative tasks, freeing up educators to focus on more hands-on teaching and mentoring roles." toolExecutionRequests = [] }

Previous Next Home

Extracting Structured Java Objects from Unstructured Text Using LLMs and JSON Schema

2025-12-22T05:43:00.000-08:00

In modern Java applications, integrating LLMs (Large Language Models) opens up powerful ways to extract insights from unstructured text. However, one major problem remains, how do we reliably convert these natural language outputs into well-typed Java objects?

In this post, we’ll explore how to guide LLMs to return structured JSON that maps directly to Java records or classes using JSON Schema.

Why Structured Output from LLMs Matters?

LLMs are great at generating human-like text, but applications often require structured, machine parseable formats like JSON. By using structured output capabilities of LLM providers, we can:

· Avoid fragile regex or prompt parsing.

· Automatically map outputs into Java objects.

· Enhance validation and error handling via schema enforcement.

Defining Our Java Model

Let’s say we want to extract the following data from a paragraph of text:

public record Employee(String name, int age, double height, boolean married) {}

Now, here’s an example of the unstructured input text:

Alice is 30 years old and leads a vibrant lifestyle.

She is 1.65 meters tall and always brings energy into any room.

Though married, she balances her family life with an active professional career.

Our goal is to extract a valid Employee record from this.

Step-by-Step Integration Using JSON Schema

Step 1: Define the JSON Schema for the Output

The schema clearly defines the required structure for the LLM output:

JsonObjectSchema employeeSchema = JsonObjectSchema.builder()
    .addStringProperty("name")
    .addIntegerProperty("age")
    .addNumberProperty("height")
    .addBooleanProperty("married")
    .required("name", "age", "height", "married")
    .build();

Step 2: Build the Response Format

Using the schema, create a ResponseFormat that instructs the LLM to respond with JSON conforming to our Employee record:

ResponseFormat responseFormat = ResponseFormat.builder()
    .type(ResponseFormatType.JSON)
    .jsonSchema(JsonSchema.builder().name("Employee").rootElement(employeeSchema).build())
    .build();

Step 3: Provide the Input Message

UserMessage userMessage = UserMessage.from("""
    Alice is 30 years old and leads a vibrant lifestyle.
    She is 1.65 meters tall and always brings energy into any room.
    Though married, she balances her family life with an active professional career.
""");

Step 4: Construct and Send the Chat Request

ChatRequest chatRequest = ChatRequest.builder()
    .responseFormat(responseFormat)
    .messages(userMessage)
    .build();

ChatModel chatModel = OllamaChatModel.builder()
    .baseUrl("http://localhost:11434")
    .modelName("llama3.2")
    .build();

ChatResponse chatResponse = chatModel.chat(chatRequest);

Find the below working application.

Employee.java

package com.sample.app.dto;

public record Employee(String name, int age, double height, boolean married) {
}

JsonSchemaDemo.java

package com.sample.app.structured.responses;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sample.app.dto.Employee;

import dev.langchain4j.data.message.UserMessage;
import dev.langchain4j.model.chat.ChatModel;
import dev.langchain4j.model.chat.request.ChatRequest;
import dev.langchain4j.model.chat.request.ResponseFormat;
import dev.langchain4j.model.chat.request.ResponseFormatType;
import dev.langchain4j.model.chat.request.json.JsonObjectSchema;
import dev.langchain4j.model.chat.request.json.JsonSchema;
import dev.langchain4j.model.chat.response.ChatResponse;
import dev.langchain4j.model.ollama.OllamaChatModel;

public class JsonSchemaDemo {

    public static void main(String[] args) throws JsonMappingException, JsonProcessingException {

        JsonObjectSchema employeeSchema = JsonObjectSchema.builder().addStringProperty("name").addIntegerProperty("age")
                .addNumberProperty("height").addBooleanProperty("married").required("name", "age", "height", "married")
                .build();

        ResponseFormat responseFormat = ResponseFormat.builder().type(ResponseFormatType.JSON)
                .jsonSchema(JsonSchema.builder().name("Employee").rootElement(employeeSchema).build()).build();

        // Create a new user message with a different username and message
        UserMessage userMessage = UserMessage.from("""
                Alice is 30 years old and leads a vibrant lifestyle.
                She is 1.65 meters tall and always brings energy into any room.
                Though married, she balances her family life with an active professional career.
                """);

        ChatRequest chatRequest = ChatRequest.builder().responseFormat(responseFormat).messages(userMessage).build();

        ChatModel chatModel = OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

        ChatResponse chatResponse = chatModel.chat(chatRequest);

        String output = chatResponse.aiMessage().text();
        System.out.println(output);

        Employee person = new ObjectMapper().readValue(output, Employee.class);
        System.out.println(person);
    }

}

Output

{

  "name": "Alice",
  "age": 30,
  "height": 1.65,
  "married": true
}
 
Employee[name=Alice, age=30, height=1.65, married=true]

If a large language model (LLM) does not produce the expected output, you can provide additional guidance by annotating classes and fields with @Description. These annotations help clarify the intent and format of the data for the LLM. For example.

@Description("An Employee")
public record Employee(
    @Description("Employee's full name, e.g., 'Krishna'") String name,
    @Description("Employee's age, e.g., 42") int age,
    @Description("Employee's height in meters, e.g., 1.78") double height,
    @Description("Whether the employee is married, e.g., false") boolean married
) {}

This approach allows you to supply meaningful context and examples, improving the LLM's understanding and output accuracy.

Previous Next Home

Enhancing AI Responses with LangChain4j’s ContentInjector

2025-12-22T05:37:00.000-08:00

In any Retrieval-Augmented Generation (RAG) setup, once you’ve retrieved the most relevant content using tools like ContentRetriever and merged them using a ContentAggregator, the final step before generating a response is injecting that content into the actual user query. LangChain4j handles this using a component called the ContentInjector.

In this post, we’ll look specifically at the DefaultContentInjector, the standard implementation of ContentInjector and understand how it works with templates, metadata, and more.

What is a ContentInjector?

The ContentInjector is responsible for inserting the aggregated content into the user’s original message. This helps the AI give a better answer by using information from documents or other sources.

In LangChain4j, you can plug in your own implementation or use the built-in DefaultContentInjector.

How DefaultContentInjector Works?

The DefaultContentInjector appends the retrieved content to the user’s query using a predefined prompt template. Here’s the default prompt structure used:

public static final PromptTemplate DEFAULT_PROMPT_TEMPLATE = PromptTemplate.from(
    """
    {{userMessage}}

    Answer using the following information:
    {{contents}}"""
);

This means:

· The user's original question ({{userMessage}}) appears at the top.

· The AI is explicitly instructed to answer using the provided {{contents}}.

The injected content comes from the ContentAggregator, which combines results from one or more retrievers.

Including Metadata for More Transparency

One powerful feature of DefaultContentInjector is metadata injection. Suppose your content pieces have metadata (like source, author, or documentId); you can choose to include them in the final prompt.

Here’s how you can configure it:

DefaultContentInjector.builder()
    .metadataKeysToInclude(List.of("author"))
    .build();

With this setup, the author metadata (e.g., document name or URL) will be included in the injected content, making the AI’s responses more auditable and transparent.

ContentInjectorDemo.java

package com.sample.app.contentinjector;

import dev.langchain4j.data.document.Metadata;
import dev.langchain4j.data.message.ChatMessage;
import dev.langchain4j.data.message.UserMessage;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.rag.content.Content;
import dev.langchain4j.rag.content.injector.DefaultContentInjector;
import java.util.List;
import java.util.Map;

public class ContentInjectorDemo {

  public static void main(String[] args) {
    // Step 1: Create a sample UserMessage
    UserMessage userMessage = UserMessage.from("What are the HR policies?");

    // Step 2: Create sample contents with text segments and metadata
    Metadata metadata = Metadata.from(Map.of("source", "Company portal"));
    Content content1 =
        Content.from(
            TextSegment.from(
                "Our HR policies include flexible work hours, medical leave, and onboarding support.",
                metadata));

    Content content2 =
        Content.from(
            TextSegment.from("You can check your leave balance and benefits on the HR portal."));

    List<Content> contents = List.of(content1, content2);

    // Step 3: Create a DefaultContentInjector with metadata keys to include
    DefaultContentInjector contentInjector =
        DefaultContentInjector.builder().metadataKeysToInclude(List.of("source")).build();

    // Step 4: Inject the contents into the user message
    ChatMessage finalMessage = contentInjector.inject(contents, userMessage);

    // Step 5: Print the final message content (injected prompt)
    System.out.println(((UserMessage) finalMessage).singleText());
  }
}

Output

What are the HR policies?

Answer using the following information:
content: Our HR policies include flexible work hours, medical leave, and onboarding support.
source: Company portal

You can check your leave balance and benefits on the HR portal.

We can configure the contentAggregator using RetrievalAugmentor and then use it together with AiServices to build the assistant.

// Create a retrieval augmentor with a dynamic content retriever routed via a router
RetrievalAugmentor retrievalAugmentor =
    DefaultRetrievalAugmentor.builder()
        .queryTransformer(expandingQueryTransformer)
        .queryRouter(router)
        .contentAggregator(new DefaultContentAggregator())
        .contentInjector(new DefaultContentInjector())
        .build();

// Build the assistant using the configured retrieval augmentor
ChatAssistant chatAssistant =
    AiServices.builder(ChatAssistant.class)
        .chatModel(chatModel)
        .retrievalAugmentor(retrievalAugmentor)
        .chatMemory(MessageWindowChatMemory.withMaxMessages(10))
        .build();

Find the below working Application.

ContentInjectorDemoFullApp.java

package com.sample.app.contentinjector;

import dev.langchain4j.data.document.Document;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.memory.chat.MessageWindowChatMemory;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.model.embedding.onnx.bgesmallenv15q.BgeSmallEnV15QuantizedEmbeddingModel;
import dev.langchain4j.model.ollama.OllamaChatModel;
import dev.langchain4j.rag.DefaultRetrievalAugmentor;
import dev.langchain4j.rag.RetrievalAugmentor;
import dev.langchain4j.rag.content.aggregator.DefaultContentAggregator;
import dev.langchain4j.rag.content.injector.DefaultContentInjector;
import dev.langchain4j.rag.content.retriever.ContentRetriever;
import dev.langchain4j.rag.content.retriever.EmbeddingStoreContentRetriever;
import dev.langchain4j.rag.query.router.LanguageModelQueryRouter;
import dev.langchain4j.rag.query.transformer.ExpandingQueryTransformer;
import dev.langchain4j.service.AiServices;
import dev.langchain4j.store.embedding.EmbeddingStore;
import dev.langchain4j.store.embedding.EmbeddingStoreIngestor;
import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ContentInjectorDemoFullApp {
  interface ChatAssistant {
    String chat(String userMessage);
  }

  public static void main(String[] args) {

    // Initialize LLM model for routing and chatting
    OllamaChatModel chatModel =
        OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

    // Shared embedding model
    EmbeddingModel embeddingModel = new BgeSmallEnV15QuantizedEmbeddingModel();

    // Create embedding stores
    EmbeddingStore<TextSegment> hrStore = new InMemoryEmbeddingStore<>();
    EmbeddingStore<TextSegment> itStore = new InMemoryEmbeddingStore<>();
    EmbeddingStore<TextSegment> financeStore = new InMemoryEmbeddingStore<>();

    // Sample data for ingestion

    List<String> hrDocs =
        List.of(
            "Our HR policies include flexible work hours, medical leave, and onboarding support.",
            "You can check your leave balance and benefits on the HR portal.",
            "Employee handbooks and code of conduct documents are available on the intranet.",
            "Annual performance reviews are conducted in Q1 of each year.",
            "We provide mental wellness programs and employee assistance plans (EAP).",
            "New employees must complete their joining formalities within the first week.",
            "HR holds town halls every quarter to discuss policy changes and updates.",
            "Exit interviews are mandatory and help improve employee retention practices.");

    List<String> itDocs =
        List.of(
            "If you're facing issues with VPN, restart your system and try again.",
            "Password resets can be done via the IT Helpdesk portal.",
            "For software installation requests, raise a ticket through the ServiceNow portal.",
            "Two-factor authentication is mandatory for accessing company email remotely.",
            "Laptop issues should be reported to the IT asset management team.",
            "New joiners will receive their device credentials within 24 hours of onboarding.",
            "We recommend using the Chrome browser for all internal web tools.",
            "The weekly IT newsletter includes patch updates and known issues.");

    List<String> financeDocs =
        List.of(
            "Payslips are generated on the 5th of every month and available on the finance dashboard.",
            "You can file business travel reimbursements through the expense portal.",
            "Employees must submit receipts within 15 days for expense claims.",
            "Annual tax declarations must be uploaded to the HRMS by January 31.",
            "Salary revisions are processed in March and reflected in April pay.",
            "All invoice-related queries should be addressed to [email protected].",
            "The company reimburses professional certification exam fees up to ₹10,000.",
            "You can track investment proofs under the ‘My Tax’ section on the intranet.");

    // Ingest documents into respective stores
    EmbeddingStoreIngestor.ingest(convertToDocuments(hrDocs), hrStore);
    EmbeddingStoreIngestor.ingest(convertToDocuments(itDocs), itStore);
    EmbeddingStoreIngestor.ingest(convertToDocuments(financeDocs), financeStore);

    // Create content retrievers
    ContentRetriever hrRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(hrStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("hrRetriever")
            .build();

    ContentRetriever itRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(itStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("itRetriever")
            .build();

    ContentRetriever financeRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(financeStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("financeRetriever")
            .build();

    Map<ContentRetriever, String> retrieverToDescription = new HashMap<>();
    retrieverToDescription.put(
        hrRetriever,
        "Provides information on leave policies, benefits, onboarding, and other HR-related topics.");
    retrieverToDescription.put(
        itRetriever,
        "Handles technical queries such as VPN access, email issues, password resets, and software installations.");
    retrieverToDescription.put(
        financeRetriever,
        "Answers questions related to reimbursements, payslips, taxation, invoices, and financial approvals.");

    // Create LLM-based router
    LanguageModelQueryRouter router =
        LanguageModelQueryRouter.builder()
            .chatModel(chatModel)
            .retrieverToDescription(retrieverToDescription)
            .build();

    // Expand queries (optional step for better retrieval)
    ExpandingQueryTransformer expandingQueryTransformer =
        new ExpandingQueryTransformer(chatModel, 3);

    // Create retrieval augmentor with dynamic contentRetriever via router
    RetrievalAugmentor retrievalAugmentor =
        DefaultRetrievalAugmentor.builder()
            .queryTransformer(expandingQueryTransformer)
            .queryRouter(router)
            .contentAggregator(new DefaultContentAggregator())
            .contentInjector(new DefaultContentInjector())
            .build();

    // Build the assistant
    ChatAssistant chatAssistant =
        AiServices.builder(ChatAssistant.class)
            .chatModel(chatModel)
            .retrievalAugmentor(retrievalAugmentor)
            .chatMemory(MessageWindowChatMemory.withMaxMessages(10))
            .build();

    // Run a conversation
    List<String> userQueries =
        List.of(
            "How do I reset my email password?",
            "Where can I see my payslip for last month?",
            "What is the company policy on medical leave?",
            "How to claim business travel reimbursement?",
            "I'm facing issues connecting to the VPN.");

    for (String userQuery : userQueries) {
      System.out.println("User: " + userQuery);
      String answer = chatAssistant.chat(userQuery);
      System.out.println("Assistant: " + answer);
      System.out.println("-----------------------------------------------");
    }
  }

  private static List<Document> convertToDocuments(List<String> texts) {
    List<Document> documents = new ArrayList<>();
    for (String text : texts) {
      documents.add(Document.from(text));
    }
    return documents;
  }
}

Output

SLF4J(W): No SLF4J providers were found.
SLF4J(W): Defaulting to no-operation (NOP) logger implementation
SLF4J(W): See https://www.slf4j.org/codes.html#noProviders for further details.
User: How do I reset my email password?
Assistant: To reset your email password, follow these steps:

1. Log in to the IT Helpdesk portal.
2. Follow the prompts to initiate a password reset request.
3. Complete any additional verification steps as required.

Note: You may need to use two-factor authentication to access company email remotely. If you're having trouble accessing the IT Helpdesk portal, try restarting your system and then attempting again.
-----------------------------------------------
User: Where can I see my payslip for last month?
Assistant: To view your payslip for last month, follow these steps:

1. Log in to the finance dashboard.
2. Look for the section related to payroll and benefits.
3. You should be able to see your payslip for the previous month.

Note: The finance dashboard is available on the intranet, so make sure you have access to it.
-----------------------------------------------
User: What is the company policy on medical leave?
Assistant: Unfortunately, the provided information does not specifically state the company's policy on medical leave. However, it does mention that medical leave is included in the HR policies, which suggests that there are guidelines and procedures in place for managing medical leaves.

To find more specific information about the company's policy on medical leave, you may need to check other resources, such as the employee handbook or speak with HR directly.
-----------------------------------------------
User: How to claim business travel reimbursement?
Assistant: To claim business travel reimbursement:

1. Log in to the expense portal.
2. Submit your receipts for eligible business expenses within 15 days of the trip.

Note: The specific steps and deadlines may vary depending on the company's policies, so it's best to check the employee handbook or consult with HR for more detailed information.
-----------------------------------------------
User: I'm facing issues connecting to the VPN.
Assistant: To resolve your issue connecting to the VPN:

1. Restart your system.
2. Try reconnecting to the VPN again.

If you're still experiencing issues, you may want to consider reaching out to the IT Helpdesk for further assistance or support.
-----------------------------------------------

Previous Next Home

Content Aggregation in LangChain4j: From Multi-Retriever Results to Ranked Relevance

2025-12-22T05:29:00.000-08:00

When building Retrieval-Augmented Generation (RAG) applications, it's common to fetch relevant content using multiple queries or content retrievers. However, this introduces challenges: how do you combine overlapping, partially relevant, or differently ranked content into a unified list that’s both accurate and non-redundant for the LLM?

LangChain4j addresses this problem with a powerful component: the ContentAggregator. This post explores its purpose, core logic, and implementations, including the DefaultContentAggregator and the ReRankingContentAggregator.

1. What Is a Content Aggregator?

In LangChain4j, the ContentAggregator interface defines a strategy for combining multiple lists of ranked content retrieved via different queries and content retrievers into a single, consistent, and sorted list of results.

This is important in RAG-based applications where:

· A single user query may be transformed into multiple sub-queries.

· Each sub-query might hit different retrievers (e.g., vector DBs, web APIs, PDF documents).

· Each retriever returns its own ranked content.

Instead of bombarding the LLM with all of this data, the ContentAggregator ensures only the most relevant and non-redundant content is passed along.

2. Interface Overview

Here’s what the ContentAggregator interface looks like:

public interface ContentAggregator {
    List<Content> aggregate(Map<Query, Collection<List<Content>>> queryToContents);
}

Here:

· Input: A map from each Query to a collection of List<Content, one list per retriever hit.

· Output: A single, relevance-ranked List<Content>.

3. ContentAggregator Implementations

LangChain4j offers two built-in implementations of the ContentAggregator interface:

· DefaultContentAggregator

· ReRankingContentAggregator

3.1 DefaultContentAggregator (Uses Reciprocal Rank Fusion Algorithm)

The DefaultContentAggregator is the standard implementation provided by LangChain4j. It performs two-stage Reciprocal Rank Fusion (RRF), a widely used technique in information retrieval.

What is Reciprocal Rank Fusion?

RRF combines multiple ranked lists by assigning scores based on the inverse of their rank positions. The idea: content that ranks high in multiple lists gets a strong overall score.

3.2 ReRankingContentAggregator (Semantic Relevance Boosted by Scoring Models)

Sometimes, structural signals like position in a list (used in RRF) aren’t enough. That’s where the ReRankingContentAggregator shines. It uses a ScoringModel (e.g., from Cohere, OpenAI, or other providers) to semantically re-score content. The re-ranking is based on how well each content chunk semantically aligns with the query.

Find the below working application.

ContentAggregatorDemo.java

package com.sample.app.contentaggregator;

import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import dev.langchain4j.rag.content.Content;
import dev.langchain4j.rag.content.aggregator.ContentAggregator;
import dev.langchain4j.rag.content.aggregator.DefaultContentAggregator;
import dev.langchain4j.rag.query.Query;

public class ContentAggregatorDemo {

    public static void main(String[] args) {

        // Step 1: Define sample queries
        Query query1 = new Query("What are the investment options available?");
        Query query2 = new Query("Tell me about the risk assessment process.");

        // Step 2: Create content chunks related to each query
        List<Content> investmentContentList1 = List.of(
                Content.from("Our company offers SIPs, mutual funds, and fixed deposits."),
                Content.from("You can diversify your investments via index funds and ETFs."));

        List<Content> investmentContentList2 = List.of(
                Content.from("Fixed income options like bonds are also available."),
                Content.from("We give personalized investment advice based on each client's risk level."));

        List<Content> riskContentList1 = List.of(
                Content.from("Risk assessment includes credit score checks and portfolio volatility analysis."),
                Content.from("We follow RBI guidelines for financial risk mitigation."));

        List<Content> riskContentList2 = List.of(
                Content.from("We use internal scoring models to assess client eligibility and investment risk."),
                Content.from("Risk is periodically reviewed based on market trends and compliance audits."));

        // Step 3: Build the map of Query -> Collection<List<Content>>
        Map<Query, Collection<List<Content>>> queryToContents = new HashMap<>();
        queryToContents.put(query1, List.of(investmentContentList1, investmentContentList2));
        queryToContents.put(query2, List.of(riskContentList1, riskContentList2));

        // Use DefaultContentAggregator (uses Reciprocal Rank Fusion)
        ContentAggregator defaultAggregator = new DefaultContentAggregator();
        List<Content> defaultAggregated = defaultAggregator.aggregate(queryToContents);

        System.out.println("=== DefaultContentAggregator Output ===");
        for (Content c : defaultAggregated) {
            System.out.println("- " + c.textSegment().text());
        }

    }
}

Output

=== DefaultContentAggregator Output ===
- Risk assessment includes credit score checks and portfolio volatility analysis.
- Our company offers SIPs, mutual funds, and fixed deposits.
- We use internal scoring models to assess client eligibility and investment risk.
- Fixed income options like bonds are also available.
- We follow RBI guidelines for financial risk mitigation.
- You can diversify your investments via index funds and ETFs.
- Risk is periodically reviewed based on market trends and compliance audits.
- We give personalized investment advice based on each client's risk level.

Let’s build a complete RAG application using content aggregator

    // Create retrieval augmentor with dynamic contentRetriever via router
    RetrievalAugmentor retrievalAugmentor =
        DefaultRetrievalAugmentor.builder()
            .queryTransformer(expandingQueryTransformer)
            .queryRouter(router)
            .contentAggregator(new DefaultContentAggregator())
            .build();

    // Build the assistant
    ChatAssistant chatAssistant =
        AiServices.builder(ChatAssistant.class)
            .chatModel(chatModel)
            .retrievalAugmentor(retrievalAugmentor)
            .chatMemory(MessageWindowChatMemory.withMaxMessages(10))
            .build();

This code snippet shows how to configure a RetrievalAugmentor in LangChain4j by setting a custom ContentAggregator.

· It uses a DefaultRetrievalAugmentor builder to configure:

o a queryTransformer (e.g., for expanding or rewriting queries),

o a queryRouter (to route queries to appropriate retrievers),

o and a DefaultContentAggregator (to merge retrieved content using Reciprocal Rank Fusion).

· This RetrievalAugmentor is then used to build a ChatAssistant, combining it with a chat model and chat memory.

· The resulting assistant can process user inputs, transform and route queries, retrieve and aggregate content, and respond accordingly.

Find the below working application.

ContentAggregatorDemoFullApp.java

package com.sample.app.contentaggregator;

import dev.langchain4j.data.document.Document;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.memory.chat.MessageWindowChatMemory;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.model.embedding.onnx.bgesmallenv15q.BgeSmallEnV15QuantizedEmbeddingModel;
import dev.langchain4j.model.ollama.OllamaChatModel;
import dev.langchain4j.rag.DefaultRetrievalAugmentor;
import dev.langchain4j.rag.RetrievalAugmentor;
import dev.langchain4j.rag.content.aggregator.DefaultContentAggregator;
import dev.langchain4j.rag.content.retriever.ContentRetriever;
import dev.langchain4j.rag.content.retriever.EmbeddingStoreContentRetriever;
import dev.langchain4j.rag.query.router.LanguageModelQueryRouter;
import dev.langchain4j.rag.query.transformer.ExpandingQueryTransformer;
import dev.langchain4j.service.AiServices;
import dev.langchain4j.store.embedding.EmbeddingStore;
import dev.langchain4j.store.embedding.EmbeddingStoreIngestor;
import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ContentAggregatorDemoFullApp {
  interface ChatAssistant {
    String chat(String userMessage);
  }

  public static void main(String[] args) {

    // Initialize LLM model for routing and chatting
    OllamaChatModel chatModel =
        OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

    // Shared embedding model
    EmbeddingModel embeddingModel = new BgeSmallEnV15QuantizedEmbeddingModel();

    // Create embedding stores
    EmbeddingStore<TextSegment> hrStore = new InMemoryEmbeddingStore<>();
    EmbeddingStore<TextSegment> itStore = new InMemoryEmbeddingStore<>();
    EmbeddingStore<TextSegment> financeStore = new InMemoryEmbeddingStore<>();

    // Sample data for ingestion

    List<String> hrDocs =
        List.of(
            "Our HR policies include flexible work hours, medical leave, and onboarding support.",
            "You can check your leave balance and benefits on the HR portal.",
            "Employee handbooks and code of conduct documents are available on the intranet.",
            "Annual performance reviews are conducted in Q1 of each year.",
            "We provide mental wellness programs and employee assistance plans (EAP).",
            "New employees must complete their joining formalities within the first week.",
            "HR holds town halls every quarter to discuss policy changes and updates.",
            "Exit interviews are mandatory and help improve employee retention practices.");

    List<String> itDocs =
        List.of(
            "If you're facing issues with VPN, restart your system and try again.",
            "Password resets can be done via the IT Helpdesk portal.",
            "For software installation requests, raise a ticket through the ServiceNow portal.",
            "Two-factor authentication is mandatory for accessing company email remotely.",
            "Laptop issues should be reported to the IT asset management team.",
            "New joiners will receive their device credentials within 24 hours of onboarding.",
            "We recommend using the Chrome browser for all internal web tools.",
            "The weekly IT newsletter includes patch updates and known issues.");

    List<String> financeDocs =
        List.of(
            "Payslips are generated on the 5th of every month and available on the finance dashboard.",
            "You can file business travel reimbursements through the expense portal.",
            "Employees must submit receipts within 15 days for expense claims.",
            "Annual tax declarations must be uploaded to the HRMS by January 31.",
            "Salary revisions are processed in March and reflected in April pay.",
            "All invoice-related queries should be addressed to [email protected].",
            "The company reimburses professional certification exam fees up to ₹10,000.",
            "You can track investment proofs under the ‘My Tax’ section on the intranet.");

    // Ingest documents into respective stores
    EmbeddingStoreIngestor.ingest(convertToDocuments(hrDocs), hrStore);
    EmbeddingStoreIngestor.ingest(convertToDocuments(itDocs), itStore);
    EmbeddingStoreIngestor.ingest(convertToDocuments(financeDocs), financeStore);

    // Create content retrievers
    ContentRetriever hrRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(hrStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("hrRetriever")
            .build();

    ContentRetriever itRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(itStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("itRetriever")
            .build();

    ContentRetriever financeRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(financeStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("financeRetriever")
            .build();

    Map<ContentRetriever, String> retrieverToDescription = new HashMap<>();
    retrieverToDescription.put(
        hrRetriever,
        "Provides information on leave policies, benefits, onboarding, and other HR-related topics.");
    retrieverToDescription.put(
        itRetriever,
        "Handles technical queries such as VPN access, email issues, password resets, and software installations.");
    retrieverToDescription.put(
        financeRetriever,
        "Answers questions related to reimbursements, payslips, taxation, invoices, and financial approvals.");

    // Create LLM-based router
    LanguageModelQueryRouter router =
        LanguageModelQueryRouter.builder()
            .chatModel(chatModel)
            .retrieverToDescription(retrieverToDescription)
            .build();

    // Expand queries (optional step for better retrieval)
    ExpandingQueryTransformer expandingQueryTransformer =
        new ExpandingQueryTransformer(chatModel, 3);

    // Create retrieval augmentor with dynamic contentRetriever via router
    RetrievalAugmentor retrievalAugmentor =
        DefaultRetrievalAugmentor.builder()
            .queryTransformer(expandingQueryTransformer)
            .queryRouter(router)
            .contentAggregator(new DefaultContentAggregator())
            .build();

    // Build the assistant
    ChatAssistant chatAssistant =
        AiServices.builder(ChatAssistant.class)
            .chatModel(chatModel)
            .retrievalAugmentor(retrievalAugmentor)
            .chatMemory(MessageWindowChatMemory.withMaxMessages(10))
            .build();

    // Run a conversation
    List<String> userQueries =
        List.of(
            "How do I reset my email password?",
            "Where can I see my payslip for last month?",
            "What is the company policy on medical leave?",
            "How to claim business travel reimbursement?",
            "I'm facing issues connecting to the VPN.");

    for (String userQuery : userQueries) {
      System.out.println("User: " + userQuery);
      String answer = chatAssistant.chat(userQuery);
      System.out.println("Assistant: " + answer);
      System.out.println("-----------------------------------------------");
    }
  }

  private static List<Document> convertToDocuments(List<String> texts) {
    List<Document> documents = new ArrayList<>();
    for (String text : texts) {
      documents.add(Document.from(text));
    }
    return documents;
  }
}

Output

User: How do I reset my email password?
Assistant: To reset your email password, follow these steps:

1. Go to the IT Helpdesk portal.
2. Log in with your company credentials.
3. Click on the "Password Reset" or "Forgot Password" option.
4. Follow the instructions provided to complete the password reset process.

Please note that you may need to provide additional verification information due to two-factor authentication requirements for accessing company email remotely.
-----------------------------------------------
User: Where can I see my payslip for last month?
Assistant: To view your payslip for last month, follow these steps:

1. Log in to the finance dashboard.
2. Check if a new month's payslips have been generated (typically around the 5th of the current month).
3. If available, click on the "View" or "Last Month's Pay" option to access your payslip for the previous month.

Please note that you should be able to view this information online through the finance dashboard, without needing any additional assistance or credentials.
-----------------------------------------------
User: What is the company policy on medical leave?
Assistant: Unfortunately, this information doesn't directly state the company's policy on medical leave. However, it does mention that "Our HR policies include flexible work hours" and provides resources for managing leaves (the HR portal) and offers mental wellness programs and EAPs. It can be inferred that the company has some form of medical leave policy, but the specific details are not provided in this information.
-----------------------------------------------
User: How to claim business travel reimbursement?
Assistant: To claim business travel reimbursement, follow these steps:

1. Use the expense portal.
2. Log in with your company credentials.
3. Navigate to the "Business Travel" or "Expense Claims" section.
4. Submit your receipts and relevant documentation for reimbursement within 15 days of the trip.

Note that specific details about reimbursement procedures, such as what expenses are eligible or any additional requirements, are not mentioned in this information.
-----------------------------------------------
User: I'm facing issues connecting to the VPN.
Assistant: Try restarting your system to resolve the issue connecting to the VPN.

If the problem persists, you can consider reaching out to IT for further assistance or requesting help from the Helpdesk via the IT Helpdesk portal.

Note that two-factor authentication and software installation requests are not directly related to resolving VPN connection issues.
-----------------------------------------------

Previous Next Home

Content Retrievers in LangChain4j: Bridging the gap between Natural Language and Data Sources

2025-10-30T08:18:00.000-07:00

Retrieval-Augmented Generation (RAG) applications rely heavily on the ability to pull relevant information from external data sources. In LangChain4j, this responsibility lies with the ContentRetriever interface. Whether your data lives in a vector store, SQL database, graph database, or even on the open web, LangChain4j provides a unified way to retrieve relevant chunks of information.

In this post, we’ll walk through the role of ContentRetriever, explore various implementations provided by LangChain4j, and understand how they help bridge the gap between unstructured queries and structured (or semi-structured) data sources.

1. What is Content Retriever?

A ContentRetriever retrieves Content from an underlying data source using a given Query.

public interface ContentRetriever {
    List<Content> retrieve(Query query);
}

At the time of writing this pos, langchain4j support following content retrievers.

LangChain4j comes with several built-in content retrievers that help get information from different types of data sources.

· The EmbeddingStoreContentRetriever finds content using vector similarity, which is great for working with embeddings and doing smart searches.

· The WebSearchContentRetriever pulls in up-to-date information from the web using a search engine.

· The AzureAiSearchContentRetriever works with Azure AI Search and supports full-text search, vector search, a mix of both, and can re-rank results to show the most useful ones first, making it good for enterprise use.

· The SqlDatabaseContentRetriever (still experimental) connects to relational databases and turns plain language questions into SQL queries to get the right data.

· Lastly, the Neo4jContentRetriever lets you search graph databases by converting natural language into Cypher queries and fetching related entities and connections.

These retrievers allow LangChain4j to link AI models with many different types of data sources.

Benefits of LangChain4j's Content Retriever Abstraction

· Pluggable architecture: Easily switch between retrievers without rewriting business logic.

· Unified Query API: Regardless of the backend (vector DB, SQL, graph), all retrievers accept the same Query object.

· Composable with RAG pipelines: Easily integrate with LangChain4j’s other RAG components (query transformers, aggregators, etc.).

Example

ContentRetriever contentRetriever = EmbeddingStoreContentRetriever.builder()
    .embeddingStore(embeddingStore)
    .embeddingModel(embeddingModel)
    .displayName("informaitonAboutIndia")
    .maxResults(3)
    .minScore(0.7)
    .build();

Above code creates an instance of ContentRetriever using the EmbeddingStoreContentRetriever. It configures how this retriever should fetch relevant content from an embedding-based data store.

Find the below working application.

ContentRetrieverDemo.java

package com.sample.app.contentretriever;

import java.util.ArrayList;
import java.util.List;

import dev.langchain4j.data.document.Document;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.model.embedding.onnx.bgesmallenv15q.BgeSmallEnV15QuantizedEmbeddingModel;
import dev.langchain4j.rag.content.Content;
import dev.langchain4j.rag.content.retriever.ContentRetriever;
import dev.langchain4j.rag.content.retriever.EmbeddingStoreContentRetriever;
import dev.langchain4j.rag.query.Query;
import dev.langchain4j.store.embedding.EmbeddingStoreIngestor;
import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;

public class ContentRetrieverDemo {

    private static List<String> indiaDocs = List.of(
            """
                    India, officially the Republic of India, is a country in South Asia. It is the seventh-largest country by area,
                    the most populous country as of 2023, and the most populous democracy in the world. Bounded by the Himalayas in
                    the north and surrounded by the Indian Ocean, Arabian Sea, and Bay of Bengal, India has a rich cultural and
                    historical heritage.
                    """,
            """
                    The Indian economy is the fifth-largest in the world by nominal GDP and the third-largest by purchasing power parity.
                    It is classified as a newly industrialized country and one of the world's fastest-growing major economies. Major
                    industries include IT services, textiles, pharmaceuticals, and agriculture.
                    """,
            """
                    India's political system is a federal parliamentary democratic republic. The President of India is the head of state,
                    while the Prime Minister is the head of government. The Indian Parliament consists of the Lok Sabha (House of the People)
                    and the Rajya Sabha (Council of States).
                    """,
            """
                    Indian culture is renowned for its diversity and depth. It includes a wide variety of languages, religions, music,
                    dance forms, cuisine, and traditions. Major religions born in India include Hinduism, Buddhism, Jainism, and Sikhism.
                    Festivals such as Diwali, Eid, Christmas, and Pongal are celebrated across the country.
                    """,
            """
                    India is home to 40 UNESCO World Heritage Sites, including the Taj Mahal, Qutub Minar, and the Western Ghats. The country
                    offers a wide variety of landscapes ranging from deserts in Rajasthan, the Himalayan mountain range, tropical rainforests
                    in the northeast, to coastal plains and fertile river valleys.
                    """,
            """
                    The Indian education system has seen significant progress, with institutes like the IITs, IIMs, and AIIMS gaining global
                    recognition. India also has a growing startup ecosystem, particularly in tech hubs such as Bengaluru, Hyderabad, and Pune.
                    """,
            """
                    India has a vibrant film industry, notably Bollywood, which produces the largest number of films in the world. Indian music
                    and dance forms such as Bharatanatyam, Kathak, Carnatic music, and Bollywood songs are famous worldwide.
                    """,
            """
                    Indian cuisine is known for its rich flavors and diverse ingredients. Each region has its own distinct dishes –
                    from butter chicken and biryani in the north, to dosa and sambar in the south, to seafood delicacies in the coastal regions.
                    Spices like turmeric, cumin, coriander, and cardamom are integral to Indian cooking.
                    """,
            """
                    The Indian space program, led by ISRO, has made remarkable achievements including the Chandrayaan lunar missions and the
                    Mars Orbiter Mission (Mangalyaan), which made India the first Asian nation to reach Mars orbit and the first in the world
                    to do so in its maiden attempt.
                    """,
            """
                    Indian sports are dominated by cricket, but other sports like hockey, badminton, wrestling, and kabaddi are also popular.
                    India has produced world-class athletes including Sachin Tendulkar, P.V. Sindhu, Neeraj Chopra, and Mary Kom. The country
                    has hosted major sporting events like the Commonwealth Games and the Cricket World Cup.
                    """);

    private static List<Document> documents = new ArrayList<>();

    private static void prepareDocuments() {
        for (String doc : indiaDocs) {
            documents.add(Document.from(doc));
        }
    }

    public static void main(String[] args) {
        prepareDocuments();

        InMemoryEmbeddingStore<TextSegment> embeddingStore = new InMemoryEmbeddingStore<>();
        EmbeddingStoreIngestor.ingest(documents, embeddingStore);

        EmbeddingModel embeddingModel = new BgeSmallEnV15QuantizedEmbeddingModel();

        ContentRetriever contentRetriever = EmbeddingStoreContentRetriever.builder().embeddingStore(embeddingStore)
                .embeddingModel(embeddingModel)
                .displayName("informaitonAboutIndia")
                .maxResults(3).minScore(0.7).build();

        String userQuery = """
                Hey, I was reading about how countries manage their political systems and got curious — can you tell me how India runs its government, who’s in charge, and what kind of structure they follow politically?

                """;

        List<Content> matchedContent = contentRetriever.retrieve(Query.from(userQuery));
        System.out.println("Total Matched Documents : " + matchedContent.size() + "\n");

        int i = 1;
        for (Content content : matchedContent) {
            System.out.println(i + ". " +content.textSegment().text() + "\n");
            i++;
        }

    }
}

Output

Total Matched Documents : 3

1. India's political system is a federal parliamentary democratic republic. The President of India is the head of state,
while the Prime Minister is the head of government. The Indian Parliament consists of the Lok Sabha (House of the People)
and the Rajya Sabha (Council of States).

2. India, officially the Republic of India, is a country in South Asia. It is the seventh-largest country by area,
the most populous country as of 2023, and the most populous democracy in the world. Bounded by the Himalayas in
the north and surrounded by the Indian Ocean, Arabian Sea, and Bay of Bengal, India has a rich cultural and
historical heritage.

3. The Indian economy is the fifth-largest in the world by nominal GDP and the third-largest by purchasing power parity.
It is classified as a newly industrialized country and one of the world's fastest-growing major economies. Major
industries include IT services, textiles, pharmaceuticals, and agriculture.

ContentRetriever takes the input from Query router in RAG pipelines.

ContentRetrieverFullApp.java

package com.sample.app.contentretriever;

import dev.langchain4j.data.document.Document;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.memory.chat.MessageWindowChatMemory;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.model.embedding.onnx.bgesmallenv15q.BgeSmallEnV15QuantizedEmbeddingModel;
import dev.langchain4j.model.ollama.OllamaChatModel;
import dev.langchain4j.rag.DefaultRetrievalAugmentor;
import dev.langchain4j.rag.RetrievalAugmentor;
import dev.langchain4j.rag.content.retriever.ContentRetriever;
import dev.langchain4j.rag.content.retriever.EmbeddingStoreContentRetriever;
import dev.langchain4j.rag.query.router.LanguageModelQueryRouter;
import dev.langchain4j.rag.query.transformer.ExpandingQueryTransformer;
import dev.langchain4j.service.AiServices;
import dev.langchain4j.store.embedding.EmbeddingStore;
import dev.langchain4j.store.embedding.EmbeddingStoreIngestor;
import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ContentRetrieverFullApp {

  interface ChatAssistant {
    String chat(String userMessage);
  }

  public static void main(String[] args) {

    // Initialize LLM model for routing and chatting
    OllamaChatModel chatModel =
        OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

    // Shared embedding model
    EmbeddingModel embeddingModel = new BgeSmallEnV15QuantizedEmbeddingModel();

    // Create embedding stores
    EmbeddingStore<TextSegment> hrStore = new InMemoryEmbeddingStore<>();
    EmbeddingStore<TextSegment> itStore = new InMemoryEmbeddingStore<>();
    EmbeddingStore<TextSegment> financeStore = new InMemoryEmbeddingStore<>();

    // Sample data for ingestion

    List<String> hrDocs =
        List.of(
            "Our HR policies include flexible work hours, medical leave, and onboarding support.",
            "You can check your leave balance and benefits on the HR portal.",
            "Employee handbooks and code of conduct documents are available on the intranet.",
            "Annual performance reviews are conducted in Q1 of each year.",
            "We provide mental wellness programs and employee assistance plans (EAP).",
            "New employees must complete their joining formalities within the first week.",
            "HR holds town halls every quarter to discuss policy changes and updates.",
            "Exit interviews are mandatory and help improve employee retention practices.");

    List<String> itDocs =
        List.of(
            "If you're facing issues with VPN, restart your system and try again.",
            "Password resets can be done via the IT Helpdesk portal.",
            "For software installation requests, raise a ticket through the ServiceNow portal.",
            "Two-factor authentication is mandatory for accessing company email remotely.",
            "Laptop issues should be reported to the IT asset management team.",
            "New joiners will receive their device credentials within 24 hours of onboarding.",
            "We recommend using the Chrome browser for all internal web tools.",
            "The weekly IT newsletter includes patch updates and known issues.");

    List<String> financeDocs =
        List.of(
            "Payslips are generated on the 5th of every month and available on the finance dashboard.",
            "You can file business travel reimbursements through the expense portal.",
            "Employees must submit receipts within 15 days for expense claims.",
            "Annual tax declarations must be uploaded to the HRMS by January 31.",
            "Salary revisions are processed in March and reflected in April pay.",
            "All invoice-related queries should be addressed to [email protected].",
            "The company reimburses professional certification exam fees up to ₹10,000.",
            "You can track investment proofs under the ‘My Tax’ section on the intranet.");

    // Ingest documents into respective stores
    EmbeddingStoreIngestor.ingest(convertToDocuments(hrDocs), hrStore);
    EmbeddingStoreIngestor.ingest(convertToDocuments(itDocs), itStore);
    EmbeddingStoreIngestor.ingest(convertToDocuments(financeDocs), financeStore);

    // Create content retrievers
    ContentRetriever hrRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(hrStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("hrRetriever")
            .build();

    ContentRetriever itRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(itStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("itRetriever")
            .build();

    ContentRetriever financeRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(financeStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("financeRetriever")
            .build();

    Map<ContentRetriever, String> retrieverToDescription = new HashMap<>();
    retrieverToDescription.put(
        hrRetriever,
        "Provides information on leave policies, benefits, onboarding, and other HR-related topics.");
    retrieverToDescription.put(
        itRetriever,
        "Handles technical queries such as VPN access, email issues, password resets, and software installations.");
    retrieverToDescription.put(
        financeRetriever,
        "Answers questions related to reimbursements, payslips, taxation, invoices, and financial approvals.");

    // Create LLM-based router
    LanguageModelQueryRouter router =
        LanguageModelQueryRouter.builder()
            .chatModel(chatModel)
            .retrieverToDescription(retrieverToDescription)
            .build();

    // Expand queries (optional step for better retrieval)
    ExpandingQueryTransformer expandingQueryTransformer =
        new ExpandingQueryTransformer(chatModel, 3);

    // Create retrieval augmentor with dynamic contentRetriever via router
    RetrievalAugmentor retrievalAugmentor =
        DefaultRetrievalAugmentor.builder()
            .queryTransformer(expandingQueryTransformer)
            .queryRouter(router)
            .build();

    // Build the assistant
    ChatAssistant chatAssistant =
        AiServices.builder(ChatAssistant.class)
            .chatModel(chatModel)
            .retrievalAugmentor(retrievalAugmentor)
            .chatMemory(MessageWindowChatMemory.withMaxMessages(10))
            .build();

    // Run a conversation
    List<String> userQueries =
        List.of(
            "How do I reset my email password?",
            "Where can I see my payslip for last month?",
            "What is the company policy on medical leave?",
            "How to claim business travel reimbursement?",
            "I'm facing issues connecting to the VPN.");

    for (String userQuery : userQueries) {
      System.out.println("User: " + userQuery);
      String answer = chatAssistant.chat(userQuery);
      System.out.println("Assistant: " + answer);
      System.out.println("-----------------------------------------------");
    }
  }

  private static List<Document> convertToDocuments(List<String> texts) {
    List<Document> documents = new ArrayList<>();
    for (String text : texts) {
      documents.add(Document.from(text));
    }
    return documents;
  }
}

Output

User: How do I reset my email password?
Assistant: Unfortunately, this information doesn't provide instructions on how to reset your email password specifically. However, I can infer that you can use the IT Helpdesk portal for assistance with resetting your email password. If you're having trouble accessing the portal or need help with the process, you may want to contact the IT department directly for support.
-----------------------------------------------
User: Where can I see my payslip for last month?
Assistant: According to the information provided, payslips for last month would be available on the finance dashboard, which is generated on the 5th of every month. This implies that if you're looking at your current month's payslip, it should be visible on the finance dashboard, but you'll need to wait until the 5th of the next month to see the previous month's payslip.
-----------------------------------------------
User: What is the company policy on medical leave?
Assistant: The company policy on medical leave is not explicitly stated, but it can be inferred that there is a provision for it as part of the "Our HR policies include flexible work hours, medical leave, and onboarding support" statement. However, the exact details of this policy are not provided in the given information.
-----------------------------------------------
User: How to claim business travel reimbursement?
Assistant: To claim business travel reimbursement, you need to use the expense portal. The exact steps or process for doing so are not provided in this information. However, it is mentioned that employees must submit receipts within 15 days for expense claims, suggesting that you will need to upload your receipts and follow any applicable procedures through the expense portal.
-----------------------------------------------
User: I'm facing issues connecting to the VPN.
Assistant: If you're facing issues connecting to the VPN, try restarting your system and then attempt to connect again. This may resolve the issue, but if not, further troubleshooting steps would be needed to identify and resolve the problem.
-----------------------------------------------

Previous Next Home

Apache Pinot vs. Druid: Which Real-Time Analytics Database Should You Choose?

2025-10-30T07:56:00.000-07:00

If you need fast analytics on live data (like dashboards or real-time reports), two open-source databases stand out: Apache Pinot and Apache Druid. Both are built for low-latency queries at scale, but they have different strengths.

1. What Are Pinot and Druid?

Both are real-time OLAP databases, meaning:

· They handle streaming data (e.g., clicks, transactions) + batch data (historical logs).

· Optimized for fast aggregations (e.g., "How many users visited today?").

· Support high concurrency (100s–1000s of queries per second).

2. Performance: Which Is Faster?

Pinot:

· Excels at high-concurrency queries (e.g., 100,000+ queries/sec).

· Used by companies like Uber Eats and Stripe for real-time dashboards.

· Requires manual tuning for best performance.

Druid:

· Handles mixed workloads better (e.g., dashboards + ad-hoc queries).

· Used by Netflix and Salesforce for analytics.

· May slow down under extreme concurrency.

Verdict:

· Need ultra-fast, predictable queries? Pinot might win.

· Need flexibility + ease of use? Druid could be better.

3. Indexing (How Data Is Organized)

Pinot:

· You choose indexes manually (like picking tools for a toolbox).

· More control but harder to set up.

Druid:

· Automatic indexing (it picks the best method for you).

· Simpler but less customizable.

Beginners might prefer Druid (less manual work), and experts might prefer Pinot (more tuning options).

4. Data Ingestion (Loading Data)

Druid:

· Supports SQL-based ingestion (transform data while loading).

· Example: You can JOIN tables during ingestion.

Pinot:

· Needs pre-processed data (e.g., via Spark or Flink).

· Less flexible for complex transformations.

5. Which Should You Choose?

Pick Druid if you:

· Want auto-indexing (less manual work).

· Need SQL-based data transformations.

· Have mixed workloads (dashboards + ad-hoc queries).

Pick Pinot if you:

· Need extreme speed (100K+ queries/sec).

· Can manually optimize indexes.

· Don’t need complex transformations during ingestion.

Previous Next Home

Designing for Scale: Inside Modern Distributed Databases

2025-10-30T07:52:00.000-07:00

This post explains how modern databases handle large amounts of data while staying fast and reliable using methods like sharding, replication, and denormalization. It looks at how databases have evolved from traditional relational systems to modern NoSQL ones like MongoDB and Cassandra, describing how they work and the pros and cons of each.

By the end, you’ll understand the main ideas behind distributed data design and how to make smart choices that balance consistency, availability, and scalability.

1. The Need for Scalable Databases

Modern applications from social media platforms to e-commerce giants and IoT networks deal with huge amounts of data and user activity. Every click, view, purchase, sensor signal, and recommendation translates into a database operation. As businesses grow, so does the need for databases that can scale seamlessly handling billions of records, millions of users, and continuous uptime expectations.

This section explains why scalability has become a first-class requirement for modern databases, where traditional relational models fall short, and how horizontal scaling and distributed systems changed the game.

1.1 The Rise of Data Volume and Concurrent Users

In the early 2000s, a typical enterprise database might have handled a few gigabytes of structured data and a few hundred concurrent users. Today, systems like Instagram, Netflix, and Amazon handle petabytes of data and millions of simultaneous users across the globe all expecting instant responses.

Example 1: Social Media Growth

· When Facebook launched in 2004, it served data from a single MySQL database.

· Within a few years, user numbers exploded from thousands to millions.

· Every user profile, post, comment, and reaction became a row in a database table.

· The relational database couldn’t handle that level of read/write concurrency and data volume.

· This led Facebook to adopt MySQL sharding and later introduce TAO, their own distributed data store for social graphs.

Example 2: E-commerce Traffic Spikes

· Consider an e-commerce site like Flipkart or Amazon during a "Big Billion Sale".

· The number of concurrent users might jump from 10,000 to 5 million within minutes.

· A single database server even a high-end one cannot handle such a surge in queries.

· Without a scalable database, users would see "Site Down" errors or slow checkout experiences.

The solution is to distribute both the data and load across multiple nodes, allowing the system to expand dynamically. These examples underline one truth that data growth is not linear, but exponential and databases must be designed to scale in step.

1.2 Why Traditional Relational Databases Hit Limits

Traditional relational databases like MySQL, PostgreSQL, and Oracle were designed in an era when data fit comfortably on a single machine. They emphasize ACID guarantees (Atomicity, Consistency, Isolation, Durability), which ensure data correctness but limit scalability.

Let’s understand their key bottlenecks:

1.2.1 Single-Server Dependency

A typical RDBMS installation runs on a single server. You can make it faster by adding more CPU, memory, or faster disks but only up to a point. Eventually, you hit hardware and I/O limits.

1.2.2 Locking and Transactions

When multiple users try to modify the same table or row, locks are used to maintain consistency. Under heavy load, this causes contention many transactions waiting for one to complete. For example, imagine 100000 users trying to update their shopping cart at the same time. The system must ensure each cart update is isolated and correct but this slows down the entire database because of locking overhead.

1.2.3 Vertical Scaling Saturation

You can upgrade to a more powerful server but after a point, adding more hardware brings diminishing returns. The CPU may be idle while the disk is busy, or the memory may be underutilized while the I/O waits on locks.

Moreover, vertical scaling (buying bigger servers) is expensive and not fault-tolerant one machine failure can take the entire database down.

1.2.4 Schema Rigidity

Relational databases enforce strict schemas. If your data model changes frequently (like in modern apps where new features evolve rapidly), schema migrations can be slow and risky, blocking production systems.

1.3 Horizontal vs Vertical Scaling

The two fundamental approaches to scaling databases are vertical scaling (scale up) and horizontal.

Horizontal Scaling

· Add more CPU, RAM, or faster storage to a single machine

· Example: Upgrading from 16-core to 64-core, 64GB to 512GB

· Pros: Simple to implement, minimal code changes sometimes no code changes at all

· Cons: Hardware limits, single point of failure and expensive

Vertical Scaling

· Add more machines (nodes) and distribute data & requests across them

· Example: Adding 10 commodity servers to share the load

· Pros: Scalable, fault-tolerant, cost-effective

· Cons: Complex coordination, consistency management

For example, suppose your MySQL server on one machine is overloaded.

· Vertical scaling: upgrade the server to 32 cores and 512GB RAM. Works temporarily, but costs rise exponentially.

· Horizontal scaling: split data across 10 smaller servers (shards), each handling a subset of data. Now you can add the 11th or 12th server as demand grows —this is true scalability.

Modern distributed databases like Cassandra, MongoDB, and CockroachDB are built natively for horizontal scaling, enabling near-infinite expansion.

1.4 Characteristics of Modern Distributed Systems

Modern databases are designed as distributed systems, a collection of independent machines working together to appear as one logical database. Let’s explore their key characteristics.

1.4.1 Fault Tolerance

Failures are inevitable like disks crash, nodes go offline, networks partition. A distributed database ensures no single point of failure. Data is replicated across nodes, so if one node goes down, another takes over seamlessly.

Example:

In MongoDB, every shard has one primary and multiple secondaries. If the primary fails, the secondaries automatically elect a new primary all without manual intervention.

1.4.2 Elasticity

Modern systems can scale elastically meaning they can add or remove resources dynamically based on load.

Cloud-native databases like Amazon DynamoDB or Google Spanner automatically scale up during traffic spikes and scale down during idle hours.

Example:

During peak shopping hours, DynamoDB might expand from 20 to 100 partitions. Once traffic drops, it automatically scales back to 20, saving cost while maintaining performance.

1.4.3 Availability

Distributed databases are built for high availability, ensuring that users can always read and write data even during failures. This is achieved by replication, automatic failover, and geo-distribution.

Example:

Cassandra replicates data across multiple nodes and even multiple data centers. If a node in Mumbai fails, requests can still be served from Delhi or Singapore replicas maintaining 24/7 uptime.

1.4.4 Eventual Consistency

To maintain availability during network failures, distributed databases often relax immediate consistency, meaning data updates may take a few seconds to propagate globally. This trade-off, known as eventual consistency, is central to systems like Cassandra and DynamoDB.

Example:

You might update your profile picture, and your friend in another region sees the old one for a few seconds that’s eventual consistency in action.

In summary, the demand for real-time, global, and large-scale applications has redefined what databases must deliver. Traditional relational databases prioritized data integrity, modern distributed systems prioritize availability, elasticity, and fault tolerance without compromising too much on consistency.

2. The Router–Shard–Replica Model

As applications grow and user requests flood in, a single database server no matter how powerful it is, eventually becomes a bottleneck. The Router–Shard–Replica model is the architectural response that allows databases to distribute both data and workload intelligently across many machines. Let’s understand how this works, step by step.

2.1 Overview of How Client Requests Are Routed

Imagine you run a global e-commerce platform where millions of customers browse, add to cart, and purchase simultaneously. When a client (say, the mobile app) sends a request to fetch a user’s order history, the system must know which server actually holds that user’s data. This is where the router or coordinator node steps in.

Instead of making every client aware of where each piece of data lives, the router acts as a traffic director:

· The client connects to the router.

· The router determines which shard (Shard holds subset of data) holds the relevant data (based on a shard key or hash function).

· It forwards the query to the correct shard and merges the results if needed.

The client sees only one endpoint, but behind the scenes, the router orchestrates the query distribution.

Example:

In MongoDB, this router is known as mongos, while in Cassandra, any node can serve as the coordinator. Both serve the same purpose simplifying the client’s view of the cluster.

2.2 The Concept of a Router (or Coordinator) Node

Think of the router as the "switchboard operator" (A switchboard operator is a person who manages incoming telephone calls, routing them to the correct person or department within an organization) of your database cluster. It doesn’t store actual data instead, it maintains metadata about where each chunk of data resides.

In MongoDB:

· The router consults config servers, which store mapping information about data chunks and shard locations.

· When a query arrives, the router checks this metadata to decide which shard(s) to contact.

· If data moves (due to rebalancing or new shard addition), the config servers update this metadata keeping the routing dynamic and consistent.

In Cassandra:

· There’s no central router. Any node can act as a coordinator by knowing the token ring layout. Read this post for more details: https://self-learning-java-tutorial.blogspot.com/2024/05/consistent-hashing-scalable-approach-to.html

· The coordinator identifies which nodes are responsible for a given partition key and forwards the request accordingly.

2.3 What a Shard Is: Horizontal Partitioning Explained

A shard is simply a subset of your total data, stored on a separate database server. Instead of one monolithic dataset living on one server, you divide it horizontally across multiple servers each responsible for a specific range or hash of data.

This is called horizontal partitioning, or sharding. Suppose an E-commerce company has 100 million customers.

· Shard 1 stores customers whose IDs start from 1–25 million

· Shard 2 stores IDs from 25–50 million

· Shard 3 stores IDs from 50–75 million

· Shard 4 stores IDs from 75–100 million

Each shard handles queries for its own range, so the load is spread evenly. Horizontal partitioning splits rows, which is more scalable for high-volume systems since it keeps schema consistent and allows parallelism.

2.3.1 Adding Shards for Capacity and Redundancy

One of the most powerful aspects of sharding is scaling by addition you don’t upgrade hardware, you just add another machine.

When an e-commecrece company grows to 200 million users, you can:

· Add new shards for new user ranges.

· Redistribute existing data across shards (a process called resharding).

· Automatically balance load so no single shard becomes a hotspot.

2.3.2 Shard Backups and Replication Strategies

Each shard must also protect against hardware failures so shards are replicated. Replication means maintaining multiple copies of the same data across different servers (often in different zones or regions).

Usually, one node acts as the primary (handling writes), while others act as secondaries (handling reads or staying in sync).

Example (MongoDB):

· Each shard is a replica set where Primary node handles all write operations and Secondary nodes replicate the data asynchronously. If the primary fails, one of the secondaries automatically becomes the new primary through an election process.

This replication provides:

· High availability: Failover without downtime.

· Disaster recovery: Even if one data center fails, another can continue serving.

· Read scalability: Secondary replicas can handle read-heavy workloads.

2.3.3 Benefits: Scalability and Resilience

The router–shard–replica model brings two major benefits:

Scalability:

· By adding shards, you add both storage and processing capacity.

· More shards = more parallelism = higher throughput.

· Horizontal scaling is near-linear when designed properly.

Resilience:

· Replicas ensure data durability and fault tolerance.

· Even in node or rack failures, data remains accessible.

· The system can self-heal via automatic failover and rebalancing.

Example:

In a replicaset of 3 nodes, if primary node crashes, a replica in another data center automatically takes over, and the router updates its routing metadata all in seconds, without user-visible downtime.

2.3.4 Challenge: Cross-Shard Joins and Data Aggregation

Distributed systems aren’t magic they trade simplicity for scale. One of the biggest trade-offs is cross-shard queries.

When data relevant to a query lives across multiple shards:

· The router must scatter the query to all shards.

· Each shard processes its local part and returns results.

· The router merges the results before sending them back to the client.

This is known as a scatter-gather operation, and it can be expensive. For example, if a report needs "total purchases across all users", every shard must contribute its partial total introducing latency and coordination overhead.

Mitigation Strategies

· Carefully design shard keys that localize related data together.

· Pre-aggregate or cache computed results.

· Use analytical systems like Spark or BigQuery for heavy aggregation workloads.

The Router–Shard–Replica model unlocks true horizontal scalability — letting databases grow seamlessly as your data and users grow. However, scaling horizontally isn’t free: it introduces data distribution, query routing, and consistency challenges. Smart data modelling, thoughtful shard key selection, and robust replication strategies are what make this architecture both powerful and reliable in real-world systems.

3. MongoDB: A Case Study in Sharded NoSQL Design

MongoDB is one of the most successful examples of a distributed database that prioritizes simplicity, scalability, and developer productivity. Its design elegantly demonstrates how the Router–Shard–Replica model works in practice, balancing high performance with fault tolerance.

Let’s explore MongoDB’s architecture piece by piece.

Introduction to MongoDB’s Distributed Architecture

At its core, MongoDB is built for horizontal scalability, meaning it can distribute data across multiple machines (shards) while keeping the application view consistent.

A typical sharded MongoDB cluster has three major components:

mongos: the query router

Shard servers: the actual data-bearing nodes

Config servers: the metadata managers

Each shard itself is not just one server but a replica set, ensuring data redundancy and high availability. Let’s see how these parts work together to form a cohesive, resilient system.

3.1 mongos: The Query Router

mongos is the entry point for all client applications. Think of it as MongoDB’s traffic controller, it knows where data resides and routes each request to the appropriate shard(s).

Here’s how it works:

· A client connects to a mongos instance instead of directly connecting to a shard.

· The mongos consults the config servers to learn where the required data chunk resides.

· It forwards the query to the correct shard or, if necessary, to multiple shards (scatter–gather).

· It aggregates the results and sends them back to the client.

This means clients don’t need to worry about where their data lives, they just query MongoDB as if it were a single large database.

Example:

If an e-comerce application stores user data sharded by userId, when a request for userId = 10234 arrives:

mongos checks metadata in config servers.

· Finds that userId 10234 is on Shard 2.

· Sends the query to that shard and returns the results.

3.2 Replica Sets: How MongoDB Ensures High Availability

Each shard in MongoDB isn’t a single point of failure, it’s a replica set, a self-healing cluster of nodes that maintain copies of the same data.

A replica set typically consists of:

· One primary node: Handles all write operations.

· Multiple secondary nodes: Replicate the data from the primary asynchronously.

This structure gives MongoDB both high availability and read scalability. If the primary node fails, MongoDB’s built-in election mechanism automatically promotes a secondary to become the new primary usually within seconds.

3.3 Primary vs Secondary Nodes

Let’s look at how reads and writes behave across these nodes.

· Writes: Always go to the primary node (to maintain a consistent order of operations).

· Reads: By default, also go to the primary for the most up-to-date data. However, applications can choose to read from secondaries using read preferences for better performance and geographic locality.

Example:

If application has a shard located in Singapore with:

· Primary: Singapore

· Secondaries: Mumbai and London

Then:

· Users in India might read from Mumbai for lower latency.

· Writes still go to the Singapore primary.

· Data replication happens asynchronously, ensuring eventual consistency.

3.4 Automatic Election Mechanism

MongoDB’s self-healing nature is one of its strongest features. If the primary node of a replica set goes down:

· Secondaries detect the loss of heartbeat from the primary.

· They initiate an election process to choose a new primary.

· The node with the most recent operations (highest replication log) is elected.

Once elected, the new primary starts accepting writes. This entire process happens automatically without manual intervention and ensure minimal downtime.

Example:

If the Singapore primary crashes, the Mumbai secondary with the latest data becomes the new primary within ~10 seconds, and all traffic is rerouted accordingly.

3.5 Config Servers: Metadata and Routing Information

Config servers are the brains of the sharded cluster they store the metadata that maps data chunks to shards.

They maintain:

· Which data ranges belong to which shards

· Cluster topology and sharding configuration

· Balancer state for redistributing chunks

In a production setup, you’ll always find three config servers to ensure redundancy and quorum-based consistency. If a new shard is added or data is rebalanced, the config servers update this metadata, and mongos routers start using the new configuration almost instantly.

3.6 Shard Key Selection and Its Impact on Performance

The shard key is the single most important design decision in a MongoDB sharded cluster. It determines how data is distributed across shards and thus, how evenly your workload is balanced.

A shard key can be:

· Hashed (e.g., hash of userId): distributes data evenly but can make range queries slower.

· Ranged (e.g., date or numeric range): supports efficient range queries but risks hotspotting if values are sequential.

Example:

In an e-commecre application, if you shard by region, and 80% of your traffic comes from Asia, your Asia shard will be overloaded and creating a hotspot. A better approach might be to shard by hash(userId) or (region, userId) composite key.

Always choose a shard key that:

· Distributes data evenly

· Matches your most common query patterns

Bad shard key selection is the #1 cause of performance issues in sharded MongoDB deployments.

In summary, MongoDB is designed for a world where networks can fail. It prioritizes Availability (A) and Partition Tolerance (P), accepting eventually consistent reads from secondary nodes to achieve this. This isn't a weakness, it's a deliberate, powerful choice.

However, MongoDB isn't designed for scenarios requiring strict, real-time cross-document transactions, for that, relational systems are better suited.

4. Cassandra: A Ring-Based Approach to Distribution

While MongoDB uses a router–shard–replica hierarchy, Cassandra takes a very different route. It was designed from day one to handle massive, always-on, global-scale workloads with no single point of failure.

Let’s explore how it achieves that through its peer-to-peer ring architecture.

The Peer-to-Peer Ring Architecture

Cassandra’s core idea is very simple, every node in the cluster is equal there’s no master, router, or coordinator. All nodes form a ring, each responsible for a specific range of data determined by a partition key.

How it works?

· Data is distributed across nodes using consistent hashing.

· Each node is assigned a range of tokens (hash values).

· When a record is written, its partition key (say, user_id) is hashed to find which node owns it.

· That node (and its replicas) handle the read and write requests for that data.

· Because every node knows the entire ring layout, any node can act as a coordinator, meaning the client can connect to any node, and it’ll route the request correctly.

Example:

Let’s say we have four nodes with these token ranges:

Node A: 0 – 25

Node B: 26 – 50

Node C: 51 – 75

Node D: 76 – 100

If a user with user_id = 43 writes data, it hashes to 43 → handled by Node B. Node B stores the data and replicates it to other nodes (as per replication strategy). Unlike MongoDB’s mongos, Cassandra doesn’t need a central router. The ring itself is the routing layer.

4.1 How Cassandra Eliminates Single Points of Failure?

Cassandra’s masterless design is the cornerstone of its fault tolerance.

· Every node can serve reads and writes.

· No node is special, removing any node doesn’t stop the cluster.

· Metadata (cluster topology, ring ownership) is gossiped among nodes, so all stay synchronized.

If one node fails:

· Its data is still available from other replicas.

· The system continues to operate without downtime.

· When the failed node recovers, it performs hinted handoff and read repair to catch up with missed writes.

Example:

· If Node B (responsible for user_id 43) goes down, Node C and Node D (replicas) still have copies of that user’s data.

· Clients can still read and write, Cassandra automatically reroutes.

· When Node B comes back, it syncs missed updates from peers.

This design gives Cassandra continuous availability, even during network partitions or regional outages.

4.2 Replication Strategies Across Nodes

Cassandra doesn't just store one copy of data, it replicates it intelligently for fault tolerance and performance. Replication is controlled by the replication factor (RF) and strategy defined per keyspace (like a database).

Common Replication Strategies

· SimpleStrategy (for single data centers): Data and its replicas are placed on the next N nodes clockwise in the ring.

· NetworkTopologyStrategy (for multiple data centers): You can specify how many replicas to keep per data center, enabling geo-redundancy.

Example

For a keyspace user_data:

CREATE KEYSPACE user_data
WITH replication = {
  'class': 'NetworkTopologyStrategy',
  'us-east': 3,
  'eu-west': 2
};

This keeps:

· 3 replicas in the US-East region

· 2 replicas in EU-West

so even if one data center fails, another can continue serving.

4.3 How Cassandra Handles Writes and Achieves Eventual Consistency

Cassandra’s write path is one of its most efficient features. Writes are fast, sequential, and always accepted even during partial failures.

Here’s what happens step by step when a client writes data:

· The client sends the write to any node (the coordinator).

· The coordinator hashes the partition key to find the responsible node(s).

· The coordinator forwards the write to all replica nodes.

Each replica:

· Logs the write to its commit log (for durability),

· Writes it to an in-memory structure called a memtable.

· The coordinator waits for acknowledgments from replicas (depending on consistency level).

· Later, data is flushed to disk as an SSTable (sorted string table).

This design makes writes append-only and non-blocking, which means Cassandra can sustain millions of writes per second.

4.4 Consistency Levels and Eventual Consistency

Cassandra is an AP (Available + Partition Tolerant) system in CAP terms. It sacrifices strict consistency in favor of availability, but allows you to tune consistency per query.

You can specify consistency levels like:

· ONE: Wait for 1 replica acknowledgment (fastest, but weakest consistency)

· QUORUM: Wait for majority replicas (balanced)

· ALL: Wait for all replicas (strongest consistency, slower)

Example

If replication factor = 3:

· Write at QUORUM waits for 2 replicas to confirm.

· Read at QUORUM ensures overlap with the write quorum, yields consistent results.

· This tunable model lets developers choose between speed and consistency per operation.

How Cassandra achieves eventual consistency?

· Hinted handoff: Temporarily stores writes for unreachable nodes and replays later.

· Read repair: During reads, if replicas have inconsistent data, Cassandra fixes them in the background.

· Anti-entropy repair: Periodically syncs SSTables between replicas.

Together, these mechanisms guarantee that even if replicas temporarily diverge, they’ll eventually converge to a consistent state.

5. Real-World Challenges in Scalable Databases

Scaling databases horizontally sounds elegant in theory, add more nodes, distribute data, and handle more load. In practice, however distributed systems face unique operational challenges that impact performance, reliability, and cost efficiency.

5.1 Resharding: Redistributing Data Safely When Scaling Out

When your data grows or your traffic spikes, you may need to add more shards (machines or partitions). But how do you move terabytes of data across the cluster without downtime or data loss?

Suppose you have 4 shards handling user data distributed by user_id:

· Shard 1 → Users 1–25M

· Shard 2 → Users 25M–50M

· Shard 3 → Users 50M–75M

· Shard 4 → Users 75M–100M

Now your user base doubles to 200M. You add 4 more shards. Resharding means repartitioning this range (e.g., 1–12.5M, 12.5–25M, etc.) and migrating chunks of data safely.

Challenges

· Data must move without blocking reads/writes.

· Hash-based sharding is hard to rebalance evenly.

· Range-based sharding can lead to hotspots if data isn’t evenly distributed.

How Systems Handle It?

· MongoDB uses a balancer process that moves chunks in the background.

· Cassandra uses consistent hashing, which automatically redistributes data when new nodes join.

· Spanner / Vitess use online resharding techniques with dual writes and data copy.

5.2 Hotspots: The "Celebrity Problem" and Uneven Data Distribution

Some data items become "celebrities", they receive disproportionate read/write traffic. Even in a well sharded system, that single record can overload one shard.

For example, In a social media app

"@elonmusk" or "@cristiano" might get millions of hits per second.

All requests for that username land on the same shard.

Impact

· One shard becomes the bottleneck.

· Latency increases.

· The rest of the cluster sits underutilized.

Solutions

· Key Salting: Add random suffixes/prefixes (e.g., “@elonmusk_1”, “@elonmusk_2”) to distribute traffic.

· Read Replicas: Serve reads from multiple replicas instead of one.

· Caching Layers: Use Redis or CDN caching to offload frequent reads.

· Adaptive Load Balancing: Detect and redistribute "hot" keys dynamically.

Hotspots are not about lack of capacity, they’re about imbalanced demand on specific partitions.

5.3 Smart Routing Algorithms and Adaptive Sharding Strategies

Routing each query to the correct shard efficiently becomes tricky as the number of shards grows.

For example, a router node (like mongos in MongoDB or a coordinator in Cassandra) must know where to find the document for user_id=73,459,182. With 100+ shards, a naive routing approach can become costly.

Techniques Used

· Consistent Hashing: Ensures minimal data movement when nodes are added or removed.

· Directory-Based Sharding: Maintain a metadata table that maps key ranges to the shard locations.

· Adaptive Routing: Systems like CockroachDB monitor query latency and dynamically adjust routing paths.

5.4 Balancing Traffic Across Shards Dynamically

Even with correct sharding, workloads change over time. Certain shards may become busier (due to regional traffic, new app features, or time-based events).

Example

· An e-commerce site sees spikes in one country during a festival.

· All users from that region are mapped to a specific shard.

· That shard becomes overloaded while others stay idle.

Solutions

· Auto-balancing: Monitor load and redistribute ranges or replicas dynamically.

· Dynamic Shard Splitting: Split a heavily loaded shard into smaller sub-shards.

· Traffic-aware Proxying: Route new sessions to less-loaded shards.

References

https://engineering.fb.com/2013/06/25/core-infra/tao-the-power-of-the-graph/

System Design Questions

Querying Apache Druid via HTTP APIs

2025-10-21T01:41:00.000-07:00

Apache Druid supports both SQL and native query APIs. Understanding both is crucial for flexible and optimized querying depending on your use case. In this post, we'll walk through how to run both SQL and native queries using curl, leveraging a sample sales dataset.

For example, I onboarded following sales_data to Druid.

sales_data.csv

timestamp,product,city,total_sales
2025-04-01T10:00:00Z,Laptop,Delhi,300
2025-04-01T10:00:00Z,Laptop,Delhi,200
2025-04-01T11:00:00Z,Tablet,Mumbai,150
2025-04-01T11:00:00Z,Tablet,Mumbai,50
2025-04-01T12:00:00Z,Mobile,Bengaluru,200
2025-04-01T13:00:00Z,Laptop,Hyderabad,250
2025-04-01T14:00:00Z,Tablet,Chennai,180
2025-04-01T15:00:00Z,Mobile,Pune,220
2025-04-01T15:00:00Z,Mobile,Pune,80

1. SQL Query via /druid/v2/sql

Druid’s SQL endpoint allows for familiar querying syntax:

Example SQL Query

Get total sales per product

SELECT product, SUM(total_sales) AS total_sales
FROM sales_data
GROUP BY product
ORDER BY total_sales DESC

Execute Using curl

Save the following to a file sql_query.json.

sql_query.json

{
  "query": "SELECT product, SUM(total_sales) AS total_sales FROM sales_data GROUP BY product ORDER BY total_sales DESC"
}

Run the query

curl -X POST http://localhost:8888/druid/v2/sql -H 'Content-Type:application/json' -d @sql_query.json

$curl -X POST http://localhost:8888/druid/v2/sql -H 'Content-Type:application/json' -d @sql_query.json
[{"product":"Laptop","total_sales":750.0},{"product":"Mobile","total_sales":500.0},{"product":"Tablet","total_sales":380.0}]

You can pass this curl output to jq command to pretty print.

$curl -X POST http://localhost:8888/druid/v2/sql -H 'Content-Type:application/json' -d @sql_query.json | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   246  100   125  100   121   9939   9621 --:--:-- --:--:-- --:--:-- 20500
[
  {
    "product": "Laptop",
    "total_sales": 750.0
  },
  {
    "product": "Mobile",
    "total_sales": 500.0
  },
  {
    "product": "Tablet",
    "total_sales": 380.0
  }
]

You can even pass the payload directly to druid/v2/sql API.

curl -X POST http://localhost:8888/druid/v2/sql -H 'Content-Type:application/json' -d '{"query":"SELECT product, SUM(total_sales) AS total_sales FROM sales_data GROUP BY product ORDER BY total_sales DESC"}'

$curl -X POST http://localhost:8888/druid/v2/sql -H 'Content-Type:application/json' -d '{"query":"SELECT product, SUM(total_sales) AS total_sales FROM sales_data GROUP BY product ORDER BY total_sales DESC"}' | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   243  100   125  100   118   9806   9257 --:--:-- --:--:-- --:--:-- 20250
[
  {
    "product": "Laptop",
    "total_sales": 750.0
  },
  {
    "product": "Mobile",
    "total_sales": 500.0
  },
  {
    "product": "Tablet",
    "total_sales": 380.0
  }
]

2. Native Query via /druid/v2

Native queries are powerful, fine-grained JSON-based constructs used for more complex operations.

Example Native Query

Equivalent native groupBy query for total sales per product:

Save to native_query.json.

Execute Using curl

native_query.json

{
  "queryType": "groupBy",
  "dataSource": "sales_data",
  "granularity": "all",
  "dimensions": ["product"],
  "aggregations": [
    {
      "type": "longSum",
      "name": "total_sales",
      "fieldName": "total_sales"
    }
  ],
  "intervals": ["2025-04-01T00:00:00.000Z/2025-04-02T00:00:00.000Z"]
}

curl -X POST http://localhost:8888/druid/v2/?pretty -H 'Content-Type:application/json' -d @native_query.json

$curl -X POST http://localhost:8888/druid/v2/?pretty -H 'Content-Type:application/json' -d @native_query.json
[ {
  "version" : "v1",
  "timestamp" : "2025-04-01T00:00:00.000Z",
  "event" : {
    "product" : "Laptop",
    "total_sales" : 750
  }
}, {
  "version" : "v1",
  "timestamp" : "2025-04-01T00:00:00.000Z",
  "event" : {
    "product" : "Mobile",
    "total_sales" : 500
  }
}, {
  "version" : "v1",
  "timestamp" : "2025-04-01T00:00:00.000Z",
  "event" : {
    "product" : "Tablet",
    "total_sales" : 380
  }
} ]

Passing Native Query payload inline

curl -X POST http://localhost:8888/druid/v2/?pretty -H 'Content-Type:application/json' -d '{"queryType":"groupBy","dataSource":"sales_data","granularity":"all","dimensions":["product"],"aggregations":[{"type":"longSum","name":"total_sales","fieldName":"sales"}],"intervals":["2025-04-01T00:00:00.000Z/2025-04-02T00:00:00.000Z"]}'

$ curl -X POST http://localhost:8888/druid/v2/?pretty -H 'Content-Type:application/json' -d '{"queryType":"groupBy","dataSource":"sales_data","granularity":"all","dimensions":["product"],"aggregations":[{"type":"longSum","name":"total_sales","fieldName":"sales"}],"intervals":["2025-04-01T00:00:00.000Z/2025-04-02T00:00:00.000Z"]}'
[ {
  "version" : "v1",
  "timestamp" : "2025-04-01T00:00:00.000Z",
  "event" : {
    "product" : "Laptop",
    "total_sales" : null
  }
}, {
  "version" : "v1",
  "timestamp" : "2025-04-01T00:00:00.000Z",
  "event" : {
    "product" : "Mobile",
    "total_sales" : null
  }
}, {
  "version" : "v1",
  "timestamp" : "2025-04-01T00:00:00.000Z",
  "event" : {
    "product" : "Tablet",
    "total_sales" : null
  }
} ]

Previous Next Home

Smart Query Routing in Langchain4j: From Default Strategies to LLM-Powered Decisions

2025-10-21T01:32:00.000-07:00

In retrieval-augmented generation (RAG) systems, deciding where to send a user’s query is just as important as how to answer it. As applications scale to use multiple data sources like internal documents, product databases, chat history, or public knowledge, the system needs a smart way to route each query to the right data sources. That’s where QueryRouter in Langchain4j comes in.

This post helps you understand, how Langchain4j’s QueryRouter interface enables intelligent query routing, compares its built-in implementations, and explores how developers can use this routing logic to fit their applications.

1. What is Query Routing in RAG Systems?

Query routing refers to the process of deciding which data sources a user's query should be sent to, especially when there are multiple potential sources of information. In the context of RAG (Retrieval-Augmented Generation) systems, this becomes critical because:

· The system doesn't know the answer itself, it must retrieve supporting content from external sources before generating a response.

· These sources could be diverse, product FAQs, code documentation, user chat history, financial records, or even web search results.

· Query routing ensures that each query is routed only to the most relevant source(s), rather than flooding all sources with every question.

Example: Imagine you’re in a library that has sections for science, history, fiction, law, and art. If you ask a question like “What is Newton’s First Law?”, it would be wasteful (and confusing) if the librarian checks every section. Instead, the librarian should route your question directly to the science section. This is exactly what query routing does in a RAG system, it plays the role of that intelligent librarian.

2. Why Irrelevant Data Retrieval is a Problem?

Without proper query routing, the system might send every query to all available retrievers or databases. This creates several issues:

· Performance Overhead: More retrievers means more computation. Fetching irrelevant documents increases processing time and costs, especially if using embeddings or LLMs for retrieval.

· Latency Increase: Retrieving from every source leads to slower response times, especially problematic in real-time chat applications.

· Answer Quality Degradation" Including unrelated documents in the context window can confuse the language model, leading to hallucinations or diluted answers.

· Wasted API/Infra Costs: Many retrieval services are usage-based. Querying unnecessary retrievers drives up costs without improving results.

3. Benefits of Effective Query Routing

Smart query routing enables:

· Higher Accuracy: By routing queries only to relevant sources, the retrieved content is more likely to be useful, improving the quality of the LLM’s response.

· Lower Latency: Querying fewer retrievers leads to faster retrieval and quicker final responses.

· Scalability: As you add more data sources (e.g., documents, APIs, databases), you can continue scaling without overwhelming the system.

· Context Optimization: It avoids context window overflow and preserving space for genuinely helpful content.

· Improved User Experience: Faster, more accurate, and less noisy responses directly improve user trust and satisfaction.

4. QueryRouter Interface in Langchain4j

The QueryRouter interface in Langchain4j is designed to route a Query object to one or more ContentRetriever instances. This decision is typically based on the nature of the query, user metadata, or semantic meaning.

Here is the contract:

public interface QueryRouter {
    Collection<ContentRetriever> route(Query query);
}

When a query arrives, the route() method decides which ContentRetrievers (e.g., vector stores, database retrievers, document retrievers) should handle the query. The goal is to avoid retrieving irrelevant content and optimize the downstream generation step.

Built-in Implementations of QueryRouter in Langchain4j

Langchain4j provides two implementations of the QueryRouter interface, each designed for different stages of application complexity and scalability:

· DefaultQueryRouter: simple and suitable for most basic use cases.

· LanguageModelQueryRouter: intelligent, LLM-powered routing for dynamic, content-aware decisions.

4.1 DefaultQueryRouter: DefaultQueryRouter is the simplest and most straightforward implementation of the QueryRouter interface. It is also the default routing mechanism used internally by the DefaultRetrievalAugmentor.

public class DefaultQueryRouter implements QueryRouter {
    // routes to all configured ContentRetrievers
}

This router sends every query to all the retrievers that were provided to it during construction. It assumes all retrievers are relevant for any query, regardless of the query’s content or context.

This implementation is ideal when:

· You have a small number of retrievers (e.g., 2–3).

· All retrievers are semantically similar or overlapping.

· You are in the early stage of building your RAG system.

· Performance and routing precision are not major concerns yet.

Limitations

· May lead to performance degradation when the number of retrievers increases.

· Fetches potentially irrelevant documents, which can confuse the language model.

· Not suitable for use cases requiring domain-specific or context-aware routing.

The class documentation clearly mentions that while DefaultQueryRouter works for most use cases now, its default behavior may evolve over time if better routing logic becomes standard.

4.2 LanguageModelQueryRouter

LanguageModelQueryRouter brings intelligent, dynamic routing to the system. It uses an LLM (via a ChatModel) to decide which retriever(s) are most relevant for a given query.

public class LanguageModelQueryRouter implements QueryRouter {
    // uses an LLM to choose retrievers based on the query
}

How It Works?

Each ContentRetriever is configured with a natural language description (e.g., “Handles queries related to HR policies.”). When a query arrives, the LLM is prompted with:

· The query itself.

· Descriptions of all available retrievers.

· The model selects one or more retrievers based on semantic understanding.

This router use following prompt to find the suitable content retrievers.

   public static final PromptTemplate DEFAULT_PROMPT_TEMPLATE = PromptTemplate.from(
            """
                    Based on the user query, determine the most suitable data source(s) \
                    to retrieve relevant information from the following options:
                    {{options}}
                    It is very important that your answer consists of either a single number \
                    or multiple numbers separated by commas and nothing else!
                    User query: {{query}}"""
    );

You can even configure a custom prompt template (promptTemplate) to control how the LLM interprets and decides routing.

This is ideal when:

· Your application deals with various domains or topics.

· You need semantic-level understanding to decide retriever relevance.

· You have many retrievers, each tied to a specific dataset.

· You want to minimize noise in the LLM’s input context.

For example, if your application has following retrievers:

· HR Retriever: "Provides info on employee benefits and leave policies."

· IT Retriever: "Handles technical issues like password resets and VPN."

· Finance Retriever: "Answers queries related to reimbursements and taxes."

And the user query is: "How do I file a reimbursement for a business trip?" The LanguageModelQueryRouter will intelligently route it only to the Finance Retriever, not HR or IT.

Sample Application

Following program demonstrates how to dynamically route user queries to the most relevant content stores using LangChain4j’s LanguageModelQueryRouter.

QueryRouterDemo.java

package com.sample.app.router;

import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.model.embedding.onnx.bgesmallenv15q.BgeSmallEnV15QuantizedEmbeddingModel;
import dev.langchain4j.model.ollama.OllamaChatModel;
import dev.langchain4j.rag.content.retriever.ContentRetriever;
import dev.langchain4j.rag.content.retriever.EmbeddingStoreContentRetriever;
import dev.langchain4j.rag.query.Query;
import dev.langchain4j.rag.query.router.LanguageModelQueryRouter;
import dev.langchain4j.store.embedding.EmbeddingStore;
import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class QueryRouterDemo {

  public static void main(String[] args) {

    // Initialize LLM model for routing
    OllamaChatModel chatModel =
        OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

    // Shared embedding model
    EmbeddingModel embeddingModel = new BgeSmallEnV15QuantizedEmbeddingModel();

    // Create embedding stores
    EmbeddingStore<TextSegment> hrStore = new InMemoryEmbeddingStore<>();
    EmbeddingStore<TextSegment> itStore = new InMemoryEmbeddingStore<>();
    EmbeddingStore<TextSegment> financeStore = new InMemoryEmbeddingStore<>();

    // Create content retrievers
    ContentRetriever hrRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(hrStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("hrRetriever")
            .build();

    ContentRetriever itRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(itStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("itRetriever")
            .build();

    ContentRetriever financeRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(financeStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("financeRetriever")
            .build();

    Map<ContentRetriever, String> retrieverToDescription = new HashMap<>();
    retrieverToDescription.put(
        hrRetriever,
        "Provides information on leave policies, benefits, onboarding, and other HR-related topics.");
    retrieverToDescription.put(
        itRetriever,
        "Handles technical queries such as VPN access, email issues, password resets, and software installations.");
    retrieverToDescription.put(
        financeRetriever,
        "Answers questions related to reimbursements, payslips, taxation, invoices, and financial approvals.");

    // Create LLM-based router
    LanguageModelQueryRouter router =
        LanguageModelQueryRouter.builder()
            .chatModel(chatModel)
            .retrieverToDescription(retrieverToDescription)
            .build();

    // Sample queries
    List<String> queries =
        List.of(
            "How do I reset my email password?",
            "Where can I see my payslip for last month?",
            "What is the company policy on medical leave?",
            "How to claim business travel reimbursement?",
            "I'm facing issues connecting to the VPN.");

    for (String userQuery : queries) {
      Query query = Query.from(userQuery);
      Collection<ContentRetriever> retrievers = router.route(query);

      System.out.println("Query: " + userQuery);
      for (ContentRetriever retriever : retrievers) {
        System.out.println("-> Routed to: " + retriever);
      }
      System.out.println("---------------------------------------------------");
    }
  }
}

Output

Query: How do I reset my email password?
-> Routed to: EmbeddingStoreContentRetriever{displayName='itRetriever'}
---------------------------------------------------
Query: Where can I see my payslip for last month?
-> Routed to: EmbeddingStoreContentRetriever{displayName='itRetriever'}
-> Routed to: EmbeddingStoreContentRetriever{displayName='financeRetriever'}
---------------------------------------------------
Query: What is the company policy on medical leave?
-> Routed to: EmbeddingStoreContentRetriever{displayName='financeRetriever'}
---------------------------------------------------
Query: How to claim business travel reimbursement?
-> Routed to: EmbeddingStoreContentRetriever{displayName='itRetriever'}
-> Routed to: EmbeddingStoreContentRetriever{displayName='financeRetriever'}
---------------------------------------------------
Query: I'm facing issues connecting to the VPN.
-> Routed to: EmbeddingStoreContentRetriever{displayName='itRetriever'}
---------------------------------------------------

Find the below complete working application.

QueryRouterFullApp.java

package com.sample.app.router;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import dev.langchain4j.data.document.Document;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.memory.chat.MessageWindowChatMemory;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.model.embedding.onnx.bgesmallenv15q.BgeSmallEnV15QuantizedEmbeddingModel;
import dev.langchain4j.model.ollama.OllamaChatModel;
import dev.langchain4j.rag.DefaultRetrievalAugmentor;
import dev.langchain4j.rag.RetrievalAugmentor;
import dev.langchain4j.rag.content.retriever.ContentRetriever;
import dev.langchain4j.rag.content.retriever.EmbeddingStoreContentRetriever;
import dev.langchain4j.rag.query.router.LanguageModelQueryRouter;
import dev.langchain4j.rag.query.transformer.ExpandingQueryTransformer;
import dev.langchain4j.service.AiServices;
import dev.langchain4j.store.embedding.EmbeddingStore;
import dev.langchain4j.store.embedding.EmbeddingStoreIngestor;
import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;


public class QueryRouterFullApp {

  interface ChatAssistant {
    String chat(String userMessage);
  }

  public static void main(String[] args) {

    // Initialize LLM model for routing and chatting
    OllamaChatModel chatModel =
        OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

    // Shared embedding model
    EmbeddingModel embeddingModel = new BgeSmallEnV15QuantizedEmbeddingModel();

    // Create embedding stores
    EmbeddingStore<TextSegment> hrStore = new InMemoryEmbeddingStore<>();
    EmbeddingStore<TextSegment> itStore = new InMemoryEmbeddingStore<>();
    EmbeddingStore<TextSegment> financeStore = new InMemoryEmbeddingStore<>();

    // Sample data for ingestion
    List<String> hrDocs =
        List.of(
            """
            Our HR policies include flexible work hours, medical leave, and onboarding support.
            """,
            """
            You can check your leave balance and benefits on the HR portal.
            """
        );

    List<String> itDocs =
        List.of(
            """
            If you're facing issues with VPN, restart your system and try again.
            """,
            """
            Password resets can be done via the IT Helpdesk portal.
            """
        );

    List<String> financeDocs =
        List.of(
            """
            Payslips are generated on the 5th of every month and available on the finance dashboard.
            """,
            """
            You can file business travel reimbursements through the expense portal.
            """
        );

    // Ingest documents into respective stores
    EmbeddingStoreIngestor.ingest(convertToDocuments(hrDocs), hrStore);
    EmbeddingStoreIngestor.ingest(convertToDocuments(itDocs), itStore);
    EmbeddingStoreIngestor.ingest(convertToDocuments(financeDocs), financeStore);

    // Create content retrievers
    ContentRetriever hrRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(hrStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("hrRetriever")
            .build();

    ContentRetriever itRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(itStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("itRetriever")
            .build();

    ContentRetriever financeRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(financeStore)
            .embeddingModel(embeddingModel)
            .maxResults(3)
            .minScore(0.7)
            .displayName("financeRetriever")
            .build();

    Map<ContentRetriever, String> retrieverToDescription = new HashMap<>();
    retrieverToDescription.put(
        hrRetriever,
        "Provides information on leave policies, benefits, onboarding, and other HR-related topics.");
    retrieverToDescription.put(
        itRetriever,
        "Handles technical queries such as VPN access, email issues, password resets, and software installations.");
    retrieverToDescription.put(
        financeRetriever,
        "Answers questions related to reimbursements, payslips, taxation, invoices, and financial approvals.");

    // Create LLM-based router
    LanguageModelQueryRouter router =
        LanguageModelQueryRouter.builder()
            .chatModel(chatModel)
            .retrieverToDescription(retrieverToDescription)
            .build();

    // Expand queries (optional step for better retrieval)
    ExpandingQueryTransformer expandingQueryTransformer = new ExpandingQueryTransformer(chatModel, 3);

    // Create retrieval augmentor with dynamic contentRetriever via router
    RetrievalAugmentor retrievalAugmentor =
        DefaultRetrievalAugmentor.builder()
            .queryTransformer(expandingQueryTransformer)
            .queryRouter(router)
            .build();

    // Build the assistant
    ChatAssistant chatAssistant =
        AiServices.builder(ChatAssistant.class)
            .chatModel(chatModel)
            .retrievalAugmentor(retrievalAugmentor)
            .chatMemory(MessageWindowChatMemory.withMaxMessages(10))
            .build();

    // Run a conversation
    List<String> userQueries =
        List.of(
            "How do I reset my email password?",
            "Where can I see my payslip for last month?",
            "What is the company policy on medical leave?",
            "How to claim business travel reimbursement?",
            "I'm facing issues connecting to the VPN."
        );

    for (String userQuery : userQueries) {
      System.out.println("User: " + userQuery);
      String answer = chatAssistant.chat(userQuery);
      System.out.println("Assistant: " + answer);
      System.out.println("-----------------------------------------------");
    }
  }

  private static List<Document> convertToDocuments(List<String> texts) {
    List<Document> documents = new ArrayList<>();
    for (String text : texts) {
      documents.add(Document.from(text));
    }
    return documents;
  }
}

Output

User: How do I reset my email password?
Assistant: To reset your email password, follow these steps:

1. Go to the IT Helpdesk portal.
2. Fill out the required details in the password reset form.
3. If you're experiencing issues with a Virtual Private Network (VPN), try restarting your system and then retrying the password reset process.

That's it!
-----------------------------------------------
User: Where can I see my payslip for last month?
Assistant: To view your payslip for last month, follow these steps:

1. Go to the finance dashboard.
2. Check if the latest month's payslip has been generated (typically available on the 5th of every month).
3. If you're experiencing issues with a VPN, restart your system and try accessing the payslip again.

Note: You can also check other HR-related information, such as leave balance and benefits, on the HR portal.
-----------------------------------------------
User: What is the company policy on medical leave?
Assistant: The company's policy on medical leave is stated in their HR policies, which include:

* Flexible work hours
* Medical leave (specific details not provided)

For more information about your specific situation, including your leave balance and benefits, you can check these details on the HR portal.

Note that none of this information provides direct guidance on the actual company policy, but rather informs where to find more information about it.
-----------------------------------------------
User: How to claim business travel reimbursement?
Assistant: To claim business travel reimbursement, follow these steps:

1. Go to the expense portal.
2. File your business travel reimbursements through this platform.

That's it! This is where you can submit your expenses for reimbursement.
-----------------------------------------------
User: I'm facing issues connecting to the VPN.
Assistant: To resolve VPN connection issues:

1. Restart your system.
2. Try reconnecting to the VPN again.

This may help resolve any temporary connectivity problems. If you continue to experience issues, you can also consider resetting your password using the IT Helpdesk portal.
-----------------------------------------------

Previous Next Home

Why Distributed Transactions Are Hard: The Hidden Cost of Scaling Relational Databases Horizontally

2025-10-13T07:45:00.000-07:00

As applications scale and demand increases, developers often look to horizontally scale their databases to handle larger workloads. While this approach helps with performance and availability, it introduces a less visible but critical challenge: managing transactions across multiple nodes.

In a traditional, single-node relational database, maintaining ACID properties is straightforward, everything happens in one place. But when data and operations span across distributed systems, transactions become complex, fragile, and costly to orchestrate.

This post helps you to understand why distributed transactions are difficult, how they impact system performance under heavy load, and what tools or architectural patterns you might consider to overcome these limitations.

1. Horizontal Scaling

Horizontal scaling refers to adding more machines (nodes) to a system to distribute the load across multiple computing resources. Each node runs an instance of the application or database and participates in processing requests.

In the context of Databases,

Vertical Scaling: Increases capacity of a single database instance by upgrading hardware (CPU, RAM, IOPS), so it can serve more traffic/user requests. It is simpler to implement, and minimal changes might be needed to the application code or DBMS configuration.

But Vertical Scaling is subject to physical hardware constraints. High availability may still require replicas, adding more hardware/procuring high end servers always incurs cost.

Horizontal Scaling: It involves distributing data and queries across multiple database instances. Following are the some common approaches used widely.

· Read Replicas: multiple read-only replicas handle read queries, and the primary node handles writes.

· Sharding: dataset is partitioned by shard key (e.g., user ID ranges), each shard resides on a different node.

· Federated Databases: A federated database system integrates several independent, often geographically distributed, databases into one virtual database. This unified view allows users and applications to query data as if it resided in a single database, without needing to be aware of the underlying complexities or physical locations of the individual databases. In this model, each node is responsible for a subset of data, with coordination at the application layer.

1.1 Technical Challenges of Horizontal Scaling

Data Partitioning (Sharding): requires a shard key strategy that minimizes cross-shard queries. If the shard key is not desingned properly, then there is a risk of uneven data distribution (hot spots).

Distributed Transactions: ACID properties guarantees across multiple nodes require protocols like 2PC (Two-Phase Commit) or saga patterns, which add latency and complexity.

Data Consistency: Achieving strong consistency across replicas or shards is non-trivial. Here CAP Theorem applies, which states that consistency, availability, and partition tolerance cannot be fully achieved simultaneously in distributed systems.

Replication Lag: In asynchronous replication, data changes made to the primary database are not immediately reflected in the read replicas. As a result, applications reading from replicas might receive stale or outdated data. This can lead to inconsistencies, especially in systems that rely on real-time accuracy.

Joins are Expensive: When data is spread across multiple shards, performing operations like joins becomes challenging. Joins across shards are either very expensive in terms of performance or not supported natively by many databases. This limitation can slow down complex queries and affect the overall efficiency of the system.

To handle these challenges, applications often require extra logic at the application level or need to use distributed query engines. These tools help coordinate data across shards and replicas, but they also add complexity to the system architecture.

2. Distributed Transactions

A distributed transaction is a database transaction that spans multiple independent systems or databases (usually across different servers or nodes), and must be executed in a coordinated way so that either all operations succeed, or all fail, and ensures atomicity and consistency across systems.

2.1 ACID Transactions in a Single Node vs Distributed Environment

In a non-distributed relational database, all components reside within a single system. Write operations are typically handled by a primary node, while read operations can be offloaded to multiple replica (or "slave") nodes. The database engine centrally manages tables, indexes, logs, buffers, and locks.

When a transaction spans multiple tables, it is coordinated internally using local ACID properties there is no need for external coordination.

BEGIN;

UPDATE accounts SET balance = balance - 500 WHERE id = 1;

UPDATE accounts SET balance = balance + 500 WHERE id = 2;

COMMIT;

This entire transaction is atomic and isolated because it is executed within a single database engine on one machine.

Whereas in a distributed environment:

· Atomicity: Every part of the transaction must either commit completely or not at all, spanning multiple nodes.

· Consistency: All nodes must maintain a consistent state (Some systems support Eventual consistency), even in the event of failures.

· Isolation: Concurrent transactions must be properly managed across distributed resources.

· Durability: Once committed, a transaction’s changes must be permanently recorded across multiple systems.

2.2 Why it is difficult to implement Distributed Transactions?

Network Latency and Failures: Implementing distributed transactions is difficult largely because they depend heavily on network communication between multiple, physically separate nodes. Unlike a single-server system where data and transaction coordination happen internally with minimal delay, distributed systems must send messages back and forth over a network. This introduces network latency, meaning every step in the transaction like preparing, committing, or rolling back changes on different nodes takes longer simply because of the time it takes for data to travel across the network. Moreover, networks are inherently unreliable, packets can be lost, delayed, duplicated, or nodes can become temporarily unreachable due to hardware failures, maintenance, or congestion. These uncertainties raise the risk of partial failures where some nodes successfully commit their part of the transaction, while others do not. Handling such inconsistencies requires complex coordination protocols and recovery mechanisms, which add overhead and further increase latency. Together, network delays and the possibility of failures make maintaining atomicity and consistency across distributed transactions a challenging problem in distributed systems.

Two-Phase Commit (2PC): It is a classic protocol designed to ensure that distributed transactions maintain ACID properties (atomicity, consistency, isolation, and durability) across multiple database nodes. It works in two main stages:

· First, in the prepare phase, the coordinator sends a request to all participating nodes asking if they are ready to commit the transaction. Each node performs all necessary checks and writes the transaction changes to a temporary log but does not finalize the commit yet. They then respond with either a “yes” (ready to commit) or “no” (abort). Only if all nodes reply “yes” does the coordinator proceed to the second stage,

· Second the commit phase, where it instructs all nodes to finalize and permanently commit the changes. If any node fails to commit, the coordinator tells all nodes to roll back the transaction or some manual intervention might be needed.

While 2PC guarantees atomicity across nodes, it has some significant drawbacks. Because all nodes must wait for each other’s responses, the protocol can become slow under heavy load, as network communication and disk writes add latency. Moreover, 2PC is a blocking protocol, if any node or the coordinator crashes after the prepare phase but before sending the commit or rollback message, the other nodes remain locked in a waiting state, unable to proceed or release resources. This can cause the entire system to stall unless complex recovery mechanisms or timeouts are implemented, which adds further complexity to distributed transaction management.

Concurrency and Locking: When a database is distributed across multiple nodes, managing concurrency, the ability for multiple transactions to run simultaneously without interfering with each other becomes significantly more challenging. Each node must enforce locks on the data it manages to ensure transactions don’t overwrite or read inconsistent data. As the number of nodes increases, so does the total number of locks held across the system. This higher volume of locks raises the chances of deadlocks. Additionally, contention increases as multiple transactions compete for the same resources, causing some to wait longer before they can proceed. This waiting leads to blocking, where transactions are forced to pause until the necessary locks are released. Together, these issues reduce the system’s overall throughput (the number of transactions completed in a given time) and add complexity to concurrency control mechanisms. Distributed systems must implement efficient algorithms to detect and resolve deadlocks, manage lock timeouts, and optimize resource access, making concurrency control a major challenge as the system scales horizontally.

Consistency Models: Maintaining strong consistency across distributed nodes means ensuring that every node always sees the same up-to-date data at any given time. Achieving this requires constant synchronization between all the nodes so that updates are immediately reflected everywhere. However, in a distributed system, network delays, failures, or partitions, where nodes become temporarily unreachable make perfect synchronization difficult.

According to the CAP theorem, a distributed system can only guarantee two out of following three properties at the same time.

· Consistency,

· Availability, and

· Partition tolerance.

Prioritizing strong consistency often means sacrificing availability because the system must wait for all nodes to synchronize before responding to requests. Alternatively, if the system remains available during network partitions, it may have to allow some nodes to operate on stale data, sacrificing strict consistency. This fundamental trade-off means that maintaining strong consistency in distributed systems is complex and often involves balancing synchronization overhead, response times, and fault tolerance.

In summary, we need an effective and robust distributed transaction manager (TM), which is a critical component for ensuring reliable coordination of transactions that span multiple nodes or services in a distributed system. The TM’s primary responsibility is to orchestrate the entire lifecycle of a distributed transaction, making sure it adheres to ACID properties across all participants. To do this, the TM communicates with all involved resource managers (such as different databases or microservices), coordinating their actions to prepare, commit, or roll back changes in a way that maintains global consistency.

The TM must also handle failure recovery gracefully. This involves detecting and resolving partial failures where some nodes may have committed while others failed, as well as managing situations where the TM or participants crash during the transaction process. It typically uses durable logs to record transaction states and decisions, enabling it to resume or roll back incomplete transactions after a failure.

In addition, the TM often implements standard protocols like Two-Phase Commit (2PC) or more advanced consensus algorithms to ensure reliable communication and coordination. It abstracts the complexity of distributed coordination away from individual resource managers, providing a unified interface that guarantees transactional integrity, even in the case of network delays, node failures, and concurrent access conflicts.

System Design Questions

Understanding ACID Properties in Database Systems

2025-10-13T07:44:00.000-07:00

In database systems, maintaining the correctness and reliability of data during transactions is essential. The ACID properties (Atomicity, Consistency, Isolation, and Durability) define a set of standards that ensure transactions are processed accurately and reliably. This blog post explains each of these properties with clear definitions and examples to help you understand how they contribute to data integrity and system reliability in transactional databases.

1. Atomicity

Atomicity refers to the "all-or-nothing" nature of a database transaction. In simple terms, this means that a transaction must be fully completed or not executed at all. If any part of the transaction fails, the entire operation is rolled back, and the database remains unchanged.

This property ensures that there is no partial execution of a transaction. It prevents scenarios where only some operations in a transaction are applied, leaving the database in an inconsistent or incomplete state.

1.1 Why is Atomicity important?

In real-world applications, a transaction often involves multiple steps or operations that are logically connected. These steps must be treated as a single unit of work. If one step fails, none of the steps should take effect. This guarantees that the system's state remains valid and predictable.

Consider a simple fund transfer between two bank accounts A and B:

· Debit ₹5,000 from Account A.

· Credit ₹5,000 to Account B.

These two operations must occur together. If the system debits Account A but fails to credit Account B (due to a crash, network issue, or any error), the customer would lose money, and the bank records would be incorrect. With atomicity, such a transaction would be rolled back entirely if both operations cannot be completed successfully.

1.2 How is Atomicity Achieved in Database Systems?

database systems use three core mechanisms to achieve Atomicity.

· Transaction Logs

· Rollback

· Commit

Let's understand this with a successful money transfer from Account A to Account B.

Let’s say a customer initiates a transaction to transfer ₹5,000 from Account A to Account B using a banking app. This transfer involves two critical updates.

· Debit ₹5,000 from Account A

· Credit ₹5,000 to Account B

Step 1: Transaction Begins

The database system starts a new transaction and assigns it a unique transaction ID (e.g., TX1234). This marks the beginning of the atomic unit of work. Nothing is changed in the actual database yet.

Step 2: Validate the Operations

The database performs preliminary validations to avoid logging unnecessary or invalid transactions.

It checks:

· If Account A has sufficient balance.

· If the accounts involved are valid and active.

· If the debit and credit operations are logically sound.

· If the transaction respects all integrity constraints.

Step 3: Write-Ahead Logging (Transaction Log)

Before any change is written to the database, the system writes detailed information about the transaction to a transaction log.

This includes:

· Transaction ID (TX1234)

· Operation types (DEBIT, CREDIT)

· Before and after values

· Current status (PENDING)

This is known as Write-Ahead Logging (WAL), means the changes must be logged before they are applied. This allows the system to recover if a failure occurs. For high reliability, some databases replicate this log to another disk or standby server. This protects against disk crashes and ensures durability.

Step 4: Execute the Operations

After logging, the debit and credit operations are carried out in memory, no actual disk operation occurs yet.

Step 5: Commit

If all operations succeed:

· The changes are written from memory to the actual database (on disk).

· A COMMIT record is added to the transaction log. The transaction is marked as successfully completed.

At this point, the updates are permanent and visible to other users. The atomicity principle is honored, both debit and credit happened together as a single unit.

2. Consistent

Consistency means that a database must always transition from one valid state to another valid state after any transaction completes. The data should never end up in a state that violates the defined rules, constraints, or logical relationships of the database.

2.1 What Does Consistency Really Mean?

Imagine the database as a system that follows strict rules about how data relates to each other. These rules include:

· Data integrity constraints like primary keys, foreign keys, and unique constraints.

· Business rules such as "an order must always belong to an existing customer."

· Validation rules ensuring data formats, ranges, or mandatory fields are correct.

A consistent database means every transaction preserves all these rules.

Example: Deleting a Customer and Their Orders

Suppose you have two tables:

· Customers: stores customer information.

· Orders: stores all orders placed by customers, linked by a foreign key to the Customers table.

If a transaction deletes a customer record, consistency requires that:

· All their orders must also be deleted (cascading delete), or

· The system must prevent the deletion if orders exist (restrict delete).

2.2 Why Is This Important?

If the system allowed deleting a customer but left behind orders that reference that now-nonexistent customer, the database would enter an inconsistent state:

Orders would have foreign keys pointing to a missing customer.

Applications reading these orders could crash or behave unpredictably.

2.3 How Is Consistency Enforced?

Database systems use several mechanisms to ensure consistency:

· Constraints: Foreign keys, unique keys, check constraints.

· Triggers: Automatically execute rules or actions to maintain consistency.

· Transaction rules: The entire transaction must commit or rollback as a whole, so partial updates can’t leave data inconsistent.

In summary, Consistency means that after every transaction, the database stays correct and follows all the rules. It makes sure the database never shows wrong or confusing information to people or programs, so the data can be trusted and is reliable.

3. Isolation

Isolation means that when two or more transactions happen at the same time, they don’t get mixed up with each other. Each transaction works by itself, without interfering with others.

For example, if two transactions try to change the same piece of data at the same time, one of them will wait until the other finishes. This way, the changes don’t cause mistakes.

Isolation keep the data accurate, even when many users are using the database at once.

3.1 Why is Isolation Important?

When many people or programs use a database at the same time, they often try to read or change the same data simultaneously. Without isolation:

· Changes made by one transaction could mix up or conflict with changes from another.

· This can cause errors, like incorrect balances, lost updates, or corrupted data.

For example, if two people try to withdraw money from the same bank account at the same time, without isolation, both might see the old balance and withdraw too much.

Isolation keeps each transaction separate and protected, so they don’t interfere with each other. This helps the database stay accurate and reliable even when many users work at once.

3.2 How is Isolation achieved?

Databases use different techniques like Locking, Transaction Scheduling, Isolation Levels etc., to keep transactions isolated.

3.2.1 Locking

When a transaction wants to read or write data, the database locks that data so others can’t change it at the same time.

For example, if Transaction A locks a customer’s account to update the balance, Transaction B must wait until Transaction A finishes.

Locks can be:

· Shared locks for reading (many can read at once).

· Exclusive locks for writing (only one can write at a time).

3.2.2 Transaction Scheduling

The database controls the order in which transactions run, making sure they don’t overlap in ways that cause conflicts. It may delay some transactions so that each one appears to run alone, even if they actually run at the same time.

3.2.3 Isolation levels

Isolation levels decide how strictly transactions are kept separate from each other when they run at the same time. The choice of level affects both the accuracy of data and the speed of the system.

Higher isolation levels keep transactions very separate and avoid almost all conflicts, but they can slow things down because transactions might have to wait longer.

Lower isolation levels let transactions overlap more, which can speed things up, but may sometimes allow incorrect or confusing data to appear temporarily.

Following are the some common isolation levels:

· Read Uncommitted

· Read Committed

· Repeatable Read

· Serializable

Read Uncommitted

Transactions can see uncommitted changes made by other transactions ("dirty reads") No locking of read data. It offers the highest performance but lowest consistency

It is rarely used in practice except for approximate analytics where absolute accuracy isn't critical

Read Committed

Transactions can only see committed changes (no dirty reads). It internally use row level locks for writes. Each read operation sees only data committed before that specific read. Following are possible:

· Non-repeatable reads (a row read twice in same transaction may differ if another transaction committed changes). For example, you read an account balance twice during a transaction and see different amounts because another transaction updated it in between.

· Phantom reads (new rows may appear in subsequent reads)

Repeatable Read

The Repeatable Read isolation level ensures that if a transaction reads a row once, it will see the same data if it reads that row again later during the same transaction. It uses locks to prevent other transactions from modifying data that has been read.

It allows Phantom reads, like if a transaction runs a query like SELECT * FROM orders WHERE amount > 100, and another transaction inserts a new matching row before the first one commits, the new row can appear in repeated queries.

Serializable (Highest Isolation)

Transactions run so strictly that it looks like they run one after another, not at the same time. It prevents all types of conflicts, including dirty reads, non-repeatable reads, and phantom reads. It is most secure but slowest, because transactions often wait for others to finish.

It is used when absolute accuracy is required, like in banking systems.

4. Durability

Durability means that once a transaction is successfully completed (committed), its changes are permanent, they will not be lost, even if the system crashes right after.

· The data is safely stored on disk.

· It will still be there if the server restarts, power goes out, or the database crashes.

Following core mechanisms helps to ensure Durability.

Write-Ahead Logging (WAL): Before applying any changes to the actual database, the database writes a log record describing the changes. In general, this log record is replicated accross multiple machines to ensure durability.

Checkpointing: A checkpoint is a snapshot operation that synchronizes in-memory data with disk storage.

In Distributed systems, they have Quorum-Based approach, where the durability is achieved via replication. Here a write is only considered successful if multiple nodes acknowledge it.

System Design Questions

Enhancing RAG Retrievals in Langchain4j with Query Transformers: Techniques & Best Practices

2025-09-16T08:21:00.000-07:00

When building Retrieval-Augmented Generation (RAG) systems, the quality of retrieved results heavily depends on how the input query is framed. Users may submit vague, ambiguous, or verbose queries, which can degrade retrieval performance. This is where Query Transformers in Langchain4j come into play, they help to transform and optimize queries to improve the relevance of retrieved documents before generating a response.

In this post, we’ll explore what Query Transformers are in Langchain4j, why they matter, and dive into various strategies such as query compression, expansion, rewriting, step-back prompting, and hypothetical document embeddings (HyDE).

1. Introduction to RAG

RAG (Retrieval-Augmented Generation) is a method that helps large language models (LLMs) to get better answers by first finding useful information from your data and adding it to the prompt before sending it to the model. This helps the model respond more accurately and reduces the chances of it making things up (called hallucinations).

There are different ways to find this useful information:

· Full-text (keyword) search: Looks for documents that match the words in your question. It uses methods like TF-IDF or BM25 to rank how well each document matches.

· Vector (semantic) search: Converts documents and questions into numbers (vectors) and compares them to find the most similar meanings, not just matching words.

· Hybrid search: Combines both full-text and vector search to get better results.

Right now, this guide mainly focuses on vector search.

2. RAG Stages

The Retrieval-Augmented Generation (RAG) process usually has two main steps: Indexing and Retrieval. LangChain4j provides helpful tools for both.

2.1 Indexing Stage

Indexing is the first step where documents are prepared so they can be quickly and accurately searched later. For vector-based search, this step includes:

· Cleaning the content to remove unnecessary or irrelevant parts

· Adding extra context or metadata if needed

· Splitting large documents into smaller parts (called chunks)

· Converting those chunks into numerical vectors using embeddings

· Saving them in a special database called an embedding store or vector database

This step is usually done in the background, not while users are interacting with the app. For example, a scheduled job (like a weekly cron job) might update the data during non peak hours, such as weekends. Sometimes, a separate service takes care of managing the vector database.

In other cases, like when users upload their own documents, the indexing needs to happen right away (in real time) and be built into the app.

For this proof of concept (POC), documents related to the fictional company called ChronoCore Industries are stored in my computer and loaded into memory. This simple setup works well for the demo.

Here is a simplified diagram of the indexing stage.

2.2 Retrieval Stage

The retrieval stage begins when a user submits a question. The system then searches through pre-indexed documents to identify the most relevant information.

For vector search, the user's query is converted into an embedding, a numerical representation of the query. The system compares this embedding to others in the embedding store to find the most similar document segments. These relevant chunks are then included in the prompt provided to the language model to help generate an accurate response.

2. Why to Transform User Queries?

In traditional RAG (Retrieval-Augmented Generation) systems, documents are typically split into chunks, embedded using a language model, and later retrieved based on their semantic similarity to a user’s query. While this naive RAG approach works in many cases, it has notable limitations that can make it difficult both the quality and relevance of retrieved information.

Key challenges include

· Irrelevant Chunk Content: The chunking process often slices documents without understanding semantic boundaries. As a result, a single chunk might contain useful information along with unrelated or distracting content, reducing retrieval accuracy.

· Poorly Formulated User Queries: Users may express their questions in vague, incomplete, or overly specific ways that don’t align well with the format or content of stored embeddings, leading to suboptimal matches.

· Lack of Structured Query Support: In more advanced use cases, retrieval isn't just about matching free-text queries. It may involve generating structured queries, such as filtering vector search results based on metadata (e.g., date, author) or even translating natural language into SQL for querying relational databases.

To overcome these limitations, a new class of techniques focuses on transforming user queries before they are sent to the retriever. These techniques improve retrieval quality by refining the intent, scope, or format of the question.

In this post, we’ll explore some of the most effective query transformation strategies, including query rewriting, expansion, compression, and more advanced techniques like step-back prompting and hypothetical document embeddings (HyDE).

2.1 Query Rewriting

In Retrieval-Augmented Generation (RAG), a system tries to answer your question by first retrieving information (from documents or a database) and then reading it using an AI model like ChatGPT. In regular flow, the system just uses your question as-is to search for information.

But here's the problem: Your original question might not be the best way to search for the right answer.

For example, imagine you asked, "What was the impact of the 2008 financial crisis on small businesses in rural areas?"

That's a complex question. If we use it directly to search, we might not get the best results. We might get the documents that talk about:

· "effects on small businesses"

· "rural economic impact"

· or just "2008 financial crisis"

So, what do we do?

We rewrite the question into simpler or more useful versions before searching.

For example:

· Impact of 2008 recession on rural small businesses

· 2008 financial crisis effect on local economies

· How small businesses were affected by the 2008 downturn

This rewritten query is more likely to match more appropriate documents.

The "Rewrite-Retrieve-Read" Framework

Here’s how the new method works step-by-step:

· Rewrite the question: make it better for searching

· Retrieve documents using the rewritten version

· Read the documents + original question to give an answer

In summary, Query rewriting means improving the user’s question before searching, so the AI can find better, more accurate information. It’s like turning a confusing search into a clearer one, and making it easier for the AI to help you.

Sample prompt template for this

References: https://arxiv.org/pdf/2305.14283

2.2 Step Back Prompting

It's a technique used to help Large Language Models (LLMs) like GPT-4 think bigger/better when solving difficult problems, especially ones that require reasoning or connecting multiple ideas.

It’s inspired by how humans solve tough problems. When we don’t know the answer right away, we step back, think about the bigger picture or the general rules, and then apply them to the specific question.

Core Idea: Two Steps

· Step Back -> Think Big

· Then Reason -> Apply to Solve

Step 1: Step Back

Instead of jumping into the answer right away, the model asks a bigger, higher-level question that helps it think better.

Examples 1:

Original question: How does pressure change with temperature and volume?

Step-back question: What physics principle explains this?

Answer: Ideal Gas Law (PV = nRT)

Example 2:

Original question: Where did Estella Leopold study in 1940s?

Step-back question: What is Estella Leopold’s education history?

This gives a broader timeline that can help answer the specific time.

Goal of this step is to help the model to gather the right facts or rules before trying to answer.

Step 2: Then Reason

Now that you have the high-level documents or principles, you use them to answer the original question.

In this phase, you essentially combine the original question with all the documents or knowledge retrieved during the Step-Back (Abstraction) phase and then ask the model to reason and answer based on that combined context.

Sample prompt template for this

You are an expert of world knowledge. I am going to ask you a question. Your response should be comprehensive and not contradicted with the following context if they are relevant. Otherwise, ignore them if they are not relevant.

{normal_context}
{step_back_context}

Original Question: {question}
Answer:

Credit goes to https://smith.langchain.com/hub/langchain-ai/stepback-answer?ref=blog.langchain.dev&__hstc=260096658.3a644e0328e94fa87e82130761cb66b8.1749469580827.1749469580827.1749469580827.1&__hssc=260096658.2.1749469580827&__hsfp=601531394&_gl=1*1gvdm1g*_gcl_au*MTU1MjgzNDUzMi4xNzQ5NDY5NTcx*_ga*MTU5ODE1NTc5MC4xNzQ5NDY5NTcz*_ga_47WX3HKKY2*czE3NDk0Njk1NzMkbzEkZzEkdDE3NDk0NzMwMDMkajYwJGwwJGgw

References: https://arxiv.org/pdf/2310.06117

2.3 Follow Up Questions

Let’s say you’re chatting with an assistant.

User: Where should I go in Japan in April?

Assistant: You can visit Kyoto for cherry blossoms, Tokyo for city life, and Hokkaido for snow festivals.

Follow-up Question: What’s the weather like?

Problem Without Query Transformation

If the system only takes “What’s the weather like?” as the search query, it has no idea you’re talking about Japan in April. So it might give you weather for another country or a random time.

Solution Using Query Transformation

Instead of embedding just the follow-up question, we use an LLM to rewrite it using the chat history.

Sample Prompt Template

Given the following conversation and a follow up question, rephrase the follow up \
question to be a standalone question.

Chat History:
{chat_history}
Follow Up Input: {question}
Standalone Question:

Credit goes to https://smith.langchain.com/hub/langchain-ai/weblangchain-search-query?ref=blog.langchain.dev&__hstc=260096658.3a644e0328e94fa87e82130761cb66b8.1749469580827.1749469580827.1749469580827.1&__hssc=260096658.2.1749469580827&__hsfp=601531394&_gl=1*oyvanp*_gcl_au*MTU1MjgzNDUzMi4xNzQ5NDY5NTcx*_ga*MTU5ODE1NTc5MC4xNzQ5NDY5NTcz*_ga_47WX3HKKY2*czE3NDk0Njk1NzMkbzEkZzEkdDE3NDk0NzQ2NzYkajYwJGwwJGgw

In summary,

· Original follow-up: “What’s the weather like?”

· After transformation: “What’s the weather like in Japan in April?”

· Benefit: More relevant results from search or retrieval

2.4 Multi-Query Retrieval Explained

Multi-Query Retrieval is a strategy where a Large Language Model (LLM) generates multiple related search queries from a single complex question. Instead of issuing just one search query, the system creates several focused queries that cover different aspects or sub-questions contained in the original question.

These multiple queries are then sent out in parallel to the search or retrieval system. The results from all these queries are collected together and used to provide a more comprehensive and accurate answer.

Why is this useful?

Some questions are complex and consist of multiple parts or layers. Searching with just one query might miss the important details or return incomplete results. By breaking the question down into multiple smaller queries, the system can cover more ground and bring back richer information.

For example, imagine a user asks: "What are the health benefits of green tea, and how does it compare to black tea in terms of caffeine content and antioxidant levels?"

This question has three distinct parts:

· What are the health benefits of green tea?

· How much caffeine does green tea have compared to black tea?

· How do antioxidant levels compare between green tea and black tea?

Without Multi-Query Retrieval

If you searched with the entire question as a single query, the results might only partially cover these topics or emphasize one part more than the others.

With Multi-Query Retrieval

An LLM can be prompted to generate multiple search queries such as:

· "Health benefits of green tea"

· "Caffeine content in green tea vs black tea"

· "Comparison of antioxidant levels in green tea and black tea"

These queries are then sent to the search system simultaneously. The results from all three are gathered, allowing the system to synthesize a complete, well-rounded answer that addresses each aspect thoroughly.

As you've seen, there are various techniques for performing query transformation, each with its own approach and benefits. While query transformation itself is not a new concept, what is new and powerful is the ability to use Large Language Models (LLMs) to carry it out.

The key difference between these methods often lies in the prompt engineering, that is, how we frame our instructions to the LLM. Writing these prompts is easier than you might think, if you can imagine the kind of question, you want the model to ask or change, you can probably write the prompt for it too.

This gives you a lot of new and exciting opportunities, now you have the power to come up with your own prompt ideas. So the real question is:

· What kinds of query transformations will you create?

· How will you use this ability in your own projects?

3. QueryTransformer interface in Langchain4j

QueryTransformer is an interface that takes a single query and returns one or more transformed queries. These new queries are often better suited for fetching relevant results from your knowledge base or embedding store.

Here’s the official definition in simple terms: QueryTransformer transforms the original user query into one or more enhanced queries to improve the quality of document retrieval.

3.1 Built-in QueryTransformer Implementations in LangChain4j

LangChain4j offers three powerful implementations of the QueryTransformer interface, each one has its own different needs in enhancing retrieval quality.

3.1.1 DefaultQueryTransformer: This is the most straightforward implementation. It simply returns the original query as-is, without making any changes or enhancements.

When to use it?

· You're just getting started and want a baseline retrieval setup.

· You don’t need any kind of transformation logic.

· Your queries are already well-structured and specific.

3.2.2. CompressingQueryTransformer

This transformer uses a ChatModel (like a language model) to compress or shorten the original query, often by removing irrelevant details or summarizing it. It also takes into account the chat memory (past interactions) to make the transformation more context-aware.

When to use it:

· The original query contains too much detail or redundant information.

· You want to refine the query to focus on what really matters.

· You’re working in a conversational context where prior interactions should influence the current query.

Prompt used by this transformer

"""
Generate {{n}} different versions of a provided user query. \
Each version should be worded differently, using synonyms or alternative sentence structures, \
but they should all retain the original meaning. \
These versions will be used to retrieve relevant documents. \
It is very important to provide each query version on a separate line, \
without enumerations, hyphens, or any additional formatting! \
User query: {{query}}
"""

CompressingQueryTransformerDemo.java

package com.sample.app;

import java.util.Collection;

import dev.langchain4j.data.message.AiMessage;
import dev.langchain4j.data.message.UserMessage;
import dev.langchain4j.memory.ChatMemory;
import dev.langchain4j.memory.chat.MessageWindowChatMemory;
import dev.langchain4j.model.ollama.OllamaChatModel;
import dev.langchain4j.rag.query.Metadata;
import dev.langchain4j.rag.query.Query;
import dev.langchain4j.rag.query.transformer.CompressingQueryTransformer;

/**
 * Demonstrates the use of {@link CompressingQueryTransformer} in Langchain4j
 * with an Ollama LLM to compress verbose user queries based on prior
 * conversation context.
 *
 * <p>
 * This example sets up:
 * <ul>
 * <li>A basic Ollama LLM chat model using llama3.2</li>
 * <li>A short chat history (ChatMemory)</li>
 * <li>A verbose user query with historical context</li>
 * <li>A query transformation step to compress the query</li>
 * </ul>
 *
 * <p>
 * Use case: Helps improve retrieval in RAG by compressing verbose or redundant
 * queries.
 */
public class CompressingQueryTransformerDemo {

    public static void main(String[] args) {

        // Step 1: Initialize the chat model (Ollama running locally)
        OllamaChatModel chatModel = OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2")
                .build();

        // Step 2: Initialize chat memory with a windowed message history (up to 5
        // messages)
        ChatMemory chatMemory = MessageWindowChatMemory.builder().id("12345").maxMessages(5).build();

        // Add previous user and AI messages to simulate conversation context
        chatMemory.add(UserMessage.from("I'm researching electric vehicles."));
        chatMemory.add(AiMessage.from("Sure! What aspect of EVs are you interested in?"));

        // Step 3: Create the verbose user query
        String userQuery = "Can you help me find battery performance comparisons for EVs released after 2022?";

        // Attach metadata including user message and chat history
        Metadata metadata = Metadata.from(UserMessage.from(userQuery), 1, chatMemory.messages());
        Query originalQuery = Query.from(userQuery, metadata);

        // Step 4: Initialize the query transformer with the chat model
        CompressingQueryTransformer transformer = new CompressingQueryTransformer(chatModel);

        // Step 5: Transform the query using prior chat context
        Collection<Query> compressedQueries = transformer.transform(originalQuery);

        // Step 6: Output the compressed query (may be 1 or more if multi-query support
        // is enabled)
        compressedQueries.forEach(compressedQuery -> System.out.println("Compressed Query: " + compressedQuery.text()));
    }
}

Output

Compressed Query: Battery performance comparisons for electric vehicles (EVs) released after 2022.

3.2.3 ExpandingQueryTransformer

In contrast to compression, this transformer expands the query. It uses a ChatModel to enrich the original query. For example, by adding related terms, inferred context, or clarifying details, making it more likely to retrieve relevant documents.

When to use it:

· The original query is too vague or under-specified.

· You want to improve by making the query broader and more inclusive.

· You're dealing with open-ended or exploratory queries.

Prompt used by this transformer

"""
Read and understand the conversation between the User and the AI. \
Then, analyze the new query from the User. \
Identify all relevant details, terms, and context from both the conversation and the new query. \
Reformulate this query into a clear, concise, and self-contained format suitable for information retrieval.

Conversation:
{{chatMemory}}

User query: {{query}}

It is very important that you provide only reformulated query and nothing else! \
Do not prepend a query with anything!
"""

Find the below working application.

ExpandingQueryTransformerDemo.java

package com.sample.app;

import java.util.Collection;

import dev.langchain4j.data.message.AiMessage;
import dev.langchain4j.data.message.UserMessage;
import dev.langchain4j.memory.ChatMemory;
import dev.langchain4j.memory.chat.MessageWindowChatMemory;
import dev.langchain4j.model.ollama.OllamaChatModel;
import dev.langchain4j.rag.query.Metadata;
import dev.langchain4j.rag.query.Query;
import dev.langchain4j.rag.query.transformer.ExpandingQueryTransformer;

public class ExpandingQueryTransformerDemo {

    public static final int EXPECTED_QUERIES = 5;

    public static void main(String[] args) {

        // Step 1: Initialize the chat model (Ollama running locally)
        OllamaChatModel chatModel = OllamaChatModel.builder()
                .baseUrl("http://localhost:11434")
                .modelName("llama3.2")
                .build();

        // Step 2: Initialize chat memory with a windowed message history (up to 5 messages)
        ChatMemory chatMemory = MessageWindowChatMemory.builder()
                .id("12345")
                .maxMessages(5)
                .build();

        // Add previous user and AI messages to simulate conversation context
        chatMemory.add(UserMessage.from("I'm researching electric vehicles."));
        chatMemory.add(AiMessage.from("Sure! What aspect of EVs are you interested in?"));

        // Step 3: Create the verbose user query
        String userQuery = "Can you help me find battery performance comparisons for EVs released after 2022?";

        // Attach metadata including user message and chat history
        Metadata metadata = Metadata.from(UserMessage.from(userQuery), 1, chatMemory.messages());
        Query originalQuery = Query.from(userQuery, metadata);

        // Step 4: Create ExpandingQueryTransformer with expected number of queries
        ExpandingQueryTransformer transformer = new ExpandingQueryTransformer(chatModel, EXPECTED_QUERIES);

        // Step 5: Transform the original query
        Collection<Query> expandedQueries = transformer.transform(originalQuery);

        // Step 6: Output the results
        System.out.println("Total Queries generated: " + expandedQueries.size());
        for (Query query : expandedQueries) {
            System.out.println(query.text());
        }
    }
}

Output

Total Queries generated: 5
I can help you find information on battery performance comparisons for electric vehicles released in the last few years
Can you provide more details about the type of comparison you're looking for regarding electric vehicle batteries?
Is there a particular aspect of battery performance you'd like to know more about, such as charging speed or range?
Can you give me an idea of what kind of information would be helpful to you when it comes to comparing different electric vehicle battery types?
Would you prefer to see comparisons between specific brands or models of electric vehicles?

Here’s how you can use ExpandingQueryTransformer inside a RAG pipeline by attaching it to a RetrievalAugmentor, which is then plugged into the ChatAssistant:

EmbeddingModel embeddingModel = new BgeSmallEnV15QuantizedEmbeddingModel();

ContentRetriever contentRetriever =
    EmbeddingStoreContentRetriever.builder()
        .embeddingStore(embeddingStore)
        .embeddingModel(embeddingModel)
        .maxResults(3)
        .minScore(0.7)
        .build();

ExpandingQueryTransformer expandingQueryTransformer =
    new ExpandingQueryTransformer(chatModel, 3);

RetrievalAugmentor retrievalAugmentor =
    DefaultRetrievalAugmentor.builder()
        .queryTransformer(expandingQueryTransformer)
        .contentRetriever(contentRetriever)
        .build();

ChatAssistant chatAssistant =
    AiServices.builder(ChatAssistant.class)
        .chatModel(chatModel)
        .retrievalAugmentor(retrievalAugmentor)
        .chatMemory(MessageWindowChatMemory.withMaxMessages(10))
        .build();

How It Works?

· ExpandingQueryTransformer generates multiple reworded versions of the user’s query using the LLM.

· These versions are passed to the ContentRetriever, improving document recall.

· The RetrievalAugmentor ties this transformation + retrieval together.

· ChatAssistant uses this full pipeline for better RAG-based answering.

Find the below working application.

QueryTransformerDemo.java

package com.sample.app.transformer;

import dev.langchain4j.data.document.Document;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.memory.chat.MessageWindowChatMemory;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.model.embedding.onnx.bgesmallenv15q.BgeSmallEnV15QuantizedEmbeddingModel;
import dev.langchain4j.model.ollama.OllamaChatModel;
import dev.langchain4j.rag.DefaultRetrievalAugmentor;
import dev.langchain4j.rag.RetrievalAugmentor;
import dev.langchain4j.rag.content.retriever.ContentRetriever;
import dev.langchain4j.rag.content.retriever.EmbeddingStoreContentRetriever;
import dev.langchain4j.rag.query.transformer.ExpandingQueryTransformer;
import dev.langchain4j.service.AiServices;
import dev.langchain4j.store.embedding.EmbeddingStoreIngestor;
import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;
import java.util.ArrayList;
import java.util.List;

public class QueryTransformerDemo {

  public static interface ChatAssistant {
    String chat(String userMessage);
  }

  private static List<String> indiaDocs =
      List.of(
          """
                    India, officially the Republic of India, is a country in South Asia. It is the seventh-largest country by area,
                    the most populous country as of 2023, and the most populous democracy in the world. Bounded by the Himalayas in
                    the north and surrounded by the Indian Ocean, Arabian Sea, and Bay of Bengal, India has a rich cultural and
                    historical heritage.
                    """,
          """
                    The Indian economy is the fifth-largest in the world by nominal GDP and the third-largest by purchasing power parity.
                    It is classified as a newly industrialized country and one of the world's fastest-growing major economies. Major
                    industries include IT services, textiles, pharmaceuticals, and agriculture.
                    """,
          """
                    India's political system is a federal parliamentary democratic republic. The President of India is the head of state,
                    while the Prime Minister is the head of government. The Indian Parliament consists of the Lok Sabha (House of the People)
                    and the Rajya Sabha (Council of States).
                    """,
          """
                    Indian culture is renowned for its diversity and depth. It includes a wide variety of languages, religions, music,
                    dance forms, cuisine, and traditions. Major religions born in India include Hinduism, Buddhism, Jainism, and Sikhism.
                    Festivals such as Diwali, Eid, Christmas, and Pongal are celebrated across the country.
                    """,
          """
                    India is home to 40 UNESCO World Heritage Sites, including the Taj Mahal, Qutub Minar, and the Western Ghats. The country
                    offers a wide variety of landscapes ranging from deserts in Rajasthan, the Himalayan mountain range, tropical rainforests
                    in the northeast, to coastal plains and fertile river valleys.
                    """,
          """
                    The Indian education system has seen significant progress, with institutes like the IITs, IIMs, and AIIMS gaining global
                    recognition. India also has a growing startup ecosystem, particularly in tech hubs such as Bengaluru, Hyderabad, and Pune.
                    """,
          """
                    India has a vibrant film industry, notably Bollywood, which produces the largest number of films in the world. Indian music
                    and dance forms such as Bharatanatyam, Kathak, Carnatic music, and Bollywood songs are famous worldwide.
                    """,
          """
                    Indian cuisine is known for its rich flavors and diverse ingredients. Each region has its own distinct dishes –
                    from butter chicken and biryani in the north, to dosa and sambar in the south, to seafood delicacies in the coastal regions.
                    Spices like turmeric, cumin, coriander, and cardamom are integral to Indian cooking.
                    """,
          """
                    The Indian space program, led by ISRO, has made remarkable achievements including the Chandrayaan lunar missions and the
                    Mars Orbiter Mission (Mangalyaan), which made India the first Asian nation to reach Mars orbit and the first in the world
                    to do so in its maiden attempt.
                    """,
          """
                    Indian sports are dominated by cricket, but other sports like hockey, badminton, wrestling, and kabaddi are also popular.
                    India has produced world-class athletes including Sachin Tendulkar, P.V. Sindhu, Neeraj Chopra, and Mary Kom. The country
                    has hosted major sporting events like the Commonwealth Games and the Cricket World Cup.
                    """);

  private static List<Document> documents = new ArrayList<>();

  private static void prepareDocuments() {
    for (String doc : indiaDocs) {
      documents.add(Document.from(doc));
    }
  }

  public static void main(String[] args) {
    prepareDocuments();

    OllamaChatModel chatModel =
        OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

    InMemoryEmbeddingStore<TextSegment> embeddingStore = new InMemoryEmbeddingStore<>();
    EmbeddingStoreIngestor.ingest(documents, embeddingStore);

    ExpandingQueryTransformer expandingQueryTransformer =
        new ExpandingQueryTransformer(chatModel, 3);
    EmbeddingModel embeddingModel = new BgeSmallEnV15QuantizedEmbeddingModel();

    ContentRetriever contentRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(embeddingStore)
            .embeddingModel(embeddingModel) // same chat model for embeddings
            .maxResults(3)
            .minScore(0.7)
            .build();

    RetrievalAugmentor retrievalAugmentor =
        DefaultRetrievalAugmentor.builder()
            .queryTransformer(expandingQueryTransformer)
            .contentRetriever(contentRetriever)
            .build();

    String prompt =
        """
                Hey, I was reading about how countries manage their political systems and got curious — can you tell me how India runs its government, who’s in charge, and what kind of structure they follow politically?

                """;

    ChatAssistant chatAssistant =
        AiServices.builder(ChatAssistant.class)
            .chatModel(chatModel)
            .retrievalAugmentor(retrievalAugmentor)
            .chatMemory(MessageWindowChatMemory.withMaxMessages(10))
            .build();

    String response = chatAssistant.chat(prompt);
    System.out.println(response);
  }
}

Output

India's government structure is a federal parliamentary democratic republic. Here's an overview of who's in charge and the kind of structure they follow:

**Head of State:** The President of India serves as the head of state. They are elected by an electoral college consisting of the members of both houses of Parliament (Lok Sabha and Rajya Sabha) and the governors of all states.

**Head of Government:** The Prime Minister is the head of government, but they do not serve as a separate executive branch. Instead, they are responsible for advising the President on matters of national importance and forming a cabinet to implement policies.

**Government Structure:** India's federal system divides power between the central government and the states (called "units"). The central government has control over areas such as defense, foreign policy, and interstate relations, while the states have autonomy over local issues like healthcare, education, and law enforcement.

**Legislative Branch:** The Indian Parliament is bicameral, consisting of two houses:

1. **Lok Sabha (House of the People):** This is the lower house of Parliament, comprising 543 elected members who serve a term of five years.
2. **Rajya Sabha (Council of States):** This is the upper house of Parliament, with 245 members who are either directly or indirectly elected by the state legislatures.

The Lok Sabha has the power to pass laws and elect the President, while the Rajya Sabha provides a check on the Lok Sabha's decisions and can also initiate legislation.

References

https://blog.langchain.dev/query-transformations/

Previous Next Home

Exploring Apache Druid Native Queries

2025-09-16T08:10:00.000-07:00

Apache Druid is a real-time analytics database designed for fast slice-and-dice analytics on large datasets. While SQL is a convenient way to query data in Druid, native queries provide greater flexibility, control, and performance tuning.

In this blog post, we’ll explore how to use Druid’s scan native query to extract raw event-level data from a sales_data datasource. We'll walk through examples using real-world sales data.

Sample Data: sales_data

I ingested the following CSV into Druid with the name sales_data.

timestamp,product,city,total_sales
2025-04-01T10:00:00Z,Laptop,Delhi,300
2025-04-01T10:00:00Z,Laptop,Delhi,200
2025-04-01T11:00:00Z,Tablet,Mumbai,150
2025-04-01T11:00:00Z,Tablet,Mumbai,50
2025-04-01T12:00:00Z,Mobile,Bengaluru,200
2025-04-01T13:00:00Z,Laptop,Hyderabad,250
2025-04-01T14:00:00Z,Tablet,Chennai,180
2025-04-01T15:00:00Z,Mobile,Pune,220
2025-04-01T15:00:00Z,Mobile,Pune,80

Each row represents a product sale at a specific timestamp, city, and quantity.

1. Scan Query in Druid

The scan query in Druid is used to fetch raw data from your datasource. Unlike groupBy or timeseries queries that aggregate data, scan queries return unprocessed rows in a customizable format.

Here’s a basic scan query that fetches all data from the sales_data datasource between two timestamps.

{
  "queryType": "scan",
  "dataSource": "sales_data",
  "resultFormat": "compactedList",
  "columns": ["__time", "product", "city", "total_sales"],
  "intervals": ["2025-04-01T10:00:00Z/2025-04-01T16:00:00Z"],
  "batchSize": 35575,
  "limit": 100
}

Following table summarizes the native query payload.

Field	Purpose
queryType	Specifies the query type. In this case, scan
dataSource	The name of the Druid datasource (your table)
resultFormat	Format of the result: compactedList, list, or valueVector
columns	List of columns to include in the response
intervals	Time interval filter, ISO 8601 format
batchSize	Number of rows fetched internally per scan
limit	Maximum number of rows returned in the result

Output json looks like below.

Output

[
  {
    "segmentId": "sales_data_2025-04-01T00:00:00.000Z_2025-04-02T00:00:00.000Z_2025-04-18T08:18:17.761Z",
    "columns": [
      "__time",
      "product",
      "city",
      "total_sales"
    ],
    "events": [
      [
        1743501600000,
        "Laptop",
        "Delhi",
        300
      ],
      [
        1743501600000,
        "Laptop",
        "Delhi",
        200
      ],
      [
        1743505200000,
        "Tablet",
        "Mumbai",
        150
      ],
      [
        1743505200000,
        "Tablet",
        "Mumbai",
        50
      ],
      [
        1743508800000,
        "Mobile",
        "Bengaluru",
        200
      ],
      [
        1743512400000,
        "Laptop",
        "Hyderabad",
        250
      ],
      [
        1743516000000,
        "Tablet",
        "Chennai",
        180
      ],
      [
        1743519600000,
        "Mobile",
        "Pune",
        220
      ],
      [
        1743519600000,
        "Mobile",
        "Pune",
        80
      ]
    ],
    "rowSignature": [
      {
        "name": "__time",
        "type": "LONG"
      },
      {
        "name": "product",
        "type": "STRING"
      },
      {
        "name": "city",
        "type": "STRING"
      },
      {
        "name": "total_sales",
        "type": "DOUBLE"
      }
    ]
  }
]

Example 1: Retrieve all rows where sales happened in a specific hour

{
  "queryType": "scan",
  "dataSource": "sales_data",
  "resultFormat": "compactedList",
  "columns": ["__time", "product", "city", "total_sales"],
  "intervals": ["2025-04-01T11:00:00Z/2025-04-01T12:00:00Z"]
}

Example 2: Filter specific columns only (e.g., product, city and total_sales)

{
  "queryType": "scan",
  "dataSource": "sales_data",
  "resultFormat": "list",
  "columns": ["product", "city", "total_sales"],
  "intervals": ["2025-04-01T11:00:00Z/2025-04-01T12:00:00Z"]
}

Example 3: Limit the number of records returned.

{
  "queryType": "scan",
  "dataSource": "sales_data",
  "resultFormat": "compactedList",
  "columns": ["__time", "product", "total_sales"],
  "intervals": ["2025-04-01T10:00:00Z/2025-04-01T16:00:00Z"],
  "limit": 3
}

Example 4: Using resultFormat as list.

{
  "queryType": "scan",
  "dataSource": "sales_data",
  "resultFormat": "list",
  "columns": ["__time", "city", "total_sales"],
  "intervals": ["2025-04-01T13:00:00Z/2025-04-01T16:00:00Z"]
}

2. Druid GroupBy Native Queries with Filters

While scan queries in Druid return raw data, groupBy queries are used to aggregate data much like SQL GROUP BY. This is useful when you want to answer questions like:

· What is the total sales per product?

· What is the sales count by city per hour?

· How much did each city sell after 2 PM?

Example 1: Total Sales by Product

{
  "queryType": "groupBy",
  "dataSource": "sales_data",
  "intervals": ["2025-04-01T10:00:00Z/2025-04-01T16:00:00Z"],
  "granularity": "all",
  "dimensions": ["product"],
  "aggregations": [
    { "type": "longSum", "name": "sales_by_product", "fieldName": "total_sales" }
  ]
}

Example 2: Sales per City after 1 PM

{
  "queryType": "groupBy",
  "dataSource": "sales_data",
  "intervals": ["2025-04-01T13:00:00Z/2025-04-01T16:00:00Z"],
  "granularity": "all",
  "dimensions": ["city"],
  "aggregations": [
    { "type": "longSum", "name": "sales_per_city", "fieldName": "total_sales" }
  ]
}

Example 3: Add a WHERE Clause (Filter by product = 'Laptop')

{
  "queryType": "groupBy",
  "dataSource": "sales_data",
  "intervals": ["2025-04-01T10:00:00Z/2025-04-01T16:00:00Z"],
  "granularity": "all",
  "filter": {
    "type": "selector",
    "dimension": "product",
    "value": "Laptop"
  },
  "dimensions": ["city"],
  "aggregations": [
    { "type": "longSum", "name": "total_sales", "fieldName": "total_sales" }
  ]
}

Previous Next Home

How to load data from ORC file into Druid?

2025-09-12T23:57:00.000-07:00

What is ORC file format?

ORC stands for Optimized Row Columnar. It's a highly efficient columnar storage format used primarily in the Hadoop ecosystem, especially with Apache Hive and Apache Spark.

Enable ORC format in Apache Druid

ORC support is not enabled by default in Apache Druid. To enable ORC format support and load ORC files into Druid, you need to manually add the ORC extension to the common.runtime.properties file and restart the Druid instance.

Step-by-Step: Enable ORC Format in Apache Druid

1. Locate the common.runtime.properties file

If you're using the Druid single-server micro-quickstart, the file path is typically:

{DRUID_HOME}/conf/druid/single-server/micro-quickstart/_common/common.runtime.properties

f you're using other modes like small, nano-quickstart, large, medium, or xlarge, locate the appropriate folder under:

{DRUID_HOME}/conf/druid/single-server/

Then navigate into the respective _common/common.runtime.properties.

2. Edit the druid.extensions.loadList property

Find the following line:

druid.extensions.loadList=["druid-hdfs-storage", "druid-kafka-indexing-service", "druid-datasketches", "druid-multi-stage-query", "druid-parquet-extensions"]

You need to add "druid-orc-extensions" to the list like this:

druid.extensions.loadList=["druid-hdfs-storage", "druid-kafka-indexing-service", "druid-datasketches", "druid-multi-stage-query", "druid-parquet-extensions", "druid-orc-extensions"]

3. Restart your Druid instance

After saving the file, restart Druid using:

bin/start-micro-quickstart

This ensures Druid loads the ORC extension during startup.

Preparing orc file

sales_data.csv

timestamp,product,city,sales
2025-04-01T10:00:00Z,Laptop,Delhi,300
2025-04-01T10:00:00Z,Laptop,Delhi,200
2025-04-01T11:00:00Z,Tablet,Mumbai,150
2025-04-01T11:00:00Z,Tablet,Mumbai,50
2025-04-01T12:00:00Z,Mobile,Bengaluru,200
2025-04-01T13:00:00Z,Laptop,Hyderabad,250
2025-04-01T14:00:00Z,Tablet,Chennai,180
2025-04-01T15:00:00Z,Mobile,Pune,220
2025-04-01T15:00:00Z,Mobile,Pune,80

Navigate to the url https://dataconverter.io/convert/csv-to-orc and convert the csv to orc file, and save it as ‘sales_data.orc’.

Loading orc file

Once Druid is running, go to the Druid Web Console http://localhost:8888.

Load data -> Batch-SQL multi-stage-query

Select Input type as ‘Local disk’.

Set the Base directory to the parent directory of orc file.

Give the file name as sales_data.orc and click on Connect data button.

You will be taken to Load data / Parse page.

Click on Next button. You will be taken to ‘Load data / Configure schema’ tab

Click on SQL tab.

Give the table name as ‘sales_data_orc’.

Click on ‘Start loading data’ button.

Upon Ingestion is successful, you will see successful message like below.

Navigate to Query tab by clicking on ‘Query: sales_data_orc’ button.

Type following sql statement, and run the query.

SELECT * FROM "sales_data_orc"

That’s it you are done…, Happy learning…

Previous Next Home

How to Ingest Data from an HTTPS Endpoint into Apache Druid?

2025-09-12T23:53:00.000-07:00

Are you working with real-time or batch data that's accessible over HTTPS and wondering how to bring it into Apache Druid for fast analytics? You're in the right place!

Apache Druid is a high-performance analytics database designed for rapid, ad-hoc queries on large datasets. While it's commonly used with Kafka, local files, or cloud storage systems like S3 or GCS, you can also pull data from an HTTPS endpoint such as a public API or your internal service and ingest it into Druid.

In this blog post, I’ll Walk through how to ingest data from an HTTPS source using Druid’s HTTPs input type.

Open the url ‘http://localhost:8888’ in browser, you will be taken Druid console.

Click on Load data -> Batch – SQL (multi-stage-query) button.

On the “Load Data / Select Input Type” page in Druid, choose HTTP(s) as your input type. This option allows you to provide one or more HTTPS URLs, separated by commas. For example, I used the following URL to ingest data into the Druid ecosystem:

https://raw.githubusercontent.com/MainakRepositor/Datasets/refs/heads/master/Weather%20Data/pressure.csv

Click on Connect data button.

You will be taken to ‘Load data / Parse’ page.

Click on Next button.

You will be taken to ‘Load data / Configure schema’ page. You can observe __time column is generated from the datetime values of the dataset.

Click on ‘Start loading data’ button.

Upon ingesting successful, navigate to Query tab and execute following sql statement.

SELECT * FROM "pressure"

That’s it, you are done….

Previous Next Home

Customizing Content Retrieval in RAG with EmbeddingStoreContentRetriever for Smarter Responses

2025-09-12T23:48:00.000-07:00

Retrieval-Augmented Generation (RAG) models enhance AI's capabilities by feeding responses from external knowledge sources. However, effective RAG systems depend heavily on how documents are retrieved. In this post, I demonstrate how to customize the retrieval process using EmbeddingStoreContentRetriever by fine-tuning parameters like maxResults and minScore, and seamlessly integrating it with a chat assistant powered by a language model.

Why to customize the Retriever?

When fetching documents for a query, we may want to

· Limit the number of returned results (maxResults)

· Filter out irrelevant results using a minimum similarity score (minScore)

· Using custom embedding models

This allows the system to return only the most relevant knowledge for response generation, reducing noise and improving trustworthiness.

Example

ContentRetriever contentRetriever =
    EmbeddingStoreContentRetriever.builder()
        .embeddingStore(embeddingStore) // Your store (e.g., in-memory or persistent)
        .embeddingModel(embeddingModel) // BGE, OpenAI, etc.
        .maxResults(5)                  // Only return top 5 similar docs
        .minScore(0.75)                 // Filter out documents below this similarity score
        .build();

ChatAssistant assistant =
    AiServices.builder(ChatAssistant.class)
        .chatModel(chatModel)          // Your LLM (OpenAI, HuggingFace, etc.)
        .contentRetriever(contentRetriever)  // Inject customized retriever
        .build();

Find the below working Application.

ChatAssistant.java

package com.sample.app.assistants;

public interface ChatAssistant {
  String chat(String userMessage);
}

InMemoryRagWithContentRetriever.java

package com.sample.app;

import com.sample.app.assistants.ChatAssistant;
import dev.langchain4j.data.document.Document;
import dev.langchain4j.data.document.loader.FileSystemDocumentLoader;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.model.embedding.onnx.bgesmallenv15q.BgeSmallEnV15QuantizedEmbeddingModel;
import dev.langchain4j.model.ollama.OllamaChatModel;
import dev.langchain4j.rag.content.retriever.ContentRetriever;
import dev.langchain4j.rag.content.retriever.EmbeddingStoreContentRetriever;
import dev.langchain4j.service.AiServices;
import dev.langchain4j.store.embedding.EmbeddingStoreIngestor;
import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;
import java.util.ArrayList;
import java.util.List;

public class InMemoryRagWithContentRetriever {

  public static void main(String[] args) {

    // Initialize local embedding model
    EmbeddingModel embeddingModel = new BgeSmallEnV15QuantizedEmbeddingModel();

    // Load all the Documents
    String resourcesFolderPath = "/Users/Shared/llm_docs";
    System.out.println("Resources Folder Path is " + resourcesFolderPath);
    List<Document> documents = FileSystemDocumentLoader.loadDocuments(resourcesFolderPath);

    InMemoryEmbeddingStore<TextSegment> embeddingStore = new InMemoryEmbeddingStore<>();
    EmbeddingStoreIngestor.ingest(documents, embeddingStore);

    OllamaChatModel chatModel =
        OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

    ContentRetriever contentRetriever =
        EmbeddingStoreContentRetriever.builder()
            .embeddingStore(embeddingStore)
            .embeddingModel(embeddingModel)
            .maxResults(5)
            .minScore(0.75)
            .build();

    ChatAssistant assistant =
        AiServices.builder(ChatAssistant.class)
            .chatModel(chatModel)
            .contentRetriever(contentRetriever)
            .build();

    List<String> questionsToAsk = new ArrayList<>();
    questionsToAsk.add("What is the tag line of ChronoCore Industries?");

    long time1 = System.currentTimeMillis();
    for (String question : questionsToAsk) {
      String answer = assistant.chat(question);
      System.out.println("----------------------------------------------------");
      System.out.println("Q: " + question);
      System.out.println("A : " + answer);
      System.out.println("----------------------------------------------------\n");
    }
    long time2 = System.currentTimeMillis();

    System.out.println("Total time taken is " + (time2 - time1));
  }
}

Output

Resources Folder Path is /Users/Shared/llm_docs
----------------------------------------------------
Q: What is the tag line of ChronoCore Industries?
A : I couldn't find the tagline explicitly stated in the provided information. However, I can provide you with a possible tagline based on the company's mission and values:

"Preserving Yesterday. Shaping Tomorrow."

This tagline is consistent with the company's mission to "unlock the fabric of time itself — responsibly, ethically, and with profound respect for the continuum that binds reality."
----------------------------------------------------

Total time taken is 4012

Previous Next Home

Document Ingestion with LangChain4j using EmbeddingStoreIngestor

2025-09-12T23:44:00.000-07:00

In this blog post, we'll explore how LangChain4j’s EmbeddingStoreIngestor works under the hood and how you can configure it to transform, split, and embed documents into an EmbeddingStore. If you're building LLM-based applications and need to semantically search or analyze documents, mastering this ingestion pipeline is a must.

The EmbeddingStoreIngestor is a utility that

· Takes in Documents

· Transforms and optionally splits them

· Embeds them using an EmbeddingModel

· Stores them in an EmbeddingStore

This modular pipeline enables efficient vectorization and storage of text data, laying the foundation for semantic search, recommendations, and knowledge retrieval in your applications.

Example

    EmbeddingStoreIngestor ingestor =
        EmbeddingStoreIngestor.builder()
            .embeddingModel(new BgeSmallEnV15QuantizedEmbeddingModel())
            .documentTransformer(
                document -> Document.from(document.text().toUpperCase(), document.metadata()))
            .textSegmentTransformer(
                segment -> TextSegment.from(segment.text().toUpperCase(), segment.metadata()))
            .embeddingStore(embeddingStore)
            .build();

Find the below working application.

EmbeddingStoreIngestorDemo.java

package com.sample.app.embeddings;

import dev.langchain4j.data.document.Document;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.model.embedding.onnx.bgesmallenv15q.BgeSmallEnV15QuantizedEmbeddingModel;
import dev.langchain4j.model.ollama.OllamaChatModel;
import dev.langchain4j.model.output.TokenUsage;
import dev.langchain4j.rag.content.retriever.EmbeddingStoreContentRetriever;
import dev.langchain4j.service.AiServices;
import dev.langchain4j.store.embedding.EmbeddingStoreIngestor;
import dev.langchain4j.store.embedding.IngestionResult;
import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;
import java.util.ArrayList;
import java.util.List;

public class EmbeddingStoreIngestorDemo {

  public interface ChatAssistant {
    String chat(String userMessage);
  }

  public static void main(String[] args) {
    List<Document> documents = new ArrayList<>();
    Document doc1 =
        Document.from("The stock market surged as tech companies reported strong earnings.");
    Document doc2 =
        Document.from("Tech giants like Apple and Amazon saw record profits this quarter.");
    Document doc3 = Document.from("Heavy rain caused flooding in coastal towns yesterday.");
    documents.add(doc1);
    documents.add(doc2);
    documents.add(doc3);

    InMemoryEmbeddingStore<TextSegment> embeddingStore = new InMemoryEmbeddingStore<>();

    EmbeddingStoreIngestor ingestor =
        EmbeddingStoreIngestor.builder()
            .embeddingModel(new BgeSmallEnV15QuantizedEmbeddingModel())
            .documentTransformer(
                document -> Document.from(document.text().toUpperCase(), document.metadata()))
            .textSegmentTransformer(
                segment -> TextSegment.from(segment.text().toUpperCase(), segment.metadata()))
            .embeddingStore(embeddingStore)
            .build();

    IngestionResult ingestionResult = ingestor.ingest(documents);
    TokenUsage tokenUsage = ingestionResult.tokenUsage();
    System.out.println("Input Token Count : " + tokenUsage.inputTokenCount());
    System.out.println("Output Token Count : " + tokenUsage.outputTokenCount());
    System.out.println("Total Token Count : " + tokenUsage.totalTokenCount());

    OllamaChatModel chatModel =
        OllamaChatModel.builder().baseUrl("http://localhost:11434").modelName("llama3.2").build();

    ChatAssistant assistant =
        AiServices.builder(ChatAssistant.class)
            .chatModel(chatModel)
            .contentRetriever(EmbeddingStoreContentRetriever.from(embeddingStore))
            .build();

    List<String> questionsToAsk = new ArrayList<>();
    questionsToAsk.add("Is tech companies stocks surged?");

    long time1 = System.currentTimeMillis();
    for (String question : questionsToAsk) {
      String answer = assistant.chat(question);
      System.out.println("----------------------------------------------------");
      System.out.println("Q: " + question);
      System.out.println("A : " + answer);
      System.out.println("----------------------------------------------------\n");
    }
    long time2 = System.currentTimeMillis();

    System.out.println("Total time taken is " + (time2 - time1));
  }
}

Output

Input Token Count : 32
Output Token Count : null
Total Token Count : 32
----------------------------------------------------
Q: Is tech companies stocks surged?
A : Yes, tech companies' stocks surged due to the strong earnings reported by these companies.
----------------------------------------------------

Total time taken is 1984

Previous Next Home

Understanding EmbeddingSearchResult and EmbeddingMatch in LangChain4j

2025-09-11T06:15:00.000-07:00

The EmbeddingSearchResult class represents the result of a semantic search in an EmbeddingStore. It holds a list of the most relevant matches to your query, each represented by an EmbeddingMatch.

Example

EmbeddingSearchResult<TextSegment> result = embeddingStore.search(request);

What is EmbeddingMatch?

Each result inside EmbeddingSearchResult is an instance of EmbeddingMatch.

public class EmbeddingMatch<Embedded> {
    private final Double score;
    private final String embeddingId;
    private final Embedding embedding;
    private final Embedded embedded;
}

Here:

· score: A relevance score (derived from cosine similarity). Higher means more relevant.

· embeddingId: The unique identifier of the embedding in the store.

· embedding: The vector representation that matched the query.

· embedded: The original content that was embedded (like a paragraph, chunk, or TextSegment).

Example

EmbeddingSearchRequest request = EmbeddingSearchRequest.builder()
    .embedding(myQueryEmbedding)
    .maxResults(2)
    .minScore(0.5)
    .build();

EmbeddingSearchResult<TextSegment> result = embeddingStore.search(request);

for (EmbeddingMatch<TextSegment> match : result.matches()) {
    System.out.println("Score: " + match.score());
    System.out.println("Original content: " + match.embedded().text());
}

Find the below working Application.

EmbeddingSearchWithFilters.java

package com.sample.app.embeddings;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import dev.langchain4j.data.document.Metadata;
import dev.langchain4j.data.embedding.Embedding;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.model.embedding.onnx.bgesmallenv15q.BgeSmallEnV15QuantizedEmbeddingModel;
import dev.langchain4j.store.embedding.EmbeddingSearchRequest;
import dev.langchain4j.store.embedding.EmbeddingSearchResult;
import dev.langchain4j.store.embedding.EmbeddingStore;
import dev.langchain4j.store.embedding.filter.Filter;
import dev.langchain4j.store.embedding.filter.comparison.IsEqualTo;
import dev.langchain4j.store.embedding.filter.comparison.IsGreaterThanOrEqualTo;
import dev.langchain4j.store.embedding.filter.comparison.IsIn;
import dev.langchain4j.store.embedding.filter.logical.And;
import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;

public class EmbeddingSearchWithFilters {

    public static void main(String[] args) {

        // Step 1: Initialize embedding model
        EmbeddingModel model = new BgeSmallEnV15QuantizedEmbeddingModel();

        // Step 2: Prepare text data with metadata
        String sentence1 = "The stock market surged as tech companies reported strong earnings.";
        Map<String, Object> metaMap1 = new HashMap<>();
        metaMap1.put("category", "finance");
        metaMap1.put("source", "news");
        metaMap1.put("relevance", 9);
        Metadata meta1 = Metadata.from(metaMap1);

        String sentence2 = "Tech giants like Apple and Amazon saw record profits this quarter.";
        Map<String, Object> metaMap2 = new HashMap<>();
        metaMap2.put("category", "technology");
        metaMap2.put("source", "news");
        metaMap2.put("relevance", 10);
        Metadata meta2 = Metadata.from(metaMap2);

        String sentence3 = "Heavy rain caused flooding in coastal towns yesterday.";
        Map<String, Object> metaMap3 = new HashMap<>();
        metaMap3.put("category", "weather");
        metaMap3.put("source", "alert");
        metaMap3.put("relevance", 5);
        Metadata meta3 = Metadata.from(metaMap3);

        try {
            // Step 3: Generate embeddings
            Embedding embedding1 = model.embed(sentence1).content();
            Embedding embedding2 = model.embed(sentence2).content();
            Embedding embedding3 = model.embed(sentence3).content();

            // Step 4: Initialize in-memory embedding store
            EmbeddingStore<TextSegment> embeddingStore = new InMemoryEmbeddingStore<>();

            // Step 5: Add embeddings to the store with metadata
            embeddingStore.add(embedding1, TextSegment.from(sentence1, meta1));
            embeddingStore.add(embedding2, TextSegment.from(sentence2, meta2));
            embeddingStore.add(embedding3, TextSegment.from(sentence3, meta3));

            System.out.println("Embeddings with metadata added to the store successfully.");

            // Step 6: Prepare a query embedding
            Embedding queryEmbedding = model.embed("Which tech companies had profits?").content();

            Filter filter = new And(
                    new And(new IsEqualTo("source", "news"), new IsIn("category", List.of("technology", "finance"))),
                    new IsGreaterThanOrEqualTo("relevance", Integer.valueOf(9)));

            // Step 8: Build search request with filter
            EmbeddingSearchRequest request = EmbeddingSearchRequest.builder().queryEmbedding(queryEmbedding)
                    .maxResults(2).minScore(0.5).filter(filter).build();

            // Step 9: Perform search
            EmbeddingSearchResult<TextSegment> result = embeddingStore.search(request);

            // Step 10: Display results
            System.out.println("\nFiltered & Ranked Results:");
            result.matches().forEach(match -> {
                TextSegment segment = match.embedded();
                System.out.printf("Score: %.3f | Text: %s | Metadata: %s%n", match.score(), segment.text(),
                        segment.metadata());
            });

        } catch (Exception e) {
            System.err.println("An error occurred: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

Output

Embeddings with metadata added to the store successfully.

Filtered & Ranked Results:
Score: 0.883 | Text: Tech giants like Apple and Amazon saw record profits this quarter. | Metadata: Metadata { metadata = {category=technology, relevance=10, source=news} }
Score: 0.882 | Text: The stock market surged as tech companies reported strong earnings. | Metadata: Metadata { metadata = {category=finance, relevance=9, source=news} }

Previous Next Home

Loading Data from Kafka into Apache Druid

2025-09-11T06:04:00.000-07:00

In the world of big data, real-time analytics is becoming very important. Apache Druid is a high-performance analytics database designed for fast slice-and-dice analytics on large datasets. Apache Kafka is a widely used distributed event streaming platform.

In this post, I will walk you through the concepts, setup, and step-by-step instructions to get your Kafka data flowing into Druid for real-time querying.

1. Setup Kafka

Apache Kafka is a powerful tool for building real-time data pipelines. It works like a high-speed messaging system, collecting data from various sources and passing it along to systems that process or analyze it.

Think of Kafka as a post office for data:

· Producers drop off messages (e.g., a user clicked a button).

· Kafka stores and delivers these messages.

· Consumers pick up the messages and do something with them, like store them in a database, trigger alerts, or update dashboards.

Kafka is used by companies like LinkedIn, Netflix, and Uber to handle millions of messages per second, making it a perfect fit for real-time data streaming.

Follow step-by-step procedure to setup Kafka.

Step 1: Go to the following link and setup Kafka.

https://kafka.apache.org/downloads

At the time of writing this post, 4.0.0 is the latest version.

Step 2: Generate a Cluster UUID by executing following statement.

Navigate to Kafka home directory and execute below command.

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

$KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
$
$echo $KAFKA_CLUSTER_ID
ImA96y2CTJGN-HzF3GSFWg

Step 3: Format log directories.

bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties

$bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties
Formatting dynamic metadata voter directory /tmp/kraft-combined-logs with metadata.version 4.0-IV3.

Step 4: Start the Kafka Server

bin/kafka-server-start.sh config/server.properties

Once the Kafka server has successfully launched, you will have a basic Kafka environment running and ready to use.

Step 5: Create a Kafka Topic for Order Events

Let’s create a Kafka topic to hold incoming order events from our e-commerce platform. These could be events like a customer placing an order, the order being shipped, or the order getting canceled.

Run the following command to create a topic named ecommerce-orders:

bin/kafka-topics.sh --create --topic ecommerce-orders --bootstrap-server localhost:9092

$ bin/kafka-topics.sh --create --topic ecommerce-orders --bootstrap-server localhost:9092
Created topic ecommerce-orders.

Step 6: Produce Some Sample Order Events

Now let’s simulate a few order events using kafka-console-producer.sh. These JSON messages represent individual e-commerce transactions.

Each message includes:

· order_id: Unique ID for the order

· user_id: ID of the customer

· product_id: ID of the product purchased

· amount: Order total in USD

· order_status: Current status (e.g., placed, shipped, delivered)

· timestamp: Order time in epoch milliseconds (used by Druid for ingestion)

Send a sample order to Kafka

echo '{"order_id":"ORD12345", "user_id":"U001", "product_id":"P1001", "amount":49.99, "order_status":"placed", "timestamp":1714567890123}' | bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic ecommerce-orders

You can send few more.

echo '{"order_id":"ORD12346", "user_id":"U002", "product_id":"P1002", "amount":89.50, "order_status":"shipped", "timestamp":1714567900000}' | bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic ecommerce-orders

echo '{"order_id":"ORD12347", "user_id":"U001", "product_id":"P1003", "amount":15.75, "order_status":"delivered", "timestamp":1714567910000}' | bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic ecommerce-orders

Step 7: Consume the Messages from the Topic

Let’s verify that Kafka is receiving these messages. Use the consumer to read from the topic:

bin/kafka-console-consumer.sh --topic ecommerce-orders --from-beginning --bootstrap-server localhost:9092

$ bin/kafka-console-consumer.sh --topic ecommerce-orders --from-beginning --bootstrap-server localhost:9092
{"order_id":"ORD12345", "user_id":"U001", "product_id":"P1001", "amount":49.99, "order_status":"placed", "timestamp":1714567890123}
{"order_id":"ORD12346", "user_id":"U002", "product_id":"P1002", "amount":89.50, "order_status":"shipped", "timestamp":1714567900000}
{"order_id":"ORD12347", "user_id":"U001", "product_id":"P1003", "amount":15.75, "order_status":"delivered", "timestamp":1714567910000}

2. Configure Druid to consume from ecommerce-orders topic

Click on Load data -> Streaming

Select ‘Apache Kafka’ and click on Connect data button.

Give Bootstrap server as localhost:9092, topic name as ecommerce-orders and click on Apply button.

Once you click on Apply button, Druid start polling the data from Kafka topic.

Click on Parse data button available at bottom right corner.

Click on Parse time button.

Here you can observe __time column took timestamp field of the records to segment the data.

Subsequently click on Transform, Filter, Configure Schema, Partition buttons.

In the Partition tab, select Segment granularity as day.

Click on Tune button. In the Tune tab, set ‘Use earliest offset’ as True.

Click on Publish button, followed by Edit spec.

In the Edit spec tab, click on Submit supervisor (running) button.

You can observe ecommerce-order task is submitted and in PENDING state.

Wait for some time and click on Refresh button. You can observe that the task is moved to RUNNING state.

Navigate to Druid Query tab and execute following query to print all the records of the datasource ecommerce-orders

SELECT * FROM "ecommerce-orders"

Let’s add one more event to the ecommerce-orders topic.

echo '{"order_id":"ORD12347", "user_id":"U003", "product_id":"P1004", "amount":99.50, "order_status":"shipped", "timestamp":1714567990900}' | bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic ecommerce-orders

Navigate to Druid Query tab and rerune the sql query, you can see new event is pulled by Druid.

That’s it, Happy learning….:)

Previous Next Home

How to delete segments in Druid?

2025-09-09T06:46:00.000-07:00

This is continuation to my previous post. I would recommend you to go through my previous post and understand segements.

Why to Delete a Segment?

Sometimes, you might want to:

· Delete old data (e.g., logs from 3 months ago)

· Fix corrupted or incorrect data

· Reingest data for a given time range

How to delete a segment permanently?

To delete a segment permanently, you must first mark it as unused, and then delete it from storage.

There are multiple ways to mark the segments as unused. To demonstrate the examples, I am using blog_visits datasource, which has 11 segments.

Option 1: Mark Specific Segment IDs as Unused

If you know the exact segment IDs, you can pass them in the payload:

Syntax

curl -X POST "http://<COORDINATOR_HOST>:<PORT>/druid/coordinator/v1/datasources/<DATASOURCE>/markUnused" \
  -H 'Content-Type: application/json' \
  -d '{
        "segmentIds": [
          "blog_visits_2025-04-18T01:00:00.000Z_2025-04-18T02:00:00.000Z_2024-07-01T10:00:00.000Z"
        ]
      }'

For example, following snippet mark two segments as unused.

$curl -X POST "http://localhost:8888/druid/coordinator/v1/datasources/blog_visits/markUnused" \
>   -H 'Content-Type: application/json' \
>   -d '{
>         "segmentIds": [
>           "blog_visits_2025-04-18T10:00:00.000Z_2025-04-18T11:00:00.000Z_2025-04-18T16:53:25.204Z",
>           "blog_visits_2025-04-18T09:00:00.000Z_2025-04-18T10:00:00.000Z_2025-04-18T16:53:25.204Z"
>         ]
>       }'
{"numChangedSegments":2,"segmentStateChanged":true}

After deletion of 2 segments, you can observe we left with 9 segments.

Option 2: Mark Segments as Unused by Time Interval

Use the markUnused endpoint with a time interval:

Syntax

curl -X POST "http://<COORDINATOR_HOST>:<PORT>/druid/coordinator/v1/datasources/<DATASOURCE>/markUnused" \
  -H 'Content-Type: application/json' \
  -d '{
        "interval": "START_INTERVAL/END_INTERVAL"
      }'

For example following statement mark the segments between intervals 2025-04-18T05:00:00.000Z and 2025-04-18T07:00:00.000Z as unused.

curl -X POST "http://localhost:8888/druid/coordinator/v1/datasources/blog_visits/markUnused" \
  -H 'Content-Type: application/json' \
  -d '{
        "interval": "2025-04-18T05:00:00.000Z/2025-04-18T07:00:00.000Z"
      }'

$curl -X POST "http://localhost:8888/druid/coordinator/v1/datasources/blog_visits/markUnused" \
>   -H 'Content-Type: application/json' \
>   -d '{
>         "interval": "2025-04-18T05:00:00.000Z/2025-04-18T07:00:00.000Z"
>       }'
{"numChangedSegments":2,"segmentStateChanged":true}

After marking another 2 segments as unused, we left with 7 segments.

Delete Segments Using a Kill Task

Now that the segments are marked as unused, they are ignored in queries but still live in deep storage. To permanently delete them, submit a kill task:

Syntax

curl -X POST http://<OVERLORD_HOST>:<PORT>/druid/indexer/v1/task \
  -H 'Content-Type: application/json' \
  -d '
    {
      "type": "kill",
      "dataSource": "DATA_SOURCE",
      "interval": "START_INTERVAL/END_INTERVAL"
    }
  '

Example

curl -X POST http://localhost:8888/druid/indexer/v1/task \
  -H 'Content-Type: application/json' \
  -d '
    {
      "type": "kill",
      "dataSource": "blog_visits",
      "interval": "2025-04-18T00:00:00.000Z/2025-04-18T11:00:00.000Z"
    }
  '

$curl -X POST http://localhost:8888/druid/indexer/v1/task \
>   -H 'Content-Type: application/json' \
>   -d '
>     {
>       "type": "kill",
>       "dataSource": "blog_visits",
>       "interval": "2025-04-18T00:00:00.000Z/2025-04-18T11:00:00.000Z"
>     }
>   '
{"task":"kill_blog_visits_hacplbld_2025-04-18T00:00:00.000Z_2025-04-18T11:00:00.000Z_2025-04-18T17:27:15.828Z"}

Previous Next Home

Understanding Segments in Apache Druid

2025-09-09T06:33:00.000-07:00

When you ingest data into Apache Druid, it's stored in segments. Segments are optimized, compressed data blocks that enable fast querying and efficient storage. Understanding how segmentation works is essential for designing performant Druid setups.

In this blog, we’ll explore what segments are, how Druid creates them based on your data and configurations and walk through an example.

Segments in Apache Druid

In Apache Druid, a segment is a chunk of data that covers a specific time interval, like an hour or a day.

· Segments are the basic unit of storage and query execution.

· They are compressed, column-oriented, and distributed across Druid nodes for parallel querying.

· By controlling segment granularity, you can optimize your ingestion and query performance.

Let’s simulate blog visit logs where each entry represents a visit to a blog post at a specific timestamp. We’ll segment the data hourly, so if your data spans 10 different hours, Druid will create 10 separate segments.

Let’s create blog_visits.csv file.

blog_visits.csv

timestamp,post_id,user_id
2025-04-18T00:10:00Z,101,u1
2025-04-18T01:05:00Z,101,u2
2025-04-18T02:15:00Z,102,u3
2025-04-18T03:30:00Z,103,u4
2025-04-18T04:25:00Z,104,u5
2025-04-18T05:55:00Z,105,u6
2025-04-18T06:45:00Z,106,u7
2025-04-18T07:20:00Z,107,u8
2025-04-18T08:10:00Z,108,u9
2025-04-18T09:40:00Z,109,u10
2025-04-18T10:01:00Z,110,u11

This dataset spans 11 hours, which will generate 11 segments when using hourly segment granularity.

Ingestion Spec

blog_visits_spec.json

{
  "type": "index",
  "spec": {
    "dataSchema": {
      "dataSource": "blog_visits",
      "timestampSpec": {
        "column": "timestamp",
        "format": "iso"
      },
      "dimensionsSpec": {
        "dimensions": [
          "post_id",
          "user_id"
        ]
      },
      "metricsSpec": [],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "HOUR",
        "queryGranularity": "NONE",
        "rollup": false
      }
    },
    "ioConfig": {
      "type": "index",
      "inputSource": {
        "type": "local",
        "baseDir": "/Users/Shared/druid_samples",
        "filter": "blog_visits.csv"
      },
      "inputFormat": {
        "type": "csv",
        "columns": [
          "timestamp",
          "post_id",
          "user_id"
        ],
        "findColumnsFromHeader": false,
        "delimiter": ","
      }
    },
    "tuningConfig": {
      "type": "index",
      "maxRowsInMemory": 10000,
      "maxBytesInMemory": 0,
      "maxRowsPerSegment": 5000000,
      "maxTotalRows": 20000000,
      "forceExtendableShardSpecs": true,
      "logParseExceptions": true,
      "resetOffsetAutomatically": true
    }
  }
}

Execute following curl statement to onboard above spec.

curl -X POST http://localhost:8081/druid/indexer/v1/task \
     -H 'Content-Type: application/json' \
     -d @blog_visits_spec.json

Upon successful onboarding of blog_visits data, navigate to Druid console -> Segments tab.

Filter by blog_visits datasource, you can see all the segments.

Best Practices to choose segment granularity

· Choose segment granularity based on data volume and query patterns.

· Use hourly for high-volume, high-frequency data like logs or metrics.

· Use daily or coarser granularity for less frequent data.

Previous Next Home

Filtering Embedding Searches in LangChain4j

2025-09-09T06:24:00.000-07:00

When working with large-scale vector databases like Pinecone, Milvus, or Qdrant, semantic search isn't just about finding similar embeddings, it's about finding the right similar items. Just like SQL how we can filter rows using WHERE clauses, modern vector databases also support metadata based filtering. LangChain4j abstracts over multiple vector stores, and its Filter API enables a consistent way to define these filters.

In this post, we’ll explore how to use Filter in LangChain4j to refine your embedding searches using rich metadata, and the types of filters available.

Why Filtering Matters in Vector Search?

Imagine a movie recommendation system. You don't just want "similar" movies, you want similar comedies, released after 2010, and with a rating above 7. That’s where metadata filtering becomes essential.

LangChain4j provides an expressive API to define filters that work across multiple embedding store implementations. These filters operate on metadata fields associated with each embedding.

Following table summarizes the list of filters available in LangChain4j.

Filter Type	Description
IsEqualTo	Checks if metadata key equals a given value
IsNotEqualTo	Opposite of IsEqualTo
IsGreaterThan	Numerical comparison
IsGreaterThanOrEqualTo	Numerical comparison
IsLessThan	Numerical comparison
IsLessThanOrEqualTo	Numerical comparison
IsIn	Checks if a value is in a given set/list
IsNotIn	Checks if a value is not in a given set/list
ContainsString	Checks if string contains a substring (limited support)
And, Or, Not	Logical combinators to build complex expressions

Example

Filter filter = new And(
	    new And(
	        new IsEqualTo("source", "news"),
	        new IsIn("category", List.of("technology", "finance"))
	    ),
	    new IsGreaterThanOrEqualTo("relevance", Integer.valueOf(9))
	);

Find the below working application.

EmbeddingSearchWithFilters.java

package com.sample.app.embeddings;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import dev.langchain4j.data.document.Metadata;
import dev.langchain4j.data.embedding.Embedding;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.model.embedding.onnx.bgesmallenv15q.BgeSmallEnV15QuantizedEmbeddingModel;
import dev.langchain4j.store.embedding.EmbeddingSearchRequest;
import dev.langchain4j.store.embedding.EmbeddingSearchResult;
import dev.langchain4j.store.embedding.EmbeddingStore;
import dev.langchain4j.store.embedding.filter.Filter;
import dev.langchain4j.store.embedding.filter.comparison.IsEqualTo;
import dev.langchain4j.store.embedding.filter.comparison.IsGreaterThanOrEqualTo;
import dev.langchain4j.store.embedding.filter.comparison.IsIn;
import dev.langchain4j.store.embedding.filter.logical.And;
import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;

public class EmbeddingSearchWithFilters {

	public static void main(String[] args) {

		// Step 1: Initialize embedding model
		EmbeddingModel model = new BgeSmallEnV15QuantizedEmbeddingModel();

		// Step 2: Prepare text data with metadata
		String sentence1 = "The stock market surged as tech companies reported strong earnings.";
		Map<String, Object> metaMap1 = new HashMap<>();
		metaMap1.put("category", "finance");
		metaMap1.put("source", "news");
		metaMap1.put("relevance", 9);
		Metadata meta1 = Metadata.from(metaMap1);

		String sentence2 = "Tech giants like Apple and Amazon saw record profits this quarter.";
		Map<String, Object> metaMap2 = new HashMap<>();
		metaMap2.put("category", "technology");
		metaMap2.put("source", "news");
		metaMap2.put("relevance", 10);
		Metadata meta2 = Metadata.from(metaMap2);

		String sentence3 = "Heavy rain caused flooding in coastal towns yesterday.";
		Map<String, Object> metaMap3 = new HashMap<>();
		metaMap3.put("category", "weather");
		metaMap3.put("source", "alert");
		metaMap3.put("relevance", 5);
		Metadata meta3 = Metadata.from(metaMap3);

		try {
			// Step 3: Generate embeddings
			Embedding embedding1 = model.embed(sentence1).content();
			Embedding embedding2 = model.embed(sentence2).content();
			Embedding embedding3 = model.embed(sentence3).content();

			// Step 4: Initialize in-memory embedding store
			EmbeddingStore<TextSegment> embeddingStore = new InMemoryEmbeddingStore<>();

			// Step 5: Add embeddings to the store with metadata
			embeddingStore.add(embedding1, TextSegment.from(sentence1, meta1));
			embeddingStore.add(embedding2, TextSegment.from(sentence2, meta2));
			embeddingStore.add(embedding3, TextSegment.from(sentence3, meta3));

			System.out.println("Embeddings with metadata added to the store successfully.");

			// Step 6: Prepare a query embedding
			Embedding queryEmbedding = model.embed("Which tech companies had profits?").content();

			Filter filter = new And(
					new And(new IsEqualTo("source", "news"), new IsIn("category", List.of("technology", "finance"))),
					new IsGreaterThanOrEqualTo("relevance", Integer.valueOf(9)));

			// Step 8: Build search request with filter
			EmbeddingSearchRequest request = EmbeddingSearchRequest.builder().queryEmbedding(queryEmbedding)
					.maxResults(2).minScore(0.5).filter(filter).build();

			// Step 9: Perform search
			EmbeddingSearchResult<TextSegment> result = embeddingStore.search(request);

			// Step 10: Display results
			System.out.println("\nFiltered & Ranked Results:");
			result.matches().forEach(match -> {
				TextSegment segment = match.embedded();
				System.out.printf("Score: %.3f | Text: %s | Metadata: %s%n", match.score(), segment.text(),
						segment.metadata());
			});

		} catch (Exception e) {
			System.err.println("An error occurred: " + e.getMessage());
			e.printStackTrace();
		}
	}
}

Output

Embeddings with metadata added to the store successfully.

Filtered & Ranked Results:
Score: 0.883 | Text: Tech giants like Apple and Amazon saw record profits this quarter. | Metadata: Metadata { metadata = {category=technology, relevance=10, source=news} }
Score: 0.882 | Text: The stock market surged as tech companies reported strong earnings. | Metadata: Metadata { metadata = {category=finance, relevance=9, source=news} }

Previous Next Home