chore(py): flatten prompt opts to kwargs (#5062)

Author: huangjeff5

py/packages/genkit/src/genkit/_ai/_prompt.py

@@ -32,6 +32,7 @@
     PromptMetadata,
 )
 from pydantic import BaseModel, ConfigDict
+from typing_extensions import Unpack
 
 from genkit._ai._generate import (
     generate_action,
@@ -295,60 +296,41 @@ async def _ensure_resolved(self) -> None:
 
     async def __call__(
         self,
-        input: InputT | None = None,
-        opts: PromptGenerateOptions | None = None,
+        input: InputT | dict[str, Any] | None = None,
+        **opts: Unpack[PromptGenerateOptions],
     ) -> ModelResponse[OutputT]:
-        """Execute the prompt and return the response."""
-        await self._ensure_resolved()
-        effective_opts: PromptGenerateOptions = opts if opts else {}
+        """Execute the prompt and return the response.
 
-        # Extract streaming callback and middleware from opts
-        on_chunk = effective_opts.get('on_chunk')
-        middleware = effective_opts.get('use') or self._use
-        context = effective_opts.get('context')
+        Args:
+            input: Template variables for rendering.
+        """
+        return await self._call_impl(input, opts)  # ty: ignore[invalid-argument-type]  # ty doesn't infer Unpack[TD] as TD in function body (PEP 692 gap)
 
+    async def _call_impl(
+        self,
+        input: InputT | dict[str, Any] | None,
+        opts: PromptGenerateOptions,
+    ) -> ModelResponse[OutputT]:
+        """Execute the prompt with resolved opts. Used by __call__ and stream."""
+        await self._ensure_resolved()
+        on_chunk = opts.get('on_chunk')
+        middleware = opts.get('use') or self._use
+        context = opts.get('context')
         result = await generate_action(
             self._registry,
-            await self.render(input=input, opts=effective_opts),
+            await self._render_impl(input, opts),
             on_chunk=on_chunk,
             middleware=middleware,
             context=context if context else ActionRunContext._current_context(),  # pyright: ignore[reportPrivateUsage]
         )
-        # Cast to preserve the generic type parameter
         return cast(ModelResponse[OutputT], result)
 
-    def stream(
-        self,
-        input: InputT | None = None,
-        opts: PromptGenerateOptions | None = None,
-        *,
-        timeout: float | None = None,
-    ) -> ModelStreamResponse[OutputT]:
-        """Stream the prompt execution, returning (stream, response_future)."""
-        effective_opts: PromptGenerateOptions = opts if opts else {}
-        channel: Channel[ModelResponseChunk, ModelResponse[OutputT]] = Channel(timeout=timeout)
-
-        # Create a copy of opts with the streaming callback
-        stream_opts: PromptGenerateOptions = {
-            **effective_opts,
-            'on_chunk': lambda c: channel.send(cast(ModelResponseChunk, c)),
-        }
-
-        resp = self.__call__(input=input, opts=stream_opts)
-        response_future: asyncio.Future[ModelResponse[OutputT]] = asyncio.create_task(resp)
-        channel.set_close_future(response_future)
-
-        return ModelStreamResponse[OutputT](channel=channel, response_future=response_future)
-
-    async def render(
+    async def _render_impl(
         self,
-        input: InputT | dict[str, Any] | None = None,
-        opts: PromptGenerateOptions | None = None,
+        input: InputT | dict[str, Any] | None,
+        opts: PromptGenerateOptions,
     ) -> GenerateActionOptions:
-        """Render the prompt template without executing, returning GenerateActionOptions."""
-        await self._ensure_resolved()
-        if opts is None:
-            opts = cast(PromptGenerateOptions, {})
+        """Render the prompt with resolved opts. Used by render() and _call_impl."""
         output_opts = opts.get('output') or {}
         context = opts.get('context')
 
@@ -499,6 +481,37 @@ def _or(opt_val: Any, default: Any) -> Any:  # noqa: ANN401
             resume=resume,
         )
 
+    def stream(
+        self,
+        input: InputT | dict[str, Any] | None = None,
+        *,
+        timeout: float | None = None,
+        **opts: Unpack[PromptGenerateOptions],
+    ) -> ModelStreamResponse[OutputT]:
+        """Stream the prompt execution, returning (stream, response_future)."""
+        channel: Channel[ModelResponseChunk, ModelResponse[OutputT]] = Channel(timeout=timeout)
+        stream_opts: PromptGenerateOptions = {
+            **opts,  # ty doesn't infer Unpack[TD] as TD in function body (PEP 692 gap)
+            'on_chunk': lambda c: channel.send(cast(ModelResponseChunk, c)),
+        }
+        resp = self._call_impl(input, stream_opts)
+        response_future: asyncio.Future[ModelResponse[OutputT]] = asyncio.create_task(resp)
+        channel.set_close_future(response_future)
+
+        return ModelStreamResponse[OutputT](channel=channel, response_future=response_future)
+
+    async def render(
+        self,
+        input: InputT | dict[str, Any] | None = None,
+        **opts: Unpack[PromptGenerateOptions],
+    ) -> GenerateActionOptions:
+        """Render the prompt template without executing, returning GenerateActionOptions.
+
+        Same keyword options as ``__call__`` (see PromptGenerateOptions).
+        """
+        await self._ensure_resolved()
+        return await self._render_impl(input, opts)  # ty: ignore[invalid-argument-type]  # ty doesn't infer Unpack[TD] as TD in function body (PEP 692 gap)
+
     async def as_tool(self) -> Action:
         """Expose this prompt as a tool.
 

py/packages/genkit/tests/genkit/ai/prompt_test.py

@@ -85,13 +85,13 @@ async def test_simple_prompt_with_override_config() -> None:
 
     my_prompt = ai.define_prompt(prompt='hi', config={'banana': True})
 
-    # New API: pass config via opts parameter - this MERGES with prompt config
-    response = await my_prompt(opts={'config': {'temperature': 12}})
+    # New API: pass config via kwargs — this MERGES with prompt config
+    response = await my_prompt(config={'temperature': 12})
 
     assert response.text == want_txt
 
-    # New API: stream also uses opts
-    result = my_prompt.stream(opts={'config': {'temperature': 12}})
+    # New API: stream also uses kwargs
+    result = my_prompt.stream(config={'temperature': 12})
 
     assert (await result.response).text == want_txt
 
@@ -243,8 +243,8 @@ async def test_prompt_rendering_dotprompt(
 
     my_prompt = ai.define_prompt(**prompt)
 
-    # New API: use opts parameter to pass config and context
-    response = await my_prompt(input, opts={'config': input_option, 'context': context})
+    # New API: use kwargs to pass config and context
+    response = await my_prompt(input, config=input_option, context=context)
 
     assert response.text == want_rendered
 
@@ -488,7 +488,7 @@ async def test_config_merge_priority() -> None:
     # New API: runtime config is MERGED with prompt config
     # - temperature: 0.9 (from opts, overrides 0.5)
     # - banana: 'yellow' (from prompt, preserved)
-    rendered = await my_prompt.render(opts={'config': {'temperature': 0.9}})
+    rendered = await my_prompt.render(config={'temperature': 0.9})
 
     assert rendered.config is not None
     # Config is now a dict after merging
@@ -509,8 +509,8 @@ async def test_opts_can_override_model() -> None:
         prompt='hello',
     )
 
-    # Override model via opts
-    response = await my_prompt(opts={'model': 'programmableModel'})
+    # Override model via kwargs
+    response = await my_prompt(model='programmableModel')
 
     # Should use programmableModel, not echoModel
     assert response.text == 'pm response'
@@ -531,8 +531,8 @@ async def test_opts_can_append_messages() -> None:
         Message(role=Role.MODEL, content=[Part(root=TextPart(text='Previous answer'))]),
     ]
 
-    # Append conversation history via opts
-    rendered = await my_prompt.render(opts={'messages': history_messages})
+    # Append conversation history via kwargs
+    rendered = await my_prompt.render(messages=history_messages)
 
     # Should have: system + history (2) + user prompt = 4 messages
     assert len(rendered.messages) == 4
@@ -583,13 +583,11 @@ class OutputSchema(BaseModel):
         output_format='text',  # Default to text
     )
 
-    # Override output via opts
+    # Override output via kwargs
     rendered = await my_prompt.render(
-        opts={
-            'output': {
-                'format': 'json',
-                'schema': OutputSchema,
-            }
+        output={
+            'format': 'json',
+            'schema': OutputSchema,
         }
     )
 
@@ -599,6 +597,31 @@ class OutputSchema(BaseModel):
     assert rendered.output.json_schema is not None
 
 
+@pytest.mark.asyncio
+async def test_executable_prompt_input_positional_opts_as_kwargs() -> None:
+    """ExecutablePrompt: input is positional, opts via kwargs after *."""
+    ai, *_ = setup_test()
+
+    my_prompt = ai.define_prompt(
+        prompt='Recipe for {{cuisine}} {{dish}}',
+        output_format='text',
+    )
+
+    # input = positional (template vars), output = kwarg (opts)
+    rendered = await my_prompt.render(
+        {'cuisine': 'Italian', 'dish': 'pasta'},
+        output={'format': 'text'},
+    )
+
+    # Template vars from input should be in the rendered prompt
+    assert any('Italian' in str(m) for m in rendered.messages)
+    assert any('pasta' in str(m) for m in rendered.messages)
+
+    # output kwarg should be respected
+    assert rendered.output is not None
+    assert rendered.output.format == 'text'
+
+
 # Tests for file-based prompt loading and two-action structure
 @pytest.mark.asyncio
 async def test_file_based_prompt_registers_two_actions() -> None: