|
|
|
|
|
import os |
|
|
import requests |
|
|
from fastapi import APIRouter |
|
|
from pydantic import BaseModel |
|
|
from typing import List |
|
|
from redis_client import redis_client as r |
|
|
from dotenv import load_dotenv |
|
|
from urllib.parse import quote_plus |
|
|
import re |
|
|
import json |
|
|
|
|
|
load_dotenv() |
|
|
|
|
|
GNEWS_API_KEY = os.getenv("GNEWS_API_KEY") |
|
|
HF_TOKEN = os.getenv("HF_TOKEN") |
|
|
|
|
|
askMe = APIRouter() |
|
|
|
|
|
class QuestionInput(BaseModel): |
|
|
question: str |
|
|
|
|
|
@askMe.post("/ask") |
|
|
async def ask_question(input: QuestionInput): |
|
|
question = input.question |
|
|
|
|
|
|
|
|
STOPWORDS = {"what", "about", "which", "would", "could", "this", "that"} |
|
|
keywords = [ |
|
|
kw for kw in re.findall(r"\b\w{4,}\b", question.lower()) |
|
|
if kw not in STOPWORDS |
|
|
] |
|
|
query_string = " AND ".join(f'"{kw}"' for kw in keywords[:7]) |
|
|
encoded_query = quote_plus(query_string) |
|
|
|
|
|
print("Query string", query_string) |
|
|
|
|
|
gnews_url = f"https://gnews.io/api/v4/search?q={query_string}&lang=en&max=3&expand=content&token={GNEWS_API_KEY}" |
|
|
|
|
|
print("GNEWS URL", gnews_url) |
|
|
|
|
|
try: |
|
|
response = requests.get(gnews_url, timeout=10) |
|
|
response.raise_for_status() |
|
|
articles = response.json().get("articles", []) |
|
|
except Exception as e: |
|
|
return {"error": f"GNews API error: {str(e)}"} |
|
|
|
|
|
print("the articles are", articles) |
|
|
|
|
|
context = "\n\n".join([ |
|
|
article.get("content") or article.get("description") or "" |
|
|
for article in articles |
|
|
])[:1500] |
|
|
|
|
|
if not context.strip(): |
|
|
return { |
|
|
"question": question, |
|
|
"answer": "Cannot answer – no relevant context found.", |
|
|
"sources": [] |
|
|
} |
|
|
|
|
|
|
|
|
hf_api_url = "https://api-inference.huggingface.co/models/mistralai/Mistral-7B-Instruct-v0.3" |
|
|
headers = { |
|
|
"Authorization": f"Bearer {HF_TOKEN}", |
|
|
"Content-Type": "application/json" |
|
|
} |
|
|
|
|
|
payload = { |
|
|
"inputs": { |
|
|
"past_user_inputs": [], |
|
|
"generated_responses": [], |
|
|
"text": f"You are an assistant that answers questions based on recent news.\n\nContext:\n{context}\n\nQuestion: {question}" |
|
|
}, |
|
|
"parameters": { |
|
|
"max_new_tokens": 256, |
|
|
"temperature": 0.7 |
|
|
} |
|
|
} |
|
|
|
|
|
try: |
|
|
response = requests.post(hf_api_url, headers=headers, data=json.dumps(payload), timeout=30) |
|
|
response.raise_for_status() |
|
|
hf_response = response.json() |
|
|
answer = hf_response.get("generated_text", "Cannot answer.") |
|
|
except Exception as e: |
|
|
return {"error": f"Hugging Face API error: {str(e)}"} |
|
|
|
|
|
return { |
|
|
"question": question, |
|
|
"answer": answer.strip(), |
|
|
"sources": [ |
|
|
{"title": a["title"], "url": a["url"]} |
|
|
for a in articles |
|
|
] |
|
|
} |
|
|
|
