Skip to content

API Reference - WebSocket Streaming to Models#

src #

intentional_websocket #

Init file for intentional_websocket.

__about__ #

Package descriptors for intentional-websocket-streaming.

bot_structure #

Websocket bot structure for Intentional.

WebsocketBotStructure #

Bases: ContinuousStreamBotStructure

Bot structure implementation for OpenAI's Realtime API and similar direct, continuous streaming LLM APIs.

Source code in plugins/intentional-websocket/src/intentional_websocket/bot_structure.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class WebsocketBotStructure(ContinuousStreamBotStructure):
    """
    Bot structure implementation for OpenAI's Realtime API and similar direct, continuous streaming LLM APIs.
    """

    name = "websocket"

    def __init__(self, config: Dict[str, Any], intent_router: IntentRouter):
        """
        Args:
            config:
                The configuration dictionary for the bot structure.
        """
        super().__init__()
        logger.debug("Loading WebsocketBotStructure from config: %s", config)

        # Init the model client
        llm_config = config.pop("llm", None)
        if not llm_config:
            raise ValueError("WebsocketBotStructure requires a 'llm' configuration key to know which model to use.")
        self.model: ContinuousStreamModelClient = load_model_client_from_dict(
            parent=self, intent_router=intent_router, config=llm_config
        )

    async def run(self) -> None:
        await self.model.run()

    async def send(self, data: Dict[str, Any]) -> None:
        await self.model.send(data)

    async def connect(self) -> None:
        await self.model.connect()

    async def disconnect(self) -> None:
        await self.model.disconnect()

    async def handle_interruption(self, lenght_to_interruption: int) -> None:
        await self.model.handle_interruption(lenght_to_interruption)
__init__(config, intent_router) #

Parameters:

Name Type Description Default
config Dict[str, Any]

The configuration dictionary for the bot structure.

 required
Source code in plugins/intentional-websocket/src/intentional_websocket/bot_structure.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(self, config: Dict[str, Any], intent_router: IntentRouter):
    """
    Args:
        config:
            The configuration dictionary for the bot structure.
    """
    super().__init__()
    logger.debug("Loading WebsocketBotStructure from config: %s", config)

    # Init the model client
    llm_config = config.pop("llm", None)
    if not llm_config:
        raise ValueError("WebsocketBotStructure requires a 'llm' configuration key to know which model to use.")
    self.model: ContinuousStreamModelClient = load_model_client_from_dict(
        parent=self, intent_router=intent_router, config=llm_config
    )