Spaces:
Running
Running
import os | |
import requests | |
import gradio as gr | |
from cerebras.cloud.sdk import Cerebras | |
from langchain_huggingface import HuggingFaceEmbeddings | |
from langchain_community.vectorstores import FAISS | |
from langchain.schema import Document | |
import numpy as np | |
from langchain_community.document_loaders import TextLoader | |
from langchain.text_splitter import RecursiveCharacterTextSplitter | |
from sentence_transformers import SentenceTransformer | |
# Initialize Cerebras API client | |
Facts = os.getenv("Facto") | |
client = Cerebras(api_key= Facts) | |
Newskey = os.getenv("News") | |
# Function to fetch latest news articles from NewsAPI | |
def get_latest_news(query): | |
api_key = Newskey | |
url = f"https://newsapi.org/v2/everything?q={query}&apiKey={api_key}" | |
response = requests.get(url) | |
data = response.json() | |
return [(article["title"], article["url"], article["source"]["name"]) for article in data.get("articles", [])[:5]] | |
# Function to update fact_checks.txt with new user input (overwrites previous content) | |
def update_fact_checks_file(query): | |
with open("fact_checks.txt", "w", encoding="utf-8") as file: | |
file.write(f"{query}\n") | |
# Function to create a FAISS retriever dynamically | |
def create_faiss_retriever(): | |
if not os.path.exists("fact_checks.txt"): | |
open("fact_checks.txt", "w").close() # Create file if it doesn't exist | |
loader = TextLoader("fact_checks.txt") | |
documents = loader.load() | |
text_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=50) | |
docs = text_splitter.split_documents(documents) | |
embedding_model = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2") | |
vector_store = FAISS.from_documents(docs, embedding_model) | |
return vector_store.as_retriever(search_type="similarity", search_kwargs={"k": 8}) | |
# Function to clear the fact_checks.txt file after execution | |
def clear_fact_checks_file(): | |
open("fact_checks.txt", "w").close() | |
# Function to perform fact-checking with Llama 3.3 | |
def fact_check_with_llama3(query): | |
# Save query to fact_checks.txt | |
update_fact_checks_file(query) | |
# Reload FAISS index with new data | |
retriever = create_faiss_retriever() | |
# Retrieve relevant facts from FAISS | |
retrieved_docs = retriever.invoke(query) | |
retrieved_texts = [doc.page_content for doc in retrieved_docs] | |
# Fetch real-time news | |
news = get_latest_news(query) | |
# Combine all retrieved context | |
context_text = "\n".join(retrieved_texts) | |
# Construct prompt for Llama 3.3 | |
prompt = f""" | |
Claim: {query} | |
Context: {context_text} | |
Based on the provided context, determine whether the claim is True, False, or Misleading. Provide a concise explanation and cite relevant sources. Don't mention any instance of your knowledge cut-off. | |
""" | |
# Call Llama 3.3 API | |
stream = client.chat.completions.create( | |
messages=[{"role": "system", "content": prompt}], | |
model="llama-3.3-70b", | |
stream=True, | |
max_completion_tokens=512, | |
temperature=0.2, | |
top_p=1 | |
) | |
# Generate AI response | |
result = "".join(chunk.choices[0].delta.content or "" for chunk in stream) | |
# Format results with sources | |
sources = "\n".join([f"{title} ({source}): {url}" for title, url, source in news]) | |
# Clear the file after execution | |
clear_fact_checks_file() | |
return result, sources if sources else "No relevant sources found." | |
# Gradio Interface | |
def fact_check_interface(query): | |
response, sources = fact_check_with_llama3(query) | |
return response, sources | |
gui = gr.Interface( | |
fn=fact_check_interface, | |
inputs=gr.Textbox(placeholder="Enter a claim to fact-check"), | |
outputs=[gr.Textbox(label="Fact-Check Result"), gr.Textbox(label="Sources")], | |
title="Facto - AI Fact-Checking System", | |
description="Enter a claim, and the system will verify it using Llama 3.3 and external knowledge sources, citing relevant sources." | |
) | |
gui.launch(debug=True) |