Spaces:
Sleeping
Sleeping
| ############################################################################### | |
| # app.py – Weather Umbrella Advisor (Streamlit + OpenWeatherMap + Claude 3) # | |
| ############################################################################### | |
| import os, json, requests, boto3, streamlit as st | |
| from dotenv import load_dotenv | |
| # --------------------------------------------------------------------------- # | |
| # 1) Load env vars (local .env or HF Space “Secrets”) # | |
| # --------------------------------------------------------------------------- # | |
| load_dotenv() | |
| OPENWEATHERMAP_API_KEY = os.getenv("OPENWEATHERMAP_API_KEY") | |
| AWS_REGION = os.getenv("AWS_REGION", "us-east-1") | |
| AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID") # may be None | |
| AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY") # may be None | |
| # --- Heuristic: auto-swap if ID looks like a secret (contains “/”) ---------- | |
| if AWS_ACCESS_KEY_ID and "/" in AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY and "/" not in AWS_SECRET_ACCESS_KEY: | |
| AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY = AWS_SECRET_ACCESS_KEY, AWS_ACCESS_KEY_ID | |
| # --------------------------------------------------------------------------- # | |
| # 2) Create Bedrock client, preferring explicit keys if they’re present # | |
| # --------------------------------------------------------------------------- # | |
| if AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY: | |
| session = boto3.Session( | |
| aws_access_key_id = AWS_ACCESS_KEY_ID, | |
| aws_secret_access_key = AWS_SECRET_ACCESS_KEY, | |
| region_name = AWS_REGION, | |
| ) | |
| else: | |
| # Falls back to IAM role or default credentials chain | |
| session = boto3.Session(region_name=AWS_REGION) | |
| bedrock = session.client("bedrock-runtime") | |
| # Quick sanity-check – will raise if creds are still wrong | |
| try: | |
| _ = bedrock.meta.region_name # simple attribute access forces no call | |
| except Exception as e: | |
| st.error(f"Credential problem: {e}") | |
| st.stop() | |
| # --------------------------------------------------------------------------- # | |
| # 3) Streamlit page style # | |
| # --------------------------------------------------------------------------- # | |
| st.set_page_config(page_title="🌤️ Umbrella Advisor", page_icon="☔", layout="centered") | |
| st.markdown(""" | |
| <div style="text-align:center"> | |
| <h1 style="color:#3c79f5;">☔ Weather Umbrella Advisor</h1> | |
| <p style="font-size:18px">Ask if you need an umbrella tomorrow – powered by <b>Claude 3</b> & <b>OpenWeatherMap</b>.</p> | |
| </div> | |
| """, unsafe_allow_html=True) | |
| # --------------------------------------------------------------------------- # | |
| # 4) Conversation state # | |
| # --------------------------------------------------------------------------- # | |
| if "messages" not in st.session_state: | |
| st.session_state.messages = [] | |
| for m in st.session_state.messages: | |
| with st.chat_message(m["role"]): | |
| st.markdown(m["content"]) | |
| # --------------------------------------------------------------------------- # | |
| # 5) Helper – get weather # | |
| # --------------------------------------------------------------------------- # | |
| def get_weather(city: str): | |
| if not city.strip(): | |
| return {"error": "Please give me a city name."} | |
| geo = f"http://api.openweathermap.org/geo/1.0/direct?q={city}&limit=1&appid={OPENWEATHERMAP_API_KEY}" | |
| try: | |
| loc = requests.get(geo, timeout=10).json() | |
| if not loc: | |
| return {"error": f"City “{city}” not found."} | |
| lat, lon = loc[0]["lat"], loc[0]["lon"] | |
| wurl = f"http://api.openweathermap.org/data/2.5/forecast?lat={lat}&lon={lon}&appid={OPENWEATHERMAP_API_KEY}&units=metric" | |
| data = requests.get(wurl, timeout=10).json() | |
| if "list" not in data: | |
| return {"error": f"No forecast for “{city}”."} | |
| fc = [{ | |
| "time": f["dt_txt"], | |
| "description": f["weather"][0]["description"].capitalize(), | |
| "rain_probability": round(f.get("pop", 0)*100, 1), | |
| "temp": f["main"]["temp"] | |
| } for f in data["list"][:8]] | |
| return {"location": city.title(), "forecast": fc} | |
| except Exception as e: | |
| return {"error": str(e)} | |
| # --------------------------------------------------------------------------- # | |
| # 6) Helper – talk to Claude (ReAct) # | |
| # --------------------------------------------------------------------------- # | |
| SYSTEM_PROMPT = """You are a helpful umbrella advisor using ReAct: | |
| 1. Think about the question. | |
| 2. If needed, act with get_weather(location). | |
| 3. Observe results. | |
| 4. Reason and answer. | |
| When you need weather data, respond EXACTLY: | |
| {"thought":"…","action":"get_weather","action_input":{"location":"City"}}""" | |
| def ask_claude(user, history=""): | |
| # First call | |
| body = { | |
| "anthropic_version": "bedrock-2023-05-31", | |
| "max_tokens": 1000, | |
| "temperature": 0.7, | |
| "top_p": 0.9, | |
| "messages": [{ | |
| "role": "user", | |
| "content": f"{SYSTEM_PROMPT}\n\nHistory:\n{history}\n\nUser: {user}" | |
| }] | |
| } | |
| raw = bedrock.invoke_model( | |
| modelId="anthropic.claude-3-sonnet-20240229-v1:0", | |
| contentType="application/json", | |
| accept="application/json", | |
| body=json.dumps(body) | |
| ) | |
| txt = json.loads(raw["body"].read())["content"][0]["text"].strip() | |
| # Try to parse ReAct JSON | |
| try: | |
| j = json.loads(txt) | |
| if j.get("action") == "get_weather": | |
| city = j["action_input"]["location"] | |
| wx = get_weather(city) | |
| if "error" in wx: return wx["error"] | |
| # Second call – reasoning | |
| reason_prompt = f"""Here is the forecast for {city}: | |
| {json.dumps(wx, indent=2)} | |
| Give a friendly answer: YES/NO umbrella, with reasoning.""" | |
| body2 = { | |
| "anthropic_version": "bedrock-2023-05-31", | |
| "max_tokens": 500, | |
| "temperature": 0.7, | |
| "messages":[{"role":"user","content":reason_prompt}] | |
| } | |
| raw2 = bedrock.invoke_model( | |
| modelId="anthropic.claude-3-sonnet-20240229-v1:0", | |
| contentType="application/json", | |
| accept="application/json", | |
| body=json.dumps(body2) | |
| ) | |
| return json.loads(raw2["body"].read())["content"][0]["text"].strip() | |
| except json.JSONDecodeError: | |
| pass | |
| return txt | |
| # --------------------------------------------------------------------------- # | |
| # 7) Chat input # | |
| # --------------------------------------------------------------------------- # | |
| def last_history(n=4): | |
| return "\n".join(f"{m['role'].capitalize()}: {m['content']}" for m in st.session_state.messages[-n:]) | |
| if prompt := st.chat_input("Ask: Do I need an umbrella tomorrow?"): | |
| 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…"): | |
| reply = ask_claude(prompt, last_history()) | |
| st.markdown(reply) | |
| st.session_state.messages.append({"role":"assistant","content":reply}) | |
| # --------------------------------------------------------------------------- # | |
| # 8) Sidebar # | |
| # --------------------------------------------------------------------------- # | |
| with st.sidebar: | |
| st.image("https://img.icons8.com/clouds/100/umbrella.png", width=100) | |
| st.markdown("### About\nUses **Claude 3 Sonnet (Bedrock)** + **OpenWeatherMap**") | |