agents.components.semantic_router

agents.components.semantic_router#

Module Contents#

Classes#

SemanticRouter

A component that routes semantic information from input topics to output topics based on pre-defined routes. The Semantic Router takes in a list of input topics, a list of routes, an optional default route, and a configuration object. It uses the database client to store and retrieve routing information.

API#

class agents.components.semantic_router.SemanticRouter(*, inputs: List[agents.ros.Topic], routes: List[agents.ros.Route], config: agents.config.SemanticRouterConfig, db_client: agents.clients.db_base.DBClient, default_route: Optional[agents.ros.Route] = None, component_name: str, **kwargs)#

Bases: agents.components.component_base.Component

A component that routes semantic information from input topics to output topics based on pre-defined routes. The Semantic Router takes in a list of input topics, a list of routes, an optional default route, and a configuration object. It uses the database client to store and retrieve routing information.

Parameters:

  • inputs (list[Topic]) – A list of input text topics that this component will subscribe to.

  • routes (list[Route]) – A list of pre-defined routes that publish incoming input to the routed output topics.

  • default_route (Optional[Route]) – An optional route that specifies the default behavior when no specific route matches up to a threshold. If not provided, the component will use the first route in the list.

  • config (SemanticRouterConfig) – The configuration object for this Semantic Router component.

  • db_client (DBClient) – A database client that is used to store and retrieve routing information.

  • component_name (str) – The name of this Semantic Router component (default: “router_component”).

  • kwargs – Additional keyword arguments.

Example usage:

input_text = Topic(name="text0", msg_type="String")
goto_route = Route(
    routes_to=goto,  # where goto is an input topic to another component
    samples=[
        "Go to the door",
        "Go to the kitchen",
        "Get me a glass",
        "Fetch a ball",
        "Go to hallway",
        "Go over there",
    ],
)
mllm_route = Route(
    routes_to=mllm_input,  # where mllm_input is an input topic to another component
    samples=[
        "Are we indoors or outdoors",
        "What do you see?",
        "Whats in front of you?",
        "Where are we",
        "Do you see any people?",
        "How many things are infront of you?",
        "Is this room occupied?",
    ],
)
config = SemanticRouterConfig(router_name="my_router")
db_client = HTTPDBClient(db=ChromaDB(host='localhost', port=8080))
semantic_router = SemanticRouter(
    inputs=[input_text],
    routes=[route1, route2],
    default_route=None,
    config=config,
    db_client=db_client
    component_name = "router"
)
custom_on_activate()#

Custom configuration for creating triggers.

create_all_subscribers()#

Override to handle trigger topics and fixed inputs. Called by parent BaseComponent

activate_all_triggers() → None#

Activates component triggers by attaching execution step to callbacks

destroy_all_subscribers() → None#

Destroys all node subscribers

trigger(trigger: Union[agents.ros.Topic, List[agents.ros.Topic], float]) → None#

Set component trigger

validate_topics(topics: Sequence[Union[agents.ros.Topic, agents.ros.FixedInput]], allowed_topic_types: Optional[Dict[str, List[Union[Type[agents.ros.SupportedType], List[Type[agents.ros.SupportedType]]]]]] = None, topics_direction: str = 'Topics')#

Verify component specific inputs or outputs using allowed topics if provided

Contents