implement gradio app

app.py

@@ -0,0 +1,66 @@
+from PIL import Image
+import gradio as gr
+
+from minerva import Minerva
+from formatter import AutoGenFormatter
+
+
+title = "Minerva: AI Guardian for Scam Protection"
+description = """
+              Built with AutoGen 0.4.0 and OpenAI. </br> 
+              Analysis might take up to 30s. </br>
+              https://github.com/dcarpintero/minerva
+              """
+inputs = gr.components.Image()
+outputs = [
+    gr.components.Textbox(label="Analysis Result"),
+    gr.HTML(label="Agentic Workflow (Streaming)")
+]
+examples = "samples"
+
+model = Minerva()
+formatter = AutoGenFormatter()
+
+def to_html(texts):
+    formatted_html = ''
+    for text in texts:
+        formatted_html += text.replace('\n', '<br>') + '<br>'
+    return f'<pre>{formatted_html}</pre>'
+
+async def predict(img):
+    try:
+        img = Image.fromarray(img)
+
+        stream = await model.analyze(img)
+
+        streams = []
+        messages = []
+        async for s in stream:
+            msg = await formatter.to_output(s)
+            streams.append(s)
+            messages.append(msg)
+            yield ["", to_html(messages)]
+        
+        if streams[-1]:
+            prediction = streams[-1].messages[-1].content
+        else:
+            prediction = "No analysis available. Try again later."
+
+        yield [prediction, to_html(messages)]
+
+    except Exception as e:
+        print(e)
+        yield ["Error during analysis. Try again later.", ""]
+
+
+with gr.Blocks() as demo:
+    with gr.Tab("Minerva: AI Guardian for Scam Protection"):
+        gr.Interface(
+            fn=predict,
+            inputs=inputs,
+            outputs=outputs,
+            examples=examples,
+            description=description,
+        ).queue(default_concurrency_limit=5)
+
+demo.launch()

formatter.py

@@ -0,0 +1,37 @@
+from typing import AsyncGenerator, List, TypeVar
+
+from autogen_agentchat.base import Response, TaskResult
+from autogen_agentchat.messages import AgentMessage, MultiModalMessage
+
+
+class AutoGenFormatter:
+    def _message_to_str(self, message: AgentMessage) -> str:
+        if isinstance(message, MultiModalMessage):
+            result: List[str] = []
+            for c in message.content:
+                if isinstance(c, str):
+                    result.append(c)
+                else:
+                    result.append("<image>")
+            return "\n".join(result)
+        else:
+            return f"{message.content}"
+
+    async def to_output(self, message: AgentMessage) -> str:
+        try:
+            if isinstance(message, TaskResult):
+                output = (
+                    f"{'-' * 10} Summary {'-' * 10}\n"
+                    f"Number of messages: {len(message.messages)}\n"
+                    f"Finish reason: {message.stop_reason}\n"
+                )
+                return output
+            
+            if isinstance(message, Response): 
+                output = f"{'-' * 10} {message.chat_message.source} {'-' * 10}\n{self._message_to_str(message.chat_message)}\n"
+                return output
+            else:
+                output = f"{'-' * 10} {message.source} {'-' * 10}\n{self._message_to_str(message)}\n"
+                return output
+        except Exception as e:
+            return ""