feat(kb): type ask/post results via binding_model keyword

CONTEXT.md

@@ -179,8 +179,20 @@ result = await kb.post(binding_set, ki_name="...")
 # Return ExchangeResult(binding_set=..., exchange_info=[ExchangeInfo, ...])
 result = await kb.ask_with_info(binding_set, ki_name="...")
 result = await kb.post_with_info(binding_set, ki_name="...")
+
+# Typed results (issue #57): Sequence[MyModel] / ExchangeResult[ResultModel]
+people = await kb.ask(binding_set, ki_name="...", binding_model=MyModel)
+result = await kb.post_with_info(
+    binding_set, ki_name="...", result_binding_model=ResultModel
+)
 ```
 
+The keyword-only `binding_model` (ASK) / `result_binding_model` (POST) argument on all
+outgoing methods (incl. `*_sync`) only types the result; parsing still uses the model
+registered for the KI, and a different model raises `ValueError`. Without it the result is
+`Sequence[BindingModel | dict[str, str]]` (PEP 696 TypeVar default). `ExchangeResult[B]` is
+generic in the binding type.
+
 `ask()` / `post()` delegate to `ask_with_info()` / `post_with_info()` and drop the exchange
 info (issue #49).  `ExchangeResult` is a frozen dataclass defined in `knowledge_interaction.py`
 (KB level, parsed bindings); `AskResult` / `PostResult` in `ke/models.py` remain the raw client

README.md

@@ -146,10 +146,18 @@ result = await kb.ask(binding_set, ki_name="...")
 result = await kb.post(binding_set, ki_name="...")
 ```
 
+If the KI has a binding model, pass it to get a typed result (no `cast()` needed). It must be the model the KI was registered with, otherwise a `ValueError` is raised:
+
+```python
+# Sequence[MyModel] / Sequence[ResultModel]
+people = await kb.ask(binding_set, ki_name="...", binding_model=MyModel)
+results = await kb.post(binding_set, ki_name="...", result_binding_model=ResultModel)
+```
+
 To also get the exchange info reported by the Knowledge Engine (which knowledge bases were involved, with status and timing), use the `*_with_info` variants. They return an `ExchangeResult` with `binding_set` and `exchange_info` (a list of `ExchangeInfo`):
 
 ```python
-result = await kb.ask_with_info(binding_set, ki_name="...")
+result = await kb.ask_with_info(binding_set, ki_name="...", binding_model=MyModel)
 for info in result.exchange_info:
     print(info.knowledge_base_id, info.status, info.failed_message)
 ```

examples/03-ask_interaction.py

@@ -45,7 +45,8 @@ async def main():
     await kb.register()
     logger.info("KB registered.")
     # ask_with_info() also returns the exchange info reported by the KE; use
-    # kb.ask() if you only need the bindings.
+    # kb.ask() if you only need the bindings. Passing binding_model types the
+    # result as PersonBindings.
     result = await kb.ask_with_info(
         [
             PersonBinding(
@@ -57,6 +58,7 @@ async def main():
             )
         ],
         "ask-ki",
+        binding_model=PersonBinding,
     )
     logger.info(f"Received result from ASK KI: {result.binding_set}")
     for info in result.exchange_info:

examples/04-post_measurement.py

@@ -80,6 +80,7 @@ async def main():
             )
         ],
         "post-measurement-ki",
+        result_binding_model=ResultBinding,
     )
     logger.info(f"Received result bindings: {result_bindings}")
     await kb.unregister()

examples/07-testing/kb.py

@@ -47,15 +47,16 @@ class ExampleBinding(BindingModel):
 
 
 async def ask_for_values_of_subject(subject_name: str) -> list[str]:
-    result_binding_set: list[ExampleBinding] = await kb.ask(
+    result_binding_set = await kb.ask(
         [
             ExampleBinding(
                 s=URIRef(f"http://example.org/knowledge-mapper/testing#{subject_name}"),
                 value=None,
             )
         ],
         "ask-ki-with-binding-model",
-    )  # pyright: ignore[reportAssignmentType]
+        binding_model=ExampleBinding,
+    )
     return (
         [str(binding.value) for binding in result_binding_set]
         if result_binding_set
@@ -98,7 +99,8 @@ async def repeat_value_post(value: str, iterations: int) -> list[URIRef]:
                     )
                 ],
                 "post-ki",
-            )  # type: ignore
+                result_binding_model=ResultBinding,
+            )
         )
     return [cast(URIRef, binding.other) for binding in result_binding_set]
 

src/knowledge_mapper/kb/knowledge_base.py

@@ -6,7 +6,7 @@
 from collections.abc import Callable, Sequence
 from enum import StrEnum
 from functools import wraps
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 from ..ke import Client
 from ..ke.client import ClientProtocol, HandleRequest, PollResult
@@ -38,6 +38,26 @@
 
 logger = logging.getLogger(__name__)
 
+# Item type of an outgoing call's result: a BindingModel subclass or a raw binding.
+type _Binding = BindingModel | dict[str, str]
+
+
+def _check_binding_model(
+    ki_ctx: KnowledgeInteractionContext[Any, ...],
+    ki_name: str,
+    binding_model: type[BindingModel | dict[str, str]] | None,
+) -> None:
+    """Raise if a caller-supplied (result) binding model differs from the one the
+    KI was registered with, since results are parsed with the registered model."""
+    if binding_model is None or binding_model is ki_ctx.validation_model:
+        return
+    registered = ki_ctx.validation_model
+    raise ValueError(
+        f"Binding model {binding_model.__name__} does not match the (result) "
+        f"binding model registered for KI '{ki_name}': "
+        f"{registered.__name__ if registered else None}."
+    )
+
 
 class KnowledgeBaseState(StrEnum):
     UNREGISTERED = "unregistered"
@@ -540,42 +560,56 @@ async def call(
             requesting_kb_id=requesting_kb_id,
         )
 
-    async def post(
+    async def post[R: _Binding = _Binding](
         self,
         binding_set: Sequence[BindingModel] | BindingSet,
         ki_name: str,
         recipients: list[KnowledgeBaseId] | None = None,
-    ) -> Sequence[BindingModel] | BindingSet:
+        *,
+        result_binding_model: type[R] | None = None,
+    ) -> Sequence[R]:
         """Invoke a POST KI by its name and return the result bindings.
 
-        Use :meth:`post_with_info` to also get the exchange info.
+        Pass the KI's ``result_binding_model`` to have the result typed as a
+        sequence of that model. Use :meth:`post_with_info` to also get the
+        exchange info.
 
         Raises:
             KeyError: If ``ki_name`` is not found in the local KI registry.
-            ValueError: If the KI is not registered at the KE runtime.
+            ValueError: If the KI is not registered at the KE runtime, or
+                ``result_binding_model`` differs from the registered one.
         """
-        result = await self.post_with_info(binding_set, ki_name, recipients)
+        result = await self.post_with_info(
+            binding_set,
+            ki_name,
+            recipients,
+            result_binding_model=result_binding_model,
+        )
         return result.binding_set
 
-    async def post_with_info(
+    async def post_with_info[R: _Binding = _Binding](
         self,
         binding_set: Sequence[BindingModel] | BindingSet,
         ki_name: str,
         recipients: list[KnowledgeBaseId] | None = None,
-    ) -> ExchangeResult:
+        *,
+        result_binding_model: type[R] | None = None,
+    ) -> ExchangeResult[R]:
         """Invoke a POST KI by its name and return the result bindings together
         with the exchange info reported by the KE.
 
         Raises:
             KeyError: If ``ki_name`` is not found in the local KI registry.
-            ValueError: If the KI is not registered at the KE runtime.
+            ValueError: If the KI is not registered at the KE runtime, or
+                ``result_binding_model`` differs from the registered one.
         """
         ki_ctx = self.ki_registry[ki_name]
         if ki_ctx.definition.type != KiTypes.POST:
             raise ValueError(
                 f"KI named '{ki_name}' is of type {ki_ctx.definition.type}, not "
                 f"POST, and cannot be called with the post() method."
             )
+        _check_binding_model(ki_ctx, ki_name, result_binding_model)
         if ki_ctx.status != KnowledgeInteractionStatus.REGISTERED:
             raise ValueError(
                 f"Cannot call KI '{ki_name}' because it is not registered. Please "
@@ -590,46 +624,60 @@ async def post_with_info(
             recipients=recipients,
         )
         return ExchangeResult(
-            binding_set=ki_ctx.parse_result(post_result.result_binding_set),
+            # Safe: parse_result() validates into the registered model, which
+            # _check_binding_model() verified equals R.
+            binding_set=cast(
+                Sequence[R], ki_ctx.parse_result(post_result.result_binding_set)
+            ),
             exchange_info=post_result.exchange_info,
         )
 
-    async def ask(
+    async def ask[B: _Binding = _Binding](
         self,
         binding_set: Sequence[BindingModel] | BindingSet,
         ki_name: str,
         recipients: list[KnowledgeBaseId] | None = None,
-    ) -> Sequence[BindingModel] | BindingSet:
+        *,
+        binding_model: type[B] | None = None,
+    ) -> Sequence[B]:
         """Invoke an ASK KI by its name and return the answer bindings.
 
-        Use :meth:`ask_with_info` to also get the exchange info.
+        Pass the KI's ``binding_model`` to have the result typed as a sequence
+        of that model. Use :meth:`ask_with_info` to also get the exchange info.
 
         Raises:
             KeyError: If ``ki_name`` is not found in the local KI registry.
-            ValueError: If the KI is not registered at the KE runtime.
+            ValueError: If the KI is not registered at the KE runtime, or
+                ``binding_model`` differs from the registered one.
         """
-        result = await self.ask_with_info(binding_set, ki_name, recipients)
+        result = await self.ask_with_info(
+            binding_set, ki_name, recipients, binding_model=binding_model
+        )
         return result.binding_set
 
-    async def ask_with_info(
+    async def ask_with_info[B: _Binding = _Binding](
         self,
         binding_set: Sequence[BindingModel] | BindingSet,
         ki_name: str,
         recipients: list[KnowledgeBaseId] | None = None,
-    ) -> ExchangeResult:
+        *,
+        binding_model: type[B] | None = None,
+    ) -> ExchangeResult[B]:
         """Invoke an ASK KI by its name and return the answer bindings together
         with the exchange info reported by the KE.
 
         Raises:
             KeyError: If ``ki_name`` is not found in the local KI registry.
-            ValueError: If the KI is not registered at the KE runtime.
+            ValueError: If the KI is not registered at the KE runtime, or
+                ``binding_model`` differs from the registered one.
         """
         ki_ctx = self.ki_registry[ki_name]
         if ki_ctx.definition.type != KiTypes.ASK:
             raise ValueError(
                 f"KI named '{ki_name}' is of type {ki_ctx.definition.type}, not "
                 f"ASK, and cannot be called with the ask() method."
             )
+        _check_binding_model(ki_ctx, ki_name, binding_model)
         if ki_ctx.status != KnowledgeInteractionStatus.REGISTERED:
             raise ValueError(
                 f"Cannot call KI '{ki_name}' because it is not registered. Please "
@@ -644,7 +692,9 @@ async def ask_with_info(
             recipients=recipients,
         )
         return ExchangeResult(
-            binding_set=ki_ctx.parse_result(ask_result.binding_set),
+            # Safe: parse_result() validates into the registered model, which
+            # _check_binding_model() verified equals B.
+            binding_set=cast(Sequence[B], ki_ctx.parse_result(ask_result.binding_set)),
             exchange_info=ask_result.exchange_info,
         )
 
@@ -662,12 +712,14 @@ def _require_loop(self) -> asyncio.AbstractEventLoop:
             )
         return loop
 
-    def ask_sync(
+    def ask_sync[B: _Binding = _Binding](
         self,
         binding_set: Sequence[BindingModel] | BindingSet,
         ki_name: str,
         recipients: list[KnowledgeBaseId] | None = None,
-    ) -> Sequence[BindingModel] | BindingSet:
+        *,
+        binding_model: type[B] | None = None,
+    ) -> Sequence[B]:
         """Blocking bridge to :meth:`ask` for use in sync handlers.
 
         Schedules the async ``ask()`` coroutine on the event loop stored by
@@ -679,16 +731,24 @@ def ask_sync(
         """
         loop = self._require_loop()
         future = asyncio.run_coroutine_threadsafe(
-            self.ask(binding_set, ki_name=ki_name, recipients=recipients), loop
+            self.ask(
+                binding_set,
+                ki_name=ki_name,
+                recipients=recipients,
+                binding_model=binding_model,
+            ),
+            loop,
         )
         return future.result()
 
-    def post_sync(
+    def post_sync[R: _Binding = _Binding](
         self,
         binding_set: Sequence[BindingModel] | BindingSet,
         ki_name: str,
         recipients: list[KnowledgeBaseId] | None = None,
-    ) -> Sequence[BindingModel] | BindingSet:
+        *,
+        result_binding_model: type[R] | None = None,
+    ) -> Sequence[R]:
         """Blocking bridge to :meth:`post` for use in sync handlers.
 
         Schedules the async ``post()`` coroutine on the event loop stored by
@@ -700,42 +760,62 @@ def post_sync(
         """
         loop = self._require_loop()
         future = asyncio.run_coroutine_threadsafe(
-            self.post(binding_set, ki_name=ki_name, recipients=recipients), loop
+            self.post(
+                binding_set,
+                ki_name=ki_name,
+                recipients=recipients,
+                result_binding_model=result_binding_model,
+            ),
+            loop,
         )
         return future.result()
 
-    def ask_with_info_sync(
+    def ask_with_info_sync[B: _Binding = _Binding](
         self,
         binding_set: Sequence[BindingModel] | BindingSet,
         ki_name: str,
         recipients: list[KnowledgeBaseId] | None = None,
-    ) -> ExchangeResult:
+        *,
+        binding_model: type[B] | None = None,
+    ) -> ExchangeResult[B]:
         """Blocking bridge to :meth:`ask_with_info` for use in sync handlers.
 
         Raises:
             RuntimeError: If called outside the handling loop context.
         """
         loop = self._require_loop()
         future = asyncio.run_coroutine_threadsafe(
-            self.ask_with_info(binding_set, ki_name=ki_name, recipients=recipients),
+            self.ask_with_info(
+                binding_set,
+                ki_name=ki_name,
+                recipients=recipients,
+                binding_model=binding_model,
+            ),
             loop,
         )
         return future.result()
 
-    def post_with_info_sync(
+    def post_with_info_sync[R: _Binding = _Binding](
         self,
         binding_set: Sequence[BindingModel] | BindingSet,
         ki_name: str,
         recipients: list[KnowledgeBaseId] | None = None,
-    ) -> ExchangeResult:
+        *,
+        result_binding_model: type[R] | None = None,
+    ) -> ExchangeResult[R]:
         """Blocking bridge to :meth:`post_with_info` for use in sync handlers.
 
         Raises:
             RuntimeError: If called outside the handling loop context.
         """
         loop = self._require_loop()
         future = asyncio.run_coroutine_threadsafe(
-            self.post_with_info(binding_set, ki_name=ki_name, recipients=recipients),
+            self.post_with_info(
+                binding_set,
+                ki_name=ki_name,
+                recipients=recipients,
+                result_binding_model=result_binding_model,
+            ),
             loop,
         )
         return future.result()