evenegas
57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
from fastapi import FastAPI, Query
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
import json
|
|
import faiss
|
|
import numpy as np
|
|
from sentence_transformers import SentenceTransformer
|
|
|
|
app = FastAPI(title="Fénix QA API")
|
|
|
|
# Cargar diccionario QA
|
|
qa_list = []
|
|
with open("qa_fenix.jsonl", encoding="utf-8") as f:
|
|
for line in f:
|
|
obj = json.loads(line)
|
|
qa_list.append({"q": obj["instruction"], "a": obj["response"]})
|
|
|
|
# Preparar encoder e índice FAISS
|
|
encoder = SentenceTransformer("all-MiniLM-L6-v2")
|
|
questions = [item["q"] for item in qa_list]
|
|
embs = encoder.encode(questions, convert_to_numpy=True, normalize_embeddings=True)
|
|
dim = embs.shape[1]
|
|
index = faiss.IndexFlatIP(dim)
|
|
index.add(embs)
|
|
|
|
# Función que devuelve respuesta y score
|
|
def buscar_respuesta(user_input: str, threshold: float = 0.65):
|
|
u_emb = encoder.encode([user_input], convert_to_numpy=True, normalize_embeddings=True)
|
|
scores, idxs = index.search(u_emb, k=1)
|
|
score, idx = float(scores[0][0]), int(idxs[0][0])
|
|
if score >= threshold:
|
|
return qa_list[idx]["a"], score
|
|
return None, score
|
|
|
|
# Modelo de entrada
|
|
class ChatRequest(BaseModel):
|
|
user_input: str
|
|
threshold: Optional[float] = 0.65
|
|
|
|
# Modelo de respuesta
|
|
class ChatResponse(BaseModel):
|
|
response: Optional[str]
|
|
score: float
|
|
threshold: float
|
|
success: bool
|
|
|
|
# Endpoint principal
|
|
@app.post("/chat", response_model=ChatResponse)
|
|
def chat(req: ChatRequest):
|
|
respuesta, score = buscar_respuesta(req.user_input, req.threshold)
|
|
return ChatResponse(
|
|
response=respuesta if respuesta else "Lo siento, no tengo esa información.",
|
|
score=score,
|
|
threshold=req.threshold,
|
|
success=respuesta is not None
|
|
)
|