feature: add new playground events

src/backend/base/langflow/alembic/versions/1eab2c3eb45e_event_error.py

@@ -0,0 +1,53 @@
+"""event_error
+
+Revision ID: 1eab2c3eb45e
+Revises: eb5e72293a8e
+Create Date: 2024-10-24 12:03:24.118937
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import sqlite
+from sqlalchemy.engine.reflection import Inspector
+
+# revision identifiers, used by Alembic.
+revision: str = '1eab2c3eb45e'
+down_revision: Union[str, None] = 'eb5e72293a8e'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = Inspector.from_engine(conn)  # type: ignore
+    table_names = inspector.get_table_names()  # noqa
+    column_names = [column["name"] for column in inspector.get_columns("message")]
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('message', schema=None) as batch_op:
+        if "properties" not in column_names:
+            batch_op.add_column(sa.Column('properties', sa.JSON(), nullable=True))
+        if "category" not in column_names:
+            batch_op.add_column(sa.Column('category', sa.Text(), nullable=True))
+        if "content_blocks" not in column_names:
+            batch_op.add_column(sa.Column('content_blocks', sa.JSON(), nullable=True))
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = Inspector.from_engine(conn)  # type: ignore
+    table_names = inspector.get_table_names()  # noqa
+    column_names = [column["name"] for column in inspector.get_columns("message")]
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('message', schema=None) as batch_op:
+        if "content_blocks" in column_names:
+            batch_op.drop_column('content_blocks')
+        if "category" in column_names:
+            batch_op.drop_column('category')
+        if "properties" in column_names:
+            batch_op.drop_column('properties')
+
+    # ### end Alembic commands ###

src/backend/base/langflow/api/v1/chat.py

@@ -177,6 +177,9 @@ async def build_graph_and_get_order() -> tuple[list[str], list[str], Graph]:
             else:
                 first_layer = graph.sort_vertices()
 
+            if inputs is not None and hasattr(inputs, "session") and inputs.session is not None:
+                graph.session_id = inputs.session
+
             for vertex_id in first_layer:
                 graph.run_manager.add_to_vertices_being_run(vertex_id)
 

src/backend/base/langflow/base/agents/callback.py

@@ -13,6 +13,33 @@ class AgentAsyncHandler(AsyncCallbackHandler):
     def __init__(self, log_function: LogFunctionType | None = None):
         self.log_function = log_function
 
+    async def on_chain_start(
+        self,
+        serialized: dict[str, Any],
+        inputs: dict[str, Any],
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        tags: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> None:
+        if self.log_function is None:
+            return
+        self.log_function(
+            {
+                "type": "chain_start",
+                "serialized": serialized,
+                "inputs": inputs,
+                "run_id": run_id,
+                "parent_run_id": parent_run_id,
+                "tags": tags,
+                "metadata": metadata,
+                **kwargs,
+            },
+            name="Chain Start",
+        )
+
     async def on_tool_start(
         self,
         serialized: dict[str, Any],

src/backend/base/langflow/base/io/chat.py

@@ -1,86 +1,15 @@
-from collections.abc import AsyncIterator, Iterator
 from typing import cast
 
 from langflow.custom import Component
 from langflow.memory import store_message
 from langflow.schema import Data
 from langflow.schema.message import Message
-from langflow.services.database.models.message.crud import update_message
-from langflow.utils.async_helpers import run_until_complete
 
 
 class ChatComponent(Component):
     display_name = "Chat Component"
     description = "Use as base for chat components."
 
-    def store_message(self, message: Message) -> Message:
-        messages = store_message(message, flow_id=self.graph.flow_id)
-        if len(messages) != 1:
-            msg = "Only one message can be stored at a time."
-            raise ValueError(msg)
-
-        stored_message = messages[0]
-        self._send_message_event(stored_message)
-
-        if self._should_stream_message(stored_message, message):
-            complete_message = self._stream_message(message, stored_message.id)
-            stored_message = self._update_stored_message(stored_message.id, complete_message)
-
-        self.status = stored_message
-        return stored_message
-
-    def _send_message_event(self, message: Message) -> None:
-        if hasattr(self, "_event_manager") and self._event_manager:
-            self._event_manager.on_message(data=message.data)
-
-    def _should_stream_message(self, stored_message: Message, original_message: Message) -> bool:
-        return bool(
-            hasattr(self, "_event_manager")
-            and self._event_manager
-            and stored_message.id
-            and not isinstance(original_message.text, str)
-        )
-
-    def _update_stored_message(self, message_id: str, complete_message: str) -> Message:
-        message_table = update_message(message_id=message_id, message={"text": complete_message})
-        updated_message = Message(**message_table.model_dump())
-        self.vertex.added_message = updated_message
-        return updated_message
-
-    def _process_chunk(self, chunk: str, complete_message: str, message: Message, message_id: str) -> str:
-        complete_message += chunk
-        if self._event_manager:
-            self._event_manager.on_token(
-                data={
-                    "text": complete_message,
-                    "chunk": chunk,
-                    "sender": message.sender,
-                    "sender_name": message.sender_name,
-                    "id": str(message_id),
-                }
-            )
-        return complete_message
-
-    async def _handle_async_iterator(self, iterator: AsyncIterator, message: Message, message_id: str) -> str:
-        complete_message = ""
-        async for chunk in iterator:
-            complete_message = self._process_chunk(chunk.content, complete_message, message, message_id)
-        return complete_message
-
-    def _stream_message(self, message: Message, message_id: str) -> str:
-        iterator = message.text
-        if not isinstance(iterator, AsyncIterator | Iterator):
-            msg = "The message must be an iterator or an async iterator."
-            raise TypeError(msg)
-
-        if isinstance(iterator, AsyncIterator):
-            return run_until_complete(self._handle_async_iterator(iterator, message, message_id))
-
-        complete_message = ""
-        for chunk in iterator:
-            complete_message = self._process_chunk(chunk.content, complete_message, message, message_id)
-        return complete_message
-
     def build_with_data(
         self,
         *,
@@ -105,9 +34,31 @@ def build_with_data(
     def _create_message(self, input_value, sender, sender_name, files, session_id) -> Message:
         if isinstance(input_value, Data):
             return Message.from_data(input_value)
-        return Message(text=input_value, sender=sender, sender_name=sender_name, files=files, session_id=session_id)
+        return Message(
+            text=input_value,
+            sender=sender,
+            sender_name=sender_name,
+            files=files,
+            session_id=session_id,
+            category="message",
+        )
 
     def _send_messages_events(self, messages) -> None:
         if hasattr(self, "_event_manager") and self._event_manager:
             for stored_message in messages:
-                self._event_manager.on_message(data=stored_message.data)
+                id_ = stored_message.id
+                self._send_message_event(message=stored_message, id_=id_)
+
+    def get_properties_from_source_component(self):
+        if self.vertex.incoming_edges:
+            source_id = self.vertex.incoming_edges[0].source_id
+            _source_vertex = self.graph.get_vertex(source_id)
+            component = _source_vertex.custom_component
+            source = component.display_name
+            icon = component.icon
+            possible_attributes = ["model_name", "model_id", "model"]
+            for attribute in possible_attributes:
+                if hasattr(component, attribute) and getattr(component, attribute):
+                    return getattr(component, attribute), icon, source, component._id
+            return source, icon, component.display_name, component._id
+        return None, None, None, None

src/backend/base/langflow/components/inputs/chat.py

@@ -55,21 +55,43 @@ class ChatInput(ChatComponent):
             advanced=True,
             is_list=True,
         ),
+        MessageTextInput(
+            name="background_color",
+            display_name="Background Color",
+            info="The background color of the icon.",
+            advanced=True,
+        ),
+        MessageTextInput(
+            name="chat_icon",
+            display_name="Icon",
+            info="The icon of the message.",
+            advanced=True,
+        ),
+        MessageTextInput(
+            name="text_color",
+            display_name="Text Color",
+            info="The text color of the name",
+            advanced=True,
+        ),
     ]
     outputs = [
         Output(display_name="Message", name="message", method="message_response"),
     ]
 
     def message_response(self) -> Message:
+        _background_color = self.background_color
+        _text_color = self.text_color
+        _icon = self.chat_icon
         message = Message(
             text=self.input_value,
             sender=self.sender,
             sender_name=self.sender_name,
             session_id=self.session_id,
             files=self.files,
+            properties={"background_color": _background_color, "text_color": _text_color, "icon": _icon},
         )
         if self.session_id and isinstance(message, Message) and self.should_store_message:
-            stored_message = self.store_message(
+            stored_message = self.send_message(
                 message,
             )
             self.message.value = stored_message

src/backend/base/langflow/components/outputs/chat.py

@@ -2,6 +2,7 @@
 from langflow.inputs import BoolInput
 from langflow.io import DropdownInput, MessageTextInput, Output
 from langflow.schema.message import Message
+from langflow.schema.properties import Properties, Source
 from langflow.utils.constants import MESSAGE_SENDER_AI, MESSAGE_SENDER_NAME_AI, MESSAGE_SENDER_USER
 
 
@@ -52,20 +53,64 @@ class ChatOutput(ChatComponent):
             advanced=True,
             info="Template to convert Data to Text. If left empty, it will be dynamically set to the Data's text key.",
         ),
+        MessageTextInput(
+            name="background_color",
+            display_name="Background Color",
+            info="The background color of the icon.",
+            advanced=True,
+        ),
+        MessageTextInput(
+            name="chat_icon",
+            display_name="Icon",
+            info="The icon of the message.",
+            advanced=True,
+        ),
+        MessageTextInput(
+            name="text_color",
+            display_name="Text Color",
+            info="The text color of the name",
+            advanced=True,
+        ),
     ]
     outputs = [
-        Output(display_name="Message", name="message", method="message_response"),
+        Output(
+            display_name="Message",
+            name="message",
+            method="message_response",
+        ),
     ]
 
+    def _build_source(self, _id: str | None, display_name: str | None, source: str | None) -> Source:
+        source_dict = {}
+        if _id:
+            source_dict["id"] = _id
+        if display_name:
+            source_dict["display_name"] = display_name
+        if source:
+            source_dict["source"] = source
+        return Source(**source_dict)
+
     def message_response(self) -> Message:
+        _source, _icon, _display_name, _source_id = self.get_properties_from_source_component()
+        _background_color = self.background_color
+        _text_color = self.text_color
+        if self.chat_icon:
+            _icon = self.chat_icon
         message = Message(
             text=self.input_value,
             sender=self.sender,
             sender_name=self.sender_name,
             session_id=self.session_id,
+            flow_id=self.graph.flow_id,
+            properties=Properties(
+                source=self._build_source(_source_id, _display_name, _source),
+                icon=_icon,
+                background_color=_background_color,
+                text_color=_text_color,
+            ),
         )
         if self.session_id and isinstance(message, Message) and self.should_store_message:
-            stored_message = self.store_message(
+            stored_message = self.send_message(
                 message,
             )
             self.message.value = stored_message