Update app.py

app.py

@@ -2,32 +2,31 @@ import datetime
 import json
 import os
 import time
+import re
+import requests
+import streamlit as st
+from dotenv import load_dotenv
+from pydantic import BaseModel
+
+# LangChain, LangGraph, and related imports
 from langchain_core.messages import SystemMessage, HumanMessage
 from langchain_community.tools.tavily_search import TavilySearchResults
 from langchain_community.document_loaders import WebBaseLoader
 from langchain_community.vectorstores import FAISS
 from langchain_text_splitters import RecursiveCharacterTextSplitter
 from langchain.tools.retriever import create_retriever_tool
-from langgraph.graph import StateGraph, START
-from langgraph.checkpoint.memory import MemorySaver
-from langgraph.graph.state import CompiledStateGraph
-from langgraph.graph import MessagesState
-from langgraph.prebuilt import ToolNode
-from langgraph.prebuilt import tools_condition
-from dotenv import load_dotenv
 from langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAIEmbeddings
-from langchain_community.tools.tavily_search import TavilySearchResults
-from langgraph.graph import END, StateGraph
-from nasapy import Nasa
-from pydantic import BaseModel
-import requests
-import streamlit as st
+from langgraph.graph import StateGraph, START, END
+from langgraph.graph.state import MessagesState
+from langgraph.prebuilt import ToolNode, tools_condition
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.graph import CompiledStateGraph
 import tavily
 
 # Load environment variables
 load_dotenv()
 
-# Initialize models and APIs
+# Initialize the LLM (using Gemini-1.5 Flash) with Google API key from env variables
 llm = ChatGoogleGenerativeAI(
     model="gemini-1.5-flash",
     google_api_key=os.getenv("GOOGLE_API_KEY")
@@ -36,77 +35,61 @@ llm = ChatGoogleGenerativeAI(
 def space_events_agent():
     """
     Fetches and formats upcoming space events from NASA's EONET API.
-
-    Args:
-        state (AgentState): The current agent state.
-
     Returns:
-        dict: A dictionary with the formatted space events or a message if no events are found.
-
-    Raises:
-        HTTPError: If the request to the NASA API fails.
-
-    Note:
-        Requires NASA API key set in the environment variable `NASA_API_KEY`.
+        dict: A dictionary with a detailed formatted message of the current space events.
     """
     base_url = "https://eonet.gsfc.nasa.gov/api/v2.1/events"
-    # Using a separate NASA API key for EONET if needed
     params = {"api_key": os.getenv("NASA_API_KEY")}
-    response = requests.get(base_url, params=params)
-    response.raise_for_status()
-    events = response.json().get("events", [])
+    try:
+        response = requests.get(base_url, params=params)
+        response.raise_for_status()
+        events = response.json().get("events", [])
+    except Exception as e:
+        return {"output": f"🚀 Space Events Agent Error: Unable to fetch events. Details: {str(e)}"}
+    
     if events:
         event_details = []
         for event in events:
-            # Safely retrieve category and coordinates
             category = (event.get('categories', [{}])[0]).get('title', 'Unknown Category')
             geometries = event.get('geometries', [])
             coordinates = geometries[0].get('coordinates', 'Unknown Location') if geometries else 'Unknown Location'
+            # Include additional detailed formatting for each event
             event_details.append(
-                f"• Event: {event.get('title', 'N/A')}\n"
-                f"  Type: {category}\n"
-                f"  Location: {coordinates}\n"
+                f"• **Event:** {event.get('title', 'N/A')}\n"
+                f"  **Type:** {category}\n"
+                f"  **Location (Lon, Lat):** {coordinates}\n"
             )
         formatted_events = "\n".join(event_details)
-        return {"output": f"🚀 Space Events Agent:\nn{formatted_events}"}
+        detailed_message = (
+            f"🚀 **Space Events Agent:**\n\n"
+            f"Here are the latest space events with detailed information:\n\n{formatted_events}\n"
+            f"*(Data retrieved from NASA's EONET API)*"
+        )
+        return {"output": detailed_message}
     else:
-        return {"output": f"🚀 Space Events Agent:No events found at the moment."}
+        return {"output": "🚀 Space Events Agent: No events found at the moment. Please check back later for updates."}
 
+# Load upcoming launches data and prepare a retriever for space news
 loader1 = WebBaseLoader("https://spacelaunchnow.me/3.3.0/launch/upcoming/")
 docs1 = loader1.load()
-documents = RecursiveCharacterTextSplitter(
-    chunk_size=1000, chunk_overlap=200
-).split_documents(docs1)
+documents = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200).split_documents(docs1)
 vector = FAISS.from_documents(documents, GoogleGenerativeAIEmbeddings(model="models/embedding-001"))
-
 retriever = vector.as_retriever()
 retriever_tool = create_retriever_tool(
     retriever,
     "spacelaunchnow",
-    "Search for information about space real time news",
+    "Search for up-to-date space launch news and real time information"
 )
 
 def astronomy_image_agent(user_input: str):
     """
-    Retrieves astronomy-related images from NASA's Image and Video Library or the Astronomy Picture of the Day (APOD) API.
-    
-    If the user query contains "galaxy", it fetches galaxy images from NASA's Image and Video Library. 
-    If no specific keyword is found, it defaults to fetching the Astronomy Picture of the Day (APOD).
-
-    Args:
-        user_input (str): The user input containing the query.
-
+    Retrieves astronomy-related images. If the query mentions 'galaxy', it fetches an image from
+    NASA's Image and Video Library; otherwise it fetches NASA’s Astronomy Picture of the Day (APOD).
     Returns:
-        dict: A dictionary containing details about the requested astronomy image, 
-              including the title, description, and URL. If an error occurs, an error message is returned.
-
-    Raises:
-        HTTPError: If the request to the NASA API fails.
-        ValueError: If no relevant image or data is found.
+        dict: A dictionary with detailed image information including title, explanation, and image URL.
     """
     user_input = user_input.lower()
     if "galaxy" in user_input:
-        # Query NASA's Image and Video Library API for galaxy images.
         search_url = "https://images-api.nasa.gov/search"
         params = {"q": "galaxy", "media_type": "image"}
         try:
@@ -121,197 +104,157 @@ def astronomy_image_agent(user_input: str):
                 title = first_item.get("data", [{}])[0].get("title", "No Title")
                 description = first_item.get("data", [{}])[0].get("description", "No Description")
                 result = (
-                    f"Title: {title}\n"
-                    f"Description: {description}\n"
-                    f"Image URL: {image_url}"
+                    f"**Title:** {title}\n"
+                    f"**Description:** {description}\n"
+                    f"**Image URL:** {image_url}\n\n"
+                    f"This image was retrieved from NASA's Image and Video Library."
                 )
-                return {"output": f"✨ Galaxy Image Agent:\n\n{result}"}
+                return {"output": f"✨ **Galaxy Image Agent:**\n\n{result}"}
             else:
-                return {"output": "No galaxy images found."}
+                return {"output": "✨ Galaxy Image Agent: No galaxy images found. Please try another query."}
         except Exception as e:
-            return {"output": f"Galaxy image search failed: {str(e)}"}
+            return {"output": f"✨ Galaxy Image Agent Error: {str(e)}"}
     else:
-        # Default to the APOD API if no keyword is detected.
+        # Default to Astronomy Picture of the Day (APOD)
         url = f"https://api.nasa.gov/planetary/apod?api_key={os.getenv('NASA_API_KEY')}"
         try:
             response = requests.get(url)
             response.raise_for_status()
             data = response.json()
             result = (
-                f"Title: {data.get('title', 'N/A')}\n"
-                f"Date: {data.get('date', 'N/A')}\n"
-                f"Explanation: {data.get('explanation', 'N/A')}\n"
-                f"Image URL: {data.get('url', 'N/A')}\n"
-                f"Media Type: {data.get('media_type', 'N/A')}\n"
-                f"Copyright: {data.get('copyright', 'Public Domain')}"
+                f"**Title:** {data.get('title', 'N/A')}\n"
+                f"**Date:** {data.get('date', 'N/A')}\n"
+                f"**Explanation:** {data.get('explanation', 'N/A')}\n"
+                f"**Image URL:** {data.get('url', 'N/A')}\n"
+                f"**Media Type:** {data.get('media_type', 'N/A')}\n"
+                f"**Copyright:** {data.get('copyright', 'Public Domain')}\n\n"
+                f"This Astronomy Picture of the Day is provided by NASA's APOD API and is updated daily."
             )
-            return {"output": f"✨ Astronomy Image Agent:\n\n{result}"}
+            return {"output": f"✨ **Astronomy Image Agent:**\n\n{result}"}
         except Exception as e:
-            return {"output": f"Astronomy Image Agent Error: {str(e)}"}
+            return {"output": f"✨ Astronomy Image Agent Error: {str(e)}"}
 
 def educational_resources_agent():
     """
-    Retrieves and formats a list of high-quality educational resources on astronomy and space science. 
-    Uses Tavily search to find trusted sources such as NASA, ESA, and other reputable platforms.
-
-    Args:
-        state (AgentState): The current agent state, containing user input.
-
+    Retrieves and formats a list of high-quality educational resources on astronomy and space science.
+    Uses Tavily search to fetch resources from trusted sources.
     Returns:
-        dict: A dictionary containing a list of educational resources with titles, descriptions, and clickable links.
-
-    Raises:
-        Exception: If an error occurs while retrieving the resources or formatting the response.
+        dict: A dictionary containing a detailed list of curated educational resources.
     """
     try:
-        # Use Tavily search to retrieve educational resources based on the query
-        resources = search.invoke({"query": "Provide a curated list of high-quality educational resources on astronomy and space science, from trusted sources like NASA, ESA, and reputable educational platforms."})
+        resources = search.invoke({"query": "Provide a curated list of high-quality educational resources on astronomy and space science from trusted sources like NASA, ESA, and reputable educational platforms."})
         if not resources:
-            return {"output": "No educational resources found. Please try again."}
+            return {"output": "🎓 Educational Resources Agent: No educational resources found. Please try again."}
         
         formatted_resources = "\n\n".join(
-            [f"📘 {r.get('title', 'Untitled Resource')}\n🔗 {r.get('url', 'No URL provided')}" for r in resources]
+            [f"📘 **{r.get('title', 'Untitled Resource')}**\n🔗 {r.get('url', 'No URL provided')}" for r in resources]
         )
-        return {"output": f"🎓 Educational Resources Agent:\n\n{formatted_resources}"}
+        detailed_message = (
+            f"🎓 **Educational Resources Agent:**\n\n"
+            f"Below is a curated list of high-quality educational resources on astronomy and space science. "
+            f"Click on the links for more detailed information:\n\n{formatted_resources}"
+        )
+        return {"output": detailed_message}
     except Exception as e:
-        return {"output": f"Educational Resources Agent Error: {str(e)}"}
+        return {"output": f"🎓 Educational Resources Agent Error: {str(e)}"}
 
 def iss_location_agent():
     """
-    Fetches astronaut data and the current position of the International Space Station (ISS).
-    
-    It fetches the list of astronauts currently on the ISS, and retrieves the current position of the ISS.
-    The data is updated at intervals and returned in a structured format.
-    
+    Fetches detailed astronaut data and the current position of the International Space Station (ISS).
     Returns:
-        dict: A dictionary containing astronaut information and ISS position updates.
-              Includes a list of people on the ISS and the positions over time.
+        dict: A dictionary containing a detailed formatted message with astronaut information and ISS position updates.
     """
     # Fetch astronaut data
     api_url = "http://api.open-notify.org/astros.json"
-    data = requests.get(api_url).json()
-
-    result = {}
-
-    if data['message'] == 'success':
-        with open('iss-astros.json', 'w') as iss_file:
-            json.dump(data, iss_file, indent=4)
-
-        result["astronauts"] = f"There are currently {data['number']} people on the ISS."
-        result["people"] = []
-        for person in data['people']:
-            result["people"].append(f"{person['name']} is currently on the ISS, Craft: {person['craft']}")
+    try:
+        data = requests.get(api_url).json()
+    except Exception as e:
+        return {"output": f"🚀 ISS Location Agent Error: Unable to fetch astronaut data. Details: {str(e)}"}
+    
+    output_lines = []
+    if data.get('message') == 'success':
+        astronauts_line = f"There are currently {data.get('number', 'N/A')} people in space."
+        people_lines = "\n".join([f"- {person['name']} aboard {person['craft']}" for person in data.get('people', [])])
+        output_lines.append(f"**Astronaut Information:**\n{astronauts_line}\n{people_lines}")
     else:
-        result["error"] = 'Failed to obtain astronauts data.'
-
-    # List to hold ISS position updates
+        output_lines.append("Error: Failed to obtain astronaut data.")
+    
+    # Fetch ISS current position data with retries
     positions = []
-
-    # fetching ISS position data
-    for attempt in range(5):  # Retry 5 times
+    for attempt in range(5):
         try:
-            data = requests.get("http://api.open-notify.org/iss-now.json", timeout=10).json()
-            if data['message'] == 'success':
-                # Extract ISS location
-                location = data["iss_position"]
-                latitude = float(location['latitude'])
-                longitude = float(location['longitude'])
-
-                position = {
-                    "latitude": latitude,
-                    "longitude": longitude,
-                    "time": str(datetime.datetime.fromtimestamp(data['timestamp']))
-                }
-                positions.append(position)
-                break  # Successfully fetched data, break the loop
+            pos_data = requests.get("http://api.open-notify.org/iss-now.json", timeout=10).json()
+            if pos_data.get('message') == 'success':
+                location = pos_data.get("iss_position", {})
+                latitude = float(location.get('latitude', 0))
+                longitude = float(location.get('longitude', 0))
+                timestamp = datetime.datetime.fromtimestamp(pos_data.get('timestamp', 0)).strftime("%Y-%m-%d %H:%M:%S")
+                positions.append(f"- Latitude: {latitude}, Longitude: {longitude} (Recorded at: {timestamp})")
+                break
         except requests.exceptions.RequestException as e:
-            result["error"] = str(e)
-            time.sleep(3)  # Wait for 3 seconds before retrying
-
-    result["positions"] = positions if positions else "Failed to retrieve ISS position data after retries."
-    return result
-
+            time.sleep(3)
+    if positions:
+        output_lines.append("**Current ISS Position:**\n" + "\n".join(positions))
+    else:
+        output_lines.append("Error: Failed to retrieve ISS position data after multiple attempts.")
+    
+    detailed_message = (
+        f"🚀 **ISS Location Agent:**\n\n" +
+        "\n\n".join(output_lines) +
+        "\n\n*(Data retrieved from open-notify.org)*"
+    )
+    return {"output": detailed_message}
+
+# Initialize Tavily search tool using the provided API key
 search = TavilySearchResults(tavily_api_key=os.getenv("TAVILY_API_KEY"))
 
+# List of tools available to the assistant
 tools = [search, iss_location_agent, space_events_agent, astronomy_image_agent, educational_resources_agent]
 llm_with_tools = llm.bind_tools(tools)
 
-# System message
-sys_msg = SystemMessage(content='''This system is designed to provide real-time astronomical data, visualizations, and educational content. Below are the key functions and tools integrated into the system and their specific purposes:
+# Updated system message instructing detailed and informative responses
+sys_msg = SystemMessage(content='''This system is designed to provide real-time astronomical data, visualizations, and educational content. The assistant must generate responses that are detailed, informative, and rich in context. Below are the key functions and tools integrated into the system and their specific purposes:
 
 1. **Tavily Search (`search`) Integration**:
-   - **Purpose**: Provides users with up-to-date, relevant space-related information from Tavily's extensive search engine.
-   - **Usage**: Enables the assistant to fetch the latest news, research articles, and educational resources related to space.
-
-2. **NASA APOD Tool (`get_nasa_apod_tool`)**:
-   - **Purpose**: Fetches the Astronomy Picture of the Day (APOD) from NASA's APOD API.
-   - **Usage**: Provides the title, explanation, and image URL of the latest astronomy image shared by NASA, offering users daily insights into the wonders of space.
+   - **Purpose**: Provides up-to-date, relevant space-related information.
+   - **Usage**: Fetches the latest news, research articles, and educational resources related to space.
 
-3. **NASA EONET Space Events (`fetch_nasa_eonet_events`)**:
-   - **Purpose**: Retrieves real-time space-related events from NASA’s EONET API.
-   - **Usage**: Provides details about ongoing or upcoming space events, including event type, title, location, and date, offering users information about significant celestial events.
+2. **NASA APOD Tool**:
+   - **Purpose**: Fetches NASA’s Astronomy Picture of the Day (APOD).
+   - **Usage**: Provides detailed information including title, explanation, and image URL.
 
-4. **ISS Location Plotting (`plot_iss_location`)**:
-   - **Purpose**: Displays the current real-time location of the International Space Station (ISS).
-   - **Usage**: Visualizes the ISS position on a latitude-longitude graph, allowing users to track its orbit across the globe in real-time.
+3. **NASA EONET Space Events**:
+   - **Purpose**: Retrieves real-time space events.
+   - **Usage**: Provides detailed data on ongoing/upcoming space events.
 
-5. **ISS Astronaut Data (`ISS_data`)**:
-   - **Purpose**: Provides data about astronauts currently aboard the ISS and their associated spacecraft.
-   - **Usage**: Retrieves information on the number of astronauts aboard, their spacecraft, and visualizes the ISS’s position on a world map.
+4. **ISS Location and Astronaut Data**:
+   - **Purpose**: Displays the current ISS location and astronaut information.
+   - **Usage**: Provides a detailed update on ISS coordinates and people in space.
 
-6. **Space Educational Fact (`educational_space_fact`)**:
-   - **Purpose**: Fetches random educational facts about space from the Space Trivia API.
-   - **Usage**: Provides fascinating trivia about celestial bodies, space phenomena, and other space-related topics, helping to educate users in an engaging manner.
+5. **Space Educational Resources**:
+   - **Purpose**: Curates educational resources on astronomy and space science.
+   - **Usage**: Provides a detailed list with titles and clickable links.
 
-7. **NEO Data Fetcher (`get_neo_data`)**:
-   - **Purpose**: Retrieves real-time data on Near-Earth Objects (NEOs) from NASA’s NEO API.
-   - **Usage**: Provides information on potentially hazardous asteroids and comets, helping users stay informed about objects that come close to Earth.
-8. **Space News Agent (`retriever_tool`)**:
-   - **Purpose**: Fetches the latest space news articles.
-   - **Usage**: Provides users with up-to-date news articles related to space exploration, satellite launches, and astronomical discoveries.
+When responding to user queries, the assistant should provide comprehensive, detailed, and accurate explanations. Visual responses should include image URLs that the interface can use to display pictures. All responses should strive for clarity and depth in explanation.
+''')
 
-### Workflow:
-- The system responds dynamically to user queries by identifying the appropriate tool for each request.  
-- If the user asks for data or visualizations related to astronomical objects or space events, the relevant NASA APIs or visualization tools are invoked.  
-- For general space education or research, tools like Tavily and Space Trivia provide comprehensive, up-to-date educational content.  
-- The system remembers the context of conversations to ensure relevant and seamless interactions with users, supported by the `MemorySaver`.
-
-### Guidelines:
-- **Scientific Accuracy**: All provided information is scientifically accurate and up-to-date.
-- **Engagement**: Educational content is easy to understand and presented in an interactive and engaging way.
-- **Visualization**: For requests requiring visual representations (such as ISS tracking or NEO data), interactive orbital plots or maps are rendered for clarity.
-- **Tool Alignment**: The system uses the most appropriate tool based on the user’s request, ensuring minimal invocation of unnecessary functions.
-
-This setup ensures that the assistant can efficiently address a wide range of user queries related to space, astronomy, and related educational content while keeping interactions intuitive and engaging.''')
-
-
-# Node
+# Define the assistant node function
 def assistant(state: MessagesState) -> MessagesState:
     return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"][-10:])]}
 
-# Build graph
+# Build the LangGraph state graph
 builder: StateGraph = StateGraph(MessagesState)
-
-# Define nodes: these do the work
 builder.add_node("assistant", assistant)
 builder.add_node("tools", ToolNode(tools))
-
-# Define edges: these determine how the control flow moves
 builder.add_edge(START, "assistant")
-builder.add_conditional_edges(
-    "assistant",
-    # If the latest message (result) from assistant is a tool call -> tools_condition routes to tools
-    # If the latest message (result) from assistant is a not a tool call -> tools_condition routes to END
-    tools_condition,
-)
+builder.add_conditional_edges("assistant", tools_condition)
 builder.add_edge("tools", "assistant")
 memory: MemorySaver = MemorySaver()
-# Compile the workflow into an application
 react_graph_memory: CompiledStateGraph = builder.compile(checkpointer=memory)
 
-
+# Streamlit App Layout
 st.title("🚀 Space Agentic Chatbot")
-st.caption("Your AI-powered space exploration assistant")
+st.caption("Your AI-powered space exploration assistant providing detailed and informative insights.")
 
 if "messages" not in st.session_state:
     st.session_state.messages = []
@@ -319,44 +262,39 @@ if "messages" not in st.session_state:
 # Display existing chat messages
 for message in st.session_state.messages:
     with st.chat_message(message["role"]):
+        # If the message content is a URL (likely an image), display it
         if message["content"].startswith("http"):
             st.image(message["content"])
         else:
             st.write(message["content"])
 
 # User input via chat
-if user_input := st.chat_input("Ask about space news, ISS location, or astronomy images"):
+if user_input := st.chat_input("Ask about space news, ISS location, astronomy images, or educational content"):
     st.session_state.messages.append({"role": "user", "content": user_input})
-    st.chat_message("user").write(user_input)
-
+    with st.chat_message("user"):
+        st.write(user_input)
     try:
-        # Prepare the messages for the model
+        # Prepare the message for the model and invoke the state graph
         messages = [HumanMessage(content=user_input)]
-
-        # Call the memory and invoke the response
         response = react_graph_memory.invoke({"messages": messages}, config={"configurable": {"thread_id": "1"}})
         assistant_response = response["messages"][-1].content
 
-        # Handle image URLs and regular text responses
+        # Check for image URL markers in a case-insensitive way
+        image_url = None
         if "image url:" in assistant_response.lower():
-            image_url = assistant_response.split("Image URL:")[1].split("\n")[0].strip()
-            
-            # Store the image URL and the response message in session state
-            st.session_state.messages.append({"role": "assistant", "content": image_url})
-            st.session_state.messages.append({"role": "assistant", "content": assistant_response})
-            with st.chat_message("assistant"):
-                st.image(image_url)  # Display the image
-                st.write(assistant_response)  # Display the accompanying text
-
-        else:
-            # If no image URL, just store the regular text response
-            st.session_state.messages.append({"role": "assistant", "content": assistant_response})
-            with st.chat_message("assistant"):
-                st.write(assistant_response)
+            # Use regex to extract the first URL after "image url:"
+            match = re.search(r"[Ii]mage [Uu][Rr][Ll]:\s*(\S+)", assistant_response)
+            if match:
+                image_url = match.group(1).strip()
 
+        # Save and display the assistant's response
+        st.session_state.messages.append({"role": "assistant", "content": assistant_response})
+        with st.chat_message("assistant"):
+            if image_url:
+                st.image(image_url)  # Display the image
+            st.write(assistant_response)
     except Exception as e:
         error_msg = f"Error processing request: {str(e)}"
-        # Store the error message in session state
         st.session_state.messages.append({"role": "assistant", "content": error_msg})
         with st.chat_message("assistant"):
             st.write(error_msg)