Prompts API¶

Template library, builders, and RAG integration for structured prompt engineering.

Overview¶

The prompts API provides a flexible system for managing, rendering, and versioning prompt templates with support for Jinja2 templating and RAG integration.

Prompt Library¶

Abstract Interface¶

dataknobs_llm.prompts.AbstractPromptLibrary ¶

Bases: ABC

Abstract base class for prompt library implementations.

All prompt libraries (filesystem, config, composite, etc.) must implement this interface to provide consistent access to prompts, message indexes, and RAG configurations.

Methods:

Name	Description
`get_system_prompt`	Retrieve a system prompt template by name.
`get_user_prompt`	Retrieve a user prompt template by name.

Functions¶

get_system_prompt `abstractmethod` ¶

get_system_prompt(name: str, **kwargs: Any) -> PromptTemplateDict | None

Retrieve a system prompt template by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	System prompt identifier	required
`**kwargs`	`Any`	Additional library-specific parameters	`{}`

Returns:

Type	Description
`PromptTemplateDict \| None`	PromptTemplateDict if found, None otherwise

Source code in packages/llm/src/dataknobs_llm/prompts/base/abstract_prompt_library.py

@abstractmethod
def get_system_prompt(
    self,
    name: str,
    **kwargs: Any
) -> PromptTemplateDict | None:
    """Retrieve a system prompt template by name.

    Args:
        name: System prompt identifier
        **kwargs: Additional library-specific parameters

    Returns:
        PromptTemplateDict if found, None otherwise
    """
    pass

get_user_prompt `abstractmethod` ¶

get_user_prompt(name: str, **kwargs: Any) -> PromptTemplateDict | None

Retrieve a user prompt template by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	User prompt identifier	required
`**kwargs`	`Any`	Additional library-specific parameters	`{}`

Returns:

Type	Description
`PromptTemplateDict \| None`	PromptTemplateDict if found, None otherwise

Source code in packages/llm/src/dataknobs_llm/prompts/base/abstract_prompt_library.py

@abstractmethod
def get_user_prompt(
    self,
    name: str,
    **kwargs: Any
) -> PromptTemplateDict | None:
    """Retrieve a user prompt template by name.

    Args:
        name: User prompt identifier
        **kwargs: Additional library-specific parameters

    Returns:
        PromptTemplateDict if found, None otherwise
    """
    pass

Implementations¶

FileSystemPromptLibrary¶

dataknobs_llm.prompts.FileSystemPromptLibrary ¶

FileSystemPromptLibrary(
    prompt_dir: Union[str, Path],
    auto_load: bool = True,
    file_extensions: List[str] | None = None,
)

Bases: BasePromptLibrary

Prompt library that loads prompts from filesystem directory.

Features: - Supports YAML and JSON file formats - Organized directory structure (system/, user/, messages/) - Caching of loaded prompts for performance - Automatic file discovery and loading - Validation config parsing from files

Example

library = FileSystemPromptLibrary(Path("prompts/")) template = library.get_system_prompt("analyze_code") print(template["template"])

Initialize filesystem prompt library.

Parameters:

Name	Type	Description	Default
`prompt_dir`	`Union[str, Path]`	Root directory containing prompt files	required
`auto_load`	`bool`	Whether to automatically load all prompts on init (default: True)	`True`
`file_extensions`	`List[str] \| None`	List of file extensions to load (default: [".yaml", ".yml", ".json"])	`None`

Methods:

Name	Description
`get_message_index`	Get a message index by name.
`get_prompt_rag_configs`	Get RAG configurations for a specific prompt.
`get_rag_config`	Get a standalone RAG configuration by name.
`get_system_prompt`	Get a system prompt by name.
`get_user_prompt`	Get a user prompt by name.
`list_message_indexes`	List all available message index names.
`list_system_prompts`	List all available system prompt names.
`list_user_prompts`	List available user prompts.
`load_all`	Load all prompts from the filesystem directory.

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/filesystem_library.py

def __init__(
    self,
    prompt_dir: Union[str, Path],
    auto_load: bool = True,
    file_extensions: List[str] | None = None
):
    """Initialize filesystem prompt library.

    Args:
        prompt_dir: Root directory containing prompt files
        auto_load: Whether to automatically load all prompts on init (default: True)
        file_extensions: List of file extensions to load (default: [".yaml", ".yml", ".json"])
    """
    super().__init__()

    self.prompt_dir = Path(prompt_dir)
    self.file_extensions = file_extensions or [".yaml", ".yml", ".json"]

    # Validate directory exists
    if not self.prompt_dir.exists():
        raise ValueError(f"Prompt directory does not exist: {self.prompt_dir}")

    if not self.prompt_dir.is_dir():
        raise ValueError(f"Prompt path is not a directory: {self.prompt_dir}")

    # Auto-load prompts if requested
    if auto_load:
        self.load_all()

Functions¶

get_message_index ¶

get_message_index(name: str, **kwargs: Any) -> MessageIndex | None

Get a message index by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Message index name	required
`**kwargs`	`Any`	Additional arguments (unused in filesystem library)	`{}`

Returns:

Type	Description
`MessageIndex \| None`	MessageIndex if found, None otherwise

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/filesystem_library.py

def get_message_index(self, name: str, **kwargs: Any) -> MessageIndex | None:
    """Get a message index by name.

    Args:
        name: Message index name
        **kwargs: Additional arguments (unused in filesystem library)

    Returns:
        MessageIndex if found, None otherwise
    """
    return self._get_cached_message_index(name)

get_prompt_rag_configs ¶

get_prompt_rag_configs(
    prompt_name: str, prompt_type: str = "user", **kwargs: Any
) -> List[RAGConfig]

Get RAG configurations for a specific prompt.

Resolves both inline RAG configs and references to standalone configs.

Parameters:

Name	Type	Description	Default
`prompt_name`	`str`	Prompt name	required
`prompt_type`	`str`	Type of prompt ("user" or "system")	`'user'`
`**kwargs`	`Any`	Additional arguments (unused)	`{}`

Returns:

Type	Description
`List[RAGConfig]`	List of RAGConfig (empty if none defined)

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/filesystem_library.py

def get_prompt_rag_configs(
    self,
    prompt_name: str,
    prompt_type: str = "user",
    **kwargs: Any
) -> List[RAGConfig]:
    """Get RAG configurations for a specific prompt.

    Resolves both inline RAG configs and references to standalone configs.

    Args:
        prompt_name: Prompt name
        prompt_type: Type of prompt ("user" or "system")
        **kwargs: Additional arguments (unused)

    Returns:
        List of RAGConfig (empty if none defined)
    """
    # Get the prompt template
    if prompt_type == "system":
        template = self.get_system_prompt(prompt_name)
    else:
        template = self.get_user_prompt(prompt_name)

    if template is None:
        return []

    configs = []

    # Get inline RAG configs from template
    if "rag_configs" in template:
        configs.extend(template["rag_configs"])

    # Resolve RAG config references
    if "rag_config_refs" in template:
        for ref_name in template["rag_config_refs"]:
            ref_config = self.get_rag_config(ref_name)
            if ref_config:
                configs.append(ref_config)
            else:
                logger.warning(
                    f"RAG config reference '{ref_name}' not found "
                    f"for prompt '{prompt_name}'"
                )

    return configs

get_rag_config ¶

get_rag_config(name: str, **kwargs: Any) -> RAGConfig | None

Get a standalone RAG configuration by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	RAG config name	required
`**kwargs`	`Any`	Additional arguments (unused in filesystem library)	`{}`

Returns:

Type	Description
`RAGConfig \| None`	RAGConfig if found, None otherwise

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/filesystem_library.py

def get_rag_config(self, name: str, **kwargs: Any) -> RAGConfig | None:
    """Get a standalone RAG configuration by name.

    Args:
        name: RAG config name
        **kwargs: Additional arguments (unused in filesystem library)

    Returns:
        RAGConfig if found, None otherwise
    """
    return self._get_cached_rag_config(name)

get_system_prompt ¶

get_system_prompt(name: str, **kwargs: Any) -> PromptTemplateDict | None

Get a system prompt by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	System prompt name	required
`**kwargs`	`Any`	Additional arguments (unused in filesystem library)	`{}`

Returns:

Type	Description
`PromptTemplateDict \| None`	PromptTemplateDict if found, None otherwise

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/filesystem_library.py

def get_system_prompt(self, name: str, **kwargs: Any) -> PromptTemplateDict | None:
    """Get a system prompt by name.

    Args:
        name: System prompt name
        **kwargs: Additional arguments (unused in filesystem library)

    Returns:
        PromptTemplateDict if found, None otherwise
    """
    return self._get_cached_system_prompt(name)

get_user_prompt ¶

get_user_prompt(name: str, **kwargs: Any) -> PromptTemplateDict | None

Get a user prompt by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	User prompt name	required
`**kwargs`	`Any`	Additional arguments (unused in filesystem library)	`{}`

Returns:

Type	Description
`PromptTemplateDict \| None`	PromptTemplateDict if found, None otherwise

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/filesystem_library.py

def get_user_prompt(self, name: str, **kwargs: Any) -> PromptTemplateDict | None:
    """Get a user prompt by name.

    Args:
        name: User prompt name
        **kwargs: Additional arguments (unused in filesystem library)

    Returns:
        PromptTemplateDict if found, None otherwise
    """
    return self._get_cached_user_prompt(name)

list_message_indexes ¶

list_message_indexes() -> List[str]

List all available message index names.

Returns:

Type	Description
`List[str]`	List of message index identifiers

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/filesystem_library.py

def list_message_indexes(self) -> List[str]:
    """List all available message index names.

    Returns:
        List of message index identifiers
    """
    return list(self._message_index_cache.keys())

list_system_prompts ¶

list_system_prompts() -> List[str]

List all available system prompt names.

Returns:

Type	Description
`List[str]`	List of system prompt identifiers

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/filesystem_library.py

def list_system_prompts(self) -> List[str]:
    """List all available system prompt names.

    Returns:
        List of system prompt identifiers
    """
    return list(self._system_prompt_cache.keys())

list_user_prompts ¶

list_user_prompts() -> List[str]

List available user prompts.

Returns:

Type	Description
`List[str]`	List of user prompt names

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/filesystem_library.py

def list_user_prompts(self) -> List[str]:
    """List available user prompts.

    Returns:
        List of user prompt names
    """
    return list(self._user_prompt_cache.keys())

load_all ¶

load_all() -> None

Load all prompts from the filesystem directory.

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/filesystem_library.py

def load_all(self) -> None:
    """Load all prompts from the filesystem directory."""
    self._load_system_prompts()
    self._load_user_prompts()
    self._load_message_indexes()
    self._load_rag_configs()

ConfigPromptLibrary¶

dataknobs_llm.prompts.ConfigPromptLibrary ¶

ConfigPromptLibrary(config: Dict[str, Any] | None = None)

Bases: BasePromptLibrary

Prompt library that loads prompts from configuration dictionaries.

Features: - Direct in-memory configuration - No filesystem dependencies - Useful for testing and programmatic definition - Supports all prompt types (system, user, messages, RAG)

Example

config = { ... "system": {"greet": {"template": "Hello {{name}}!"}}, ... "user": {"ask": {"template": "Tell me about {{topic}}"}} ... } library = ConfigPromptLibrary(config) template = library.get_system_prompt("greet")

Initialize configuration-based prompt library.

Parameters:

Name	Type	Description	Default
`config`	`Dict[str, Any] \| None`	Configuration dictionary with structure: { "system": {name: PromptTemplateDict, ...}, "user": {name: PromptTemplateDict, ...}, "messages": {name: MessageIndex, ...}, "rag": {name: RAGConfig, ...} }	`None`

Methods:

Name	Description
`add_message_index`	Add or update a message index.
`add_rag_config`	Add or update a RAG configuration.
`add_system_prompt`	Add or update a system prompt.
`add_user_prompt`	Add or update a user prompt.
`get_message_index`	Get a message index by name.
`get_prompt_rag_configs`	Get RAG configurations for a specific prompt.
`get_rag_config`	Get a standalone RAG configuration by name.
`get_system_prompt`	Get a system prompt by name.
`get_user_prompt`	Get a user prompt by name.
`list_message_indexes`	List all available message index names.
`list_system_prompts`	List all available system prompt names.
`list_user_prompts`	List available user prompts.

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def __init__(self, config: Dict[str, Any] | None = None):
    """Initialize configuration-based prompt library.

    Args:
        config: Configuration dictionary with structure:
            {
                "system": {name: PromptTemplateDict, ...},
                "user": {name: PromptTemplateDict, ...},
                "messages": {name: MessageIndex, ...},
                "rag": {name: RAGConfig, ...}
            }
    """
    super().__init__()
    self._config = config or {}

    # Load prompts from config
    self._load_from_config()

Functions¶

add_message_index ¶

add_message_index(name: str, message_index: MessageIndex) -> None

Add or update a message index.

Parameters:

Name	Type	Description	Default
`name`	`str`	Message index name	required
`message_index`	`MessageIndex`	Message index to add	required

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def add_message_index(self, name: str, message_index: MessageIndex) -> None:
    """Add or update a message index.

    Args:
        name: Message index name
        message_index: Message index to add
    """
    self._cache_message_index(name, message_index)
    logger.debug(f"Added/updated message index: {name}")

add_rag_config ¶

add_rag_config(name: str, rag_config: RAGConfig) -> None

Add or update a RAG configuration.

Parameters:

Name	Type	Description	Default
`name`	`str`	RAG config name	required
`rag_config`	`RAGConfig`	RAG configuration to add	required

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def add_rag_config(self, name: str, rag_config: RAGConfig) -> None:
    """Add or update a RAG configuration.

    Args:
        name: RAG config name
        rag_config: RAG configuration to add
    """
    self._cache_rag_config(name, rag_config)
    logger.debug(f"Added/updated RAG config: {name}")

add_system_prompt ¶

add_system_prompt(name: str, template: PromptTemplateDict) -> None

Add or update a system prompt.

Parameters:

Name	Type	Description	Default
`name`	`str`	System prompt name	required
`template`	`PromptTemplateDict`	Prompt template to add	required

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def add_system_prompt(self, name: str, template: PromptTemplateDict) -> None:
    """Add or update a system prompt.

    Args:
        name: System prompt name
        template: Prompt template to add
    """
    self._cache_system_prompt(name, template)
    logger.debug(f"Added/updated system prompt: {name}")

add_user_prompt ¶

add_user_prompt(name: str, template: PromptTemplateDict) -> None

Add or update a user prompt.

Parameters:

Name	Type	Description	Default
`name`	`str`	User prompt name	required
`template`	`PromptTemplateDict`	Prompt template to add	required

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def add_user_prompt(self, name: str, template: PromptTemplateDict) -> None:
    """Add or update a user prompt.

    Args:
        name: User prompt name
        template: Prompt template to add
    """
    self._cache_user_prompt(name, template)
    logger.debug(f"Added/updated user prompt: {name}")

get_message_index ¶

get_message_index(name: str, **kwargs: Any) -> MessageIndex | None

Get a message index by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Message index name	required
`**kwargs`	`Any`	Additional arguments (unused)	`{}`

Returns:

Type	Description
`MessageIndex \| None`	MessageIndex if found, None otherwise

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def get_message_index(self, name: str, **kwargs: Any) -> MessageIndex | None:
    """Get a message index by name.

    Args:
        name: Message index name
        **kwargs: Additional arguments (unused)

    Returns:
        MessageIndex if found, None otherwise
    """
    return self._get_cached_message_index(name)

get_prompt_rag_configs ¶

get_prompt_rag_configs(
    prompt_name: str, prompt_type: str = "user", **kwargs: Any
) -> List[RAGConfig]

Get RAG configurations for a specific prompt.

Resolves both inline RAG configs and references to standalone configs.

Parameters:

Name	Type	Description	Default
`prompt_name`	`str`	Prompt name	required
`prompt_type`	`str`	Type of prompt ("user" or "system")	`'user'`
`**kwargs`	`Any`	Additional arguments (unused)	`{}`

Returns:

Type	Description
`List[RAGConfig]`	List of RAGConfig (empty if none defined)

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def get_prompt_rag_configs(
    self,
    prompt_name: str,
    prompt_type: str = "user",
    **kwargs: Any
) -> List[RAGConfig]:
    """Get RAG configurations for a specific prompt.

    Resolves both inline RAG configs and references to standalone configs.

    Args:
        prompt_name: Prompt name
        prompt_type: Type of prompt ("user" or "system")
        **kwargs: Additional arguments (unused)

    Returns:
        List of RAGConfig (empty if none defined)
    """
    # Get the prompt template
    if prompt_type == "system":
        template = self.get_system_prompt(prompt_name)
    else:
        template = self.get_user_prompt(prompt_name)

    if template is None:
        return []

    configs = []

    # Get inline RAG configs from template
    if "rag_configs" in template:
        configs.extend(template["rag_configs"])

    # Resolve RAG config references
    if "rag_config_refs" in template:
        for ref_name in template["rag_config_refs"]:
            ref_config = self.get_rag_config(ref_name)
            if ref_config:
                configs.append(ref_config)
            else:
                logger.warning(
                    f"RAG config reference '{ref_name}' not found "
                    f"for prompt '{prompt_name}'"
                )

    return configs

get_rag_config ¶

get_rag_config(name: str, **kwargs: Any) -> RAGConfig | None

Get a standalone RAG configuration by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	RAG config name	required
`**kwargs`	`Any`	Additional arguments (unused)	`{}`

Returns:

Type	Description
`RAGConfig \| None`	RAGConfig if found, None otherwise

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def get_rag_config(self, name: str, **kwargs: Any) -> RAGConfig | None:
    """Get a standalone RAG configuration by name.

    Args:
        name: RAG config name
        **kwargs: Additional arguments (unused)

    Returns:
        RAGConfig if found, None otherwise
    """
    return self._get_cached_rag_config(name)

get_system_prompt ¶

get_system_prompt(name: str, **kwargs: Any) -> PromptTemplateDict | None

Get a system prompt by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	System prompt name	required
`**kwargs`	`Any`	Additional arguments (unused)	`{}`

Returns:

Type	Description
`PromptTemplateDict \| None`	PromptTemplateDict if found, None otherwise

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def get_system_prompt(self, name: str, **kwargs: Any) -> PromptTemplateDict | None:
    """Get a system prompt by name.

    Args:
        name: System prompt name
        **kwargs: Additional arguments (unused)

    Returns:
        PromptTemplateDict if found, None otherwise
    """
    return self._get_cached_system_prompt(name)

get_user_prompt ¶

get_user_prompt(name: str, **kwargs: Any) -> PromptTemplateDict | None

Get a user prompt by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	User prompt name	required
`**kwargs`	`Any`	Additional arguments (unused)	`{}`

Returns:

Type	Description
`PromptTemplateDict \| None`	PromptTemplateDict if found, None otherwise

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def get_user_prompt(self, name: str, **kwargs: Any) -> PromptTemplateDict | None:
    """Get a user prompt by name.

    Args:
        name: User prompt name
        **kwargs: Additional arguments (unused)

    Returns:
        PromptTemplateDict if found, None otherwise
    """
    return self._get_cached_user_prompt(name)

list_message_indexes ¶

list_message_indexes() -> List[str]

List all available message index names.

Returns:

Type	Description
`List[str]`	List of message index identifiers

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def list_message_indexes(self) -> List[str]:
    """List all available message index names.

    Returns:
        List of message index identifiers
    """
    return list(self._message_index_cache.keys())

list_system_prompts ¶

list_system_prompts() -> List[str]

List all available system prompt names.

Returns:

Type	Description
`List[str]`	List of system prompt identifiers

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def list_system_prompts(self) -> List[str]:
    """List all available system prompt names.

    Returns:
        List of system prompt identifiers
    """
    return list(self._system_prompt_cache.keys())

list_user_prompts ¶

list_user_prompts() -> List[str]

List available user prompts.

Returns:

Type	Description
`List[str]`	List of user prompt names

Source code in packages/llm/src/dataknobs_llm/prompts/implementations/config_library.py

def list_user_prompts(self) -> List[str]:
    """List available user prompts.

    Returns:
        List of user prompt names
    """
    return list(self._user_prompt_cache.keys())

Prompt Builders¶

PromptBuilder (Sync)¶

dataknobs_llm.prompts.PromptBuilder ¶

PromptBuilder(
    library: AbstractPromptLibrary,
    adapters: Dict[str, ResourceAdapter] | None = None,
    default_validation: ValidationLevel = ValidationLevel.WARN,
    raise_on_rag_error: bool = False,
)

Bases: BasePromptBuilder

Synchronous prompt builder for constructing prompts with RAG and validation.

This class provides a high-level API for building prompts by: 1. Retrieving prompt templates from a library 2. Resolving parameters from adapters and runtime values 3. Executing RAG searches via adapters 4. Injecting RAG content into templates 5. Rendering final prompts with validation

Example

library = ConfigPromptLibrary(config) adapters = { ... 'config': DictResourceAdapter(config_data), ... 'docs': DataknobsBackendAdapter(docs_db) ... } builder = PromptBuilder(library=library, adapters=adapters)

Render a system prompt¶

result = builder.render_system_prompt( ... 'analyze_code', ... params={'code': code_snippet, 'language': 'python'} ... )

Initialize the synchronous prompt builder.

Parameters:

Name	Type	Description	Default
`library`	`AbstractPromptLibrary`	Prompt library to retrieve templates from	required
`adapters`	`Dict[str, ResourceAdapter] \| None`	Dictionary of named resource adapters for parameter resolution and RAG searches	`None`
`default_validation`	`ValidationLevel`	Default validation level for templates without explicit validation configuration	`WARN`
`raise_on_rag_error`	`bool`	If True, raise exceptions on RAG failures; if False (default), log warning and continue	`False`

Raises:

Type	Description
`TypeError`	If any adapter is async (use AsyncPromptBuilder instead)

Methods:

Name	Description
`render_system_prompt`	Render a system prompt with parameters and optional RAG content.
`render_user_prompt`	Render a user prompt with parameters and optional RAG content.

Source code in packages/llm/src/dataknobs_llm/prompts/builders/prompt_builder.py

def __init__(
    self,
    library: AbstractPromptLibrary,
    adapters: Dict[str, ResourceAdapter] | None = None,
    default_validation: ValidationLevel = ValidationLevel.WARN,
    raise_on_rag_error: bool = False
):
    """Initialize the synchronous prompt builder.

    Args:
        library: Prompt library to retrieve templates from
        adapters: Dictionary of named resource adapters for parameter
                 resolution and RAG searches
        default_validation: Default validation level for templates without
                          explicit validation configuration
        raise_on_rag_error: If True, raise exceptions on RAG failures;
                          if False (default), log warning and continue

    Raises:
        TypeError: If any adapter is async (use AsyncPromptBuilder instead)
    """
    super().__init__(library, adapters, default_validation, raise_on_rag_error)
    self._validate_adapters()

Functions¶

render_system_prompt ¶

render_system_prompt(
    name: str,
    params: Dict[str, Any] | None = None,
    include_rag: bool = True,
    validation_override: ValidationLevel | None = None,
    return_rag_metadata: bool = False,
    cached_rag: Dict[str, Any] | None = None,
    **kwargs: Any,
) -> RenderResult

Render a system prompt with parameters and optional RAG content.

Parameters:

Name	Type	Description	Default
`name`	`str`	System prompt identifier	required
`params`	`Dict[str, Any] \| None`	Runtime parameters to use in rendering	`None`
`include_rag`	`bool`	Whether to include RAG content (default: True)	`True`
`validation_override`	`ValidationLevel \| None`	Override validation level for this render	`None`
`return_rag_metadata`	`bool`	If True, attach RAG metadata to result	`False`
`cached_rag`	`Dict[str, Any] \| None`	If provided, use these cached RAG results instead of executing new searches	`None`
`**kwargs`	`Any`	Additional parameters passed to library	`{}`

Returns:

Type	Description
`RenderResult`	RenderResult with rendered content and metadata

Raises:

Type	Description
`ValueError`	If prompt not found or validation fails

Example

Capture RAG metadata¶

result = builder.render_system_prompt( ... 'code_question', ... params={'language': 'python'}, ... return_rag_metadata=True ... ) print(result.rag_metadata)

Reuse cached RAG¶

result2 = builder.render_system_prompt( ... 'code_question', ... params={'language': 'python'}, ... cached_rag=result.rag_metadata ... )

Source code in packages/llm/src/dataknobs_llm/prompts/builders/prompt_builder.py

def render_system_prompt(
    self,
    name: str,
    params: Dict[str, Any] | None = None,
    include_rag: bool = True,
    validation_override: ValidationLevel | None = None,
    return_rag_metadata: bool = False,
    cached_rag: Dict[str, Any] | None = None,
    **kwargs: Any
) -> RenderResult:
    """Render a system prompt with parameters and optional RAG content.

    Args:
        name: System prompt identifier
        params: Runtime parameters to use in rendering
        include_rag: Whether to include RAG content (default: True)
        validation_override: Override validation level for this render
        return_rag_metadata: If True, attach RAG metadata to result
        cached_rag: If provided, use these cached RAG results instead
                   of executing new searches
        **kwargs: Additional parameters passed to library

    Returns:
        RenderResult with rendered content and metadata

    Raises:
        ValueError: If prompt not found or validation fails

    Example:
        >>> # Capture RAG metadata
        >>> result = builder.render_system_prompt(
        ...     'code_question',
        ...     params={'language': 'python'},
        ...     return_rag_metadata=True
        ... )
        >>> print(result.rag_metadata)
        >>>
        >>> # Reuse cached RAG
        >>> result2 = builder.render_system_prompt(
        ...     'code_question',
        ...     params={'language': 'python'},
        ...     cached_rag=result.rag_metadata
        ... )
    """
    params = params or {}

    # Retrieve template from library
    template_dict = self.library.get_system_prompt(name, **kwargs)
    if template_dict is None:
        raise ValueError(f"System prompt not found: {name}")

    # Render the prompt
    return self._render_prompt_impl(
        prompt_name=name,
        prompt_type="system",
        template_dict=template_dict,
        runtime_params=params,
        include_rag=include_rag,
        validation_override=validation_override,
        return_rag_metadata=return_rag_metadata,
        cached_rag=cached_rag,
        **kwargs
    )

render_user_prompt ¶

render_user_prompt(
    name: str,
    params: Dict[str, Any] | None = None,
    include_rag: bool = True,
    validation_override: ValidationLevel | None = None,
    return_rag_metadata: bool = False,
    cached_rag: Dict[str, Any] | None = None,
    **kwargs: Any,
) -> RenderResult

Render a user prompt with parameters and optional RAG content.

Parameters:

Name	Type	Description	Default
`name`	`str`	User prompt identifier	required
`params`	`Dict[str, Any] \| None`	Runtime parameters to use in rendering	`None`
`include_rag`	`bool`	Whether to include RAG content (default: True)	`True`
`validation_override`	`ValidationLevel \| None`	Override validation level for this render	`None`
`return_rag_metadata`	`bool`	If True, attach RAG metadata to result	`False`
`cached_rag`	`Dict[str, Any] \| None`	If provided, use these cached RAG results instead of executing new searches	`None`
`**kwargs`	`Any`	Additional parameters passed to library	`{}`

Returns:

Type	Description
`RenderResult`	RenderResult with rendered content and metadata

Raises:

Type	Description
`ValueError`	If prompt not found or validation fails

Source code in packages/llm/src/dataknobs_llm/prompts/builders/prompt_builder.py

def render_user_prompt(
    self,
    name: str,
    params: Dict[str, Any] | None = None,
    include_rag: bool = True,
    validation_override: ValidationLevel | None = None,
    return_rag_metadata: bool = False,
    cached_rag: Dict[str, Any] | None = None,
    **kwargs: Any
) -> RenderResult:
    """Render a user prompt with parameters and optional RAG content.

    Args:
        name: User prompt identifier
        params: Runtime parameters to use in rendering
        include_rag: Whether to include RAG content (default: True)
        validation_override: Override validation level for this render
        return_rag_metadata: If True, attach RAG metadata to result
        cached_rag: If provided, use these cached RAG results instead
                   of executing new searches
        **kwargs: Additional parameters passed to library

    Returns:
        RenderResult with rendered content and metadata

    Raises:
        ValueError: If prompt not found or validation fails
    """
    params = params or {}

    # Retrieve template from library
    template_dict = self.library.get_user_prompt(name, **kwargs)
    if template_dict is None:
        raise ValueError(f"User prompt not found: {name}")

    # Render the prompt
    return self._render_prompt_impl(
        prompt_name=name,
        prompt_type="user",
        template_dict=template_dict,
        runtime_params=params,
        include_rag=include_rag,
        validation_override=validation_override,
        return_rag_metadata=return_rag_metadata,
        cached_rag=cached_rag,
        **kwargs
    )

AsyncPromptBuilder¶

dataknobs_llm.prompts.AsyncPromptBuilder ¶

AsyncPromptBuilder(
    library: AbstractPromptLibrary,
    adapters: Dict[str, AsyncResourceAdapter] | None = None,
    default_validation: ValidationLevel = ValidationLevel.WARN,
    raise_on_rag_error: bool = False,
)

Bases: BasePromptBuilder

Asynchronous prompt builder for constructing prompts with RAG and validation.

This class provides a high-level async API for building prompts by: 1. Retrieving prompt templates from a library 2. Resolving parameters from adapters and runtime values (concurrently) 3. Executing RAG searches via adapters (in parallel) 4. Injecting RAG content into templates 5. Rendering final prompts with validation

Example

library = ConfigPromptLibrary(config) adapters = { ... 'config': AsyncDictResourceAdapter(config_data), ... 'docs': AsyncDataknobsBackendAdapter(docs_db) ... } builder = AsyncPromptBuilder(library=library, adapters=adapters)

Render a system prompt¶

result = await builder.render_system_prompt( ... 'analyze_code', ... params={'code': code_snippet, 'language': 'python'} ... )

Initialize the asynchronous prompt builder.

Parameters:

Name	Type	Description	Default
`library`	`AbstractPromptLibrary`	Prompt library to retrieve templates from	required
`adapters`	`Dict[str, AsyncResourceAdapter] \| None`	Dictionary of named async resource adapters for parameter resolution and RAG searches	`None`
`default_validation`	`ValidationLevel`	Default validation level for templates without explicit validation configuration	`WARN`
`raise_on_rag_error`	`bool`	If True, raise exceptions on RAG failures; if False (default), log warning and continue	`False`

Raises:

Type	Description
`TypeError`	If any adapter is sync (use PromptBuilder instead)

Methods:

Name	Description
`render_system_prompt`	Render a system prompt with parameters and optional RAG content.
`render_user_prompt`	Render a user prompt with parameters and optional RAG content.

Source code in packages/llm/src/dataknobs_llm/prompts/builders/async_prompt_builder.py

def __init__(
    self,
    library: AbstractPromptLibrary,
    adapters: Dict[str, AsyncResourceAdapter] | None = None,
    default_validation: ValidationLevel = ValidationLevel.WARN,
    raise_on_rag_error: bool = False
):
    """Initialize the asynchronous prompt builder.

    Args:
        library: Prompt library to retrieve templates from
        adapters: Dictionary of named async resource adapters for parameter
                 resolution and RAG searches
        default_validation: Default validation level for templates without
                          explicit validation configuration
        raise_on_rag_error: If True, raise exceptions on RAG failures;
                          if False (default), log warning and continue

    Raises:
        TypeError: If any adapter is sync (use PromptBuilder instead)
    """
    super().__init__(library, adapters, default_validation, raise_on_rag_error)
    self._validate_adapters()

Functions¶

render_system_prompt `async` ¶

render_system_prompt(
    name: str,
    params: Dict[str, Any] | None = None,
    include_rag: bool = True,
    validation_override: ValidationLevel | None = None,
    return_rag_metadata: bool = False,
    cached_rag: Dict[str, Any] | None = None,
    **kwargs: Any,
) -> RenderResult

Render a system prompt with parameters and optional RAG content.

Parameters:

Name	Type	Description	Default
`name`	`str`	System prompt identifier	required
`params`	`Dict[str, Any] \| None`	Runtime parameters to use in rendering	`None`
`include_rag`	`bool`	Whether to include RAG content (default: True)	`True`
`validation_override`	`ValidationLevel \| None`	Override validation level for this render	`None`
`return_rag_metadata`	`bool`	If True, attach RAG metadata to result	`False`
`cached_rag`	`Dict[str, Any] \| None`	If provided, use these cached RAG results instead of executing new searches	`None`
`**kwargs`	`Any`	Additional parameters passed to library	`{}`

Returns:

Type	Description
`RenderResult`	RenderResult with rendered content and metadata

Raises:

Type	Description
`ValueError`	If prompt not found or validation fails

Example

Capture RAG metadata¶

result = await builder.render_system_prompt( ... 'code_question', ... params={'language': 'python'}, ... return_rag_metadata=True ... ) print(result.rag_metadata)

Reuse cached RAG¶

result2 = await builder.render_system_prompt( ... 'code_question', ... params={'language': 'python'}, ... cached_rag=result.rag_metadata ... )

Source code in packages/llm/src/dataknobs_llm/prompts/builders/async_prompt_builder.py

async def render_system_prompt(
    self,
    name: str,
    params: Dict[str, Any] | None = None,
    include_rag: bool = True,
    validation_override: ValidationLevel | None = None,
    return_rag_metadata: bool = False,
    cached_rag: Dict[str, Any] | None = None,
    **kwargs: Any
) -> RenderResult:
    """Render a system prompt with parameters and optional RAG content.

    Args:
        name: System prompt identifier
        params: Runtime parameters to use in rendering
        include_rag: Whether to include RAG content (default: True)
        validation_override: Override validation level for this render
        return_rag_metadata: If True, attach RAG metadata to result
        cached_rag: If provided, use these cached RAG results instead
                   of executing new searches
        **kwargs: Additional parameters passed to library

    Returns:
        RenderResult with rendered content and metadata

    Raises:
        ValueError: If prompt not found or validation fails

    Example:
        >>> # Capture RAG metadata
        >>> result = await builder.render_system_prompt(
        ...     'code_question',
        ...     params={'language': 'python'},
        ...     return_rag_metadata=True
        ... )
        >>> print(result.rag_metadata)
        >>>
        >>> # Reuse cached RAG
        >>> result2 = await builder.render_system_prompt(
        ...     'code_question',
        ...     params={'language': 'python'},
        ...     cached_rag=result.rag_metadata
        ... )
    """
    params = params or {}

    # Retrieve template from library
    template_dict = self.library.get_system_prompt(name, **kwargs)
    if template_dict is None:
        raise ValueError(f"System prompt not found: {name}")

    # Render the prompt
    return await self._render_prompt_impl(
        prompt_name=name,
        prompt_type="system",
        template_dict=template_dict,
        runtime_params=params,
        include_rag=include_rag,
        validation_override=validation_override,
        return_rag_metadata=return_rag_metadata,
        cached_rag=cached_rag,
        **kwargs
    )

render_user_prompt `async` ¶

render_user_prompt(
    name: str,
    params: Dict[str, Any] | None = None,
    include_rag: bool = True,
    validation_override: ValidationLevel | None = None,
    return_rag_metadata: bool = False,
    cached_rag: Dict[str, Any] | None = None,
    **kwargs: Any,
) -> RenderResult

Render a user prompt with parameters and optional RAG content.

Parameters:

Name	Type	Description	Default
`name`	`str`	User prompt identifier	required
`params`	`Dict[str, Any] \| None`	Runtime parameters to use in rendering	`None`
`include_rag`	`bool`	Whether to include RAG content (default: True)	`True`
`validation_override`	`ValidationLevel \| None`	Override validation level for this render	`None`
`return_rag_metadata`	`bool`	If True, attach RAG metadata to result	`False`
`cached_rag`	`Dict[str, Any] \| None`	If provided, use these cached RAG results instead of executing new searches	`None`
`**kwargs`	`Any`	Additional parameters passed to library	`{}`

Returns:

Type	Description
`RenderResult`	RenderResult with rendered content and metadata

Raises:

Type	Description
`ValueError`	If prompt not found or validation fails

Source code in packages/llm/src/dataknobs_llm/prompts/builders/async_prompt_builder.py

async def render_user_prompt(
    self,
    name: str,
    params: Dict[str, Any] | None = None,
    include_rag: bool = True,
    validation_override: ValidationLevel | None = None,
    return_rag_metadata: bool = False,
    cached_rag: Dict[str, Any] | None = None,
    **kwargs: Any
) -> RenderResult:
    """Render a user prompt with parameters and optional RAG content.

    Args:
        name: User prompt identifier
        params: Runtime parameters to use in rendering
        include_rag: Whether to include RAG content (default: True)
        validation_override: Override validation level for this render
        return_rag_metadata: If True, attach RAG metadata to result
        cached_rag: If provided, use these cached RAG results instead
                   of executing new searches
        **kwargs: Additional parameters passed to library

    Returns:
        RenderResult with rendered content and metadata

    Raises:
        ValueError: If prompt not found or validation fails
    """
    params = params or {}

    # Retrieve template from library
    template_dict = self.library.get_user_prompt(name, **kwargs)
    if template_dict is None:
        raise ValueError(f"User prompt not found: {name}")

    # Render the prompt
    return await self._render_prompt_impl(
        prompt_name=name,
        prompt_type="user",
        template_dict=template_dict,
        runtime_params=params,
        include_rag=include_rag,
        validation_override=validation_override,
        return_rag_metadata=return_rag_metadata,
        cached_rag=cached_rag,
        **kwargs
    )

Resource Adapters¶

Resource adapters provide data for RAG and template variables.

DictResourceAdapter¶

dataknobs_llm.prompts.DictResourceAdapter ¶

DictResourceAdapter(
    data: Dict[str, Any],
    name: str = "dict_adapter",
    case_sensitive: bool = False,
)

Bases: ResourceAdapter

Synchronous adapter for Python dictionary resources.

Features: - Nested key access using dot notation (e.g., "user.name") - Simple text-based search across values - Optional case-insensitive search - Filtering and deduplication via BaseSearchLogic

Example

data = { ... "user": {"name": "Alice", "age": 30}, ... "settings": {"theme": "dark"} ... } adapter = DictResourceAdapter(data, name="config") adapter.get_value("user.name") "Alice" adapter.search("Alice") [{'content': "Alice", 'key': "user.name", 'score': 1.0}]

Initialize dictionary adapter.

Parameters:

Name	Type	Description	Default
`data`	`Dict[str, Any]`	Dictionary to wrap as a resource	required
`name`	`str`	Name identifier for this adapter	`'dict_adapter'`
`case_sensitive`	`bool`	Whether search should be case-sensitive (default: False)	`False`

Methods:

Name	Description
`get_value`	Retrieve a value by key from the dictionary.
`search`	Perform text-based search across dictionary values.

Source code in packages/llm/src/dataknobs_llm/prompts/adapters/dict_adapter.py

def __init__(
    self,
    data: Dict[str, Any],
    name: str = "dict_adapter",
    case_sensitive: bool = False
):
    """Initialize dictionary adapter.

    Args:
        data: Dictionary to wrap as a resource
        name: Name identifier for this adapter
        case_sensitive: Whether search should be case-sensitive (default: False)
    """
    super().__init__(name=name)
    self._data = data
    self._case_sensitive = case_sensitive

Functions¶

get_value ¶

get_value(
    key: str, default: Any = None, context: Dict[str, Any] | None = None
) -> Any

Retrieve a value by key from the dictionary.

Supports nested key access using dot notation. Dot-separated keys traverse nested dictionaries (e.g., a.b.c accesses nested values).

Parameters:

Name	Type	Description	Default
`key`	`str`	Key to look up (supports dot notation for nested access)	required
`default`	`Any`	Value to return if key is not found	`None`
`context`	`Dict[str, Any] \| None`	Optional context (unused in dict adapter)	`None`

Returns:

Type	Description
`Any`	Value at the key, or default if not found

Source code in packages/llm/src/dataknobs_llm/prompts/adapters/dict_adapter.py

def get_value(
    self,
    key: str,
    default: Any = None,
    context: Dict[str, Any] | None = None
) -> Any:
    """Retrieve a value by key from the dictionary.

    Supports nested key access using dot notation. Dot-separated keys
    traverse nested dictionaries (e.g., a.b.c accesses nested values).

    Args:
        key: Key to look up (supports dot notation for nested access)
        default: Value to return if key is not found
        context: Optional context (unused in dict adapter)

    Returns:
        Value at the key, or default if not found
    """
    # Handle dot notation for nested keys
    if '.' in key:
        parts = key.split('.')
        value = self._data
        for part in parts:
            if isinstance(value, dict) and part in value:
                value = value[part]
            else:
                return default
        return value
    else:
        return self._data.get(key, default)

search ¶

search(
    query: str, k: int = 5, filters: Dict[str, Any] | None = None, **kwargs: Any
) -> List[Dict[str, Any]]

Perform text-based search across dictionary values.

Searches through all values in the dictionary (including nested values) and returns items where the query string appears in the value.

Parameters:

Name	Type	Description	Default
`query`	`str`	Search query string	required
`k`	`int`	Maximum number of results to return	`5`
`filters`	`Dict[str, Any] \| None`	Optional filters to apply (passed to BaseSearchLogic)	`None`
`**kwargs`	`Any`	Additional search options: - min_score: Minimum score threshold (default: 0.0) - deduplicate: Whether to deduplicate results (default: False)	`{}`

Returns:

Type	Description
`List[Dict[str, Any]]`	List of search results with structure:
`List[Dict[str, Any]]`	{ 'content': , 'key': , 'score': , 'metadata': {}
`List[Dict[str, Any]]`	}

Source code in packages/llm/src/dataknobs_llm/prompts/adapters/dict_adapter.py

def search(
    self,
    query: str,
    k: int = 5,
    filters: Dict[str, Any] | None = None,
    **kwargs: Any
) -> List[Dict[str, Any]]:
    """Perform text-based search across dictionary values.

    Searches through all values in the dictionary (including nested values)
    and returns items where the query string appears in the value.

    Args:
        query: Search query string
        k: Maximum number of results to return
        filters: Optional filters to apply (passed to BaseSearchLogic)
        **kwargs: Additional search options:
            - min_score: Minimum score threshold (default: 0.0)
            - deduplicate: Whether to deduplicate results (default: False)

    Returns:
        List of search results with structure:
        {
            'content': <value>,
            'key': <key path>,
            'score': <relevance score>,
            'metadata': {<additional metadata>}
        }
    """
    results = []

    # Normalize query for case-insensitive search
    search_query = query if self._case_sensitive else query.lower()

    # Flatten dictionary and search
    for key, value in self._flatten_dict(self._data).items():
        value_str = str(value)
        search_value = value_str if self._case_sensitive else value_str.lower()

        if search_query in search_value:
            # Simple scoring: exact match = 1.0, contains = 0.8
            score = 1.0 if search_query == search_value else 0.8

            result = BaseSearchLogic.format_search_result(
                value_str,
                score=score,
                metadata={"key": key}
            )
            result["key"] = key  # Add key to top level for easier access
            results.append(result)

            if len(results) >= k:
                break

    # Apply filters if provided
    if filters:
        results = BaseSearchLogic.filter_results(results, filters=filters)

    # Apply min_score filter if provided
    min_score = kwargs.get('min_score', 0.0)
    if min_score > 0:
        results = BaseSearchLogic.filter_results(results, min_score=min_score)

    # Deduplicate if requested
    if kwargs.get('deduplicate', False):
        results = BaseSearchLogic.deduplicate_results(results, key='content')

    return results[:k]

InMemoryAdapter¶

dataknobs_llm.prompts.InMemoryAdapter ¶

InMemoryAdapter(
    search_results: List[Dict[str, Any]] | None = None,
    data: Dict[str, Any] | None = None,
    name: str = "inmemory",
)

Bases: InMemoryAdapterBase, ResourceAdapter

Synchronous in-memory adapter with predefined search results.

This adapter is designed for testing, demos, and examples. It returns predefined search results and optionally stores key-value pairs.

Features: - Predefined search results for predictable testing - Optional key-value storage via get_value() - Simple and fast (no external I/O) - Search count tracking for testing

Example

Simple usage with search results¶

adapter = InMemoryAdapter( ... search_results=[ ... {'content': "Python is a programming language", 'score': 0.9}, ... {'content': "Python was created by Guido van Rossum", 'score': 0.8} ... ], ... name="docs" ... ) results = adapter.search("python") len(results) 2 first_result = results[0] first_result.get('content') 'Python is a programming language' adapter.search_count # Track how many times search was called 1

With key-value storage¶

adapter = InMemoryAdapter( ... data={'language': "Python", 'version': "3.11"}, ... name="config" ... ) adapter.get_value('language') 'Python'

Initialize synchronous in-memory adapter.

Parameters:

Name	Type	Description	Default
`search_results`	`List[Dict[str, Any]] \| None`	List of results to return from search()	`None`
`data`	`Dict[str, Any] \| None`	Optional dictionary for key-value storage	`None`
`name`	`str`	Name identifier for this adapter	`'inmemory'`

Methods:

Name	Description
`get_value`	Retrieve a value by key from the in-memory data.
`search`	Return predefined search results.

Source code in packages/llm/src/dataknobs_llm/prompts/adapters/inmemory_adapter.py

def __init__(
    self,
    search_results: List[Dict[str, Any]] | None = None,
    data: Dict[str, Any] | None = None,
    name: str = "inmemory"
):
    """Initialize synchronous in-memory adapter.

    Args:
        search_results: List of results to return from search()
        data: Optional dictionary for key-value storage
        name: Name identifier for this adapter
    """
    super().__init__(search_results=search_results, data=data, name=name)

Functions¶

get_value ¶

get_value(
    key: str, default: Any = None, context: Dict[str, Any] | None = None
) -> Any

Retrieve a value by key from the in-memory data.

Parameters:

Name	Type	Description	Default
`key`	`str`	Key to look up	required
`default`	`Any`	Value to return if key is not found	`None`
`context`	`Dict[str, Any] \| None`	Optional context (unused in in-memory adapter)	`None`

Returns:

Type	Description
`Any`	Value at the key, or default if not found

Source code in packages/llm/src/dataknobs_llm/prompts/adapters/inmemory_adapter.py

def get_value(
    self,
    key: str,
    default: Any = None,
    context: Dict[str, Any] | None = None
) -> Any:
    """Retrieve a value by key from the in-memory data.

    Args:
        key: Key to look up
        default: Value to return if key is not found
        context: Optional context (unused in in-memory adapter)

    Returns:
        Value at the key, or default if not found
    """
    return self._get_value_impl(key, default, context)

search ¶

search(
    query: str, k: int = 5, filters: Dict[str, Any] | None = None, **kwargs: Any
) -> List[Dict[str, Any]]

Return predefined search results.

Parameters:

Name	Type	Description	Default
`query`	`str`	Search query (unused - returns all configured results)	required
`k`	`int`	Maximum number of results to return	`5`
`filters`	`Dict[str, Any] \| None`	Optional filters to apply via BaseSearchLogic	`None`
`**kwargs`	`Any`	Additional options: - min_score: Minimum score threshold	`{}`

Returns:

Type	Description
`List[Dict[str, Any]]`	List of search results (up to k items)

Source code in packages/llm/src/dataknobs_llm/prompts/adapters/inmemory_adapter.py

def search(
    self,
    query: str,
    k: int = 5,
    filters: Dict[str, Any] | None = None,
    **kwargs: Any
) -> List[Dict[str, Any]]:
    """Return predefined search results.

    Args:
        query: Search query (unused - returns all configured results)
        k: Maximum number of results to return
        filters: Optional filters to apply via BaseSearchLogic
        **kwargs: Additional options:
            - min_score: Minimum score threshold

    Returns:
        List of search results (up to k items)
    """
    return self._search_impl(query, k, filters, **kwargs)

DataknobsBackendAdapter¶

dataknobs_llm.prompts.DataknobsBackendAdapter ¶

DataknobsBackendAdapter(
    database: SyncDatabase,
    name: str = "dataknobs_backend",
    text_field: str = "content",
    metadata_field: str | None = None,
)

Bases: ResourceAdapter

Synchronous adapter for dataknobs database backends.

Wraps a dataknobs SyncDatabase instance to provide resource adapter functionality.

Features: - Record retrieval by ID using database.read() - Search using database.search() with Query objects - Field extraction with dot-notation support - Score-based ranking from search results

Example

from dataknobs_data.backends import SyncMemoryDatabase db = SyncMemoryDatabase()

... populate database ...¶

adapter = DataknobsBackendAdapter(db, name="memory") record = adapter.get_value("record_id_123") results = adapter.search("query text")

Initialize dataknobs backend adapter.

Parameters:

Name	Type	Description	Default
`database`	`SyncDatabase`	SyncDatabase instance to wrap	required
`name`	`str`	Name identifier for this adapter	`'dataknobs_backend'`
`text_field`	`str`	Field name to use as primary content (default: "content")	`'content'`
`metadata_field`	`str \| None`	Optional field to extract as metadata	`None`

Methods:

Name	Description
`get_value`	Retrieve a record or field value by ID.
`search`	Perform search using database backend.

Source code in packages/llm/src/dataknobs_llm/prompts/adapters/dataknobs_backend_adapter.py

def __init__(
    self,
    database: "SyncDatabase",
    name: str = "dataknobs_backend",
    text_field: str = "content",
    metadata_field: str | None = None
):
    """Initialize dataknobs backend adapter.

    Args:
        database: SyncDatabase instance to wrap
        name: Name identifier for this adapter
        text_field: Field name to use as primary content (default: "content")
        metadata_field: Optional field to extract as metadata
    """
    super().__init__(name=name)
    self._database = database
    self._text_field = text_field
    self._metadata_field = metadata_field

Functions¶

get_value ¶

get_value(
    key: str, default: Any = None, context: Dict[str, Any] | None = None
) -> Any

Retrieve a record or field value by ID.

Supports field extraction using dot notation: - Simple key: Returns entire record as dict - "record_id.field_name": Returns specific field value - "record_id.field.nested": Returns nested field value

Parameters:

Name	Type	Description	Default
`key`	`str`	Record ID or "record_id.field" notation	required
`default`	`Any`	Value to return if record/field not found	`None`
`context`	`Dict[str, Any] \| None`	Optional context with additional parameters	`None`

Returns:

Type	Description
`Any`	Record dict, field value, or default if not found

Source code in packages/llm/src/dataknobs_llm/prompts/adapters/dataknobs_backend_adapter.py

def get_value(
    self,
    key: str,
    default: Any = None,
    context: Dict[str, Any] | None = None
) -> Any:
    """Retrieve a record or field value by ID.

    Supports field extraction using dot notation:
    - Simple key: Returns entire record as dict
    - "record_id.field_name": Returns specific field value
    - "record_id.field.nested": Returns nested field value

    Args:
        key: Record ID or "record_id.field" notation
        default: Value to return if record/field not found
        context: Optional context with additional parameters

    Returns:
        Record dict, field value, or default if not found
    """
    try:
        # Parse key for potential field extraction
        if '.' in key:
            parts = key.split('.', 1)
            record_id = parts[0]
            field_path = parts[1]
        else:
            record_id = key
            field_path = None

        # Read record from database
        record = self._database.read(record_id)

        if record is None:
            return default

        # Extract field if specified
        if field_path:
            return record.get_value(field_path, default=default)
        else:
            # Return full record as dict
            return record.to_dict(include_metadata=True)

    except Exception:
        # Log error if needed, return default
        return default

search ¶

search(
    query: str, k: int = 5, filters: Dict[str, Any] | None = None, **kwargs: Any
) -> List[Dict[str, Any]]

Perform search using database backend.

Creates a Query object with LIKE filter for text search. Results are formatted according to BaseSearchLogic standards.

Parameters:

Name	Type	Description	Default
`query`	`str`	Search query string (searches text_field using LIKE)	required
`k`	`int`	Maximum number of results to return	`5`
`filters`	`Dict[str, Any] \| None`	Optional additional filters for the search	`None`
`**kwargs`	`Any`	Additional search options: - min_score: Minimum relevance score (default: 0.0) - deduplicate: Whether to deduplicate results (default: False)	`{}`

Returns:

Type	Description
`List[Dict[str, Any]]`	List of search results with structure:
`List[Dict[str, Any]]`	{ "content": , "score": , "metadata": {}
`List[Dict[str, Any]]`	}

Source code in packages/llm/src/dataknobs_llm/prompts/adapters/dataknobs_backend_adapter.py

def search(
    self,
    query: str,
    k: int = 5,
    filters: Dict[str, Any] | None = None,
    **kwargs: Any
) -> List[Dict[str, Any]]:
    """Perform search using database backend.

    Creates a Query object with LIKE filter for text search.
    Results are formatted according to BaseSearchLogic standards.

    Args:
        query: Search query string (searches text_field using LIKE)
        k: Maximum number of results to return
        filters: Optional additional filters for the search
        **kwargs: Additional search options:
            - min_score: Minimum relevance score (default: 0.0)
            - deduplicate: Whether to deduplicate results (default: False)

    Returns:
        List of search results with structure:
        {
            "content": <text content>,
            "score": <relevance score>,
            "metadata": {<record metadata>}
        }
    """
    try:
        from dataknobs_data.query import Query, Filter, Operator

        # Build filter for text search using LIKE operator
        # This searches for the query string anywhere in the text field
        search_filter = Filter(
            field=self._text_field,
            operator=Operator.LIKE,
            value=f"%{query}%"
        )

        # Build query object with filter and limit
        query_obj = Query(
            filters=[search_filter],
            limit_value=k
        )

        # Execute search
        records = self._database.search(query_obj)

        # Format results
        results = []
        for record in records:
            # Extract content field
            content = record.get_value(self._text_field, default="")

            # Get score from metadata if available
            score = record.metadata.get("score", record.metadata.get("_score", 1.0))

            # Extract metadata
            metadata = {}
            if self._metadata_field:
                metadata_value = record.get_value(self._metadata_field)
                if metadata_value is not None:
                    metadata["metadata_field"] = metadata_value

            # Add record ID and other metadata
            if hasattr(record, 'storage_id') and record.storage_id:
                metadata["record_id"] = record.storage_id

            # Merge with record metadata
            metadata.update(record.metadata)

            # Format result
            result = BaseSearchLogic.format_search_result(
                content,
                score=score,
                metadata=metadata
            )
            results.append(result)

        # Apply filters if provided
        if filters:
            results = BaseSearchLogic.filter_results(results, filters=filters)

        # Apply min_score filter
        min_score = kwargs.get('min_score', 0.0)
        if min_score > 0:
            results = BaseSearchLogic.filter_results(results, min_score=min_score)

        # Deduplicate if requested
        if kwargs.get('deduplicate', False):
            results = BaseSearchLogic.deduplicate_results(results, key='content')

        return results[:k]

    except Exception:
        # Log error if needed
        return []

Validation¶

ValidationLevel¶

dataknobs_llm.prompts.ValidationLevel ¶

Bases: Enum

Validation strictness levels for template rendering.

Attributes:

Name	Type	Description
`ERROR`		Raise exception for missing required parameters
`WARN`		Log warning for missing required parameters (default)
`IGNORE`		Silently ignore missing required parameters

ValidationConfig¶

dataknobs_llm.prompts.ValidationConfig `dataclass` ¶

ValidationConfig(
    level: ValidationLevel | None = None,
    required_params: List[str] | None = None,
    optional_params: List[str] | None = None,
)

Configuration for template parameter validation.

Attributes:

Name	Type	Description
`level`	`ValidationLevel \| None`	Validation strictness level (None to inherit from context)
`required_params`	`set[str]`	Set of parameter names that must be provided
`optional_params`	`set[str]`	Set of parameter names that are optional

Initialize validation configuration.

Parameters:

Name	Type	Description	Default
`level`	`ValidationLevel \| None`	Validation strictness level (None to inherit from context)	`None`
`required_params`	`List[str] \| None`	List of required parameter names	`None`
`optional_params`	`List[str] \| None`	List of optional parameter names	`None`

Source code in packages/llm/src/dataknobs_llm/prompts/base/types.py

def __init__(
    self,
    level: ValidationLevel | None = None,
    required_params: List[str] | None = None,
    optional_params: List[str] | None = None
):
    """Initialize validation configuration.

    Args:
        level: Validation strictness level (None to inherit from context)
        required_params: List of required parameter names
        optional_params: List of optional parameter names
    """
    self.level = level
    self.required_params = set(required_params or [])
    self.optional_params = set(optional_params or [])

Functions¶

Prompt Types¶

PromptTemplateDict¶

dataknobs_llm.prompts.PromptTemplateDict ¶

Bases: TypedDict

TypedDict structure for prompt template configuration.

This defines the schema for prompt template definitions used throughout the prompt library system. It supports template inheritance, validation, RAG configuration, and flexible template composition.

Attributes:

Name	Type	Description
`template`	`str`	The template string with {{variables}} and ((conditionals))
`defaults`	`Dict[str, Any]`	Default values for template parameters
`validation`	`ValidationConfig`	Validation configuration for this template
`metadata`	`Dict[str, Any]`	Additional metadata (author, version, etc.)
`sections`	`Dict[str, str]`	Section definitions for template composition
`extends`	`str`	Name of base template to inherit from
`rag_config_refs`	`List[str]`	References to standalone RAG configurations
`rag_configs`	`List[RAGConfig]`	Inline RAG configurations
`template_mode`	`str`	Template rendering mode ("mixed" or "jinja2")

RAGConfig¶

dataknobs_llm.prompts.RAGConfig ¶

Bases: TypedDict

Configuration for RAG (Retrieval-Augmented Generation) searches.

Attributes:

Name	Type	Description
`adapter_name`	`str`	Name of the adapter to use for RAG search
`query`	`str`	RAG search query (may contain {{variables}})
`k`	`int`	Number of results to retrieve
`filters`	`Dict[str, Any]`	Additional filters for the search
`placeholder`	`str`	Placeholder name in template (e.g., "RAG_CONTENT")
`header`	`str`	Header text to prepend to RAG results
`item_template`	`str`	Template for formatting individual RAG items

Usage Examples¶

Basic Template Loading¶

from dataknobs_llm.prompts import FileSystemPromptLibrary, AsyncPromptBuilder
from pathlib import Path

# Load from filesystem
library = FileSystemPromptLibrary(prompt_dir=Path("prompts/"))

# Create builder
builder = AsyncPromptBuilder(library=library)

# Render prompt
result = await builder.render_user_prompt(
    "code_review",
    params={"language": "python", "code": "def foo(): pass"}
)
print(result)

In-Memory Library¶

from dataknobs_llm.prompts import InMemoryPromptLibrary, PromptTemplateDict

# Create templates
templates = {
    "greeting": PromptTemplateDict(
        template="Hello {{name}}!",
        defaults={"name": "User"},
        validation={"required": ["name"]}
    )
}

# Create library
library = InMemoryPromptLibrary(prompts={"system": templates})

# Use with builder
builder = AsyncPromptBuilder(library=library)
result = await builder.render_system_prompt("greeting", {"name": "Alice"})

RAG Integration¶

RAG (Retrieval-Augmented Generation) is configured in prompt templates using RAGConfig:

# In your prompt YAML file (e.g., user/code_question.yaml)
template: |
  Answer this question about {{language}}:
  {{question}}

  Relevant documentation:
  {{RAG_DOCS}}

rag_configs:
  - adapter_name: docs
    query_template: "{{language}} programming"
    k: 3
    placeholder: "RAG_DOCS"

For comprehensive RAG documentation, including caching and configuration, see: - Location: packages/llm/docs/RAG_CACHING.md - Location: packages/llm/docs/USER_GUIDE.md (RAG section)

Template Modes¶

# String formatting mode
template = PromptTemplateDict(
    template_mode="string",
    template="Hello {name}!"
)

# Jinja2 mode (default)
template = PromptTemplateDict(
    template_mode="jinja2",
    template="Hello {{name | upper}}!"
)

# Conditional mode
template = PromptTemplateDict(
    template_mode="conditional",
    template="Hello {{name}}!",
    conditional_blocks=[
        {
            "condition": "is_premium",
            "content": "Welcome to premium support!"
        }
    ]
)

Validation¶

# Define validation rules
template = PromptTemplateDict(
    template="Code review for {{language}}: {{code}}",
    validation={
        "required": ["language", "code"],
        "types": {
            "language": "str",
            "code": "str"
        }
    }
)

# Validation happens automatically during render
try:
    result = await builder.render_user_prompt(
        "code_review",
        params={"language": "python"}  # Missing 'code'
    )
except ValueError as e:
    print(f"Validation error: {e}")

Jinja2 Filters¶

# Use built-in filters
template = """
Name: {{name | upper}}
Date: {{date | format_date('%Y-%m-%d')}}
List: {{items | join(', ')}}
JSON: {{data | tojson}}
"""

result = await builder.render_user_prompt(
    "example",
    params={
        "name": "alice",
        "date": datetime.now(),
        "items": ["a", "b", "c"],
        "data": {"key": "value"}
    }
)

Prompt File Format¶

YAML Format¶

# system/code_reviewer.yaml
template: |
  You are an expert code reviewer.
  Focus on {{language}} best practices.

  Review the following code:
  {{code}}

defaults:
  language: python

validation:
  required:
    - code
  types:
    language: str
    code: str

rag_configs:
  - adapter_name: docs
    query_template: "{{language}} best practices"
    k: 5
    placeholder: "BEST_PRACTICES"

metadata:
  version: "1.0.0"
  author: "team"

Prompts API¶

Overview¶

Prompt Library¶

Abstract Interface¶

dataknobs_llm.prompts.AbstractPromptLibrary ¶

Functions¶

get_system_prompt abstractmethod ¶

get_user_prompt abstractmethod ¶

Implementations¶

FileSystemPromptLibrary¶

dataknobs_llm.prompts.FileSystemPromptLibrary ¶

Functions¶

get_message_index ¶

get_prompt_rag_configs ¶

get_rag_config ¶

get_system_prompt ¶

get_user_prompt ¶

list_message_indexes ¶

list_system_prompts ¶

list_user_prompts ¶

load_all ¶

ConfigPromptLibrary¶

dataknobs_llm.prompts.ConfigPromptLibrary ¶

Functions¶

add_message_index ¶

add_rag_config ¶

add_system_prompt ¶

add_user_prompt ¶

get_message_index ¶

get_prompt_rag_configs ¶

get_rag_config ¶

get_system_prompt ¶

get_user_prompt ¶

list_message_indexes ¶

list_system_prompts ¶

list_user_prompts ¶

Prompt Builders¶

PromptBuilder (Sync)¶

dataknobs_llm.prompts.PromptBuilder ¶

Render a system prompt¶

Functions¶

render_system_prompt ¶

Capture RAG metadata¶

Reuse cached RAG¶

render_user_prompt ¶

AsyncPromptBuilder¶

dataknobs_llm.prompts.AsyncPromptBuilder ¶

Render a system prompt¶

Functions¶

render_system_prompt async ¶

Capture RAG metadata¶

Reuse cached RAG¶

render_user_prompt async ¶

Resource Adapters¶

DictResourceAdapter¶

dataknobs_llm.prompts.DictResourceAdapter ¶

Functions¶

get_value ¶

search ¶

InMemoryAdapter¶

dataknobs_llm.prompts.InMemoryAdapter ¶

Simple usage with search results¶

With key-value storage¶

Functions¶

get_value ¶

search ¶

DataknobsBackendAdapter¶

dataknobs_llm.prompts.DataknobsBackendAdapter ¶

... populate database ...¶

Functions¶

get_value ¶

search ¶

Validation¶

ValidationLevel¶

dataknobs_llm.prompts.ValidationLevel ¶

ValidationConfig¶

dataknobs_llm.prompts.ValidationConfig dataclass ¶

Functions¶

Prompt Types¶

PromptTemplateDict¶

get_system_prompt `abstractmethod` ¶

get_user_prompt `abstractmethod` ¶

render_system_prompt `async` ¶

render_user_prompt `async` ¶

dataknobs_llm.prompts.ValidationConfig `dataclass` ¶