Create app.py

app.py

@@ -0,0 +1,362 @@
+import datetime
+import json
+import os
+import time
+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
+import tavily
+
+# Load environment variables
+load_dotenv()
+
+# Initialize models and APIs
+llm = ChatGoogleGenerativeAI(
+    model="gemini-1.5-flash",
+    google_api_key=os.getenv("GOOGLE_API_KEY")
+)
+
+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`.
+    """
+    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", [])
+    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: {coordinates}\n"
+            )
+        formatted_events = "\n".join(event_details)
+        return {"output": f"🚀 Space Events Agent:\nn{formatted_events}"}
+    else:
+        return {"output": f"🚀 Space Events Agent:No events found at the moment."}
+
+loader1 = WebBaseLoader("https://spacelaunchnow.me/3.3.0/launch/upcoming/")
+docs1 = loader1.load()
+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",
+)
+
+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.
+
+    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.
+    """
+    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:
+            response = requests.get(search_url, params=params)
+            response.raise_for_status()
+            data = response.json()
+            items = data.get("collection", {}).get("items", [])
+            if items:
+                first_item = items[0]
+                links = first_item.get("links", [])
+                image_url = links[0]["href"] if links else "No URL available"
+                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}"
+                )
+                return {"output": f"✨ Galaxy Image Agent:\n\n{result}"}
+            else:
+                return {"output": "No galaxy images found."}
+        except Exception as e:
+            return {"output": f"Galaxy image search failed: {str(e)}"}
+    else:
+        # 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')}"
+            )
+            return {"output": f"✨ Astronomy Image Agent:\n\n{result}"}
+        except Exception as 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.
+
+    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.
+    """
+    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."})
+        if not resources:
+            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]
+        )
+        return {"output": f"🎓 Educational Resources Agent:\n\n{formatted_resources}"}
+    except Exception as 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.
+    
+    Returns:
+        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"
+    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:
+        result["error"] = 'Failed to obtain astronauts data.'
+
+    # List to hold ISS position updates
+    positions = []
+
+    # fetching ISS position data
+    for attempt in range(5):  # Retry 5 times
+        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
+        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
+
+search = TavilySearchResults(tavily_api_key=os.getenv("TAVILY_API_KEY"))
+
+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:
+
+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.
+
+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.
+
+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.
+
+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.
+
+### 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
+def assistant(state: MessagesState) -> MessagesState:
+    return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"][-10:])]}
+
+# 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",
+    # 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_edge("tools", "assistant")
+memory: MemorySaver = MemorySaver()
+# Compile the workflow into an application
+react_graph_memory: CompiledStateGraph = builder.compile(checkpointer=memory)
+
+
+st.title("🚀 Space Agentic Chatbot")
+st.caption("Your AI-powered space exploration assistant")
+
+if "messages" not in st.session_state:
+    st.session_state.messages = []
+
+# Display existing chat messages
+for message in st.session_state.messages:
+    with st.chat_message(message["role"]):
+        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"):
+    st.session_state.messages.append({"role": "user", "content": user_input})
+    st.chat_message("user").write(user_input)
+
+    try:
+        # 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
+
+        # Handle image URLs and regular text responses
+        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)
+
+    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)