RuleRouter

Base classes

Name	Children	Inherits
AgentRouter llmling_agent.delegation.router Base class for routing messages between agents.	CallbackRouter RuleRouter	object

⋔ Inheritance diagram

graph TD
  94350422105280["router.RuleRouter"]
  94350422103296["router.AgentRouter"]
  140709601677504["builtins.object"]
  94350422103296 --> 94350422105280
  140709601677504 --> 94350422103296

🛈 DocStrings

Bases: AgentRouter

Router using predefined rules.

Source code in src/llmling_agent/delegation/router.py

class RuleRouter(AgentRouter):
    """Router using predefined rules."""

    def __init__(self, pool: AgentPool, config: RoutingConfig):
        self.pool = pool
        self.config = config

    async def decide(self, message: str) -> Decision:
        """Make decision based on configured rules."""
        msg = message if self.config.case_sensitive else message.lower()

        # Check each rule in priority order
        for rule in sorted(self.config.rules, key=lambda r: r.priority):
            keyword = rule.keyword if self.config.case_sensitive else rule.keyword.lower()

            if keyword not in msg:
                continue

            # Skip if target doesn't exist
            if rule.target not in self.pool.list_agents():
                msg = "Target agent %s not available for rule: %s"
                logger.debug(msg, rule.target, rule.keyword)
                continue

            # Skip if capability required but not available
            if rule.requires_capability:
                agent = self.pool.get_agent(rule.target)
                if not agent.context.capabilities.has_capability(
                    rule.requires_capability
                ):
                    msg = "Agent %s missing required capability: %s"
                    logger.debug(msg, rule.target, rule.requires_capability)
                    continue

            # Create appropriate decision using base class methods
            if rule.wait_for_response:
                return self.get_wait_decision(target=rule.target, reason=rule.reason)
            return self.get_route_decision(target=rule.target, reason=rule.reason)

        # Use default route if configured
        if self.config.default_target:
            return self.get_wait_decision(
                target=self.config.default_target,
                reason=self.config.default_reason,
            )

        # End if no route found
        return self.get_end_decision(reason="No matching rule or default route")

decide async

decide(message: str) -> Decision

Make decision based on configured rules.

Source code in src/llmling_agent/delegation/router.py

async def decide(self, message: str) -> Decision:
    """Make decision based on configured rules."""
    msg = message if self.config.case_sensitive else message.lower()

    # Check each rule in priority order
    for rule in sorted(self.config.rules, key=lambda r: r.priority):
        keyword = rule.keyword if self.config.case_sensitive else rule.keyword.lower()

        if keyword not in msg:
            continue

        # Skip if target doesn't exist
        if rule.target not in self.pool.list_agents():
            msg = "Target agent %s not available for rule: %s"
            logger.debug(msg, rule.target, rule.keyword)
            continue

        # Skip if capability required but not available
        if rule.requires_capability:
            agent = self.pool.get_agent(rule.target)
            if not agent.context.capabilities.has_capability(
                rule.requires_capability
            ):
                msg = "Agent %s missing required capability: %s"
                logger.debug(msg, rule.target, rule.requires_capability)
                continue

        # Create appropriate decision using base class methods
        if rule.wait_for_response:
            return self.get_wait_decision(target=rule.target, reason=rule.reason)
        return self.get_route_decision(target=rule.target, reason=rule.reason)

    # Use default route if configured
    if self.config.default_target:
        return self.get_wait_decision(
            target=self.config.default_target,
            reason=self.config.default_reason,
        )

    # End if no route found
    return self.get_end_decision(reason="No matching rule or default route")

Show source on GitHub