Create a semantic router to route text queries between different components

Create a semantic router to route text queries between different components#

While semantic routing can be implemented with an LLM component, ROS Agents also provides a convenient SemanticRouter component that works directly with text encoding distances and can be utilized with a vector DB.

In this example we will use the SemanticRouter component to route text queries between two components, a general purpose LLM and a Go-to-X component that we built in the previous example. Lets start by setting up our components.

Setting up the components#

In the following code snippet we will setup our two components.

from agents.components import LLM
from agents.clients.ollama import OllamaClient
from agents.clients.roboml import HTTPModelClient
from agents.models import Idefics2, Llama3_1
from agents.config import LLMConfig
from agents.ros import Topic

# Create a llama3.1 client using Ollama
llama = Llama3_1(name="llama")
ollama_client = OllamaClient(llama)

# Make a generic LLM component using the Llama3_1 model
llm_in = Topic(name="llm_in", msg_type="String")
llm_out = Topic(name="llm_out", msg_type="String")

llm = LLM(
    inputs=[llm_in],
    outputs=[llm_out],
    model_client=llama_client,
    trigger=[llm_in],
)

# Make a Go-to-X component using the same Llama3_1 model
goto_in = Topic(name="goto_in", msg_type="String")
goal_point = Topic(name="goal_point", msg_type="PoseStamped")

config = LLMConfig(enable_rag=True,
                   collection_name="map",
                   distance_func="l2",
                   n_results=1,
                   add_metadata=True)

goto = LLM(
    inputs=[goto_in],
    outputs=[goal_point],
    model_client=llama_client,
    db_client=chroma_client,
    trigger=goto_in,
    component_name='go_to_x'
)

# set a component prompt
goto.set_component_prompt(
    template="""From the given metadata, extract coordinates and provide
    the coordinates in the following json format:\n {"position": coordinates}"""
)

# pre-process the output before publishing to a topic of msg_type PoseStamped
def llm_answer_to_goal_point(output: str) -> Optional[np.ndarray]:
    # extract the json part of the output string (including brackets)
    # one can use sophisticated regex parsing here but we'll keep it simple
    json_string = output[output.find("{"):output.find("}") + 1]

    # load the string as a json and extract position coordinates
    # if there is an error, return None, i.e. no output would be published to goal_point
    try:
        json_dict = json.loads(json_string)
        return np.array(json_dict['position'])
    except Exception:
        return

# add the pre-processing function to the goal_point output topic
goto.add_publisher_preprocessor(goal_point, llm_answer_to_goal_point)

Note

Note that we have reused the same model and its client for both components.

Note

For a detailed explanation of the code for setting up the Go-to-X component, check the previous example.

Caution

In the code block above we are using the same DB client that was setup in this example.

Creating the SemanticRouter#

The SemanticRouter takes an input String topic and sends whatever is published on that topic to a Route. A Route is a thin wrapper around Topic and takes in the name of a topic to publish on and example queries, that would match a potential query that should be published to a particular topic. For example, if we ask our robot a general question, like “Whats the capital of France?”, we do not want that question to be routed to a Go-to-X component, but to a generic LLM. Thus in its route, we would provide examples of general questions. The SemanticRouter component works by storing these examples in a vector DB. Distance is calculated between an incoming query’s embedding and the embeddings of example queries to determine which Route(Topic) the query should be sent on. Lets start by creating our routes for the input topics of the two components above.

from agents.ros import Route

# Create the input topic for the router
query_topic = Topic(name="question", msg_type="String")

# Define a route to a topic that processes go-to-x commands
goto_route = Route(routes_to=goto_in,
    samples=["Go to the door", "Go to the kitchen",
        "Get me a glass", "Fetch a ball", "Go to hallway"])

# Define a route to a topic that is input to an LLM component
llm_route = Route(routes_to=llm_in,
    samples=["What is the capital of France?", "Is there life on Mars?",
        "How many tablespoons in a cup?", "How are you today?", "Whats up?"])

For the database client we will use the ChromaDB client setup in this example. We will specify a router name in our router config, which will act as a collection_name in the database.

from agents.components import SemanticRouter
from agents.config import SemanticRouterConfig

router_config = SemanticRouterConfig(router_name="go-to-router", distance_func="l2")
# Initialize the router component
router = SemanticRouter(
    inputs=[query_topic],
    routes=[llm_route, goto_route],
    default_route=llm_route,  # If none of the routes fall within a distance threshold
    config=router_config,
    db_client=chroma_client,  # reusing the db_client from the previous example
)

And that is it. Whenever something is published on the input topic question, it will be routed, either to a Go-to-X component or an LLM component. We can now expose this topic to our command interface. The complete code for setting up the router is given below:

Semantic Routing#
  1from typing import Optional
  2import json
  3import numpy as np
  4from agents.components import LLM, SemanticRouter
  5from agents.models import Llama3_1
  6from agents.vectordbs import ChromaDB
  7from agents.config import LLMConfig, SemanticRouterConfig
  8from agents.clients.roboml import HTTPDBClient
  9from agents.clients.ollama import OllamaClient
 10from agents.ros import Launcher, Topic, Route
 11
 12
 13# Start a Llama3.1 based llm component using ollama client
 14llama = Llama3_1(name="llama")
 15llama_client = OllamaClient(llama)
 16
 17# Initialize a vector DB that will store our routes
 18chroma = ChromaDB(name="MainDB")
 19chroma_client = HTTPDBClient(db=chroma)
 20
 21
 22# Make a generic LLM component using the Llama3_1 model
 23llm_in = Topic(name="llm_in", msg_type="String")
 24llm_out = Topic(name="llm_out", msg_type="String")
 25
 26llm = LLM(
 27    inputs=[llm_in],
 28    outputs=[llm_out],
 29    model_client=llama_client,
 30    trigger=llm_in
 31)
 32
 33
 34# Define LLM input and output topics including goal_point topic of type PoseStamped
 35goto_in = Topic(name="goto_in", msg_type="String")
 36goal_point = Topic(name="goal_point", msg_type="PoseStamped")
 37
 38config = LLMConfig(enable_rag=True,
 39                   collection_name="map",
 40                   distance_func="l2",
 41                   n_results=1,
 42                   add_metadata=True)
 43
 44# initialize the component
 45goto = LLM(
 46    inputs=[goto_in],
 47    outputs=[goal_point],
 48    model_client=llama_client,
 49    db_client=chroma_client,  # check the previous example where we setup this database client
 50    trigger=goto_in,
 51    component_name='go_to_x'
 52)
 53
 54# set a component prompt
 55goto.set_component_prompt(
 56    template="""From the given metadata, extract coordinates and provide
 57    the coordinates in the following json format:\n {"position": coordinates}"""
 58)
 59
 60
 61# pre-process the output before publishing to a topic of msg_type PoseStamped
 62def llm_answer_to_goal_point(output: str) -> Optional[np.ndarray]:
 63    # extract the json part of the output string (including brackets)
 64    # one can use sophisticated regex parsing here but we'll keep it simple
 65    json_string = output[output.find("{"):output.find("}") + 1]
 66
 67    # load the string as a json and extract position coordinates
 68    # if there is an error, return None, i.e. no output would be published to goal_point
 69    try:
 70        json_dict = json.loads(json_string)
 71        return np.array(json_dict['position'])
 72    except Exception:
 73        return
 74
 75
 76# add the pre-processing function to the goal_point output topic
 77goto.add_publisher_preprocessor(goal_point, llm_answer_to_goal_point)
 78
 79# Create the input topic for the router
 80query_topic = Topic(name="question", msg_type="String")
 81
 82# Define a route to a topic that processes go-to-x commands
 83goto_route = Route(routes_to=goto_in,
 84    samples=["Go to the door", "Go to the kitchen",
 85        "Get me a glass", "Fetch a ball", "Go to hallway"])
 86
 87# Define a route to a topic that is input to an LLM component
 88llm_route = Route(routes_to=llm_in,
 89    samples=["What is the capital of France?", "Is there life on Mars?",
 90        "How many tablespoons in a cup?", "How are you today?", "Whats up?"])
 91
 92router_config = SemanticRouterConfig(router_name="go-to-router", distance_func="l2")
 93# Initialize the router component
 94router = SemanticRouter(
 95    inputs=[query_topic],
 96    routes=[llm_route, goto_route],
 97    default_route=llm_route,  # If none of the routes fall within a distance threshold
 98    config=router_config,
 99    db_client=chroma_client,  # reusing the db_client from the previous example
100)
101
102# Launch the components
103launcher = Launcher()
104launcher.add_pkg(
105    components=[llm, goto, router],
106    activate_all_components_on_start=True)
107launcher.bringup()
Contents