import streamlit as st import requests import json # API URL API_URL = "https://startrz-devi.hf.space/api/v1/prediction/e54adffc-ae77-42e5-9fc0-c4584e081093" def safe_get(obj, *keys, default=None): """ Safely navigate through nested dictionaries Args: obj: Starting object *keys: Sequence of keys to navigate default: Value to return if navigation fails Returns: Value at the specified path or default """ try: for key in keys: obj = obj[key] return obj except (TypeError, KeyError, IndexError): return default def parse_text_content(data): """ Parse text content from the API response Args: data (dict): Full API response Returns: dict: Parsed text content with title and content """ try: # Try to parse the text field as JSON text_content = json.loads(safe_get(data, "text", default="{}")) return { "title": text_content.get("title", "No Title Available"), "content": text_content.get("content", "No Content Available"), "outline": text_content.get("outline", []) } except (json.JSONDecodeError, TypeError): # Fallback if parsing fails return { "title": "Analysis Result", "content": safe_get(data, "text", default="No content available"), "outline": [] } def parse_tool_details(tool): """ Parse tool details with robust handling of different input types Args: tool (dict): A single tool dictionary from the API response Returns: dict: Parsed tool details with consistent structure """ # Ensure tool is a dictionary if not isinstance(tool, dict): return { "tool": "Unknown Tool", "toolInput": "Invalid tool data", "toolOutput": "No output available" } # Parse toolInput input_value = "" tool_input = tool.get("toolInput", {}) if isinstance(tool_input, dict): input_value = tool_input.get("input", "") # Fallback to full input dict as string if no 'input' key if not input_value: try: input_value = json.dumps(tool_input, indent=2) except Exception: input_value = str(tool_input) elif isinstance(tool_input, str): input_value = tool_input else: input_value = str(tool_input) if tool_input is not None else "No input details" # Parse toolOutput output_value = tool.get("toolOutput") # Flexible output handling if output_value is None: output_value = "No output available" elif isinstance(output_value, (list, dict)): # Convert to formatted JSON string for better readability try: output_value = json.dumps(output_value, indent=2) except Exception: output_value = str(output_value) else: # Convert to string for any other type output_value = str(output_value) return { "tool": tool.get("tool", "Unknown Tool"), "toolInput": input_value, "toolOutput": output_value } def query(payload): """ Query the API and process the response Args: payload (dict): Question payload to send to the API Returns: dict: Processed response with tool details """ try: # Send POST request to the API response = requests.post(API_URL, json=payload) response.raise_for_status() # Parse the JSON response data = response.json() # Extract text content text_content = parse_text_content(data) # Extract tool details tool_details = [] # Handle different potential response structures agent_reasoning = safe_get(data, "agentReasoning", default=[]) for reasoning in agent_reasoning: # Safely extract used tools used_tools = safe_get(reasoning, "usedTools", default=[]) for tool in used_tools: if tool is not None: parsed_tool = parse_tool_details(tool) tool_details.append(parsed_tool) return { "raw_response": data, "text_content": text_content, "tool_details": tool_details } except requests.exceptions.RequestException as e: return {"error": f"API Request Error: {str(e)}"} except json.JSONDecodeError as e: return {"error": f"JSON Parsing Error: {str(e)}"} except Exception as e: return {"error": f"Unexpected Error: {str(e)}"} def display_outline(outline): """ Display the document outline in an expandable section Args: outline (list): List of outline sections """ if not outline: return with st.expander("📋 Document Outline"): for section in outline: st.markdown(f"### {section.get('section_title', 'Untitled Section')}") key_points = section.get('key_points', []) for point in key_points: st.markdown(f"- {point}") def main(): """ Main Streamlit application function """ st.set_page_config( page_title="DEVI Research Assistant", page_icon="🔍", layout="wide" ) st.title("🔬 DEVI RESEARCH ASSISTANT") st.write("Explore insights by asking a research question!") # User input section user_input = st.text_input( "What would you like to research?", placeholder="Enter your research query here..." ) # Submit button if st.button("Explore Insights", type="primary"): if user_input: # Progress spinner during API call with st.spinner("Gathering research insights..."): response = query({"question": user_input}) # Error handling if "error" in response: st.error(response["error"]) return # Display Text Content st.header("📄 Research Insights") # Extract text content text_content = response.get("text_content", {}) # Display Title st.subheader(text_content.get("title", "Research Analysis")) # Display Content st.write(text_content.get("content", "No content available")) # Display Outline display_outline(text_content.get("outline", [])) # Display Online Resources st.header("🌐 Online Resources") tool_details = response.get("tool_details", []) if tool_details: # Create tabs for each resource tabs = st.tabs([ f"{idx+1}. {tool.get('tool', 'Unknown')}" for idx, tool in enumerate(tool_details) ]) # Populate each tab with resource details for idx, (tool, tab) in enumerate(zip(tool_details, tabs)): with tab: st.subheader("Research Name") st.code(tool.get('toolInput', 'No input'), language=None) st.subheader("Research Findings") # Use st.code for better formatting st.code(tool.get('toolOutput', 'No output'), language=None) else: st.info("No resources found for this query.") # Raw response in expander for advanced users with st.expander("🔍 Advanced: Full API Response"): st.json(response.get("raw_response", {})) else: st.warning("Please enter a research question!") if __name__ == "__main__": main()