Spaces:
Sleeping
Sleeping
| import os | |
| import json | |
| import requests | |
| import boto3 | |
| import streamlit as st | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| # Environment config | |
| OPENWEATHERMAP_API_KEY = os.getenv("OPENWEATHERMAP_API_KEY") | |
| AWS_REGION = os.getenv("AWS_REGION", "us-east-1") | |
| # Bedrock Claude client | |
| bedrock = boto3.client("bedrock-runtime", region_name=AWS_REGION) | |
| # App UI | |
| st.title("Weather Assistant - Umbrella Advisor") | |
| st.markdown("Ask me if you should carry an umbrella tomorrow!") | |
| # Session state for chat | |
| if "messages" not in st.session_state: | |
| st.session_state.messages = [] | |
| # Chat history display | |
| for msg in st.session_state.messages: | |
| with st.chat_message(msg["role"]): | |
| st.markdown(msg["content"]) | |
| def get_weather(location): | |
| """Get weather forecast for a specific location""" | |
| print(f"Getting weather for: {location}") | |
| # Validate location input | |
| if not location or location.strip() == "": | |
| return {"error": "Please specify a valid location/city name."} | |
| location = location.strip() | |
| geo_url = f"http://api.openweathermap.org/geo/1.0/direct?q={location}&limit=1&appid={OPENWEATHERMAP_API_KEY}" | |
| try: | |
| geo_resp = requests.get(geo_url).json() | |
| if not geo_resp or len(geo_resp) == 0: | |
| return {"error": f"Location '{location}' not found. Please check the spelling and try again."} | |
| lat, lon = geo_resp[0]['lat'], geo_resp[0]['lon'] | |
| except (KeyError, IndexError, requests.RequestException) as e: | |
| return {"error": f"Error getting location data for '{location}': {str(e)}"} | |
| try: | |
| weather_url = f"http://api.openweathermap.org/data/2.5/forecast?lat={lat}&lon={lon}&appid={OPENWEATHERMAP_API_KEY}&units=metric" | |
| weather_data = requests.get(weather_url).json() | |
| if 'list' not in weather_data: | |
| return {"error": f"Unable to get weather forecast for '{location}'."} | |
| forecast = [] | |
| for f in weather_data['list'][:8]: # Next 24 hours | |
| forecast.append({ | |
| "time": f["dt_txt"], | |
| "description": f["weather"][0]["description"], | |
| "rain_probability": f.get("pop", 0) * 100, | |
| "temp": f["main"]["temp"], | |
| "humidity": f["main"]["humidity"] | |
| }) | |
| return { | |
| "location": location, | |
| "forecast": forecast | |
| } | |
| except (KeyError, requests.RequestException) as e: | |
| return {"error": f"Error getting weather forecast for '{location}': {str(e)}"} | |
| def generate_react_response(user_input, conversation_history=""): | |
| """Generate response using ReAct (Reasoning + Acting) approach""" | |
| system_prompt = """You are a helpful weather assistant that uses ReAct (Reasoning + Acting) methodology to help users decide about carrying umbrellas. | |
| Follow this process: | |
| 1. **Think**: Analyze what the user is asking | |
| 2. **Act**: Use available tools if needed | |
| 3. **Observe**: Process the results | |
| 4. **Reason**: Draw conclusions and provide advice | |
| Available tools: | |
| - get_weather(location): Gets weather forecast for tomorrow | |
| When you need to get weather data, respond with this JSON format: | |
| { | |
| "thought": "I need to get weather data for [location] to advise about umbrella", | |
| "action": "get_weather", | |
| "action_input": {"location": "city_name"} | |
| } | |
| When you have all needed information, provide a conversational response that includes: | |
| - The location | |
| - Your reasoning based on weather conditions | |
| - Clear umbrella advice | |
| Example: "You do not need to carry an umbrella tomorrow as the weather in New York will be sunny with no chance of rain." | |
| If the user doesn't specify a location, ask them to specify it conversationally.""" | |
| # Build conversation context | |
| messages = [ | |
| {"role": "user", "content": f"{system_prompt}\n\nConversation history: {conversation_history}\n\nUser: {user_input}"} | |
| ] | |
| claude_body = { | |
| "anthropic_version": "bedrock-2023-05-31", | |
| "max_tokens": 1000, | |
| "temperature": 0.7, | |
| "top_p": 0.9, | |
| "messages": messages | |
| } | |
| response = bedrock.invoke_model( | |
| modelId="anthropic.claude-3-sonnet-20240229-v1:0", | |
| contentType="application/json", | |
| accept="application/json", | |
| body=json.dumps(claude_body), | |
| ) | |
| content = json.loads(response["body"].read())["content"][0]["text"].strip() | |
| # Try to parse as ReAct JSON | |
| try: | |
| react_response = json.loads(content) | |
| if react_response.get("action") == "get_weather": | |
| location = react_response.get("action_input", {}).get("location", "").strip() | |
| thought = react_response.get("thought", "") | |
| # Validate location before calling weather function | |
| if not location: | |
| return "I need to know which city or location you're asking about. Could you please specify the location?" | |
| # Get weather data | |
| weather_data = get_weather(location) | |
| if "error" in weather_data: | |
| return weather_data["error"] | |
| # Process weather data and generate final reasoning | |
| try: | |
| reasoning_prompt = f"""Based on this weather data for {location}, provide your final umbrella recommendation: | |
| Weather forecast: {json.dumps(weather_data, indent=2)} | |
| Your previous thought: {thought} | |
| Provide a conversational response that includes: | |
| 1. The location | |
| 2. Your reasoning based on the weather conditions | |
| 3. Clear umbrella advice | |
| Format like: "You [do/do not] need to carry an umbrella tomorrow as the weather in [location] will be [conditions and reasoning]." | |
| """ | |
| final_messages = [{"role": "user", "content": reasoning_prompt}] | |
| final_body = { | |
| "anthropic_version": "bedrock-2023-05-31", | |
| "max_tokens": 500, | |
| "temperature": 0.7, | |
| "messages": final_messages | |
| } | |
| final_response = bedrock.invoke_model( | |
| modelId="anthropic.claude-3-sonnet-20240229-v1:0", | |
| contentType="application/json", | |
| accept="application/json", | |
| body=json.dumps(final_body), | |
| ) | |
| final_content = json.loads(final_response["body"].read())["content"][0]["text"].strip() | |
| return final_content | |
| except Exception as e: | |
| return f"Error processing weather data: {str(e)}" | |
| except json.JSONDecodeError: | |
| # If not JSON, return the content as is (probably asking for location) | |
| pass | |
| return content | |
| def build_conversation_history(): | |
| """Build conversation history for context""" | |
| history = [] | |
| for msg in st.session_state.messages[-4:]: # Last 4 messages for context | |
| history.append(f"{msg['role'].capitalize()}: {msg['content']}") | |
| return "\n".join(history) | |
| if prompt := st.chat_input("Type your question here..."): | |
| st.session_state.messages.append({"role": "user", "content": prompt}) | |
| with st.chat_message("user"): | |
| st.markdown(prompt) | |
| with st.chat_message("assistant"): | |
| with st.spinner("Thinking..."): | |
| conversation_history = build_conversation_history() | |
| reply = generate_react_response(prompt, conversation_history) | |
| st.markdown(reply) | |
| st.session_state.messages.append({"role": "assistant", "content": reply}) | |
| with st.sidebar: | |
| st.header("About") | |
| st.markdown(""" | |
| - Powered by **AWS Bedrock (Claude Sonnet)** | |
| - Uses **ReAct (Reasoning + Acting)** methodology | |
| - Retrieves real-time data from **OpenWeatherMap** | |
| - Provides step-by-step reasoning for umbrella advice | |
| """) | |
| st.subheader("Sample Prompts") | |
| st.markdown(""" | |
| - Should I bring an umbrella tomorrow? | |
| - Will it rain in Delhi tomorrow? | |
| - Do I need an umbrella in Tokyo? | |
| - Should I carry an umbrella tomorrow in London? | |
| """) | |
| st.subheader("ReAct Process") | |
| st.markdown(""" | |
| 1. **Think**: Analyze your question | |
| 2. **Act**: Get weather data if needed | |
| 3. **Observe**: Process weather information | |
| 4. **Reason**: Provide umbrella advice with explanation | |
| """) |