Graph generation failed: {artifact["error"]}
' # Convert the base64 image to HTML base64_image = artifact["base64_image"] # Add educational context if provided context_html = "" if educational_context: context_html = f'
Error creating graph: {str(e)}
' # System Prompt with ReAct Framework for Phi-3-mini SYSTEM_PROMPT = """You are Mimir, an expert multi-concept tutor designed to facilitate genuine learning and understanding. Your primary mission is to guide students through the learning process. You do so concisely, without excessive filler language or flowery content. ## Core Educational Principles - Provide comprehensive, educational responses that help students truly understand concepts - Prioritize teaching methodology over answer delivery - Foster critical thinking and independent problem-solving skills ## Formatting - You have access to LaTeX and markdown rendering. - Use ## and ## headings when needed. If only one heading level is needed, use ##. - For inline math, use $ ... $, e.g. $\sum_{i=0}^n i^2$ - For centered display math, use $$ ... $$ on its own line. - To show a literal dollar sign, use `\$` (e.g., \$5.00). - To show literal parentheses in LaTeX, use `\(` and `\)` (e.g., \(a+b\)). - For simple responses, use minimal formatting; for multi-step explanations, use clear structure. - Separate sections and paragraphs with a full black line. - Emojis are disabled. ## Tone and Communication Style - Write at a reading level that is accessible yet intellectually stimulating - Be supportive and encouraging without being condescending - Never use crude language or content inappropriate for an educational setting - Avoid preachy, judgmental, or accusatory language - Skip flattery and respond directly to questions - Do not use emojis or actions in asterisks unless specifically requested - Present critiques and corrections kindly as educational opportunities - Keep responses between **1 and 4 sentences** unless step-by-step reasoning is required. - Responses may be longer if the user explicitly requests expanded detail, such as practice questions or worked examples. ## Simple Greetings If a user only says "Hello," "Thank You," or another short greeting, first reciprocate in a professional, friendly way, then ask what you can help with today. ### Tool Usage Instructions You are equipped with a sophisticated data visualization tool, `generate_plot`, designed to create precise, publication-quality charts. Your primary function is to assist users in data analysis and interpretation by generating visual representations of their data. When a user's query involves numerical data that would benefit from visualization, you must invoke this tool. **Tool Signature:** `generate_plot(data: Dict[str, float], plot_type: Literal["bar", "line", "pie"], title: str, labels: List[str], x_label: str, y_label: str)` **Parameter Guide:** * `data` **(Required)**: A dictionary where keys are string labels and values are the corresponding numeric data points. * *Example:* `{"Experiment A": 88.5, "Experiment B": 92.1}` * `plot_type` **(Required)**: The specific type of chart to generate. This **must** be one of `"bar"`, `"line"`, or `"pie"`. * `title` (Optional): A formal title for the plot. * `x_label` (Optional): The label for the horizontal axis (for `bar` and `line` charts). * `y_label` (Optional): The label for the vertical axis (for `bar` and `line` charts). * `labels` (Optional): A list of strings to use as custom labels, overriding the keys from the `data` dictionary if necessary for specific ordering or formatting. **When to Use This Tool:** Invoke the `generate_plot` tool to address analytical and academic queries, such as: * **Trend Analysis:** Visualizing data points over a sequence to identify trends, growth, or decay (use a `line` chart). * **Comparative Analysis:** Comparing discrete quantities or categories against each other (use a `bar` chart). * **Proportional Distribution:** Illustrating the component parts of a whole, typically as percentages (use a `pie` chart). **Example Scenarios:** * **User Query:** "I need help practicing interpretation of trends in line graphs. To analyze the efficacy of a new fertilizer, I have recorded crop yield in kilograms over a five-week period. Please generate a line graph to visualize this growth trend and label the axes appropriately as 'Week' and 'Crop Yield (kg)'." * **Your Tool Call:** * `data`: `{"Week 1": 120, "Week 2": 155, "Week 3": 190, "Week 4": 210, "Week 5": 245}` * `plot_type`: `"line"` * `title`: `"Efficacy of New Fertilizer on Crop Yield"` * `x_label`: `"Week"` * `y_label`: `"Crop Yield (kg)"` * **User Query:** "I am studying for my ACT, and I am at a loss on interpreting the charts. For practice, consider this: a study surveyed the primary mode of transportation for 1000 commuters. The results were: 450 drive, 300 use public transit, 150 cycle, and 100 walk. Construct a pie chart to illustrate the proportional distribution of these methods." * **Your Tool Call:** * `data`: `{"Driving": 450, "Public Transit": 300, "Cycling": 150, "Walking": 100}` * `plot_type`: `"pie"` * `title`: `"Proportional Distribution of Commuter Transportation Methods"` NOTE: If specific data to use is not supplied, create reasonable data to create your charts. ## Academic Integrity and Response Guidelines - Do not provide full solutions. Instead: - **Guide through processes**: Break down problems into conceptual components - **Ask clarifying questions**: Understand what the student knows - **Provide similar examples**: Work through analogous problems - **Encourage original thinking**: Help students develop reasoning skills - **Suggest study strategies**: Recommend effective learning approaches - **Math problems**: Explain concepts and guide through steps without computing final answers - **Multiple-choice questions**: Discuss concepts being tested rather than identifying correct choices - **Essays**: Discuss research strategies and organizational techniques - **Factual questions**: Provide educational context and encourage synthesis ## Practice Question Templates **Multiple Choice** 1. 1 to 4 sentence question OPTIONAL, IF NEEDED. only INCLUDE A GRAPH, LINKED AS IMAGE, OR TABLE, NEVER BOTH.  | Example C1 | Example C2 |... | :---------------: | :----------------: |... | Content...... | Content....... |... A. Option B. Option C. Option D. Option --- **All That Apply** 1. 1 to 4 sentence question OPTIONAL, IF NEEDED. only INCLUDE A GRAPH, LINKED AS IMAGE, OR TABLE, NEVER BOTH.  | Example C1 | Example C2 |... | :---------------: | :----------------: |... | Content...... | Content....... |... - [ ] A. Option - [ ] B. Option - [ ] C. Option - [ ] D. Option --- **Written Response** 1. 1 to 4 sentence question OPTIONAL, IF NEEDED. only INCLUDE A GRAPH, LINKED AS IMAGE, OR TABLE, NEVER BOTH.  | Example C1 | Example C2 |... | :---------------: | :----------------: |... | Content...... | Content....... |... Prompt the user, in one sentence, to write their response """ # --- Stop Criteria --- class StopOnSequence(StoppingCriteria): def __init__(self, tokenizer, stop_sequence): self.tokenizer = tokenizer self.stop_sequence = tokenizer.encode(stop_sequence, add_special_tokens=False) def __call__(self, input_ids, scores, **kwargs): if input_ids[0, -len(self.stop_sequence):].tolist() == self.stop_sequence: return True return False # --- LLM Class with Phi-3 Mini --- class Phi3MiniEducationalLLM(Runnable): """LLM class optimized for Microsoft Phi-3-mini-4k-instruct with 4-bit quantization""" def __init__(self, model_path: str = "microsoft/Phi-3-mini-4k-instruct"): super().__init__() logger.info(f"Loading Phi-3-mini model with 4-bit quantization: {model_path}") start_Loading_Model_time = time.perf_counter() current_time = datetime.now() self.model_name = model_path try: # Load tokenizer (can be done on CPU) self.tokenizer = AutoTokenizer.from_pretrained( model_path, trust_remote_code=True, token=hf_token, use_fast=False ) # Configure 4-bit quantization self.quantization_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_compute_dtype=torch.bfloat16, bnb_4bit_quant_type="nf4", # NormalFloat 4-bit bnb_4bit_use_double_quant=True, # Nested quantization for extra savings ) # Store model path - model will be loaded inside GPU context self.model_path = model_path self.model = None except Exception as e: logger.error(f"Failed to initialize Phi-3-mini model {model_path}: {e}") raise # Ensure pad token exists if self.tokenizer.pad_token is None: self.tokenizer.pad_token = self.tokenizer.eos_token self.streamer = None def _load_model_if_needed(self): """Load model with 4-bit quantization only when needed inside GPU context""" if self.model is None: logger.info("Loading model with 4-bit quantization...") try: self.model = AutoModelForCausalLM.from_pretrained( self.model_path, quantization_config=self.quantization_config, torch_dtype=torch.bfloat16, trust_remote_code=True, low_cpu_mem_usage=True, token=hf_token, attn_implementation="eager", device_map="auto" ) logger.info(f"Model loaded successfully. Memory footprint reduced to ~2.2GB with 4-bit quantization") except Exception as e: logger.error(f"Failed to load quantized model: {e}") raise return self.model def _format_chat_template(self, prompt: str) -> str: """Format prompt using Phi-3's chat template""" try: messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": prompt} ] # Use Phi-3's chat template formatted_text = self.tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) return formatted_text except Exception as e: logger.warning(f"Chat template failed, using fallback format: {e}") # Fallback to manual Phi-3 format return f"<|system|>\n{SYSTEM_PROMPT}<|end|>\n<|user|>\n{prompt}<|end|>\n<|assistant|>\n" @spaces.GPU(duration=180) def invoke(self, input: Input, config=None) -> Output: """Main invoke method optimized for 4-bit quantized Phi‑3‑mini""" start_invoke_time = time.perf_counter() current_time = datetime.now() # Handle different input types if isinstance(input, dict): if 'input' in input: prompt = input['input'] elif 'messages' in input: prompt = str(input['messages']) else: prompt = str(input) else: prompt = str(input) try: model = self._load_model_if_needed() text = self._format_chat_template(prompt) try: max_input_length = 2000 - 400 inputs = self.tokenizer( text, return_tensors="pt", padding=True, truncation=True, max_length=max_input_length ) if 'input_ids' not in inputs: logger.error("Tokenizer did not return input_ids") return "I encountered an error processing your request. Please try again." except Exception as tokenizer_error: logger.error(f"Tokenization error: {tokenizer_error}") return "I encountered an error processing your request. Please try again." try: inputs = {k: v.to(model.device) for k, v in inputs.items()} except Exception as device_error: logger.error(f"Device transfer error: {device_error}") return "I encountered an error processing your request. Please try again." # Define stopping criteria after tokenizer initialization stop_criteria = StoppingCriteriaList([StopOnSequence(self.tokenizer, "User:")]) with torch.no_grad(): try: outputs = model.generate( input_ids=inputs['input_ids'], attention_mask=inputs.get('attention_mask', None), max_new_tokens=250, do_sample=True, temperature=0.4, top_p=0.9, top_k=50, repetition_penalty=1.1, pad_token_id=self.tokenizer.eos_token_id, use_cache=False, past_key_values=None, stopping_criteria=stop_criteria ) except Exception as generation_error: logger.error(f"Generation error: {generation_error}") return "I encountered an error generating the response. Please try again." try: new_tokens = outputs[0][len(inputs['input_ids'][0]):] result = self.tokenizer.decode(new_tokens, skip_special_tokens=True).strip() # Apply soft-stop cleanup for stop_word in ["User:", "\n\n", "###"]: if stop_word in result: result = result.split(stop_word)[0].strip() break except Exception as decode_error: logger.error(f"Decoding error: {decode_error}") return "I encountered an error processing the response. Please try again." end_invoke_time = time.perf_counter() invoke_time = end_invoke_time - start_invoke_time log_metric( f"LLM Invoke time (4‑bit): {invoke_time:0.4f} seconds. " f"Input length: {len(prompt)} chars. " f"Model: {self.model_name}. " f"Timestamp: {current_time:%Y‑%m‑%d %H:%M:%S}" ) return result if result else "I'm still learning how to respond to that properly." except Exception as e: logger.error(f"Generation error with 4‑bit model: {e}") end_invoke_time = time.perf_counter() invoke_time = end_invoke_time - start_invoke_time log_metric( f"LLM Invoke time (error): {invoke_time:0.4f} seconds. " f"Model: {self.model_name}. " f"Timestamp: {current_time:%Y‑%m‑%d %H:%M:%S}" ) return f"I encountered an error: {str(e)}" @spaces.GPU(duration=240) def stream_generate(self, input: Input, config=None): """Streaming generation with 4‑bit quantized model and expanded context""" start_stream_time = time.perf_counter() current_time = datetime.now() logger.info("Starting stream_generate with 4‑bit quantized model...") # Handle input properly if isinstance(input, dict): prompt = input.get('input', str(input)) else: prompt = str(input) try: model = self._load_model_if_needed() if torch.cuda.is_available(): torch.cuda.empty_cache() text = self._format_chat_template(prompt) try: inputs = self.tokenizer( text, return_tensors="pt", padding=True, truncation=True, max_length=2000 ) if 'input_ids' not in inputs: yield "I encountered an error processing your request. Please try again." return except Exception as tokenizer_error: logger.error(f"Streaming tokenization error: {tokenizer_error}") yield "I encountered an error processing your request. Please try again." return try: inputs = {k: v.to(model.device) for k, v in inputs.items()} except Exception as device_error: logger.error(f"Streaming device transfer error: {device_error}") yield "I encountered an error processing your request. Please try again." return streamer = TextIteratorStreamer( self.tokenizer, skip_prompt=True, skip_special_tokens=True ) generation_kwargs = { "input_ids": inputs['input_ids'], "attention_mask": inputs.get('attention_mask', None), "max_new_tokens": 250, "do_sample": True, "temperature": 0.7, "top_p": 0.9, "top_k": 50, "repetition_penalty": 1.2, "pad_token_id": self.tokenizer.eos_token_id, "streamer": streamer, "use_cache": False, "past_key_values": None } generation_thread = threading.Thread( target=model.generate, kwargs=generation_kwargs ) generation_thread.start() generated_text = "" consecutive_repeats = 0 last_chunk = "" try: for new_token_text in streamer: if not new_token_text: continue generated_text += new_token_text if new_token_text == last_chunk: consecutive_repeats += 1 if consecutive_repeats >= 5: logger.warning("Repetitive generation detected, stopping early") break else: consecutive_repeats = 0 last_chunk = new_token_text yield generated_text except Exception as e: logger.error(f"Error in streaming iteration: {e}") if not generated_text.strip(): generated_text = "I apologize, but I'm having trouble generating a response. Please try rephrasing your question." yield generated_text generation_thread.join() if not generated_text.strip(): generated_text = "I apologize, but I'm having trouble generating a response. Please try rephrasing your question." yield generated_text end_stream_time = time.perf_counter() stream_time = end_stream_time - start_stream_time log_metric( f"LLM Stream time (4‑bit): {stream_time:0.4f} seconds. " f"Generated length: {len(generated_text)} chars. " f"Model: {self.model_name}. " f"Timestamp: {current_time:%Y‑%m‑%d %H:%M:%S}" ) except Exception as e: logger.error(f"4‑bit streaming generation error: {e}") end_stream_time = time.perf_counter() stream_time = end_stream_time - start_stream_time log_metric( f"LLM Stream time (error): {stream_time:0.4f} seconds. " f"Model: {self.model_name}. " f"Timestamp: {current_time:%Y‑%m‑%d %H:%M:%S}" ) yield "I encountered an error generating the response. Please try again." @property def InputType(self) -> Type[Input]: return str @property def OutputType(self) -> Type[Output]: return str # LangGraph Agent Implementation with Tool Calling class Educational_Agent: """Modern LangGraph-based educational agent with Phi-3-mini and improved tool calling""" def __init__(self): start_init_and_langgraph_time = time.perf_counter() current_time = datetime.now() self.llm = Phi3MiniEducationalLLM(model_path="microsoft/Phi-3-mini-4k-instruct") self.tool_decision_engine = Tool_Decision_Engine(self.llm) # Create LangGraph workflow self.app = self._create_langgraph_workflow() end_init_and_langgraph_time = time.perf_counter() init_and_langgraph_time = end_init_and_langgraph_time - start_init_and_langgraph_time log_metric(f"Init and LangGraph workflow setup time: {init_and_langgraph_time:0.4f} seconds. Timestamp: {current_time:%Y-%m-%d %H:%M:%S}") def _create_langgraph_workflow(self): """Create the complete LangGraph workflow with improved tool calling""" # Use the updated Create_Graph_Tool tools = [Create_Graph_Tool] tool_node = ToolNode(tools) def call_model(state: EducationalAgentState) -> dict: """Call the LLM to generate a response""" start_call_model_time = time.perf_counter() current_time = datetime.now() messages = state["messages"] # Get the latest human message user_query = "" for msg in reversed(messages): if isinstance(msg, HumanMessage): user_query = msg.content break if not user_query: return {"messages": [AIMessage(content="I didn't receive a question. Please ask me something!")]} try: # Check if tools are needed based on state needs_tools = state.get("needs_tools", False) if needs_tools: # Create tool prompt that guides the model to use structured parameters tool_prompt = f""" You are an educational AI assistant. The user has asked: "{user_query}" This query would benefit from a visualization. Please call the Create_Graph_Tool with appropriate structured parameters. For the data parameter, create a meaningful dictionary with string keys and numeric values that illustrate the concept being discussed. Choose the appropriate plot_type: - "bar" for comparing categories or discrete data - "line" for showing trends over time or continuous relationships - "pie" for showing parts of a whole or proportions Create a descriptive title and appropriate axis labels. Include an educational_context explaining why this visualization helps learning. Call the tool with these structured parameters, don't format as JSON. """ prompt = tool_prompt else: prompt = user_query # Bind tools to LLM if needed if needs_tools: model_with_tools = self.llm # For Phi-3, we need to manually bind tools if supported try: if hasattr(self.llm, 'bind_tools'): model_with_tools = self.llm.bind_tools(tools) response = model_with_tools.invoke(prompt) except: # Fallback if tool binding not supported response = self.llm.invoke(prompt) else: response = self.llm.invoke(prompt) # Create AI message ai_message = AIMessage(content=response) end_call_model_time = time.perf_counter() call_model_time = end_call_model_time - start_call_model_time log_metric(f"Call model time: {call_model_time:0.4f} seconds. Timestamp: {current_time:%Y-%m-%d %H:%M:%S}") return {"messages": [ai_message]} except Exception as e: logger.error(f"Error in call_model: {e}") end_call_model_time = time.perf_counter() call_model_time = end_call_model_time - start_call_model_time log_metric(f"Call model time (error): {call_model_time:0.4f} seconds. Timestamp: {current_time:%Y-%m-%d %H:%M:%S}") error_message = AIMessage(content=f"I encountered an error generating a response: {str(e)}") return {"messages": [error_message]} def should_continue(state: EducationalAgentState) -> str: """Route to tools or end based on the last message""" last_message = state["messages"][-1] # Check if the last message has tool calls if hasattr(last_message, "tool_calls") and last_message.tool_calls: return "tools" else: return END def make_tool_decision(state: EducationalAgentState) -> dict: """Decide whether tools are needed and update state""" start_tool_decision_time = time.perf_counter() current_time = datetime.now() messages = state["messages"] # Get the latest human message user_query = "" for msg in reversed(messages): if isinstance(msg, HumanMessage): user_query = msg.content break if not user_query: return {"needs_tools": False} # Use the tool decision engine needs_visualization = self.tool_decision_engine.should_use_visualization(user_query) end_tool_decision_time = time.perf_counter() tool_decision_time = end_tool_decision_time - start_tool_decision_time log_metric(f"Tool decision workflow time: {tool_decision_time:0.4f} seconds. Decision: {needs_visualization}. Timestamp: {current_time:%Y-%m-%d %H:%M:%S}") return {"needs_tools": needs_visualization} # Create the workflow graph workflow = StateGraph(EducationalAgentState) # Add nodes workflow.add_node("decide_tools", make_tool_decision) workflow.add_node("call_model", call_model) workflow.add_node("tools", tool_node) # Add edges workflow.add_edge(START, "decide_tools") workflow.add_edge("decide_tools", "call_model") # Add conditional edge from call_model workflow.add_conditional_edges( "call_model", should_continue, {"tools": "tools", END: END} ) # After tools, go back to call_model for final response workflow.add_edge("tools", "call_model") # Compile the workflow return workflow.compile(checkpointer=MemorySaver()) def process_query(self, user_input: str, thread_id: str = "default") -> str: """Process a user query through the LangGraph workflow""" start_process_query_time = time.perf_counter() current_time = datetime.now() try: # Create initial state initial_state = { "messages": [HumanMessage(content=user_input)], "needs_tools": False, "educational_context": None } # Run the workflow config = {"configurable": {"thread_id": thread_id}} result = self.app.invoke(initial_state, config) # Extract the final response messages = result["messages"] # Combine AI message and tool results response_parts = [] for msg in messages: if isinstance(msg, AIMessage): # Clean up the response - remove JSON blocks if tools were used content = msg.content if "```json" in content and result.get("needs_tools", False): # Remove JSON blocks from display since tools handle visualization content = re.sub(r'```json.*?```', '', content, flags=re.DOTALL) content = content.strip() response_parts.append(content) elif isinstance(msg, ToolMessage): response_parts.append(msg.content) final_response = "\n\n".join(response_parts).strip() end_process_query_time = time.perf_counter() process_query_time = end_process_query_time - start_process_query_time log_metric(f"Total query processing time: {process_query_time:0.4f} seconds. Input: '{user_input[:50]}...'. Timestamp: {current_time:%Y-%m-%d %H:%M:%S}") return final_response if final_response else "I'm having trouble generating a response. Please try rephrasing your question." except Exception as e: logger.error(f"Error in process_query: {e}") end_process_query_time = time.perf_counter() process_query_time = end_process_query_time - start_process_query_time log_metric(f"Total query processing time (error): {process_query_time:0.4f} seconds. Timestamp: {current_time:%Y-%m-%d %H:%M:%S}") return f"I encountered an error processing your request: {str(e)}" def stream_query(self, user_input: str, thread_id: str = "default"): """Stream a response for a user query""" start_stream_query_time = time.perf_counter() current_time = datetime.now() try: # For streaming, we'll use the LLM directly with tool decision needs_tools = self.tool_decision_engine.should_use_visualization(user_input) if needs_tools: # Create tool prompt tool_prompt = f""" You are an educational AI assistant. The user has asked: "{user_input}" This query would benefit from a visualization. Please provide a helpful educational response AND include a JSON configuration for creating a graph or chart. Format your response with explanatory text followed by a JSON block like this: ```json {{ "data": {{"Category 1": value1, "Category 2": value2}}, "plot_type": "bar|line|pie", "title": "Descriptive Title", "x_label": "X Axis Label", "y_label": "Y Axis Label", "educational_context": "Explanation of why this visualization helps learning" }} ``` Provide your educational response followed by the JSON configuration. """ prompt = tool_prompt else: prompt = user_input # Stream the response full_response = "" for chunk in self.llm.stream_generate(prompt): full_response = chunk yield chunk # Process tools if needed after streaming completes if needs_tools and "```json" in full_response: json_pattern = r'```json\s*\n?(.*?)\n?```' json_matches = re.findall(json_pattern, full_response, re.DOTALL) if json_matches: json_config = json_matches[0].strip() try: # Validate and process the JSON config_dict = json.loads(json_config) required_keys = ['data', 'plot_type', 'title'] if all(key in config_dict for key in required_keys): # Generate the visualization tool_result = Create_Graph_Tool.invoke({"graph_config": json_config}) # Clean response and add visualization cleaned_response = re.sub(r'```json.*?```', '', full_response, flags=re.DOTALL).strip() final_response = f"{cleaned_response}\n\n{tool_result}" yield final_response except (json.JSONDecodeError, Exception) as e: logger.error(f"Error processing streamed JSON: {e}") end_stream_query_time = time.perf_counter() stream_query_time = end_stream_query_time - start_stream_query_time log_metric(f"Stream query total time: {stream_query_time:0.4f} seconds. Input: '{user_input[:50]}...'. Timestamp: {current_time:%Y-%m-%d %H:%M:%S}") except Exception as e: logger.error(f"Error in stream_query: {e}") end_stream_query_time = time.perf_counter() stream_query_time = end_stream_query_time - start_stream_query_time log_metric(f"Stream query total time (error): {stream_query_time:0.4f} seconds. Timestamp: {current_time:%Y-%m-%d %H:%M:%S}") yield f"I encountered an error: {str(e)}" # Gradio Interface Functions def warmup_agent(): """Warm up the agent with a simple test query""" try: logger.info("Warming up Phi-3-mini educational agent...") start_warmup_time = time.perf_counter() # Simple warmup query warmup_response = agent.process_query("Hello", thread_id="warmup") end_warmup_time = time.perf_counter() warmup_time = end_warmup_time - start_warmup_time log_metric(f"Agent warmup completed in {warmup_time:.2f} seconds") logger.info(f"Warmup response: {warmup_response[:100]}...") except Exception as e: logger.error(f"Warmup failed: {e}") def add_user_message(message, chat_history, conversation_state): """Add user message to state and display immediately""" if not message.strip(): return "", chat_history, conversation_state # Add to conversation state conversation_state.append({"role": "user", "content": message}) # Update chat display chat_history.append({"role": "user", "content": message}) return "", chat_history, conversation_state def add_thinking_indicator(chat_history, conversation_state): """Add thinking indicator to chat display""" if not conversation_state: return chat_history, conversation_state # Add simple animated dots to chat display (not permanent state) thinking_html = get_thinking_dots() chat_history.append({"role": "assistant", "content": thinking_html}) return chat_history, conversation_state def generate_response(chat_history, conversation_state): """Generate streaming response from the agent""" if not conversation_state: return chat_history, conversation_state # Get the last user message last_user_message = "" for msg in reversed(conversation_state): if msg["role"] == "user": last_user_message = msg["content"] break if not last_user_message: return chat_history, conversation_state try: # Stream the response full_response = "" for chunk in agent.stream_query(last_user_message): full_response = chunk # Update the last message in chat display (replace thinking indicator) if chat_history and chat_history[-1]["role"] == "assistant": chat_history[-1]["content"] = full_response else: chat_history.append({"role": "assistant", "content": full_response}) yield chat_history, conversation_state # Add final response to permanent conversation state conversation_state.append({"role": "assistant", "content": full_response}) yield chat_history, conversation_state except Exception as e: logger.error(f"Error in generate_response: {e}") error_msg = f"I encountered an error: {str(e)}" # Update display if chat_history and chat_history[-1]["role"] == "assistant": chat_history[-1]["content"] = error_msg else: chat_history.append({"role": "assistant", "content": error_msg}) # Add to permanent state conversation_state.append({"role": "assistant", "content": error_msg}) yield chat_history, conversation_state def reset_conversation(): """Reset both chat display and conversation state""" return [], [] # --- UI: Interface Creation --- def create_interface(): """Creates and configures the complete Gradio interface with proper state management.""" start_create_interface_time = time.perf_counter() current_time = datetime.now() # Read CSS file custom_css = "" try: with open("styles.css", "r", encoding="utf-8") as css_file: custom_css = css_file.read() except FileNotFoundError: logger.warning("styles.css file not found, using default styling") except Exception as e: logger.warning(f"Error reading styles.css: {e}") try: with open("styles.css", "r", encoding="utf-8") as css_file: custom_css = css_file.read() logger.info(f"CSS loaded successfully, length: {len(custom_css)} characters") # Check if dots CSS is actually in the file if ".thinking-indicator" in custom_css: logger.info("Dots CSS found in file") else: logger.warning("Dots CSS NOT found in file") except FileNotFoundError: logger.warning("styles.css file not found, using default styling") with gr.Blocks( title="Mimir", fill_width=True, fill_height=True, theme=gr.themes.Origin() ) as demo: # Add head content gr.HTML(html_head_content) gr.HTML(force_light_mode) # State management - this is the key addition conversation_state = gr.State([]) # Persistent conversation memory with gr.Column(elem_classes=["main-container"]): # Title Section gr.HTML('