usage.md

Usage

There are several different ways to leverage the IO Mapper functions in Python. There is an agentic interface using models that can be invoked on different AI platforms and a imperative interface that does deterministic JSON remapping without using any AI models.

Key features

The Agent IO Mapper uses an LLM/model to transform the inputs (typically output of the first agent) to match the desired output (typically the input of a second agent). As such, it additionally supports specifying the model prompts for the translation. The configuration object provides a specification for the system and default user prompts:

Example Agent IO mapping

LangGraph Example 1

This project supports specifying model interations using LangGraph.

Define an agent io mapper metadata

metadata = IOMappingAgentMetadata(
    input_fields=["selected_users", "campaign_details.name"],
    output_fields=["stats.status"],
)

The abobe instruction directs the IO mapper agent to utilize the selected_users and name from the campaign_details field and map them to the stats.status. Here is an example to illustrate this further. No further information is needed since the type information can be derived from the input data which is a pydantic model.

Bellow is a table that explain each fields of the IOMappingAgentMetadata class and how to use each

Field	Description	Required	Example
input_fields	an array of json paths	✅	["state.fiedl1", "state.field2", "state"]
output_fields	an array of json paths	✅	["state.output_fiedl1"]
input_schema	defines the schema of the input data	➖	{ "type": "object", "properties": { "title": {"type": "string"}, "ingredients": {"type": "array", "items": {"type": "string"}}, "instructions": {"type": "string"}, }, "required": ["title", "ingredients, instructions"], } OR from pydantic import TypeAdapter TypeAdapter(GraphState).json_schema()
output_schema	defines the schema for the output data	➖	same as input_schema
output_description_prompt	A prompt structured using a Jinja template that can be used by the llm in the mapping definition	➖	"""Output as JSON with this structure: {{ "name": "Campaign Name", "content": "Campaign Content", "is_urgent": "yes/no" }} """

Define an Instance of the Agent

mapping_agent = IOMappingAgent(metadata=metadata, llm=llm)

Bellow is the tablex explaining the interface of the IOMappingAgent class

Field	Description	Required	Example
metadata		✅	IOMappingAgentMetadata( input_fields=["documents.0.page_content"], output_fields=["recipe"], input_schema=TypeAdapter(GraphState).json_schema(), output_schema={ "type": "object", "properties": { "title": {"type": "string"}, "ingredients": {"type": "array", "items": {"type": "string"}}, "instructions": {"type": "string"}, }, "required": ["title", "ingredients, instructions"], }, )
llm	An instance of the large language model to be used	✅	AzureChatOpenAI( model=model_version, api_version=api_version, seed=42, temperature=0, )

Add the node to the LangGraph graph

workflow.add_node(
    "io_mapping",
    mapping_agent.langgraph_node,
)

Finally add the edge and you can run the your LangGraph graph

workflow.add_edge("create_communication", "io_mapping")
workflow.add_edge("io_mapping", "send_communication")

Here is a flow chart of io mapper in a langgraph graph of the discussed application

flowchart TD
    A[create_communication] -->|input in specific format| B(IO Mapper Agent)
    B -->|output expected format| D[send_communication]

Loading

LangGraph Example 2

This example involves a multi-agent software system designed to process a list of ingredients. It interacts with an agent specialized in recipe books to identify feasible recipes based on the provided ingredients. The information is then relayed to an IO mapper, which converts it into a format suitable for display to the user.

Define an agent io mapper metadata

metadata = IOMappingAgentMetadata(
    input_fields=["documents.0.page_content"],
    output_fields=["recipe"],
    input_schema=TypeAdapter(GraphState).json_schema(),
    output_schema={
        "type": "object",
        "properties": {
            "title": {"type": "string"},
            "ingredients": {"type": "array", "items": {"type": "string"}},
            "instructions": {"type": "string"},
        },
        "required": ["title", "ingredients, instructions"],
    },
)

Define an Instance of the Agent

mapping_agent = IOMappingAgent(metadata=metadata, llm=llm)

Add the node to the LangGraph graph

graph.add_node(
    "recipe_io_mapper",
    mapping_agent.langgraph_node,
)

Finally add the edge and you can run the your LangGraph graph

graph.add_edge("recipe_expert", "recipe_io_mapper")

LlamaIndex

This project supports specifying model interations using LangGraph.

LlamaIndex AgentWorkflow

Use Imperative / Deterministic IO Mapper

The code snippet below illustrates a fully functional deterministic mapping that transforms the output of one agent into input for a second agent. The code for the agents is omitted.

 # define schema for the origin agent
 input_schema = {"question": {"type": "string"}}

 # define schema to witch the input should be converted to
 output_schema = {
     "quiz": {
         "type": "object",
         "properties": {
             "prof_question": {"type": "string"},
             "due_date": {"type": "string"},
         },
     }
 }

 # the mapping object using jsonpath, note: the value of the mapping
 # can be either a jsonpath or a function
 mapping_object = {
     "prof_question": "$.question",
     "due_date": lambda _: datetime.now().strftime("%x"),
 }

 input = IOMapperInput(
     input=ArgumentsDescription(
         json_schema=Schema.model_validate(input_schema)
     ),
     output=ArgumentsDescription(
         json_schema=Schema.model_validate(output_schema)
     ),
     data={"question": output_prof},
 )
 # instantiate the mapper
 imperative_mapp = ImperativeIOMapper(
     field_mapping=mapping_object,
 )
 # get the mapping result and send to the other agent
 mapping_result = imperative_mapp.invoke(input=input)

Use Examples

1. To run the examples we strongly recommend that a virtual environment is created
2. Install the requirements file
3. From within examples folder run:

make run_imperative_example