Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import logging | |
| from app_config import create_agent, create_task, get_app_info | |
| from smolagents import MultiStepAgent | |
| from smolagents.gradio_ui import GradioUI, stream_to_gradio | |
| # Configure logging | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='%(asctime)s - %(levelname)s - %(message)s', | |
| datefmt='%Y-%m-%d %H:%M:%S' | |
| ) | |
| logger = logging.getLogger(__name__) | |
| class EggTartAgentUI(GradioUI): | |
| def __init__(self, agent, app_name, app_description, example_questions): | |
| super().__init__(agent) | |
| self.name = app_name | |
| self.description = app_description | |
| self.example_questions = example_questions | |
| def validate_question_length(self, question): | |
| if len(question.strip()) < 20: | |
| raise gr.Error("Please enter a question with at least 20 characters.") | |
| def interact_with_agent(self, prompt, messages, session_state): | |
| self.validate_question_length(prompt) | |
| # Get the agent type from the template agent | |
| if "agent" not in session_state: | |
| session_state["agent"] = self.agent | |
| try: | |
| messages.append(gr.ChatMessage(role="user", content=prompt, metadata={"status": "done"})) | |
| yield messages | |
| task = create_task(question=prompt) | |
| for msg in stream_to_gradio(session_state["agent"], task=task, reset_agent_memory=False): | |
| if isinstance(msg, gr.ChatMessage): | |
| messages[-1].metadata["status"] = "done" | |
| messages.append(msg) | |
| elif isinstance(msg, str): # Then it's only a completion delta | |
| msg = msg.replace("<", r"\<").replace(">", r"\>") # HTML tags seem to break Gradio Chatbot | |
| if messages[-1].metadata["status"] == "pending": | |
| messages[-1].content = msg | |
| else: | |
| messages.append(gr.ChatMessage(role="assistant", content=msg, metadata={"status": "pending"})) | |
| yield messages | |
| yield messages | |
| except Exception as e: | |
| yield messages | |
| raise gr.Error(f"Error in interaction: {str(e)}") | |
| def ask(self, question: str) -> str: | |
| """ | |
| This function serves as the Model Control Panel (MCP) API endpoint for the Egg Tart agent, | |
| which is designed to answer questions about Hong Kong using over 1,800 datasets from data.gov.hk. | |
| She provides concise, friendly, and markdown-formatted answers to a wide range of queries, | |
| including weather, transportation,demographics, public holidays, and economic statistics. | |
| Args: | |
| question (str): The user's question about Hong Kong (e.g., weather, transport, population, events, or statistics). | |
| Returns: | |
| str: A concise, markdown-formatted answer generated by the agent, including relevant data and citations. | |
| """ | |
| logger.info(f"New task received: {question}") | |
| response = agent.run( | |
| task=create_task(question=question), | |
| max_steps=15 | |
| ) | |
| return response.strip() | |
| def create_app(self): | |
| with gr.Blocks(theme=gr.themes.Ocean(), fill_height=True, title=self.name) as demo: | |
| # Add session state to store session-specific data | |
| session_state = gr.State({}) | |
| stored_messages = gr.State([]) | |
| file_uploads_log = gr.State([]) | |
| with gr.Sidebar(): | |
| gr.Markdown( | |
| f"# {self.name}\n\n{self.description}" | |
| ) | |
| with gr.Group(): | |
| gr.Markdown("**Your Question**", container=True) | |
| text_input = gr.Textbox( | |
| lines=3, | |
| label="Chat Message", | |
| container=False, | |
| placeholder="Enter your question here and press Shift+Enter or press the button", | |
| ) | |
| submit_btn = gr.Button("Submit", variant="primary") | |
| # If an upload folder is provided, enable the upload feature | |
| if self.file_upload_folder is not None: | |
| upload_file = gr.File(label="Upload a file") | |
| upload_status = gr.Textbox(label="Upload Status", interactive=False, visible=False) | |
| upload_file.change( | |
| self.upload_file, | |
| [upload_file, file_uploads_log], | |
| [upload_status, file_uploads_log], | |
| ) | |
| gr.Examples(self.example_questions,text_input) | |
| mcp_btn = gr.Button( | |
| "MCP Server Trigger", | |
| variant="stop", | |
| visible=False | |
| ) | |
| mcp_result = gr.Markdown( | |
| visible=False, | |
| ) | |
| gr.HTML( | |
| "<br><br><h4><center>Powered by <a target='_blank' href='https://github.com/huggingface/smolagents'><b>smolagents</b></a></center></h4>" | |
| ) | |
| # Main chat interface | |
| chatbot = gr.Chatbot( | |
| label="Running Log", | |
| type="messages", | |
| resizeable=True, | |
| scale=1, | |
| latex_delimiters=[ | |
| {"left": r"$$", "right": r"$$", "display": True}, | |
| {"left": r"$", "right": r"$", "display": False}, | |
| {"left": r"\[", "right": r"\]", "display": True}, | |
| {"left": r"\(", "right": r"\)", "display": False}, | |
| ], | |
| ) | |
| # Set up event handlers | |
| text_input.submit( | |
| self.log_user_message, | |
| [text_input, file_uploads_log], | |
| [stored_messages, text_input, submit_btn], | |
| show_api=False, | |
| ).then( | |
| self.interact_with_agent, | |
| [stored_messages, chatbot, session_state], | |
| [chatbot], | |
| show_api=False, | |
| ).then( | |
| lambda: ( | |
| gr.Textbox( | |
| interactive=True, placeholder="Enter your question here and press Shift+Enter or the button" | |
| ), | |
| gr.Button(interactive=True), | |
| ), | |
| None, | |
| [text_input, submit_btn], | |
| show_api=False, | |
| ) | |
| mcp_btn.click( | |
| self.ask, | |
| [text_input], | |
| [mcp_result], | |
| show_api=True | |
| ) | |
| submit_btn.click( | |
| self.log_user_message, | |
| [text_input, file_uploads_log], | |
| [stored_messages, text_input, submit_btn], | |
| show_api=False, | |
| ).then( | |
| self.interact_with_agent, | |
| [stored_messages, chatbot, session_state], | |
| [chatbot], | |
| show_api=False, | |
| ).then( | |
| lambda: ( | |
| gr.Textbox( | |
| interactive=True, placeholder="Enter your question here and press Shift+Enter or the button" | |
| ), | |
| gr.Button(interactive=True), | |
| ), | |
| None, | |
| [text_input, submit_btn], | |
| show_api=False, | |
| ) | |
| return demo | |
| if __name__ == "__main__": | |
| logger.info("Starting Egg Tart Agent app initialization") | |
| agent = create_agent() | |
| app_name, app_description, example_questions = get_app_info() | |
| logger.info(f"Egg Tart Agent initialized with app name: {app_name}") | |
| ui = EggTartAgentUI(agent, app_name, app_description, example_questions) | |
| ui.launch(share=False, mcp_server=True) |