feat(prompts): add a central prompt module

Author: magurh

src/flare_ai_rag/prompts/__init__.py

@@ -0,0 +1,5 @@
+from .library import PromptLibrary
+from .schemas import SemanticRouterResponse
+from .service import PromptService
+
+__all__ = ["PromptLibrary", "PromptService", "SemanticRouterResponse"]

src/flare_ai_rag/prompts/library.py

@@ -0,0 +1,201 @@
+"""
+Prompt Library Module for Flare AI RAG
+
+This module provides a centralized management system for AI prompts used throughout
+the Flare AI RAG application. It handles the organization, storage, and retrieval
+of various prompt templates used for different operations like token transactions,
+account generation, and user interactions.
+
+The module implements a PromptLibrary class that maintains a collection of Prompt
+objects, each representing a specific type of interaction or operation template.
+Prompts are categorized for easy management and retrieval.
+"""
+
+import structlog
+
+from flare_ai_rag.prompts.schemas import (
+    Prompt,
+    RAGRouterResponse,
+    SemanticRouterResponse,
+)
+from flare_ai_rag.prompts.templates import (
+    CONVERSATIONAL,
+    RAG_RESPONDER,
+    RAG_ROUTER,
+    REMOTE_ATTESTATION,
+    SEMANTIC_ROUTER,
+)
+
+logger = structlog.get_logger(__name__)
+
+
+class PromptLibrary:
+    """
+    A library for managing and organizing AI prompts used in the Flare AI RAG.
+
+    This class serves as a central repository for all prompt templates used in
+    the application. It provides functionality to add, retrieve, and categorize
+    prompts for various operations such as token transactions, account management,
+    and user interactions.
+
+    Attributes:
+        prompts (dict[str, Prompt]): Dictionary storing prompt objects
+            with their names as keys.
+    """
+
+    def __init__(self) -> None:
+        """
+        Initialize a new PromptLibrary instance.
+
+        Creates an empty prompt dictionary and populates it with default prompts
+        through the _initialize_default_prompts method.
+        """
+        self.prompts: dict[str, Prompt] = {}
+        self._initialize_default_prompts()
+
+    def _initialize_default_prompts(self) -> None:
+        """
+        Initialize the library with a set of default prompts.
+
+        Creates and adds the following default prompts:
+        - semantic_router: For routing user queries
+        - token_send: For token transfer operations
+        - token_swap: For token swap operations
+        - generate_account: For wallet generation
+        - conversational: For general user interactions
+        - request_attestation: For remote attestation requests
+        - tx_confirmation: For transaction confirmation
+
+        This method is called automatically during instance initialization.
+        """
+        default_prompts = [
+            Prompt(
+                name="semantic_router",
+                description="Route user query based on user input",
+                template=SEMANTIC_ROUTER,
+                required_inputs=["user_input"],
+                response_mime_type="text/x.enum",
+                response_schema=SemanticRouterResponse,
+                category="router",
+            ),
+            Prompt(
+                name="conversational",
+                description="Converse with a user",
+                template=CONVERSATIONAL,
+                required_inputs=["user_input"],
+                response_schema=None,
+                response_mime_type=None,
+                category="conversational",
+            ),
+            Prompt(
+                name="rag_router",
+                description="The ",
+                template=RAG_ROUTER,
+                required_inputs=["user_input"],
+                response_mime_type="application/json",
+                response_schema=RAGRouterResponse,
+                category="rag-router",
+            ),
+            Prompt(
+                name="rag_responder",
+                description="The ",
+                template=RAG_RESPONDER,
+                required_inputs=["user_input"],
+                response_schema=None,
+                response_mime_type=None,
+                category="conversational",
+            ),
+            Prompt(
+                name="request_attestation",
+                description="User has requested a remote attestation",
+                template=REMOTE_ATTESTATION,
+                required_inputs=None,
+                response_schema=None,
+                response_mime_type=None,
+                category="conversational",
+            ),
+        ]
+
+        for prompt in default_prompts:
+            self.add_prompt(prompt)
+
+    def add_prompt(self, prompt: Prompt) -> None:
+        """
+        Add a new prompt to the library.
+
+        Args:
+            prompt (Prompt): The prompt object to add to the library.
+
+        Logs:
+            Debug log entry when prompt is successfully added.
+
+        Example:
+            ```python
+            custom_prompt = Prompt(name="custom", template="...", category="misc")
+            library.add_prompt(custom_prompt)
+            ```
+        """
+        self.prompts[prompt.name] = prompt
+        logger.debug("prompt_added", name=prompt.name, category=prompt.category)
+
+    def get_prompt(self, name: str) -> Prompt:
+        """
+        Retrieve a prompt by its name.
+
+        Args:
+            name (str): The name of the prompt to retrieve.
+
+        Returns:
+            Prompt: The requested prompt object.
+
+        Raises:
+            KeyError: If the prompt name doesn't exist in the library.
+
+        Example:
+            ```python
+            try:
+                prompt = library.get_prompt("token_send")
+            except KeyError:
+                print("Prompt not found")
+            ```
+        """
+        if name not in self.prompts:
+            logger.error("prompt_not_found", name=name)
+            msg = f"Prompt '{name}' not found in library"
+            raise KeyError(msg)
+        return self.prompts[name]
+
+    def get_prompts_by_category(self, category: str) -> list[Prompt]:
+        """
+        Get all prompts in a specific category.
+
+        Args:
+            category (str): The category to filter prompts by.
+
+        Returns:
+            list[Prompt]: A list of all prompts in the specified category.
+        """
+        return [
+            prompt for prompt in self.prompts.values() if prompt.category == category
+        ]
+
+    def list_categories(self) -> list[str]:
+        """
+        List all available prompt categories.
+
+        Returns:
+            list[str]: A list of unique category names used in the library.
+
+        Example:
+            ```python
+            categories = library.list_categories()
+            print("Available categories:", categories)
+            ```
+        """
+        return list(
+            {
+                prompt.category
+                for prompt in self.prompts.values()
+                if prompt.category is not None
+            }
+        )

src/flare_ai_rag/prompts/schemas.py

@@ -0,0 +1,157 @@
+"""
+Schema Definitions for Flare AI DeFAI Prompts
+
+This module defines the core data structures and types used for managing prompts
+and their responses in the Flare AI DeFAI system. It includes enums for semantic
+routing, type definitions for token operations, and a dataclass for prompt
+templates.
+
+The module provides type safety and structured data handling for AI prompt
+interactions, ensuring consistency in prompt formatting and response handling
+across the application.
+"""
+
+from dataclasses import dataclass
+from enum import Enum
+from string import Template
+from typing import TypedDict
+
+
+class SemanticRouterResponse(str, Enum):
+    """
+    Enumeration of possible semantic routing outcomes for user queries.
+
+    This enum defines the various types of operations that can be triggered
+    based on user input analysis. Each value represents a specific action
+    or response type that the system can handle.
+
+    Attributes:
+        REQUEST_ATTESTATION: Route to attestation request handling
+        CONVERSATIONAL: Route to general conversational response
+        RAG_ROUTER: Router to RAG pipeline router
+        RAG_RESPONDER: Router to RAG pipeline responder
+    """
+
+    REQUEST_ATTESTATION = "RequestAttestation"
+    CONVERSATIONAL = "Conversational"
+    RAG_ROUTER = "RagRouter"
+    RAG_RESPONDER = "RagResponder"
+
+
+class RAGRouterResponse(TypedDict):
+    """
+    Type definition for RAG router response type.
+
+    Defines the required fields for a RAG routing operation,
+
+    Attributes:
+        classification (str): The response class
+    """
+
+    classification: str
+
+
+class PromptInputs(TypedDict, total=False):
+    """
+    Type definition for various types of prompt inputs.
+
+    A flexible TypedDict that defines possible input types for prompts.
+    The total=False flag indicates that all fields are optional.
+
+    Attributes:
+        user_input (str): Raw user input text
+        text (str): Processed or formatted text
+        content (str): General content string
+        code (str): Code snippet or related content
+    """
+
+    user_input: str
+    text: str
+    content: str
+    code: str
+
+
+@dataclass
+class Prompt:
+    """
+    A dataclass representing an AI prompt template with its metadata
+    and formatting logic.
+
+    This class encapsulates all information needed to define and use an AI prompt,
+    including its template text, required inputs, response handling, and metadata.
+
+    Attributes:
+        name (str): Unique identifier for the prompt
+        description (str): Human-readable description of the prompt's purpose
+        template (str): The prompt template text with optional placeholder variables
+        required_inputs (list[str] | None): List of required input variable names
+        response_schema (type | None): Expected response type/schema
+        response_mime_type (str | None): MIME type of the expected response
+        examples (list[dict[str, str]] | None): Example usages of the prompt
+        category (str | None): Grouping category for the prompt
+        version (str): Version string for the prompt template
+
+    Example:
+        ```python
+        prompt = Prompt(
+            name="token_send",
+            description="Format a token send request",
+            template="Send ${amount} tokens to ${address}",
+            required_inputs=["amount", "address"],
+            response_schema=TokenSendResponse,
+            response_mime_type="application/json",
+        )
+        formatted = prompt.format(amount="100", address="0x123...")
+        ```
+    """
+
+    name: str
+    description: str
+    template: str
+    required_inputs: list[str] | None
+    response_schema: type | None
+    response_mime_type: str | None
+    examples: list[dict[str, str]] | None = None
+    category: str | None = None
+    version: str = "1.0"
+
+    def format(self, **kwargs: str | PromptInputs) -> str:
+        """
+        Format the prompt template with provided input values.
+
+        This method uses string.Template to substitute variables in the prompt
+        template with provided values. It validates that all required inputs
+        are provided before formatting.
+
+        Args:
+            **kwargs: Keyword arguments containing values for template variables.
+                     Can be strings or PromptInputs objects.
+
+        Returns:
+            str: The formatted prompt string with all variables substituted.
+
+        Raises:
+            ValueError: If any required inputs are missing from kwargs.
+            KeyError: If template substitution fails due to missing keys.
+
+        Example:
+            ```python
+            prompt = Prompt(
+                template="Hello ${name}!",
+                required_inputs=["name"],
+                ...
+            )
+            result = prompt.format(name="Alice")
+            ```
+        """
+        if not self.required_inputs:
+            return self.template
+
+        try:
+            return Template(self.template).safe_substitute(**kwargs)
+        except KeyError as e:
+            missing_keys = set(self.required_inputs) - set(kwargs.keys())
+            if missing_keys:
+                msg = f"Missing required inputs: {missing_keys}"
+                raise ValueError(msg) from e
+            raise