Spaces:
Sleeping
Sleeping
| # genesis/providers.py | |
| import os | |
| import requests | |
| from typing import List, Dict | |
| OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") | |
| HF_TOKEN = os.getenv("HF_TOKEN") | |
| GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") | |
| NCBI_API_KEY = os.getenv("NCBI_API_KEY") | |
| NCBI_EMAIL = os.getenv("NCBI_EMAIL") | |
| # -------- DeepSeek Summary -------- | |
| def run_deepseek_summary(prompt: str) -> str: | |
| """Run a dense scientific summary using DeepSeek API.""" | |
| try: | |
| url = "https://api.deepseek.com/v1/chat/completions" | |
| headers = {"Authorization": f"Bearer {HF_TOKEN}", "Content-Type": "application/json"} | |
| payload = { | |
| "model": "deepseek-science", | |
| "messages": [{"role": "user", "content": prompt}], | |
| "temperature": 0.2 | |
| } | |
| r = requests.post(url, headers=headers, json=payload, timeout=60) | |
| r.raise_for_status() | |
| return r.json()["choices"][0]["message"]["content"] | |
| except Exception as e: | |
| print(f"[DeepSeek] Failed: {e}") | |
| return prompt | |
| # -------- Gemini Polish -------- | |
| def run_gemini_polish(text: str) -> str: | |
| """Polish the summary using Gemini for clarity.""" | |
| if not GEMINI_API_KEY: | |
| return text | |
| try: | |
| url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent?key={GEMINI_API_KEY}" | |
| payload = {"contents": [{"parts": [{"text": f"Polish and clarify this research report:\n\n{text}"}]}]} | |
| r = requests.post(url, json=payload, timeout=30) | |
| r.raise_for_status() | |
| return r.json()["candidates"][0]["content"]["parts"][0]["text"] | |
| except Exception as e: | |
| print(f"[Gemini] Failed: {e}") | |
| return text | |
| # -------- Image Generation (OpenAI → HF Fallback) -------- | |
| def run_openai_image(query: str) -> str: | |
| """Generate a research diagram using OpenAI image API, fallback to HF Stable Diffusion.""" | |
| # Try OpenAI first | |
| if OPENAI_API_KEY: | |
| try: | |
| url = "https://api.openai.com/v1/images/generations" | |
| headers = {"Authorization": f"Bearer {OPENAI_API_KEY}"} | |
| payload = {"model": "gpt-image-1", "prompt": f"Scientific diagram about: {query}", "size": "1024x1024"} | |
| r = requests.post(url, headers=headers, json=payload, timeout=60) | |
| r.raise_for_status() | |
| return r.json()["data"][0]["url"] | |
| except Exception as e: | |
| print(f"[OpenAI Image] Failed: {e}") | |
| # Hugging Face fallback | |
| if HF_TOKEN: | |
| try: | |
| print("[Image] Falling back to Hugging Face Stable Diffusion...") | |
| url = "https://api-inference.huggingface.co/models/stabilityai/stable-diffusion-xl-base-1.0" | |
| headers = {"Authorization": f"Bearer {HF_TOKEN}"} | |
| payload = {"inputs": f"highly detailed scientific diagram about: {query}"} | |
| r = requests.post(url, headers=headers, json=payload, timeout=60) | |
| r.raise_for_status() | |
| img_path = f"/tmp/{query.replace(' ', '_')}.png" | |
| with open(img_path, "wb") as f: | |
| f.write(r.content) | |
| return img_path | |
| except Exception as e: | |
| print(f"[HF Image] Failed: {e}") | |
| return None | |
| # -------- PubMed Fallback -------- | |
| def pubmed_fallback_search(query: str, api_key: str, email: str) -> List[Dict]: | |
| """Search PubMed for relevant articles if no citations found.""" | |
| results = [] | |
| if not api_key or not email: | |
| return results | |
| try: | |
| base_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi" | |
| params = { | |
| "db": "pubmed", | |
| "term": query, | |
| "retmax": 3, | |
| "api_key": api_key, | |
| "email": email, | |
| "retmode": "json" | |
| } | |
| r = requests.get(base_url, params=params, timeout=10) | |
| r.raise_for_status() | |
| ids = r.json().get("esearchresult", {}).get("idlist", []) | |
| for pmid in ids: | |
| results.append({ | |
| "type": "PMID", | |
| "id": pmid, | |
| "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/" | |
| }) | |
| except Exception as e: | |
| print(f"[PubMed] Failed: {e}") | |
| return results | |