FIX: SPS Relatred Refactoring and Fixed in Prompt Service (#550)

* Refactored the prompt service related to SPS

* Improved column names for manage documents modal

---------

Co-authored-by: Gayathri <142381512+gaya3-zipstack@users.noreply.github.com>

Author: tahierhussain

frontend/src/components/custom-tools/manage-docs-modal/ManageDocsModal.jsx

@@ -299,7 +299,7 @@ function ManageDocsModal({
     {
       title: (
         <Space className="w-100">
-          <Typography.Text>Index</Typography.Text>
+          <Typography.Text>Raw View</Typography.Text>
           <Typography.Text type="secondary">
             {"(" + getLlmProfileName(rawLlmProfile) + ")"}
           </Typography.Text>
@@ -308,13 +308,13 @@ function ManageDocsModal({
       ),
       dataIndex: "index",
       key: "index",
-      width: 260,
+      width: 300,
     },
     {
-      title: "Index",
+      title: "Actions",
       dataIndex: "reindex",
       key: "reindex",
-      width: 260,
+      width: 200,
     },
     {
       title: "",
@@ -340,7 +340,7 @@ function ManageDocsModal({
       ),
       dataIndex: "summary",
       key: "summary",
-      width: 260,
+      width: 300,
     });
   }
 

prompt-service/src/unstract/prompt_service/helper.py

@@ -7,6 +7,10 @@
 from dotenv import load_dotenv
 from flask import Flask, current_app
 from unstract.prompt_service.authentication_middleware import AuthenticationMiddleware
+from unstract.prompt_service.constants import PromptServiceContants as PSKeys
+from unstract.prompt_service.exceptions import APIError, RateLimitError
+from unstract.sdk.exceptions import RateLimitError as SdkRateLimitError
+from unstract.sdk.llm import LLM
 
 load_dotenv()
 
@@ -152,3 +156,115 @@ def _get_key_and_item(row: tuple) -> tuple[str, dict[str, Any]]:
 def _format_float_positional(value: float, precision: int = 10) -> str:
     formatted: str = f"{value:.{precision}f}"
     return formatted.rstrip("0").rstrip(".") if "." in formatted else formatted
+
+
+def extract_variable(
+    structured_output: dict[str, Any],
+    variable_names: list[Any],
+    output: dict[str, Any],
+    promptx: str,
+) -> str:
+    logger: Logger = current_app.logger
+    for variable_name in variable_names:
+        if promptx.find(f"%{variable_name}%") >= 0:
+            if variable_name in structured_output:
+                promptx = promptx.replace(
+                    f"%{variable_name}%",
+                    str(structured_output[variable_name]),
+                )
+            else:
+                raise ValueError(
+                    f"Variable {variable_name} not found " "in structured output"
+                )
+
+    if promptx != output[PSKeys.PROMPT]:
+        logger.info(f"Prompt after variable replacement: {promptx}")
+    return promptx
+
+
+def construct_and_run_prompt(
+    tool_settings: dict[str, Any],
+    output: dict[str, Any],
+    llm: LLM,
+    context: str,
+    prompt: str,
+) -> tuple[str, dict[str, Any]]:
+    prompt = construct_prompt(
+        preamble=tool_settings.get(PSKeys.PREAMBLE, ""),
+        prompt=output[prompt],
+        postamble=tool_settings.get(PSKeys.POSTAMBLE, ""),
+        grammar_list=tool_settings.get(PSKeys.GRAMMAR, []),
+        context=context,
+    )
+    return run_completion(
+        llm=llm,
+        prompt=prompt,
+    )
+
+
+def construct_prompt(
+    preamble: str,
+    prompt: str,
+    postamble: str,
+    grammar_list: list[dict[str, Any]],
+    context: str,
+) -> str:
+    logger: Logger = current_app.logger
+    # Let's cleanup the context. Remove if 3 consecutive newlines are found
+    context_lines = context.split("\n")
+    new_context_lines = []
+    empty_line_count = 0
+    for line in context_lines:
+        if line.strip() == "":
+            empty_line_count += 1
+        else:
+            if empty_line_count >= 3:
+                empty_line_count = 3
+            for i in range(empty_line_count):
+                new_context_lines.append("")
+            empty_line_count = 0
+            new_context_lines.append(line.rstrip())
+    context = "\n".join(new_context_lines)
+    logger.info(
+        f"Old context length: {len(context_lines)}, "
+        f"New context length: {len(new_context_lines)}"
+    )
+
+    prompt = (
+        f"{preamble}\n\nContext:\n---------------{context}\n"
+        f"-----------------\n\nQuestion or Instruction: {prompt}\n"
+    )
+    if grammar_list is not None and len(grammar_list) > 0:
+        prompt += "\n"
+        for grammar in grammar_list:
+            word = ""
+            synonyms = []
+            if PSKeys.WORD in grammar:
+                word = grammar[PSKeys.WORD]
+                if PSKeys.SYNONYMS in grammar:
+                    synonyms = grammar[PSKeys.SYNONYMS]
+            if len(synonyms) > 0 and word != "":
+                prompt += f'\nNote: You can consider that the word {word} is same as \
+                    {", ".join(synonyms)} in both the quesiton and the context.'  # noqa
+    prompt += f"\n\n{postamble}"
+    prompt += "\n\nAnswer:"
+    return prompt
+
+
+def run_completion(
+    llm: LLM,
+    prompt: str,
+) -> tuple[str, dict[str, Any]]:
+    logger: Logger = current_app.logger
+    try:
+        completion = llm.complete(prompt)
+
+        answer: str = completion[PSKeys.RESPONSE].text
+        return answer
+    # TODO: Catch and handle specific exception here
+    except SdkRateLimitError as e:
+        raise RateLimitError(f"Rate limit error. {str(e)}") from e
+    except Exception as e:
+        logger.error(f"Error fetching response for prompt: {e}.")
+        # TODO: Publish this error as a FE update
+        raise APIError(str(e)) from e

prompt-service/src/unstract/prompt_service/main.py

@@ -11,21 +11,18 @@
 from unstract.prompt_service.config import create_app
 from unstract.prompt_service.constants import PromptServiceContants as PSKeys
 from unstract.prompt_service.constants import RunLevel
-from unstract.prompt_service.exceptions import (
-    APIError,
-    ErrorResponse,
-    NoPayloadError,
-    RateLimitError,
-)
+from unstract.prompt_service.exceptions import APIError, ErrorResponse, NoPayloadError
 from unstract.prompt_service.helper import (
     EnvLoader,
+    construct_and_run_prompt,
+    extract_variable,
     plugin_loader,
     query_usage_metadata,
+    run_completion,
 )
 from unstract.prompt_service.prompt_ide_base_tool import PromptServiceBaseTool
 from unstract.sdk.constants import LogLevel
 from unstract.sdk.embedding import Embedding
-from unstract.sdk.exceptions import RateLimitError as SdkRateLimitError
 from unstract.sdk.exceptions import SdkError
 from unstract.sdk.index import Index
 from unstract.sdk.llm import LLM
@@ -81,54 +78,6 @@ def _publish_log(
     )
 
 
-def construct_prompt(
-    preamble: str,
-    prompt: str,
-    postamble: str,
-    grammar_list: list[dict[str, Any]],
-    context: str,
-) -> str:
-    # Let's cleanup the context. Remove if 3 consecutive newlines are found
-    context_lines = context.split("\n")
-    new_context_lines = []
-    empty_line_count = 0
-    for line in context_lines:
-        if line.strip() == "":
-            empty_line_count += 1
-        else:
-            if empty_line_count >= 3:
-                empty_line_count = 3
-            for i in range(empty_line_count):
-                new_context_lines.append("")
-            empty_line_count = 0
-            new_context_lines.append(line.rstrip())
-    context = "\n".join(new_context_lines)
-    app.logger.info(
-        f"Old context length: {len(context_lines)}, "
-        f"New context length: {len(new_context_lines)}"
-    )
-
-    prompt = (
-        f"{preamble}\n\nContext:\n---------------{context}\n"
-        f"-----------------\n\nQuestion or Instruction: {prompt}\n"
-    )
-    if grammar_list is not None and len(grammar_list) > 0:
-        prompt += "\n"
-        for grammar in grammar_list:
-            word = ""
-            synonyms = []
-            if PSKeys.WORD in grammar:
-                word = grammar[PSKeys.WORD]
-                if PSKeys.SYNONYMS in grammar:
-                    synonyms = grammar[PSKeys.SYNONYMS]
-            if len(synonyms) > 0 and word != "":
-                prompt += f'\nNote: You can consider that the word {word} is same as \
-                    {", ".join(synonyms)} in both the quesiton and the context.'  # noqa
-    prompt += f"\n\n{postamble}"
-    prompt += "\n\nAnswer:"
-    return prompt
-
-
 def authentication_middleware(func: Any) -> Any:
     def wrapper(*args: Any, **kwargs: Any) -> Any:
         token = AuthenticationMiddleware.get_token_from_auth_header(request)
@@ -736,67 +685,6 @@ def _retrieve_context(output, doc_id, vector_index, answer) -> str:
     return text
 
 
-def construct_and_run_prompt(
-    tool_settings: dict[str, Any],
-    output: dict[str, Any],
-    llm: LLM,
-    context: str,
-    prompt: str,
-) -> tuple[str, dict[str, Any]]:
-    prompt = construct_prompt(
-        preamble=tool_settings.get(PSKeys.PREAMBLE, ""),
-        prompt=output[prompt],
-        postamble=tool_settings.get(PSKeys.POSTAMBLE, ""),
-        grammar_list=tool_settings.get(PSKeys.GRAMMAR, []),
-        context=context,
-    )
-    return run_completion(
-        llm=llm,
-        prompt=prompt,
-    )
-
-
-def run_completion(
-    llm: LLM,
-    prompt: str,
-) -> tuple[str, dict[str, Any]]:
-    try:
-        completion = llm.complete(prompt)
-
-        answer: str = completion[PSKeys.RESPONSE].text
-        return answer
-    # TODO: Catch and handle specific exception here
-    except SdkRateLimitError as e:
-        raise RateLimitError(f"Rate limit error. {str(e)}") from e
-    except Exception as e:
-        app.logger.error(f"Error fetching response for prompt: {e}.")
-        # TODO: Publish this error as a FE update
-        raise APIError(str(e)) from e
-
-
-def extract_variable(
-    structured_output: dict[str, Any],
-    variable_names: list[Any],
-    output: dict[str, Any],
-    promptx: str,
-) -> str:
-    for variable_name in variable_names:
-        if promptx.find(f"%{variable_name}%") >= 0:
-            if variable_name in structured_output:
-                promptx = promptx.replace(
-                    f"%{variable_name}%",
-                    str(structured_output[variable_name]),
-                )
-            else:
-                raise ValueError(
-                    f"Variable {variable_name} not found " "in structured output"
-                )
-
-    if promptx != output[PSKeys.PROMPT]:
-        app.logger.info(f"Prompt after variable replacement: {promptx}")
-    return promptx
-
-
 def enable_plugins() -> None:
     """Enables plugins if available."""
     single_pass_extration_plugin: dict[str, Any] = plugins.get(