Update app.py

app.py

@@ -3,29 +3,32 @@ 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 langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAIEmbeddings
-from langgraph.graph import StateGraph, START, END
-from langgraph.prebuilt import ToolNode, tools_condition
+from langgraph.graph import StateGraph, START
 from langgraph.checkpoint.memory import MemorySaver
-from langgraph.graph import CompiledStateGraph
+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
 import tavily
 
 # Load environment variables
 load_dotenv()
 
-# Initialize the LLM (using Gemini-1.5 Flash) with Google API key from env variables
+# Initialize models and APIs
 llm = ChatGoogleGenerativeAI(
     model="gemini-1.5-flash",
     google_api_key=os.getenv("GOOGLE_API_KEY")
@@ -34,60 +37,77 @@ 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 a detailed formatted message of the current space events.
+        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`.
     """
     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")}
-    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)}"}
-    
+    response = requests.get(base_url, params=params)
+    response.raise_for_status()
+    events = response.json().get("events", [])
     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'
             event_details.append(
-                f"• **Event:** {event.get('title', 'N/A')}\n"
-                f"  **Type:** {category}\n"
-                f"  **Location (Lon, Lat):** {coordinates}\n"
+                f"• Event: {event.get('title', 'N/A')}\n"
+                f"  Type: {category}\n"
+                f"  Location: {coordinates}\n"
             )
         formatted_events = "\n".join(event_details)
-        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}
+        return {"output": f"🚀 Space Events Agent:\n\n{formatted_events}"}
     else:
-        return {"output": "🚀 Space Events Agent: No events found at the moment. Please check back later for updates."}
+        return {"output": f"🚀 Space Events Agent: No events found at the moment."}
 
-# 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 up-to-date space launch news and real time information"
+    "Search for information about space real time news",
 )
 
 def astronomy_image_agent(user_input: str):
     """
-    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).
+    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.
+
     Returns:
-        dict: A dictionary with detailed image information including title, explanation, and image URL.
+        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.
     """
     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:
@@ -102,157 +122,196 @@ 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}\n\n"
-                    f"This image was retrieved from NASA's Image and Video Library."
+                    f"Title: {title}\n"
+                    f"Description: {description}\n"
+                    f"Image URL: {image_url}"
                 )
-                return {"output": f"✨ **Galaxy Image Agent:**\n\n{result}"}
+                return {"output": f"✨ Galaxy Image Agent:\n\n{result}"}
             else:
-                return {"output": "✨ Galaxy Image Agent: No galaxy images found. Please try another query."}
+                return {"output": "No galaxy images found."}
         except Exception as e:
-            return {"output": f"✨ Galaxy Image Agent Error: {str(e)}"}
+            return {"output": f"Galaxy image search failed: {str(e)}"}
     else:
-        # Default to Astronomy Picture of the Day (APOD)
+        # Default to the APOD API if no keyword is detected.
         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')}\n\n"
-                f"This Astronomy Picture of the Day is provided by NASA's APOD API and is updated daily."
+                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')}"
             )
-            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 fetch resources from trusted sources.
+    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.
+
     Returns:
-        dict: A dictionary containing a detailed list of curated educational resources.
+        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.
     """
     try:
-        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."})
+        # 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."})
         if not resources:
-            return {"output": "🎓 Educational Resources Agent: No educational resources found. Please try again."}
+            return {"output": "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]
         )
-        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}
+        return {"output": f"🎓 Educational Resources Agent:\n\n{formatted_resources}"}
     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 detailed astronaut data and the current position of the International Space Station (ISS).
+    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.
+    
     Returns:
-        dict: A dictionary containing a detailed formatted message with astronaut information and ISS position updates.
+        dict: A dictionary containing astronaut information and ISS position updates.
+              Includes a list of people on the ISS and the positions over time.
     """
     # Fetch astronaut data
     api_url = "http://api.open-notify.org/astros.json"
-    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}")
+    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']}")
     else:
-        output_lines.append("Error: Failed to obtain astronaut data.")
-    
-    # Fetch ISS current position data with retries
+        result["error"] = 'Failed to obtain astronauts data.'
+
+    # List to hold ISS position updates
     positions = []
-    for attempt in range(5):
+
+    # fetching ISS position data
+    for attempt in range(5):  # Retry 5 times
         try:
-            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
+            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
         except requests.exceptions.RequestException as e:
-            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
+            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
+
 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)
 
-# 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:
+# 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:
 
 1. **Tavily Search (`search`) Integration**:
-   - **Purpose**: Provides up-to-date, relevant space-related information.
-   - **Usage**: Fetches the latest news, research articles, and educational resources related to space.
+   - **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.
+
+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.
+
+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.
+
+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.
 
-2. **NASA APOD Tool**:
-   - **Purpose**: Fetches NASA’s Astronomy Picture of the Day (APOD).
-   - **Usage**: Provides detailed information including title, explanation, and image URL.
+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.
 
-3. **NASA EONET Space Events**:
-   - **Purpose**: Retrieves real-time space events.
-   - **Usage**: Provides detailed data on ongoing/upcoming space events.
+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.
 
-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.
+### 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`.
 
-5. **Space Educational Resources**:
-   - **Purpose**: Curates educational resources on astronomy and space science.
-   - **Usage**: Provides a detailed list with titles and clickable links.
+### 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.
 
-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.
+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.
 ''')
 
-# Define the assistant node function; we use a plain dictionary as the state type
-def assistant(state: dict) -> dict:
+# Node
+def assistant(state: MessagesState) -> MessagesState:
     return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"][-10:])]}
 
-# Build the LangGraph state graph using dict as the state type
-builder: StateGraph = StateGraph(dict)
+# Build 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", tools_condition)
+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 not a tool call -> tools_condition routes to END
+    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 providing detailed and informative insights.")
+st.caption("Your AI-powered space exploration assistant")
 
 if "messages" not in st.session_state:
     st.session_state.messages = []
@@ -260,37 +319,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, astronomy images, or educational content"):
+if user_input := st.chat_input("Ask about space news, ISS location, or astronomy images"):
     st.session_state.messages.append({"role": "user", "content": user_input})
-    with st.chat_message("user"):
-        st.write(user_input)
+    st.chat_message("user").write(user_input)
+
     try:
-        # Prepare the message for the model and invoke the state graph
+        # Prepare the messages for the model
         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
 
-        # Check for image URL markers in a case-insensitive way
-        image_url = None
-        if "image url:" in assistant_response.lower():
-            # 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()
+        # Use regex to detect image URLs in the assistant response
+        image_pattern = re.compile(r'(https?://\S+\.(?:png|jpg|jpeg|gif))', re.IGNORECASE)
+        match = image_pattern.search(assistant_response)
+        if match:
+            image_url = match.group(1)
+            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)
+                st.write(assistant_response)
+        else:
+            st.session_state.messages.append({"role": "assistant", "content": assistant_response})
+            with st.chat_message("assistant"):
+                st.write(assistant_response)
 
-        # 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)}"
         st.session_state.messages.append({"role": "assistant", "content": error_msg})