handle null value exceptions when arguments are missing or Null in caling RAG pipeline

Signed-off-by: Xun Zhang <xunzh@amazon.com>

Author: Zhangxunmt

search-processors/src/main/java/org/opensearch/searchpipelines/questionanswering/generative/GenerativeQAResponseProcessor.java

@@ -27,6 +27,7 @@
 import java.util.Map;
 import java.util.function.BooleanSupplier;
 
+import org.opensearch.OpenSearchException;
 import org.opensearch.action.search.SearchRequest;
 import org.opensearch.action.search.SearchResponse;
 import org.opensearch.client.Client;
@@ -58,7 +59,8 @@
  */
 @Log4j2
 public class GenerativeQAResponseProcessor extends AbstractProcessor implements SearchResponseProcessor {
-
+    public static String IllegalArgumentMessage =
+        "Please check the provided generative_qa_parameters are complete and non-null(https://opensearch.org/docs/latest/search-plugins/conversational-search/#rag-pipeline). Messages in the memory can not have Null value for input and response";
     private static final int DEFAULT_CHAT_HISTORY_WINDOW = 10;
 
     private static final int DEFAULT_PROCESSOR_TIME_IN_SECONDS = 30;
@@ -148,37 +150,51 @@ public SearchResponse processResponse(SearchRequest request, SearchResponse resp
         log.info("system_prompt: {}", systemPrompt);
         log.info("user_instructions: {}", userInstructions);
         start = Instant.now();
-        ChatCompletionOutput output = llm
-            .doChatCompletion(
-                LlmIOUtil
-                    .createChatCompletionInput(systemPrompt, userInstructions, llmModel, llmQuestion, chatHistory, searchResults, timeout)
-            );
-        log.info("doChatCompletion complete. ({})", getDuration(start));
-
-        String answer = null;
-        String errorMessage = null;
-        String interactionId = null;
-        if (output.isErrorOccurred()) {
-            errorMessage = output.getErrors().get(0);
-        } else {
-            answer = (String) output.getAnswers().get(0);
-
-            if (conversationId != null) {
-                start = Instant.now();
-                interactionId = memoryClient
-                    .createInteraction(
-                        conversationId,
-                        llmQuestion,
-                        PromptUtil.getPromptTemplate(systemPrompt, userInstructions),
-                        answer,
-                        GenerativeQAProcessorConstants.RESPONSE_PROCESSOR_TYPE,
-                        Collections.singletonMap("metadata", jsonArrayToString(searchResults))
-                    );
-                log.info("Created a new interaction: {} ({})", interactionId, getDuration(start));
+        try {
+            ChatCompletionOutput output = llm
+                .doChatCompletion(
+                    LlmIOUtil
+                        .createChatCompletionInput(
+                            systemPrompt,
+                            userInstructions,
+                            llmModel,
+                            llmQuestion,
+                            chatHistory,
+                            searchResults,
+                            timeout
+                        )
+                );
+            log.info("doChatCompletion complete. ({})", getDuration(start));
+
+            String answer = null;
+            String errorMessage = null;
+            String interactionId = null;
+            if (output.isErrorOccurred()) {
+                errorMessage = output.getErrors().get(0);
+            } else {
+                answer = (String) output.getAnswers().get(0);
+
+                if (conversationId != null) {
+                    start = Instant.now();
+                    interactionId = memoryClient
+                        .createInteraction(
+                            conversationId,
+                            llmQuestion,
+                            PromptUtil.getPromptTemplate(systemPrompt, userInstructions),
+                            answer,
+                            GenerativeQAProcessorConstants.RESPONSE_PROCESSOR_TYPE,
+                            Collections.singletonMap("metadata", jsonArrayToString(searchResults))
+                        );
+                    log.info("Created a new interaction: {} ({})", interactionId, getDuration(start));
+                }
             }
-        }
 
-        return insertAnswer(response, answer, errorMessage, interactionId);
+            return insertAnswer(response, answer, errorMessage, interactionId);
+        } catch (NullPointerException nullPointerException) {
+            throw new IllegalArgumentException(IllegalArgumentMessage);
+        } catch (Exception e) {
+            throw new OpenSearchException("GenerativeQAResponseProcessor failed in precessing response");
+        }
     }
 
     long getDuration(Instant start) {

search-processors/src/test/java/org/opensearch/searchpipelines/questionanswering/generative/GenerativeQAResponseProcessorTests.java

@@ -22,6 +22,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+import static org.opensearch.searchpipelines.questionanswering.generative.GenerativeQAResponseProcessor.IllegalArgumentMessage;
 
 import java.time.Instant;
 import java.util.Collections;
@@ -407,4 +408,57 @@ public void testProcessorFeatureOffOnOff() throws Exception {
         }
         assertTrue(secondExceptionThrown);
     }
+
+    public void testProcessResponseNullValueInteractions() throws Exception {
+        exceptionRule.expect(IllegalArgumentException.class);
+        exceptionRule.expectMessage(IllegalArgumentMessage);
+
+        Client client = mock(Client.class);
+        Map<String, Object> config = new HashMap<>();
+        config.put(GenerativeQAProcessorConstants.CONFIG_NAME_MODEL_ID, "dummy-model");
+        config.put(GenerativeQAProcessorConstants.CONFIG_NAME_CONTEXT_FIELD_LIST, List.of("text"));
+
+        GenerativeQAResponseProcessor processor = (GenerativeQAResponseProcessor) new GenerativeQAResponseProcessor.Factory(
+            client,
+            alwaysOn
+        ).create(null, "tag", "desc", true, config, null);
+
+        ConversationalMemoryClient memoryClient = mock(ConversationalMemoryClient.class);
+        when(memoryClient.getInteractions(any(), anyInt()))
+            .thenReturn(List.of(new Interaction("0", Instant.now(), "1", null, null, null, null, null)));
+        processor.setMemoryClient(memoryClient);
+
+        SearchRequest request = new SearchRequest();
+        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
+        int contextSize = 5;
+        GenerativeQAParameters params = new GenerativeQAParameters("12345", "llm_model", "You are kind.", contextSize, null, null);
+        GenerativeQAParamExtBuilder extBuilder = new GenerativeQAParamExtBuilder();
+        extBuilder.setParams(params);
+        request.source(sourceBuilder);
+        sourceBuilder.ext(List.of(extBuilder));
+
+        int numHits = 10;
+        SearchHit[] hitsArray = new SearchHit[numHits];
+        for (int i = 0; i < numHits; i++) {
+            XContentBuilder sourceContent = JsonXContent
+                .contentBuilder()
+                .startObject()
+                .field("_id", String.valueOf(i))
+                .field("text", "passage" + i)
+                .field("title", "This is the title for document " + i)
+                .endObject();
+            hitsArray[i] = new SearchHit(i, "doc" + i, Map.of(), Map.of());
+            hitsArray[i].sourceRef(BytesReference.bytes(sourceContent));
+        }
+
+        SearchHits searchHits = new SearchHits(hitsArray, null, 1.0f);
+        SearchResponseSections internal = new SearchResponseSections(searchHits, null, null, false, false, null, 0);
+        SearchResponse response = new SearchResponse(internal, null, 1, 1, 0, 1, null, null, null);
+
+        Llm llm = mock(Llm.class);
+        when(llm.doChatCompletion(any())).thenThrow(new NullPointerException("Null Pointer in Interactions"));
+        processor.setLlm(llm);
+
+        SearchResponse res = processor.processResponse(request, response);
+    }
 }