feat: add ability to imperatively defining the mapping between input ouput

Author: CostaRegi

Diff for: README.md

@@ -15,15 +15,15 @@ Communication between agents is not possible if there are discrepancies between
 
 Ensuring that agents are semantically compatible, i.e., the output of the one agent contains the information needed
 by later agents, is an problem of composition or planning in the application. This project, the IO Mapper Agent,
-addresses level 2 and 3 compatibility.  It is a component, implemented as an agent, that can make use of an LLM
+addresses level 2 and 3 compatibility. It is a component, implemented as an agent, that can make use of an LLM
 to transform the output of one agent to become compatible to the input of another agent. Note that this may mean
 many different things, for example:
 
-* JSON structure transcoding: A JSON dictionary needs to be remapped into another JSON dictionary
-* Text summarisation: A  text needs to be summarised or some information needs to be removed
-* Text translation: A text needs to be translated from one language to another
-* Text manipulation: Part of the information of one text needs to be reformulated into another text
-* Any combination of the above
+- JSON structure transcoding: A JSON dictionary needs to be remapped into another JSON dictionary
+- Text summarisation: A text needs to be summarised or some information needs to be removed
+- Text translation: A text needs to be translated from one language to another
+- Text manipulation: Part of the information of one text needs to be reformulated into another text
+- Any combination of the above
 
 The IO mapper Agent can be fed the schema definitions of inputs and outputs as defined by the [Agent Connect Protocol](https://github.com/agntcy/acp-spec).
 

Diff for: agntcy_iomapper/__init__.py

@@ -12,3 +12,7 @@
     IOMapperConfig,
     IOModelArgs,
 )
+
+from .imperative import ImperativeIOMapper
+
+___all__ = [ "ImperativeIOMapper", "AgentIOMapper", "IOMapperOutput", "IOMapperInput"]

Diff for: agntcy_iomapper/base.py

@@ -1,11 +1,11 @@
 # SPDX-FileCopyrightText: Copyright (c) 2025 Cisco and/or its affiliates.
 # SPDX-License-Identifier: Apache-2.0
-from abc import abstractmethod, ABC
-from pydantic import BaseModel, model_validator, Field
-from typing import Any
+from abc import ABC, abstractmethod
+from typing import Any, TypedDict
+
 from openapi_pydantic import Schema
+from pydantic import BaseModel, Field, model_validator
 from typing_extensions import Self
-from typing import TypedDict
 
 
 class ArgumentsDescription(BaseModel):

Diff for: agntcy_iomapper/imperative.py

@@ -0,0 +1,136 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 Cisco and/or its affiliates.
+# SPDX-License-Identifier: Apache-2.0"
+"""
+The deterministic I/O mapper is a component
+designed to translate specific inputs into
+corresponding outputs in a predictable and consistent manner.
+When configured with a JSONPath definition,
+this mapper utilizes the JSONPath query language
+to extract data from JSON-formatted input,
+transforming it into a structured output based on predefined rules.
+The deterministic nature of the mapper ensures that given the same input and
+JSONPath configuration, the output will always be the same,
+providing reliability and repeatability.
+This is particularly useful in scenarios where
+consistent data transformation is required.
+"""
+
+import json
+import logging
+from typing import Any, Callable, Dict, Union
+
+import jsonschema
+from jsonpath_ng.ext import parse
+
+from agntcy_iomapper.base import BaseIOMapper, IOMapperInput, IOMapperOutput
+
+logger = logging.getLogger(__name__)
+
+
+class ImperativeIOMapper(BaseIOMapper):
+    field_mapping: Dict[str, Union[str, Callable]]
+    """A dictionary for where the keys are fields of the output object
+    and values are JSONPath (strings) representing how the mapping
+    """
+
+    def __init__(self, field_mapping: Dict[str, Union[str, Callable]]) -> None:
+        super().__init__()
+        self.field_mapping = field_mapping
+
+    def invoke(self, input: IOMapperInput) -> IOMapperOutput | None:
+        if input.data is None:
+            return None
+        if self.field_mapping is None:
+            return IOMapperOutput(data=input.data)
+
+        data = self._imperative_map(input)
+        return IOMapperOutput(data=data)
+
+    def ainvoke(self, input: IOMapperInput) -> IOMapperOutput | None:
+        return self.invoke(input)
+
+    def _imperative_map(self, input_definition: IOMapperInput) -> Any:
+        """
+        Converts input data to a desired output type.
+
+        This function attempts to convert the provided data into the specified
+        target type. It performs validation using a JSON schema and raises a
+        ValidationError if the data does not conform to the expected schema for
+        the target type.
+
+        Parameters:
+        ----------
+        data : Any
+            The input data to be converted. This can be of any type.
+        Returns:
+        -------
+        Any
+            The converted data in the desired output type.
+        Raises:
+        ------
+        ValidationError
+            If the input data does not conform to the expected schema for the
+            target type.
+        Notes:
+        -----
+        The function assumes that the caller provides a valid `input_schema`.
+        Unsupported target types should be handled as needed within the function.
+        """
+        data = input_definition.data
+        input_schema = input_definition.input.json_schema
+
+        jsonschema.validate(
+            instance=data,
+            schema=input_schema.model_dump(exclude_none=True, mode="json"),
+        )
+
+        mapped_output = {}
+
+        for output_field, json_path_or_func in self.field_mapping.items():
+            if isinstance(json_path_or_func, str):
+                jsonpath_expr = parse(json_path_or_func)
+                match = jsonpath_expr.find(data)
+                expect_value = match[0].value if match else None
+            elif callable(json_path_or_func):
+                expect_value = json_path_or_func(data)
+            else:
+                raise TypeError(
+                    "Mapping values must be strings (JSONPath) or callables (functions)."
+                )
+
+            self._set_jsonpath(mapped_output, output_field, expect_value)
+        jsonschema.validate(
+            instance=mapped_output,
+            schema=input_definition.output.json_schema.model_dump(
+                exclude_none=True, mode="json"
+            ),
+        )
+        # return a serialized version of the object
+        return json.dumps(mapped_output)
+
+    def _set_jsonpath(
+        self, data: dict[str, Any], path: str, value: Any
+    ) -> dict[str, Any]:
+        """set value for field based on its json path
+        Args:
+            data: Data so far
+            path: the json path
+            value: the value to set the json path to
+        Returns:
+        -----
+        dict[str,Any]
+            The mapped filed with the value
+        """
+        copy_data: dict[str, Any] = data
+        # Split the path into parts and remove the leading root
+        parts = path.strip("$.").split(".")
+        # Add value to corresponding path
+        for part in parts[:-1]:
+            if part not in copy_data:
+                copy_data[part] = {}
+
+            copy_data = copy_data[part]
+
+        copy_data[parts[-1]] = value
+
+        return copy_data

Diff for: examples/Makefile

@@ -0,0 +1,7 @@
+
+.PHONY: run_imperative_example
+
+EXAMPLES ?= .
+
+run_imperative_example:
+	python ./imperative.py

Diff for: examples/__init__.py

@@ -0,0 +1,108 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 Cisco and/or its affiliates.
+# SPDX-License-Identifier: Apache-2.0
+import json
+import logging
+from datetime import datetime
+
+from langchain_core.language_models import FakeListChatModel
+from langchain_core.messages import HumanMessage
+from pydantic import BaseModel
+
+from agntcy_iomapper import ImperativeIOMapper
+
+logger = logging.getLogger(__name__)
+
+
+class ProfessorAgent:
+    """
+    This agent mission is to test it's students knowledges
+    """
+
+    predefined_questions = [
+        "What is the capital of France?",
+        "Who wrote 'To Kill a Mockingbird'?",
+        "What is the largest planet in our solar system?",
+        "What is the chemical symbol for gold?",
+        "Who painted the Mona Lisa?",
+    ]
+
+    def __init__(self) -> None:
+        self.model = FakeListChatModel(responses=self.predefined_questions)
+
+    def ask_question(self) -> str:
+        response = self.model.invoke([HumanMessage(content="Generate a question")])
+        return str(response.content)
+
+
+class InputQuiz(BaseModel):
+    prof_question: str
+    due_date: str
+
+
+class StudentAgent:
+    """
+    This agent mission is to answer questions
+    """
+
+    predefined_answers = [
+        "The capital of France is Paris.",
+        "Harper Lee wrote 'To Kill a Mockingbird'.",
+        "The largest planet in our solar system is Jupiter.",
+        "The chemical symbol for gold is Au.",
+        "Leonardo da Vinci painted the Mona Lisa.",
+    ]
+
+    def __init__(self) -> None:
+        self.model = FakeListChatModel(responses=self.predefined_answers)
+
+    def answer(self, quiz) -> str:
+        # q = InputQuiz.parse_raw(quiz)
+        # print(f"here is the quiz {q}")
+        response = self.model.invoke([HumanMessage(content=quiz.prof_question)])
+        return str(response.content)
+
+
+class MultiAgentApp:
+    @staticmethod
+    def run_app():
+        agent_prof = ProfessorAgent()
+        agent_student = StudentAgent()
+
+        output_prof = agent_prof.ask_question()
+
+        prof_agent_output_schema = {"question": {"type": "string"}}
+        student_agent_schema = {
+            "quiz": {
+                "type": "object",
+                "properties": {
+                    "prof_question": {"type": "string"},
+                    "due_date": {"type": "string"},
+                },
+            }
+        }
+
+        mapping_object = {
+            "prof_question": "$.question",
+            "due_date": lambda _: datetime.now().strftime("%x"),
+        }
+
+        imerative_mapp = ImperativeIOMapper(
+            input_schema=prof_agent_output_schema,
+            output_schema=student_agent_schema,
+            field_mapping=mapping_object,
+        )
+
+        print(f"professors question was {output_prof}")
+
+        mapping_result = imerative_mapp.invoke({"question": output_prof})
+
+        print(f"the mapping_result was {mapping_result}")
+
+        response = agent_student.answer(InputQuiz(**(json.loads(mapping_result))))
+
+        print(f"student response was {response}")
+        # map data between agents
+
+
+def run():
+    MultiAgentApp.run_app()

Diff for: examples/imperative.py

@@ -0,0 +1,3 @@
+langchain-core==0.3.35
+langgraph==0.2.72
+pydantic==2.10.6