Multiple messenger interface support

dff/messengers/common/interface.py

@@ -27,6 +27,10 @@ class MessengerInterface(abc.ABC):
     It is responsible for connection between user and pipeline, as well as for request-response transactions.
     """
 
+    def __init__(self, name: Optional[str] = None):
+        self.name = name if name is not None else str(type(self).__name__)
+
+
     @abc.abstractmethod
     async def connect(self, pipeline_runner: PipelineRunnerFunction):
         """
@@ -44,6 +48,9 @@ class PollingMessengerInterface(MessengerInterface):
     Polling message interface runs in a loop, constantly asking users for a new input.
     """
 
+    def __init__(self, name: Optional[str] = None):
+        MessengerInterface.__init__(self, name)
+
     @abc.abstractmethod
     def _request(self) -> List[Tuple[Message, Hashable]]:
         """
@@ -119,8 +126,9 @@ class CallbackMessengerInterface(MessengerInterface):
     Callback message interface is waiting for user input and answers once it gets one.
     """
 
-    def __init__(self):
+    def __init__(self, name: Optional[str] = None):
         self._pipeline_runner: Optional[PipelineRunnerFunction] = None
+        MessengerInterface.__init__(self, name)
 
     async def connect(self, pipeline_runner: PipelineRunnerFunction):
         self._pipeline_runner = pipeline_runner
@@ -156,19 +164,20 @@ def __init__(
         prompt_request: str = "request: ",
         prompt_response: str = "response: ",
         out_descriptor: Optional[TextIO] = None,
+        name: Optional[str] = None
     ):
-        super().__init__()
+        PollingMessengerInterface.__init__(self, name)
         self._ctx_id: Optional[Hashable] = None
         self._intro: Optional[str] = intro
         self._prompt_request: str = prompt_request
         self._prompt_response: str = prompt_response
         self._descriptor: Optional[TextIO] = out_descriptor
 
     def _request(self) -> List[Tuple[Message, Any]]:
-        return [(Message(input(self._prompt_request)), self._ctx_id)]
+        return [(Message(input(self._prompt_request), interface=self.name), self._ctx_id)]
 
     def _respond(self, responses: List[Context]):
-        print(f"{self._prompt_response}{responses[0].last_response.text}", file=self._descriptor)
+        print(f"{self._prompt_response}{responses[0].last_response_to(self.name).text}", file=self._descriptor)
 
     async def connect(self, pipeline_runner: PipelineRunnerFunction, **kwargs):
         """

dff/messengers/telegram/interface.py

@@ -114,7 +114,9 @@ def __init__(
         timeout: int = 20,
         long_polling_timeout: int = 20,
         messenger: Optional[TelegramMessenger] = None,
+        name: Optional[str] = None
     ):
+        super().__init__(name)
         self.messenger = (
             messenger if messenger is not None else TelegramMessenger(token, suppress_middleware_excepions=True)
         )

dff/pipeline/pipeline/actor.py

@@ -149,10 +149,13 @@ async def __call__(self, pipeline: Pipeline, ctx: Context):
         await self._run_pre_response_processing(ctx, pipeline)
         await self._run_handlers(ctx, pipeline, ActorStage.RUN_PRE_RESPONSE_PROCESSING)
 
+        last_interface = ctx.last_request.interface
+
         # create response
         ctx.framework_states["actor"]["response"] = await self.run_response(
             ctx.framework_states["actor"]["pre_response_processed_node"].response, ctx, pipeline
         )
+        ctx.framework_states["actor"]["response"].interface = last_interface
         await self._run_handlers(ctx, pipeline, ActorStage.CREATE_RESPONSE)
         ctx.add_response(ctx.framework_states["actor"]["response"])
 

dff/pipeline/pipeline/pipeline.py

@@ -16,7 +16,8 @@
 
 import asyncio
 import logging
-from typing import Union, List, Dict, Optional, Hashable, Callable
+from typing import Iterable, Union, List, Dict, Optional, Hashable, Callable
+from uuid import uuid4
 
 from dff.context_storages import DBContextStorage
 from dff.script import Script, Context, ActorStage
@@ -62,7 +63,7 @@ class Pipeline:
         - key: :py:class:`~dff.script.ActorStage` - Stage in which the handler is called.
         - value: List[Callable] - The list of called handlers for each stage. Defaults to an empty `dict`.
 
-    :param messenger_interface: An `AbsMessagingInterface` instance for this pipeline.
+    :param messenger_interfaces: An `AbsMessagingInterface` instance for this pipeline.
     :param context_storage: An :py:class:`~.DBContextStorage` instance for this pipeline or
         a dict to store dialog :py:class:`~.Context`.
     :param services: (required) A :py:data:`~.ServiceGroupBuilder` object,
@@ -92,7 +93,7 @@ def __init__(
         condition_handler: Optional[Callable] = None,
         verbose: bool = True,
         handlers: Optional[Dict[ActorStage, List[Callable]]] = None,
-        messenger_interface: Optional[MessengerInterface] = None,
+        messenger_interface: Optional[Union[MessengerInterface, Iterable[MessengerInterface]]] = None,
         context_storage: Optional[Union[DBContextStorage, Dict]] = None,
         before_handler: Optional[ExtraHandlerBuilder] = None,
         after_handler: Optional[ExtraHandlerBuilder] = None,
@@ -101,7 +102,6 @@ def __init__(
         parallelize_processing: bool = False,
     ):
         self.actor: Actor = None
-        self.messenger_interface = CLIMessengerInterface() if messenger_interface is None else messenger_interface
         self.context_storage = {} if context_storage is None else context_storage
         self._services_pipeline = ServiceGroup(
             components,
@@ -110,6 +110,18 @@ def __init__(
             timeout=timeout,
         )
 
+        if messenger_interface is None:
+            interface = CLIMessengerInterface()
+            self.messenger_interfaces = {interface.name: interface}
+        elif isinstance(messenger_interface, MessengerInterface):
+            self.messenger_interfaces = {messenger_interface.name: interface}
+        else:
+            self.messenger_interfaces = dict()
+            for iface in messenger_interface:
+                if iface.name in self.messenger_interfaces.keys():
+                    logger.warning(f"Messenger interface name '{iface.name}' duplicated!")
+                self.messenger_interfaces.update({iface.name: iface})
+
         self._services_pipeline.name = "pipeline"
         self._services_pipeline.path = ".pipeline"
         actor_exists = finalize_service_group(self._services_pipeline, path=self._services_pipeline.path)
@@ -188,7 +200,9 @@ def info_dict(self) -> dict:
         """
         return {
             "type": type(self).__name__,
-            "messenger_interface": f"Instance of {type(self.messenger_interface).__name__}",
+            "messenger_interfaces": {
+                k: f"Instance of {type(v).__name__}" for k, v in self.messenger_interfaces.items()
+            },
             "context_storage": f"Instance of {type(self.context_storage).__name__}",
             "services": [self._services_pipeline.info_dict],
         }
@@ -217,7 +231,7 @@ def from_script(
         parallelize_processing: bool = False,
         handlers: Optional[Dict[ActorStage, List[Callable]]] = None,
         context_storage: Optional[Union[DBContextStorage, Dict]] = None,
-        messenger_interface: Optional[MessengerInterface] = None,
+        messenger_interfaces: Optional[Iterable[MessengerInterface]] = None,
         pre_services: Optional[List[Union[ServiceBuilder, ServiceGroupBuilder]]] = None,
         post_services: Optional[List[Union[ServiceBuilder, ServiceGroupBuilder]]] = None,
     ) -> "Pipeline":
@@ -249,7 +263,7 @@ def from_script(
 
         :param context_storage: An :py:class:`~.DBContextStorage` instance for this pipeline
             or a dict to store dialog :py:class:`~.Context`.
-        :param messenger_interface: An instance for this pipeline.
+        :param messenger_interfaces: An instance for this pipeline.
         :param pre_services: List of :py:data:`~.ServiceBuilder` or
             :py:data:`~.ServiceGroupBuilder` that will be executed before Actor.
         :type pre_services: Optional[List[Union[ServiceBuilder, ServiceGroupBuilder]]]
@@ -270,7 +284,7 @@ def from_script(
             verbose=verbose,
             parallelize_processing=parallelize_processing,
             handlers=handlers,
-            messenger_interface=messenger_interface,
+            messenger_interfaces=messenger_interfaces,
             context_storage=context_storage,
             components=[*pre_services, ACTOR, *post_services],
         )
@@ -369,7 +383,7 @@ def run(self):
         This method can be both blocking and non-blocking. It depends on current `messenger_interface` nature.
         Message interfaces that run in a loop block current thread.
         """
-        asyncio.run(self.messenger_interface.connect(self._run_pipeline))
+        asyncio.run(asyncio.gather(*[iface.connect(self._run_pipeline) for iface in self.messenger_interfaces.values()]))
 
     def __call__(
         self, request: Message, ctx_id: Optional[Hashable] = None, update_ctx_misc: Optional[dict] = None

dff/pipeline/types.py

@@ -252,7 +252,7 @@ class ExtraHandlerRuntimeInfo(BaseModel):
 PipelineBuilder: TypeAlias = TypedDict(
     "PipelineBuilder",
     {
-        "messenger_interface": NotRequired[Optional["MessengerInterface"]],
+        "messenger_interfaces": NotRequired[Optional[Union["MessengerInterface", Iterable["MessengerInterface"]]]],
         "context_storage": NotRequired[Optional[Union[DBContextStorage, Dict]]],
         "components": ServiceGroupBuilder,
         "before_handler": NotRequired[Optional[ExtraHandlerBuilder]],

dff/script/conditions/__init__.py

@@ -9,6 +9,7 @@
     all,
     negation,
     has_last_labels,
+    from_interface,
     true,
     false,
     agg,

dff/script/conditions/std_conditions.py

@@ -9,12 +9,13 @@
 or other factors that may affect the conversation flow.
 """
 
-from typing import Callable, Pattern, Union, List, Optional
+from typing import Callable, Pattern, Type, Union, List, Optional
 import logging
 import re
 
 from pydantic import validate_call
 
+from dff.messengers.common.interface import MessengerInterface
 from dff.pipeline import Pipeline
 from dff.script import NodeLabel2Type, Context, Message
 
@@ -193,6 +194,21 @@ def has_last_labels_condition_handler(ctx: Context, pipeline: Pipeline) -> bool:
     return has_last_labels_condition_handler
 
 
+def from_interface(iface: Optional[Type[MessengerInterface]] = None, name: Optional[str] = None) -> Callable[[Context, Pipeline], bool]:
+    def is_from_interface_type(ctx: Context, pipeline: Pipeline) -> bool:
+        if ctx.last_request is None:
+            return False
+        latest_interface = ctx.last_request.interface
+        for interface_name, interface_object in pipeline.messenger_interfaces.items():
+            if interface_name == latest_interface:
+                name_match = name is None or interface_name == name
+                type_match = iface is None or isinstance(interface_object, iface)
+                return name_match and type_match
+        return False
+
+    return is_from_interface_type
+
+
 @validate_call
 def true() -> Callable[[Context, Pipeline], bool]:
     """

dff/script/core/context.py

@@ -223,6 +223,13 @@ def last_label(self) -> Optional[NodeLabel2Type]:
         last_index = get_last_index(self.labels)
         return self.labels.get(last_index)
 
+    def last_response_to(self, interface_id: Optional[str]) -> Optional[Message]:
+        for index in list(self.responses)[::-1]:
+            response = self.responses.get(index)
+            if response is not None and response.interface == interface_id:
+                return response
+        return None
+
     @property
     def last_response(self) -> Optional[Message]:
         """
@@ -232,6 +239,13 @@ def last_response(self) -> Optional[Message]:
         last_index = get_last_index(self.responses)
         return self.responses.get(last_index)
 
+    def last_request_from(self, interface_id: Optional[str]) -> Optional[Message]:
+        for index in list(self.requests)[::-1]:
+            request = self.requests.get(index)
+            if request is not None and request.interface == interface_id:
+                return request
+        return None
+
     @last_response.setter
     def last_response(self, response: Optional[Message]):
         """

dff/script/core/message.py

@@ -196,6 +196,7 @@ class level variables to store message information.
     attachments: Optional[Attachments] = None
     annotations: Optional[dict] = None
     misc: Optional[dict] = None
+    interface: Optional[str] = None
     # commands and state options are required for integration with services
     # that use an intermediate backend server, like Yandex's Alice
     # state: Optional[Session] = Session.ACTIVE
@@ -208,10 +209,11 @@ def __init__(
         attachments: Optional[Attachments] = None,
         annotations: Optional[dict] = None,
         misc: Optional[dict] = None,
+        interface: Optional[str] = None,
         **kwargs,
     ):
         super().__init__(
-            text=text, commands=commands, attachments=attachments, annotations=annotations, misc=misc, **kwargs
+            text=text, commands=commands, attachments=attachments, annotations=annotations, misc=misc, interface=interface, **kwargs
         )
 
     def __eq__(self, other):

tests/pipeline/test_messenger_interface.py

@@ -41,7 +41,8 @@ def test_cli_messenger_interface(monkeypatch):
     monkeypatch.setattr("builtins.input", lambda _: "Ping")
     sys.path.append(str(pathlib.Path(__file__).parent.absolute()))
 
-    pipeline.messenger_interface = CLIMessengerInterface(intro="Hi, it's DFF powered bot, let's chat!")
+    interface = CLIMessengerInterface(intro="Hi, it's DFF powered bot, let's chat!")
+    pipeline.messenger_interfaces = {interface.name: interface}
 
     def loop() -> bool:
         loop.runs_left -= 1
@@ -50,12 +51,12 @@ def loop() -> bool:
     loop.runs_left = 5
 
     # Literally what happens in pipeline.run()
-    asyncio.run(pipeline.messenger_interface.connect(pipeline._run_pipeline, loop=loop))
+    asyncio.run(interface.connect(pipeline._run_pipeline, loop=loop))
 
 
 def test_callback_messenger_interface(monkeypatch):
     interface = CallbackMessengerInterface()
-    pipeline.messenger_interface = interface
+    pipeline.messenger_interfaces = {interface.name: interface}
 
     pipeline.run()
 

tutorials/messengers/multiple_interfaces.py

@@ -0,0 +1,82 @@
+# %% [markdown]
+"""
+# Multiple interfaces
+"""
+
+# %pip install dff[telegram]
+
+# %%
+import os
+
+from dff.messengers.common.interface import CLIMessengerInterface
+from dff.script import conditions as cnd
+from dff.script import RESPONSE, TRANSITIONS, Message
+from dff.messengers.telegram import PollingTelegramInterface
+from dff.pipeline import Pipeline
+from dff.utils.testing.common import is_interactive_mode
+
+
+# %% [markdown]
+"""
+"""
+
+
+# %%
+script = {
+    "greeting_flow": {
+        "start_node": {
+            TRANSITIONS: {"greeting_node": cnd.exact_match(Message("/start"))},
+        },
+        "greeting_node": {
+            RESPONSE: Message("Check out responses from different interfaces!"),
+            TRANSITIONS: {
+                "console_node": cnd.from_interface(CLIMessengerInterface),
+                "telegram_node": cnd.from_interface(PollingTelegramInterface)
+            },
+        },
+        "console_node": {
+            RESPONSE: Message("Hi from CLI!"),
+            TRANSITIONS: {"greeting_node": cnd.true()}
+        },
+        "telegram_node": {
+            RESPONSE: Message("Hi from Telegram!"),
+            TRANSITIONS: {"greeting_node": cnd.true()}
+        },
+        "fallback_node": {
+            RESPONSE: Message("Please, repeat the request"),
+            TRANSITIONS: {"greeting_node": cnd.exact_match(Message("/start"))},
+        },
+    }
+}
+
+# this variable is only for testing
+happy_path = (
+    (Message("/start"), Message("Hi")),
+    (Message("Hi"), Message("Hi")),
+    (Message("Bye"), Message("Hi")),
+)
+
+
+# %%
+telegram_interface = PollingTelegramInterface(token=os.environ["TG_BOT_TOKEN"])
+
+console_interface = CLIMessengerInterface()
+
+
+# %%
+pipeline = Pipeline.from_script(
+    script=script,
+    start_label=("greeting_flow", "start_node"),
+    fallback_label=("greeting_flow", "fallback_node"),
+    messenger_interfaces=[telegram_interface, console_interface],
+    # The interface can be passed as a pipeline argument.
+)
+
+
+def main():
+    pipeline.run()
+
+
+if __name__ == "__main__" and is_interactive_mode():
+    # prevent run during doc building
+    main()