Skip to content

registry

Class info

Classes

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

          🛈 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 
           60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
          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: %s", 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: %s", 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: %s", 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 
          69
70
71
72
73
74
75
          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)

          _handle_message_flow 

          _handle_message_flow(event: ConnectionProcessed)

          Forward message flow to global stream.

          Source code in src/llmling_agent/talk/registry.py 
          93
94
95
          def _handle_message_flow(self, event: Talk.ConnectionProcessed):
    """Forward message flow to global stream."""
    self.message_flow.emit(event)

          _on_talk_added 

          _on_talk_added(name: str, talk: Talk)

          Handle new talk being added to registry.

          Source code in src/llmling_agent/talk/registry.py 
          77
78
79
80
          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: %s", name)

          _on_talk_changed 

          _on_talk_changed(name: str, old_talk: Talk, new_talk: Talk)

          Handle talk being replaced in registry.

          Source code in src/llmling_agent/talk/registry.py 
          87
88
89
90
91
          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: %s", name)

          _on_talk_removed 

          _on_talk_removed(name: str, talk: Talk)

          Handle talk being removed from registry.

          Source code in src/llmling_agent/talk/registry.py 
          82
83
84
85
          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: %s", name)

          _validate_item 

          _validate_item(item: Any) -> Talk

          Ensure only Talk instances can be registered.

          Source code in src/llmling_agent/talk/registry.py 
          101
102
103
104
105
106
107
          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

          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 
          109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
          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 
          25
26
          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 
          29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
          @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 
          49
50
51
52
53
54
55
56
57
          @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.