feat(py): loading prompts automatically

py/packages/genkit/src/genkit/ai/_aio.py

@@ -23,6 +23,7 @@ class while customizing it with any plugins.
 import uuid
 from asyncio import Future
 from collections.abc import AsyncIterator
+from pathlib import Path
 from typing import Any
 
 from genkit.aio import Channel
@@ -38,7 +39,7 @@ class while customizing it with any plugins.
     GenerateResponseWrapper,
     ModelMiddleware,
 )
-from genkit.blocks.prompt import PromptConfig, to_generate_action_options
+from genkit.blocks.prompt import PromptConfig, load_prompt_folder, to_generate_action_options
 from genkit.blocks.retriever import IndexerRef, IndexerRequest, RetrieverRef
 from genkit.core.action import ActionRunContext
 from genkit.core.action.types import ActionKind
@@ -72,18 +73,30 @@ def __init__(
         self,
         plugins: list[Plugin] | None = None,
         model: str | None = None,
+        prompt_dir: str | Path | None = None,
         reflection_server_spec: ServerSpec | None = None,
     ) -> None:
         """Initialize a new Genkit instance.
 
         Args:
             plugins: List of plugins to initialize.
             model: Model name to use.
+            prompt_dir: Directory to automatically load prompts from.
+                If not provided, defaults to loading from './prompts' if it exists.
             reflection_server_spec: Server spec for the reflection
                 server.
         """
         super().__init__(plugins=plugins, model=model, reflection_server_spec=reflection_server_spec)
 
+        load_path = prompt_dir
+        if load_path is None:
+            default_prompts_path = Path('./prompts')
+            if default_prompts_path.is_dir():
+                load_path = default_prompts_path
+
+        if load_path:
+            load_prompt_folder(self.registry, dir_path=load_path)
+
     async def generate(
         self,
         model: str | None = None,

py/packages/genkit/src/genkit/ai/_registry.py

@@ -51,7 +51,11 @@
 from genkit.blocks.evaluator import BatchEvaluatorFn, EvaluatorFn
 from genkit.blocks.formats.types import FormatDef
 from genkit.blocks.model import ModelFn, ModelMiddleware
-from genkit.blocks.prompt import define_prompt
+from genkit.blocks.prompt import (
+    define_helper,
+    define_prompt,
+    lookup_prompt,
+)
 from genkit.blocks.retriever import IndexerFn, RetrieverFn
 from genkit.blocks.tools import ToolRunContext
 from genkit.codec import dump_dict
@@ -168,6 +172,15 @@ def sync_wrapper(*args, **kwargs):
 
         return wrapper
 
+    def define_helper(self, name: str, fn: Callable) -> None:
+        """Define a Handlebars helper function in the registry.
+
+        Args:
+            name: The name of the helper function.
+            fn: The helper function to register.
+        """
+        define_helper(self.registry, name, fn)
+
     def tool(self, name: str | None = None, description: str | None = None) -> Callable[[Callable], Callable]:
         """Decorator to register a function as a tool.
 

py/packages/genkit/src/genkit/blocks/model.py

@@ -36,8 +36,8 @@ def my_model(request: GenerateRequest) -> GenerateResponse:
 
 from pydantic import BaseModel, Field
 
-from genkit.ai import ActionKind
 from genkit.core.action import ActionMetadata, ActionRunContext
+from genkit.core.action.types import ActionKind
 from genkit.core.extract import extract_json
 from genkit.core.schema import to_json_schema
 from genkit.core.typing import (

py/packages/genkit/src/genkit/blocks/prompt.py

@@ -659,10 +659,11 @@ async def render_dotprompt_to_parts(
     Raises:
         Exception: If the template produces more than one message.
     """
-    merged_input = input_
+    # Flatten input and context for template resolution
+    flattened_data = {**(context or {}), **(input_ or {})}
     rendered = await prompt_function(
         data=DataArgument[dict[str, Any]](
-            input=merged_input,
+            input=flattened_data,
             context=context,
         ),
         options=options,
@@ -718,9 +719,11 @@ async def render_message_prompt(
         if isinstance(options.messages, list):
             messages_ = [e.model_dump() for e in options.messages]
 
+        # Flatten input and context for template resolution
+        flattened_data = {**(context or {}), **(input or {})}
         rendered = await prompt_cache.messages(
             data=DataArgument[dict[str, Any]](
-                input=input,
+                input=flattened_data,
                 context=context,
                 messages=messages_,
             ),
@@ -841,7 +844,7 @@ def define_helper(registry: Registry, name: str, fn: Callable) -> None:
     logger.debug(f'Registered Dotprompt helper "{name}"')
 
 
-def load_prompt(registry: Registry, path: Path, filename: str, prefix: str = '', ns: str = 'dotprompt') -> None:
+def load_prompt(registry: Registry, path: Path, filename: str, prefix: str = '', ns: str = '') -> None:
     """Load a single prompt file and register it in the registry.
 
     This function loads a .prompt file, parses it, and registers it as a lazy-loaded
@@ -1091,6 +1094,13 @@ def load_prompt_folder_recursively(registry: Registry, dir_path: Path, ns: str,
                     partial_name = entry.name[1:-7]  # Remove "_" prefix and ".prompt" suffix
                     with open(entry.path, 'r', encoding='utf-8') as f:
                         source = f.read()
+
+                    # Strip frontmatter if present
+                    if source.startswith('---'):
+                        end_frontmatter = source.find('---', 3)
+                        if end_frontmatter != -1:
+                            source = source[end_frontmatter + 3 :].strip()
+
                     define_partial(registry, partial_name, source)
                     logger.debug(f'Registered Dotprompt partial "{partial_name}" from "{entry.path}"')
                 else:
@@ -1107,7 +1117,7 @@ def load_prompt_folder_recursively(registry: Registry, dir_path: Path, ns: str,
         logger.error(f'Error loading prompts from {full_path}: {e}')
 
 
-def load_prompt_folder(registry: Registry, dir_path: str | Path = './prompts', ns: str = 'dotprompt') -> None:
+def load_prompt_folder(registry: Registry, dir_path: str | Path = './prompts', ns: str = '') -> None:
     """Load all prompt files from a directory.
 
     This is the main entry point for loading prompts from a directory.

py/packages/genkit/tests/genkit/blocks/prompt_test.py

@@ -465,9 +465,9 @@ async def test_file_based_prompt_registers_two_actions() -> None:
         # Load prompts from directory
         load_prompt_folder(ai.registry, prompt_dir)
 
-        # Actions are registered with registry_definition_key (e.g., "dotprompt/filePrompt")
+        # Actions are registered with registry_definition_key (e.g., "filePrompt")
         # We need to look them up by kind and name (without the /prompt/ prefix)
-        action_name = 'dotprompt/filePrompt'  # registry_definition_key format
+        action_name = 'filePrompt'  # registry_definition_key format
 
         prompt_action = ai.registry.lookup_action(ActionKind.PROMPT, action_name)
         executable_prompt_action = ai.registry.lookup_action(ActionKind.EXECUTABLE_PROMPT, action_name)
@@ -491,7 +491,7 @@ async def test_prompt_and_executable_prompt_return_types() -> None:
         prompt_file.write_text('hello {{name}}')
 
         load_prompt_folder(ai.registry, prompt_dir)
-        action_name = 'dotprompt/testPrompt'
+        action_name = 'testPrompt'
 
         prompt_action = ai.registry.lookup_action(ActionKind.PROMPT, action_name)
         executable_prompt_action = ai.registry.lookup_action(ActionKind.EXECUTABLE_PROMPT, action_name)
@@ -540,7 +540,73 @@ async def test_prompt_function_uses_lookup_prompt() -> None:
 
         load_prompt_folder(ai.registry, prompt_dir)
 
-        # Use prompt() function to look up the file-based prompt
-        executable = await prompt(ai.registry, 'promptFuncTest')
-        response = await executable({'name': 'World'})
-        assert 'World' in response.text
+        # Use ai.prompt() to look up the file-based prompt
+        executable = await ai.prompt('promptFuncTest')
+
+        # Verify it can be executed
+        response = await executable({'name': 'Genkit'})
+        assert 'Genkit' in response.text
+
+
+@pytest.mark.asyncio
+async def test_automatic_prompt_loading():
+    """Test that Genkit automatically loads prompts from a directory."""
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        # Create a prompt file
+        prompt_content = """---
+name: testPrompt
+---
+Hello {{name}}!
+"""
+        prompt_file = Path(tmp_dir) / 'test.prompt'
+        prompt_file.write_text(prompt_content)
+
+        # Initialize Genkit with the temporary directory
+        ai = Genkit(prompt_dir=tmp_dir)
+
+        # Verify the prompt is registered
+        # File-based prompts are registered with an empty namespace by default
+        actions = ai.registry.list_serializable_actions()
+        assert '/prompt/test' in actions
+        assert '/executable-prompt/test' in actions
+
+
+@pytest.mark.asyncio
+async def test_automatic_prompt_loading_default_none():
+    """Test that Genkit does not load prompts if prompt_dir is None."""
+    ai = Genkit(prompt_dir=None)
+    actions = ai.registry.list_serializable_actions()
+
+    # Check that no prompts are registered (assuming a clean environment)
+    dotprompts = [key for key in actions.keys() if '/prompt/' in key or '/executable-prompt/' in key]
+    assert len(dotprompts) == 0
+
+
+@pytest.mark.asyncio
+async def test_automatic_prompt_loading_defaults_mock():
+    """Test that Genkit defaults to ./prompts when prompt_dir is not specified and dir exists."""
+    from unittest.mock import ANY, MagicMock, patch
+
+    with patch('genkit.ai._aio.load_prompt_folder') as mock_load, patch('genkit.ai._aio.Path') as mock_path:
+        # Setup mock to simulate ./prompts existing
+        mock_path_instance = MagicMock()
+        mock_path_instance.is_dir.return_value = True
+        mock_path.return_value = mock_path_instance
+
+        Genkit()
+        mock_load.assert_called_once_with(ANY, dir_path=mock_path_instance)
+
+
+@pytest.mark.asyncio
+async def test_automatic_prompt_loading_defaults_missing():
+    """Test that Genkit skips loading when ./prompts is missing."""
+    from unittest.mock import ANY, MagicMock, patch
+
+    with patch('genkit.ai._aio.load_prompt_folder') as mock_load, patch('genkit.ai._aio.Path') as mock_path:
+        # Setup mock to simulate ./prompts missing
+        mock_path_instance = MagicMock()
+        mock_path_instance.is_dir.return_value = False
+        mock_path.return_value = mock_path_instance
+
+        Genkit()
+        mock_load.assert_not_called()

py/samples/evaluator-demo/src/eval_demo.py

@@ -16,7 +16,7 @@
 
 import json
 import os
-from typing import Any
+from typing import Any, List
 
 import pytest
 import structlog
@@ -27,7 +27,7 @@
 
 logger = structlog.get_logger(__name__)
 
-ai = Genkit(plugins=[GoogleAI()])
+ai = Genkit(plugins=[GoogleAI()], model='googleai/gemini-2.5-flash')
 
 
 async def substring_match(datapoint: BaseDataPoint, options: Any | None):
@@ -54,15 +54,19 @@ async def substring_match(datapoint: BaseDataPoint, options: Any | None):
 )
 
 
+
 #  Define a flow that programmatically runs the evaluation
 @ai.flow()
-async def run_eval_demo(input: Any = None):
-    # Load dataset
-    data_path = os.path.join(os.path.dirname(__file__), '..', 'data', 'dataset.json')
-    with open(data_path, 'r') as f:
-        raw_data = json.load(f)
-
-    dataset = [BaseDataPoint(**d) for d in raw_data]
+async def run_eval_demo(dataset_input: List[BaseDataPoint] | None = None):
+    if dataset_input:
+        dataset = dataset_input
+    else:
+        # Load dataset
+        data_path = os.path.join(os.path.dirname(__file__), '..', 'data', 'dataset.json')
+        with open(data_path, 'r') as f:
+            raw_data = json.load(f)
+
+        dataset = [BaseDataPoint(**d) for d in raw_data]
 
     logger.info('Running evaluation...', count=len(dataset))
 

py/samples/prompt_demo/README.md

@@ -19,6 +19,6 @@ genkit start -- uv run src/prompt_demo.py
 
 ## Prompt Structure
 
-- `data/`: Contains `.prompt` files (using [Dotprompt](https://genkit.dev/docs/dotprompt)).
-- `data/_shared_partial.prompt`: A partial that can be included in other prompts.
-- `data/nested/nested_hello.prompt`: A prompt demonstrating nested structure and partial inclusion.
+- `prompts/`: Contains `.prompt` files (using [Dotprompt](https://genkit.dev/docs/dotprompt)).
+- `prompts/_shared_partial.prompt`: A partial that can be included in other prompts.
+- `prompts/nested/nested_hello.prompt`: A prompt demonstrating nested structure and partial inclusion.

py/samples/prompt_demo/data/_shared_partial.prompt → py/samples/prompt_demo/prompts/_shared_partial.prompt

@@ -1,4 +1,4 @@
 ---
-model: googleai/gemini-1.5-flash
+model: googleai/gemini-2.5-flash
 ---
 This is a PARTIAL that says: {{my_helper "Partial content with helper"}}

py/samples/prompt_demo/prompts/hello.prompt

@@ -0,0 +1,8 @@
+---
+model: googleai/gemini-2.5-flash
+input:
+  schema:
+    name: string
+---
+
+Hello {{name}}!

py/samples/prompt_demo/data/nested/nested_hello.prompt → py/samples/prompt_demo/prompts/nested/nested_hello.prompt

@@ -1,5 +1,5 @@
 ---
-model: googleai/gemini-1.5-flash
+model: googleai/gemini-2.5-flash
 input:
   schema:
     name: string

py/samples/prompt_demo/src/prompt_demo.py

@@ -20,22 +20,27 @@
 import structlog
 
 from genkit.ai import Genkit
-from genkit.blocks.prompt import load_prompt_folder
 from genkit.plugins.google_genai import GoogleAI
 
 logger = structlog.get_logger(__name__)
 
 
-# Initialize with GoogleAI plugin
-ai = Genkit(plugins=[GoogleAI()])
+current_dir = Path(__file__).resolve().parent
+prompts_path = current_dir.parent / 'prompts'
 
+ai = Genkit(plugins=[GoogleAI()], model='googleai/gemini-2.5-flash', prompt_dir=prompts_path)
 
-async def main():
-    # Load the prompts from the directory (data)
-    current_dir = Path(__file__).resolve().parent
-    prompts_path = current_dir.parent / 'data'
-    load_prompt_folder(ai.registry, prompts_path)
 
+def my_helper(content, *_, **__):
+    if isinstance(content, list):
+        content = content[0] if content else ''
+    return f'*** {content} ***'
+
+
+ai.define_helper('my_helper', my_helper)
+
+
+async def main():
     # List actions to verify loading
     actions = ai.registry.list_serializable_actions()
 
@@ -47,6 +52,13 @@ async def main():
 
     if not prompts:
         await logger.awarning('No prompts found! Check directory structure.')
+        return
+
+    # Execute the 'hello' prompt
+    hello_prompt = await ai.prompt('hello')
+    response = await hello_prompt(input={'name': 'Genkit User'})
+
+    await logger.ainfo('Prompt Execution Result', text=response.text)
 
 
 if __name__ == '__main__':