registry

Class info

Classes

Name	Children	Inherits
ConnectionRegistry llmling_agent.talk.registry Registry for managing named connections.		BaseRegistry
ConnectionRegistryError llmling_agent.talk.registry Errors related to connection registration.		LLMLingError
EventContext llmling_agent.talk.registry Base context for all condition/event operations.	TriggerContext	Generic
Talk llmling_agent.talk.talk Manages message flow between agents/groups.		Generic
TriggerContext llmling_agent.talk.registry Context for trigger events, extending base context with event information.		EventContext Generic

🛈 DocStrings

Manages message flow between agents/groups.

ConnectionRegistry

Bases: BaseRegistry[str, Talk]

Registry for managing named connections.

Allows looking up Talk instances by their name. Only named connections get registered.

Source code in src/llmling_agent/talk/registry.py

class ConnectionRegistry(BaseRegistry[str, Talk]):
    """Registry for managing named connections.

    Allows looking up Talk instances by their name. Only named
    connections get registered.
    """

    message_flow = Signal(Talk.ConnectionProcessed)

    def __init__(self, *args, **kwargs):
        """Initialize registry and connect event handlers."""
        super().__init__(*args, **kwargs)
        # Connect handlers to EventedDict events
        self._items.events.added.connect(self._on_talk_added)
        self._items.events.removed.connect(self._on_talk_removed)
        self._items.events.changed.connect(self._on_talk_changed)

    def _on_talk_added(self, name: str, talk: Talk):
        """Handle new talk being added to registry."""
        talk.connection_processed.connect(self._handle_message_flow)
        logger.debug("Connected signal for talk", name=name)

    def _on_talk_removed(self, name: str, talk: Talk):
        """Handle talk being removed from registry."""
        talk.connection_processed.disconnect(self._handle_message_flow)
        logger.debug("Disconnected signal for talk", name=name)

    def _on_talk_changed(self, name: str, old_talk: Talk, new_talk: Talk):
        """Handle talk being replaced in registry."""
        old_talk.connection_processed.disconnect(self._handle_message_flow)
        new_talk.connection_processed.connect(self._handle_message_flow)
        logger.debug("Reconnected signal for talk", name=name)

    def _handle_message_flow(self, event: Talk.ConnectionProcessed):
        """Forward message flow to global stream."""
        self.message_flow.emit(event)

    @property
    def _error_class(self) -> type[ConnectionRegistryError]:
        return ConnectionRegistryError

    def _validate_item(self, item: Any) -> Talk:
        """Ensure only Talk instances can be registered."""
        if not isinstance(item, Talk):
            msg = f"Expected Talk instance, got {type(item)}"
            raise self._error_class(msg)

        return item

    def register_auto(self, talk: Talk[Any], base_name: str | None = None) -> str:
        """Register talk with auto-generated unique name.

        Args:
            talk: Talk instance to register
            base_name: Optional base name to use (defaults to talk.name)

        Returns:
            The actual name used for registration
        """
        base = base_name or talk.name
        counter = 1
        name = base

        while name in self:
            name = f"{base}_{counter}"
            counter += 1
        talk.name = name
        self.register(name, talk)
        return name

__init__

__init__(*args, **kwargs)

Initialize registry and connect event handlers.

Source code in src/llmling_agent/talk/registry.py

def __init__(self, *args, **kwargs):
    """Initialize registry and connect event handlers."""
    super().__init__(*args, **kwargs)
    # Connect handlers to EventedDict events
    self._items.events.added.connect(self._on_talk_added)
    self._items.events.removed.connect(self._on_talk_removed)
    self._items.events.changed.connect(self._on_talk_changed)

register_auto

register_auto(talk: Talk[Any], base_name: str | None = None) -> str

Register talk with auto-generated unique name.

Parameters:

Name	Type	Description	Default
talk	Talk[Any]	Talk instance to register	required
base_name	str | None	Optional base name to use (defaults to talk.name)	None

Returns:

Type	Description
str	The actual name used for registration

Source code in src/llmling_agent/talk/registry.py

def register_auto(self, talk: Talk[Any], base_name: str | None = None) -> str:
    """Register talk with auto-generated unique name.

    Args:
        talk: Talk instance to register
        base_name: Optional base name to use (defaults to talk.name)

    Returns:
        The actual name used for registration
    """
    base = base_name or talk.name
    counter = 1
    name = base

    while name in self:
        name = f"{base}_{counter}"
        counter += 1
    talk.name = name
    self.register(name, talk)
    return name

ConnectionRegistryError

Bases: LLMLingError

Errors related to connection registration.

Source code in src/llmling_agent/talk/registry.py

class ConnectionRegistryError(LLMLingError):
    """Errors related to connection registration."""

EventContext dataclass

Base context for all condition/event operations.

Source code in src/llmling_agent/talk/registry.py

@dataclass(frozen=True)
class EventContext[TMessageContent]:
    """Base context for all condition/event operations."""

    message: ChatMessage[TMessageContent]
    """The message being processed."""

    target: MessageNode
    """The target node this message is being sent to."""

    stats: TalkStats
    """Statistics for the current connection."""

    registry: ConnectionRegistry | None
    """Registry of all named connections."""

    talk: Talk
    """The Talk instance handling this message flow."""

message instance-attribute

message: ChatMessage[TMessageContent]

The message being processed.

registry instance-attribute

registry: ConnectionRegistry | None

Registry of all named connections.

stats instance-attribute

stats: TalkStats

Statistics for the current connection.

talk instance-attribute

talk: Talk

The Talk instance handling this message flow.

target instance-attribute

target: MessageNode

The target node this message is being sent to.

TriggerContext dataclass

Bases: EventContext[TMessageContent]

Context for trigger events, extending base context with event information.

Source code in src/llmling_agent/talk/registry.py

@dataclass(frozen=True)
class TriggerContext[TMessageContent](EventContext[TMessageContent]):
    """Context for trigger events, extending base context with event information."""

    event_type: Literal["condition_met", "message_processed", "disconnected"]
    """Type of event that triggered this call."""

    condition: ConnectionCondition
    """The condition that was triggered (if event_type is condition_met)."""

condition instance-attribute

condition: ConnectionCondition

The condition that was triggered (if event_type is condition_met).

event_type instance-attribute

event_type: Literal['condition_met', 'message_processed', 'disconnected']

Type of event that triggered this call.