feat: Added sidebar nav to docs. Moved json validation to base class (though code is not there yet). Updated (a little) the usage docs. Working on LangGraph mapper example.

Author: jnapper7

Diff for: agntcy_iomapper/__init__.py

@@ -9,6 +9,7 @@
 )
 from .base import (
     BaseIOMapper,
+    BaseIOMapperConfig,
     BaseIOMapperInput,
     BaseIOMapperOutput,
 )
@@ -17,18 +18,3 @@
     ImperativeIOMapperInput,
     ImperativeIOMapperOutput,
 )
-from .pydantic_ai import (
-    PydanticAIAgentIOMapperConfig,
-    PydanticAIAgentIOMapperInput,
-    PydanticAIAgentIOMapperOutput,
-    PydanticAIIOAgentIOMapper,
-)
-
-"""
-from .langgraph import (
-    LangGraphAgentIOMapper,
-    LangGraphIOMapperConfig,
-    LangGraphIOMapperInput,
-    LangGraphIOMapperOutput,
-)
-"""

Diff for: agntcy_iomapper/agent_iomapper.py

@@ -5,12 +5,18 @@
 import re
 from abc import abstractmethod
 from typing import ClassVar
+
 import jsonschema
 from jinja2 import Environment
 from jinja2.sandbox import SandboxedEnvironment
-from pydantic import BaseModel, Field
+from pydantic import Field
 
-from .base import BaseIOMapper, BaseIOMapperInput, BaseIOMapperOutput
+from .base import (
+    BaseIOMapper,
+    BaseIOMapperConfig,
+    BaseIOMapperInput,
+    BaseIOMapperOutput,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -26,13 +32,7 @@ class AgentIOMapperInput(BaseIOMapperInput):
 AgentIOMapperOutput = BaseIOMapperOutput
 
 
-class AgentIOMapperConfig(BaseModel):
-    validate_json_input: bool = Field(
-        default=False, description="Validate input against JSON schema."
-    )
-    validate_json_output: bool = Field(
-        default=False, description="Validate output against JSON schema."
-    )
+class AgentIOMapperConfig(BaseIOMapperConfig):
     system_prompt_template: str = Field(
         max_length=4096,
         default="You are a translation machine. You translate both natural language and object formats for computers.",
@@ -52,14 +52,13 @@ class AgentIOMapper(BaseIOMapper):
 
     def __init__(
         self,
-        config: AgentIOMapperConfig,
-        *,
+        config: AgentIOMapperConfig | None = None,
         jinja_env: Environment | None = None,
         jinja_env_async: Environment | None = None,
     ):
-        super().__init__()
-
-        self.config = config
+        if config is None:
+            config = AgentIOMapperConfig()
+        super().__init__(config)
 
         if jinja_env is not None and jinja_env.is_async:
             raise ValueError("Async Jinja env passed to jinja_env argument")
@@ -167,7 +166,8 @@ def invoke(self, input: AgentIOMapperInput, **kwargs) -> AgentIOMapperOutput:
             user_template = self.user_template
         user_prompt = user_template.render(render_env)
 
-        outputs = self._invoke(input,
+        outputs = self._invoke(
+            input,
             messages=[
                 {"role": "system", "content": system_prompt},
                 {"role": "user", "content": user_prompt},
@@ -180,7 +180,9 @@ def invoke(self, input: AgentIOMapperInput, **kwargs) -> AgentIOMapperOutput:
         return output
 
     @abstractmethod
-    def _invoke(self, input: AgentIOMapperInput, messages: list[dict[str, str]], **kwargs) -> str:
+    def _invoke(
+        self, input: AgentIOMapperInput, messages: list[dict[str, str]], **kwargs
+    ) -> str:
         """Invoke internal model to process messages.
         Args:
             messages: the messages to send to the LLM
@@ -201,7 +203,8 @@ async def ainvoke(self, input: AgentIOMapperInput, **kwargs) -> AgentIOMapperOut
             user_template_async = self.user_template_async
         user_prompt = await user_template_async.render_async(render_env)
 
-        outputs = await self._ainvoke(input,
+        outputs = await self._ainvoke(
+            input,
             messages=[
                 {"role": "system", "content": system_prompt},
                 {"role": "user", "content": user_prompt},
@@ -214,7 +217,9 @@ async def ainvoke(self, input: AgentIOMapperInput, **kwargs) -> AgentIOMapperOut
         return output
 
     @abstractmethod
-    async def _ainvoke(self, input: AgentIOMapperInput, messages: list[dict[str, str]], **kwargs) -> str:
+    async def _ainvoke(
+        self, input: AgentIOMapperInput, messages: list[dict[str, str]], **kwargs
+    ) -> str:
         """Async invoke internal model to process messages.
         Args:
             messages: the messages to send to the LLM

Diff for: agntcy_iomapper/base.py

@@ -38,11 +38,26 @@ class BaseIOMapperOutput(BaseModel):
     )
 
 
+class BaseIOMapperConfig(BaseModel):
+    validate_json_input: bool = Field(
+        default=False, description="Validate input against JSON schema."
+    )
+    validate_json_output: bool = Field(
+        default=False, description="Validate output against JSON schema."
+    )
+
+
 class BaseIOMapper(ABC):
     """Abstract base class for interfacing with io mapper.
     All io mappers wrappers inherited from BaseIOMapper.
     """
 
+    def __init__(
+        self,
+        config: BaseIOMapperConfig | None = None,
+    ):
+        self.config = config if config is not None else BaseIOMapperConfig()
+
     @abstractmethod
     def invoke(self, input: BaseIOMapperInput) -> BaseIOMapperOutput:
         """Pass input data

Diff for: agntcy_iomapper/imperative.py

@@ -22,7 +22,12 @@
 import jsonschema
 from jsonpath_ng.ext import parse
 
-from .base import BaseIOMapper, BaseIOMapperInput, BaseIOMapperOutput
+from .base import (
+    BaseIOMapper,
+    BaseIOMapperConfig,
+    BaseIOMapperInput,
+    BaseIOMapperOutput,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -37,8 +42,12 @@ class ImperativeIOMapper(BaseIOMapper):
     and values are JSONPath (strings) representing how the mapping
     """
 
-    def __init__(self, field_mapping: dict[str, Union[str, Callable]] | None) -> None:
-        super().__init__()
+    def __init__(
+        self,
+        field_mapping: dict[str, Union[str, Callable]] | None,
+        config: BaseIOMapperConfig | None = None,
+    ) -> None:
+        super().__init__(config)
         self.field_mapping = field_mapping
 
     def invoke(self, input: ImperativeIOMapperInput) -> ImperativeIOMapperOutput | None:

Diff for: agntcy_iomapper/langgraph.py

@@ -1,17 +1,19 @@
 # SPDX-FileCopyrightText: Copyright (c) 2025 Cisco and/or its affiliates.
 # SPDX-License-Identifier: Apache-2.0
 import logging
-
 from typing import Any
+
+from langchain.chat_models import init_chat_model
 from langchain_core.language_models import BaseChatModel
 from langchain_core.runnables import Runnable, RunnableConfig
 from langgraph.utils.runnable import RunnableCallable
+from pydantic import Field
 
 from .agent_iomapper import (
+    AgentIOMapper,
     AgentIOMapperConfig,
     AgentIOMapperInput,
     AgentIOMapperOutput,
-    AgentIOMapper,
 )
 
 logger = logging.getLogger(__name__)
@@ -21,47 +23,72 @@
 
 
 class LangGraphIOMapperConfig(AgentIOMapperConfig):
-    llm: BaseChatModel | str
+    llm: BaseChatModel | str = Field(
+        default="anthropic:claude-3-5-sonnet-latest",
+        description="Model to use for translation as LangChain description or model class.",
+    )
 
 
 class _LangGraphAgentIOMapper(AgentIOMapper):
     def __init__(
         self,
-        config: LangGraphIOMapperConfig,
+        config: LangGraphIOMapperConfig | None = None,
         **kwargs,
     ):
+        if config is None:
+            config = LangGraphIOMapperConfig()
         super().__init__(config, **kwargs)
+        if isinstance(config.llm, str):
+            self.llm = init_chat_model(config.llm)
+        else:
+            self.llm = config.llm
 
-    def _invoke(self, input: LangGraphIOMapperInput, messages: list[dict[str, str]], *, config: RunnableConfig | None = None, **kwargs) -> str:
-        response = self.config.llm.invoke(messages, config, **kwargs)
+    def _invoke(
+        self,
+        input: LangGraphIOMapperInput,
+        messages: list[dict[str, str]],
+        *,
+        config: RunnableConfig | None = None,
+        **kwargs,
+    ) -> str:
+        response = self.llm.invoke(messages, config, **kwargs)
         return response.content
 
-    async def _ainvoke(self, input: LangGraphIOMapperOutput, messages: list[dict[str, str]], *, config: RunnableConfig | None = None, **kwargs) -> str:
-        response = await self.config.llm.ainvoke(messages, config, **kwargs)
+    async def _ainvoke(
+        self,
+        input: LangGraphIOMapperOutput,
+        messages: list[dict[str, str]],
+        *,
+        config: RunnableConfig | None = None,
+        **kwargs,
+    ) -> str:
+        response = await self.llm.ainvoke(messages, config, **kwargs)
         return response.content
 
 
 class LangGraphIOMapper:
     def __init__(self, config: LangGraphIOMapperConfig):
         self._iomapper = _LangGraphAgentIOMapper(config)
 
-    async def ainvoke(self, state: dict[str,Any], config: RunnableConfig) -> dict:
+    async def ainvoke(self, state: dict[str, Any], config: RunnableConfig) -> dict:
         response = await self._iomapper.ainvoke(input=state["input"], config=config)
         if response is not None:
-            return { "output": response }
+            return {"output": response}
         else:
             return {}
 
-    def invoke(self, state: dict[str,Any], config: RunnableConfig) -> dict:
+    def invoke(self, state: dict[str, Any], config: RunnableConfig) -> dict:
         response = self._iomapper.invoke(input=state["input"], config=config)
         if response is not None:
-            return { "output": response }
+            return {"output": response}
         else:
             return {}
 
     def as_runnable(self):
         return RunnableCallable(self.invoke, self.ainvoke, name="extract", trace=False)
 
 
-def create_iomapper(config: LangGraphIOMapperConfig) -> Runnable[LangGraphIOMapperInput, LangGraphIOMapperOutput]:
+def create_iomapper(
+    config: LangGraphIOMapperConfig,
+) -> Runnable[LangGraphIOMapperInput, LangGraphIOMapperOutput]:
     return LangGraphIOMapper(config).as_runnable()

Diff for: agntcy_iomapper/pydantic_ai.py

@@ -144,7 +144,7 @@ def _get_model_settings(self, input: PydanticAIAgentIOMapperInput):
             model_name = input.model
         else:
             model_name = self.config.default_model
-        
+
         if model_name not in self.config.models:
             raise ValueError(f"requested model {model_name} not found")
         elif hasattr(input, "model_settings") and input.model_settings is not None:
@@ -197,7 +197,12 @@ def _get_prompts(
 
         return (system_prompt, user_prompt, message_history)
 
-    def _invoke(self, input: PydanticAIAgentIOMapperInput, messages: list[dict[str, str]], **kwargs) -> str:
+    def _invoke(
+        self,
+        input: PydanticAIAgentIOMapperInput,
+        messages: list[dict[str, str]],
+        **kwargs,
+    ) -> str:
         system_prompt, user_prompt, message_history = self._get_prompts(messages)
 
         agent = self._get_agent(input, system_prompt)
@@ -208,7 +213,12 @@ def _invoke(self, input: PydanticAIAgentIOMapperInput, messages: list[dict[str,
         )
         return response.data
 
-    async def _ainvoke(self, input: PydanticAIAgentIOMapperInput, messages: list[dict[str, str]], **kwargs) -> str:
+    async def _ainvoke(
+        self,
+        input: PydanticAIAgentIOMapperInput,
+        messages: list[dict[str, str]],
+        **kwargs,
+    ) -> str:
         system_prompt, user_prompt, message_history = self._get_prompts(messages)
 
         agent = self._get_agent(input, system_prompt)

Diff for: docs/_navbar.md

@@ -0,0 +1,6 @@
+* [Getting Started](#getting-started)
+* [Usage](usage#usage)
+    * [Pydantic-AI Agentic IOMapper](usage#pydantic-ai)
+    * [LangGraph Agentic IOMapper](usage#langgraph)
+    * [JSON Deterministic IOMapper](usage#use-imperative--deterministic-mapper)
+* [Contributing](CONTRIBUTING#how-to-contribute)

Diff for: docs/index.html

@@ -15,7 +15,8 @@
       name: 'iomapper-agnt',
       repo: 'agntcy/iomapper-agnt',
       homepage   : 'README.md',
-      coverpage: true
+      coverpage: true,
+      loadNavbar: true
     }
   </script>
   <!-- Docsify v4 -->