added adaptive cards

innovation_pathfinder_ai/backend/app/schemas/adaptive_cards_schema.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+from typing import Any
+
+
+class ICreateMediaAC(BaseModel):
+    media_object: Any
+    media_type: str
+    url: str

innovation_pathfinder_ai/backend/app/utils/adaptive_cards/cards.py

@@ -0,0 +1,141 @@
+import adaptive_cards.card_types as types
+from adaptive_cards.card import AdaptiveCard
+from adaptive_cards.elements import TextBlock, Image, Media
+from adaptive_cards.containers import Container, ImageSet
+from adaptive_cards.actions import ActionSubmit
+import re
+
+from app.schemas.adaptive_cards_schema import ICreateMediaAC
+
+
+def custom_media(anwser):
+    regex_http = r'https?://[^\s"]+'
+    url_search = re.search(regex_http, anwser)
+    url_search_all = re.findall(regex_http, anwser)
+
+    regex_image = r"\b(https?|ftp):\/\/[^\s/$.?#].[^\s]*\.(jpg|jpeg|png|gif|webp)\b"
+    url_image_search = re.search(regex_image, anwser)
+
+    if url_image_search:
+        url_image = url_search.group()
+        if url_image.endswith(")"):
+            url_image = url_image[:-1]
+        if url_image.endswith(")."):
+            url_image = url_image[:-2]
+        media = Image(url=url_image)
+        return None
+        return ICreateMediaAC(media_object=media, media_type="image", url=url_image)
+
+    regex_audio = r"\b(https?|ftp):\/\/[^\s/$.?#].[^\s]*\.(mp3|wav|ogg)\b"
+    url_search_audio = re.search(regex_audio, anwser)
+    if url_search_audio:
+        url_audio = url_search_audio.group()
+        media = Media(
+            sources=[{"mimeType": "audio/mp3", "url": url_audio}],
+            poster="https://adaptivecards.io/content/poster-audio.jpg",
+        )
+        return ICreateMediaAC(media_object=media, media_type="audio", url=url_audio)
+
+    regex_video = r"\b(https?|ftp):\/\/[^\s/$.?#].[^\s]*\.(mp4|webm|ogg)\b"
+    url_search_video = re.search(regex_video, anwser)
+    if url_search_video:
+        url_video = url_search_video.group()
+        media = Media(
+            sources=[{"mimeType": "video/mp4", "url": url_video}],
+            # poster="https://adaptivecards.io/content/poster-video.png",
+            poster="https://douglasgreen.com/wp-content/uploads/2014/03/video-play-btn-featured.png",
+        )
+        return ICreateMediaAC(media_object=media, media_type="video", url=url_video)
+
+    regex_youtube_video = (
+        r"(https?://)?(www\.)?"
+        "(youtube|youtu|youtube-nocookie)\\.(com|be)/"
+        "(watch\\?v=|embed/|v/|.+\\?v=)?([^&=%\\?]{11})"
+    )
+    url_search_youtube_video = re.search(regex_youtube_video, anwser)
+
+    if url_search_youtube_video:
+        url_youtube_video = url_search_youtube_video.group()
+        media = Media(
+            sources=[{"mimeType": "video/mp4", "url": url_youtube_video}],
+        )
+        return ICreateMediaAC(
+            media_object=media, media_type="youtube_video", url=url_youtube_video
+        )
+
+    if len(url_search_all) > 0:
+        list_media_element = []
+        for photo in url_search_all:
+            if "https://images.unsplash.com" in photo:
+                media = Image(url=photo)
+                list_media_element.append(media)
+        body_container_images = ImageSet(images=list_media_element)
+        return None
+        return ICreateMediaAC(
+            media_object=body_container_images, media_type="image", url=""
+        )
+
+    return None
+
+
+def create_hidden_video_card(url):
+    return Media(
+        sources=[
+            {
+                "mimeType": "video/mp4",
+                "url": url,
+            }
+        ],
+        is_visible=False,
+    )
+
+
+def create_adaptive_card(answer: str, actions: list[str] = []) -> AdaptiveCard:
+    custom_media_element: ICreateMediaAC | None = custom_media(answer)
+    custom_media_item = (
+        custom_media_element.media_object if custom_media_element else None
+    )
+    hidden_video_youtube = None
+    # if custom_media_element.media_type == "youtube_video":
+    #     hidden_video_youtube = create_hidden_video_card(custom_media_element.url)
+    if custom_media_element and custom_media_element.media_type == "youtube_video":
+        hidden_video_youtube = create_hidden_video_card(custom_media_element.url)
+
+    # if custom_media_element:
+    #     answer = answer.replace(custom_media_element.url, "")
+
+    description_text = TextBlock(text=answer, wrap=True)
+    items = [
+        description_text,
+        custom_media_item,
+        hidden_video_youtube,
+    ]
+    body_container = Container(items=items)
+
+    # crear action
+    actions = [ActionSubmit(title=action) for action in actions]
+
+    # Crear Adaptive Card
+    adaptive_card = AdaptiveCard(body=[body_container], actions=actions, version="1.5")
+
+    return adaptive_card
+
+
+def create_image_card(image_url: str) -> AdaptiveCard:
+    image = Image(url=image_url)
+    body_container = Container(items=[image])
+    adaptive_card = AdaptiveCard(body=[body_container], version="1.5")
+
+    return adaptive_card
+
+
+def create_loading_card(image_url: str) -> AdaptiveCard:
+    image = Image(
+        url=image_url,
+        size="small",
+        horizontal_alignment=types.HorizontalAlignment.LEFT,
+    )
+    body_container = Container(items=[image])
+    adaptive_card = AdaptiveCard(body=[body_container], version="1.5")
+
+    return adaptive_card

innovation_pathfinder_ai/backend/app/utils/callback.py

@@ -1,4 +1,6 @@
 from app.schemas.message_schema import IChatResponse
+from app.utils.adaptive_cards.cards import create_adaptive_card, create_image_card
+from app.utils.chains import get_suggestions_questions
 from langchain.callbacks.base import AsyncCallbackHandler
 from app.utils.utils import generate_uuid
 from fastapi import WebSocket
@@ -51,6 +53,10 @@ class CustomAsyncCallbackHandler(AsyncCallbackHandler):
         self.message_id: str = message_id
         self.text: str = ""
         self.started: bool = False
+        self.loading_card = create_image_card(
+            "https://res.cloudinary.com/dnv0qwkrk/image/upload/v1691005682/Alita/Ellipsis-2.4s-81px_1_nja8hq.gif"
+        )
+        self.adaptive_card = self.loading_card
 
         if answer_prefix_tokens is None:
             self.answer_prefix_tokens = DEFAULT_ANSWER_PREFIX_TOKENS
@@ -88,6 +94,7 @@ class CustomAsyncCallbackHandler(AsyncCallbackHandler):
         self.append_to_last_tokens(token)
 
         self.text += f"{token}"
+        self.adaptive_card = create_adaptive_card(self.text)
         resp = IChatResponse(
             # id=generate_uuid(),
             id="",
@@ -96,7 +103,7 @@ class CustomAsyncCallbackHandler(AsyncCallbackHandler):
             message=self.adaptive_card.to_dict(),
             type="stream",
         )
-        await self.websocket.send_json(resp.dict())
+        await self.websocket.send_json(resp.model_dump())
 
     async def on_llm_end(
         self,
@@ -115,4 +122,157 @@ class CustomAsyncCallbackHandler(AsyncCallbackHandler):
             message=self.adaptive_card.to_dict(),
             type="end",
         )
-        await self.websocket.send_json(resp.dict())
+        await self.websocket.send_json(resp.model_dump())
+
+
+class CustomFinalStreamingStdOutCallbackHandler(AsyncCallbackHandler):
+    """Callback handler for streaming in agents.
+    Only works with agents using LLMs that support streaming.
+
+    Only the final output of the agent will be streamed.
+    """
+
+    def append_to_last_tokens(self, token: str) -> None:
+        self.last_tokens.append(token)
+        self.last_tokens_stripped.append(token.strip())
+        if len(self.last_tokens) > len(self.answer_prefix_tokens):
+            self.last_tokens.pop(0)
+            self.last_tokens_stripped.pop(0)
+
+    def check_if_answer_reached(self) -> bool:
+        if self.strip_tokens:
+            return self.last_tokens_stripped == self.answer_prefix_tokens_stripped
+        else:
+            return self.last_tokens == self.answer_prefix_tokens
+
+    def update_message_id(self, message_id: str = generate_uuid()):
+        self.message_id = message_id
+
+    def __init__(
+        self,
+        websocket: WebSocket,
+        *,
+        message_id: str = generate_uuid(),
+        answer_prefix_tokens: list[str] | None = None,
+        strip_tokens: bool = True,
+        stream_prefix: bool = False,
+    ) -> None:
+        """Instantiate FinalStreamingStdOutCallbackHandler.
+
+        Args:
+            answer_prefix_tokens: Token sequence that prefixes the answer.
+                Default is ["Final", "Answer", ":"]
+            strip_tokens: Ignore white spaces and new lines when comparing
+                answer_prefix_tokens to last tokens? (to determine if answer has been
+                reached)
+            stream_prefix: Should answer prefix itself also be streamed?
+        """
+        self.websocket: WebSocket = websocket
+        self.message_id: str = message_id
+        self.text: str = ""
+        self.started: bool = False
+        self.loading_card = create_image_card(
+            "https://res.cloudinary.com/dnv0qwkrk/image/upload/v1691005682/Alita/Ellipsis-2.4s-81px_1_nja8hq.gif"
+        )
+        self.adaptive_card = self.loading_card
+
+        if answer_prefix_tokens is None:
+            self.answer_prefix_tokens = DEFAULT_ANSWER_PREFIX_TOKENS
+        else:
+            self.answer_prefix_tokens = answer_prefix_tokens
+        if strip_tokens:
+            self.answer_prefix_tokens_stripped = [
+                token.strip() for token in self.answer_prefix_tokens
+            ]
+        else:
+            self.answer_prefix_tokens_stripped = self.answer_prefix_tokens
+        self.last_tokens = [""] * len(self.answer_prefix_tokens)
+        self.last_tokens_stripped = [""] * len(self.answer_prefix_tokens)
+        self.strip_tokens = strip_tokens
+        self.stream_prefix = stream_prefix
+        self.answer_reached = False
+
+    async def on_llm_start(
+        self, serialized: dict[str, Any], prompts: list[str], **kwargs: Any
+    ) -> None:
+        """Run when LLM starts running."""
+        if self.started == False:
+            self.started = True
+            resp = IChatResponse(
+                id="",
+                message_id=self.message_id,
+                sender="bot",
+                message=self.loading_card.to_dict(),
+                type="start",
+            )
+            await self.websocket.send_json(resp.model_dump())
+
+    async def on_agent_finish(
+        self,
+        finish: AgentFinish,
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        """Run on agent end."""
+        message: str = (
+            self.text
+            if self.text != ""
+            # else "😕 Lo siento no he podido hallar lo que buscabas"
+            else finish.return_values["output"]
+        )
+        self.adaptive_card = create_adaptive_card(message)
+
+        resp = IChatResponse(
+            id="",
+            message_id=self.message_id,
+            sender="bot",
+            message=self.adaptive_card.to_dict(),
+            type="stream",
+        )
+        await self.websocket.send_json(resp.dict())
+
+        suggested_responses = await get_suggestions_questions(message)
+        if len(suggested_responses) > 0:
+            self.adaptive_card = create_adaptive_card(
+                answer=message,
+            )
+        medium_resp = IChatResponse(
+            id="",
+            message_id=self.message_id,
+            sender="bot",
+            message=self.adaptive_card.to_dict(),
+            type="end",
+            suggested_responses=suggested_responses,
+        )
+        await self.websocket.send_json(medium_resp.model_dump())
+
+        # Reset values
+        self.text = ""
+        self.answer_reached = False
+        self.started = False
+
+    async def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
+        """Run on new LLM token. Only available when streaming is enabled."""
+        # Remember the last n tokens, where n = len(answer_prefix_tokens)
+        self.append_to_last_tokens(token)
+
+        # Check if the last n tokens match the answer_prefix_tokens list ...
+        if self.check_if_answer_reached():
+            self.answer_reached = True
+            return
+
+        # ... if yes, then print tokens from now on
+        if self.answer_reached:
+            self.text += f"{token}"
+            self.adaptive_card = create_adaptive_card(self.text)
+
+            resp = IChatResponse(
+                id="",
+                message_id=self.message_id,
+                sender="bot",
+                message=self.adaptive_card.to_dict(),
+                type="stream",
+            )
+            await self.websocket.send_json(resp.model_dump())

innovation_pathfinder_ai/backend/app/utils/chains.py

@@ -0,0 +1,35 @@
+from langchain import LLMChain, PromptTemplate
+from langchain_community.llms import HuggingFaceEndpoint
+
+import re
+
+
+async def get_suggestions_questions(input: str) -> list[str]:
+    """Get suggestions questions."""
+
+    llm = HuggingFaceEndpoint(repo_id="mistralai/Mixtral-8x7B-Instruct-v0.1", 
+                                    temperature=0.1, 
+                                    max_new_tokens=1024,
+                                    repetition_penalty=1.2,
+                                    return_full_text=False
+                )
+
+    prompt_is_farewell_topic_chain = PromptTemplate(
+        input_variables=["input"],
+        template="Determinate if the '{input}' is related to the topic of farewell and return True or False",
+    )
+    prompt = PromptTemplate(
+        input_variables=["input"],
+        template="Create three good suggestions questions about this topic of: {input}. Return the suggestions like a list.",
+    )
+    is_farewell_topic_chain = LLMChain(llm=llm, prompt=prompt_is_farewell_topic_chain)
+    is_farewell_topic_response = await is_farewell_topic_chain.arun(input)
+    suggested_responses = []
+
+    if "False" in is_farewell_topic_response:
+        chain = LLMChain(llm=llm, prompt=prompt)
+        response_chain = await chain.arun(input)
+        suggested_responses = re.findall(r"\d+\.\s(.*?\?)", response_chain)
+        suggested_responses = suggested_responses[:3]
+
+    return suggested_responses

innovation_pathfinder_ai/frontend/app.py

@@ -57,6 +57,7 @@ if __name__ == "__main__":
         #     response_w_sources = response['output']+"\n\n\n Sources:  \n\n\n Internal knowledge base"
         # else:
         #     response_w_sources = response['output']+"\n\n\n Sources: \n\n\n"+src_list
+        print(response)
         history[-1][1] = response['output']
        # all_sources.clear()
         return history
@@ -69,7 +70,8 @@ if __name__ == "__main__":
                 "message": question,
                 "history": history
             }
-            await websocket.send(json.dumps(message_data))
+            json_data = json.dumps(message_data)
+            await websocket.send(json_data)
 
             # Wait for the response
             response_data = await websocket.recv()
@@ -135,8 +137,7 @@ if __name__ == "__main__":
             with gr.Accordion("Open for More!", open=False):
                 gr.Markdown("Nothing yet...")
 
-    demo.queue()
-    demo.launch(debug=True, favicon_path="assets/favicon.ico", share=True)
+    demo.queue().launch(debug=True, favicon_path="assets/favicon.ico", share=True)
 
     x = 0 # for debugging purposes
     app = gr.mount_gradio_app(app, demo, path="/")

requirements.txt

@@ -13,4 +13,6 @@ sqlmodel
 rich
 fastapi
 uvicorn
-sentence-transformers
+sentence-transformers
+fastapi-pagination
+adaptive-cards-py