feat(backend): add FastAPI routes with SSE chat and reindex guard
This commit is contained in:
omp committed
1 parent
99bc46f053
commit
7a0396422f
3 files changed
+555
No files matched your search
@@ -60,6 +60,9 @@ class Settings(BaseSettings):
|
||||
# Limites de ingestão
|
||||
max_pdf_bytes: int = 200 * 1024 * 1024
|
||||
|
||||
# Indexação automática na inicialização
|
||||
auto_index: bool = True
|
||||
|
||||
@property
|
||||
def ocr_cache_dir(self) -> Path:
|
||||
return self.vector_db_dir / "ocr_cache"
|
||||
|
||||
@@ -0,0 +1,337 @@
|
||||
"""API HTTP do Pastor Online.
|
||||
|
||||
Rotas:
|
||||
|
||||
* `GET /api/health` — verificação de saúde
|
||||
* `GET /api/status` — estado da indexação (PDFs lidos e vetores indexados)
|
||||
* `POST /api/reindex` — força a reindexação dos PDFs
|
||||
* `POST /api/chat` — chat em streaming (SSE)
|
||||
* `GET /api/sources` — últimos trechos recuperados (inspeção)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from contextlib import asynccontextmanager
|
||||
from dataclasses import dataclass
|
||||
from typing import AsyncIterator
|
||||
|
||||
from fastapi import FastAPI, Header, HTTPException, Request
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import StreamingResponse
|
||||
|
||||
from .bible_parser import BibleParser
|
||||
from .chat import ChatService
|
||||
from .config import Settings, get_settings
|
||||
from .embedding import Embedder, HashEmbedder, SentenceTransformerEmbedder
|
||||
from .extraction import PdfExtractor
|
||||
from .indexer import Indexer
|
||||
from .llm import LLMProvider, LLMProviderError, build_provider
|
||||
from .models import (
|
||||
ChatRequest,
|
||||
HealthResponse,
|
||||
ReindexResponse,
|
||||
Source,
|
||||
SourcesResponse,
|
||||
StatusResponse,
|
||||
)
|
||||
from .vector_store import InMemoryVectorStore, VectorStore
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Services:
|
||||
"""Dependências de longa duração da aplicação."""
|
||||
|
||||
settings: Settings
|
||||
store: VectorStore
|
||||
embedder: Embedder
|
||||
provider: LLMProvider
|
||||
indexer: Indexer
|
||||
chat: ChatService
|
||||
last_sources: list[Source]
|
||||
|
||||
|
||||
def _build_embedder(settings: Settings) -> Embedder:
|
||||
"""Cria o embedder conforme o modelo configurado.
|
||||
|
||||
`hash` e modelos desconhecidos usam o embedder determinístico; os demais
|
||||
usam sentence-transformers.
|
||||
"""
|
||||
model = settings.embedding_model.strip().lower()
|
||||
if model in ("", "hash", "hash-test", "none"):
|
||||
return HashEmbedder()
|
||||
try:
|
||||
return SentenceTransformerEmbedder(
|
||||
settings.embedding_model, batch_size=settings.embedding_batch_size
|
||||
)
|
||||
except Exception as exc: # pragma: no cover - falha de download do modelo
|
||||
logger.error(
|
||||
"Não foi possível carregar o modelo de embedding %s: %s. "
|
||||
"Usando embedder determinístico.",
|
||||
settings.embedding_model,
|
||||
exc,
|
||||
)
|
||||
return HashEmbedder()
|
||||
|
||||
|
||||
def create_app(
|
||||
settings: Settings | None = None,
|
||||
store: VectorStore | None = None,
|
||||
embedder: Embedder | None = None,
|
||||
provider: LLMProvider | None = None,
|
||||
indexer: Indexer | None = None,
|
||||
) -> FastAPI:
|
||||
"""Cria a aplicação FastAPI. As dependências podem ser injetadas nos testes."""
|
||||
settings = settings or get_settings()
|
||||
|
||||
if store is None:
|
||||
store = VectorStore(
|
||||
persist_dir=settings.vector_db_dir,
|
||||
collection_name=settings.collection_name,
|
||||
embedder=embedder,
|
||||
)
|
||||
if embedder is None:
|
||||
embedder = getattr(store, "embedder", None) or _build_embedder(settings)
|
||||
if provider is None:
|
||||
provider = build_provider(settings)
|
||||
|
||||
if indexer is None:
|
||||
extractor = PdfExtractor(
|
||||
cache_dir=settings.ocr_cache_dir,
|
||||
dpi=settings.ocr_dpi,
|
||||
ocr_lang=settings.ocr_language,
|
||||
ocr_timeout=settings.ocr_timeout,
|
||||
min_legible_ratio=settings.min_legible_ratio,
|
||||
)
|
||||
indexer = Indexer(
|
||||
settings=settings,
|
||||
extractor=extractor,
|
||||
parser=BibleParser(),
|
||||
embedder=embedder,
|
||||
store=store,
|
||||
)
|
||||
|
||||
chat = ChatService(
|
||||
settings=settings, store=store, embedder=embedder, provider=provider
|
||||
)
|
||||
|
||||
services = Services(
|
||||
settings=settings,
|
||||
store=store,
|
||||
embedder=embedder,
|
||||
provider=provider,
|
||||
indexer=indexer,
|
||||
chat=chat,
|
||||
last_sources=[],
|
||||
)
|
||||
reindex_lock = asyncio.Lock()
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
"""Dispara a indexação inicial sem bloquear o startup."""
|
||||
if settings.auto_index:
|
||||
asyncio.create_task(_run_initial_index(services))
|
||||
yield
|
||||
close = getattr(provider, "aclose", None)
|
||||
if close is not None:
|
||||
await close()
|
||||
|
||||
app = FastAPI(
|
||||
title="Pastor Online",
|
||||
description="Sistema RAG para consulta bíblica com agente pastoral.",
|
||||
version="1.0.0",
|
||||
lifespan=lifespan,
|
||||
)
|
||||
app.state.services = services
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=settings.cors_origin_list,
|
||||
allow_credentials=False,
|
||||
allow_methods=["GET", "POST"],
|
||||
allow_headers=["Content-Type", "X-Reindex-Token"],
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Rotas
|
||||
# ------------------------------------------------------------------ #
|
||||
|
||||
@app.get("/api/health", response_model=HealthResponse)
|
||||
async def health() -> HealthResponse:
|
||||
"""Verificação de saúde do serviço."""
|
||||
state = services.indexer.status().state
|
||||
return HealthResponse(
|
||||
status="ok",
|
||||
llm_provider=settings.llm_provider,
|
||||
indexing="indexing" if state == "indexing" else "ready",
|
||||
)
|
||||
|
||||
@app.get("/api/status", response_model=StatusResponse)
|
||||
async def status() -> StatusResponse:
|
||||
"""Estado da indexação: PDFs lidos e vetores indexados."""
|
||||
current = services.indexer.status()
|
||||
return StatusResponse(
|
||||
state=current.state,
|
||||
pdfs_total=current.pdfs_total,
|
||||
pdfs_indexed=current.pdfs_indexed,
|
||||
chunks=current.chunks,
|
||||
model=current.model,
|
||||
last_indexed_at=current.last_indexed_at,
|
||||
errors=current.errors,
|
||||
)
|
||||
|
||||
@app.post("/api/reindex", response_model=ReindexResponse, status_code=202)
|
||||
async def reindex(
|
||||
x_reindex_token: str | None = Header(default=None),
|
||||
) -> ReindexResponse:
|
||||
"""Força a releitura e reindexação dos PDFs."""
|
||||
if settings.reindex_token:
|
||||
if x_reindex_token != settings.reindex_token:
|
||||
raise HTTPException(status_code=401, detail="Token inválido")
|
||||
|
||||
if reindex_lock.locked():
|
||||
raise HTTPException(
|
||||
status_code=409, detail="Uma reindexação já está em andamento"
|
||||
)
|
||||
|
||||
# O lock é adquirido e liberado pela tarefa de background: a rota apenas
|
||||
# agenda o trabalho. Liberá-lo aqui permitiria reindexações concorrentes.
|
||||
await reindex_lock.acquire()
|
||||
asyncio.create_task(_run_reindex(services, reindex_lock))
|
||||
|
||||
return ReindexResponse(state="indexing")
|
||||
|
||||
@app.post("/api/chat")
|
||||
async def chat(request: ChatRequest) -> StreamingResponse:
|
||||
"""Responde a uma pergunta em streaming (SSE)."""
|
||||
if len(request.message) > settings.max_message_length:
|
||||
raise HTTPException(status_code=422, detail="Mensagem muito longa")
|
||||
|
||||
history = [
|
||||
{"role": entry.role, "content": entry.content} for entry in request.history
|
||||
]
|
||||
|
||||
async def event_stream() -> AsyncIterator[str]:
|
||||
try:
|
||||
result = await services.chat.answer(request.message, history)
|
||||
except LLMProviderError as exc:
|
||||
logger.error("Falha no provedor de LLM: %s", exc)
|
||||
yield _sse("error", {"message": "Provedor de LLM indisponível."})
|
||||
return
|
||||
except Exception as exc: # pragma: no cover - erro inesperado
|
||||
logger.exception("Erro ao processar a pergunta")
|
||||
yield _sse("error", {"message": "Erro interno ao processar a pergunta."})
|
||||
return
|
||||
|
||||
sources = [
|
||||
{
|
||||
"reference": source.reference,
|
||||
"text": source.text,
|
||||
"book": source.book,
|
||||
"chapter": source.chapter,
|
||||
"verse": source.verse,
|
||||
"page": source.page,
|
||||
"file": source.file,
|
||||
"score": round(source.score, 4),
|
||||
}
|
||||
for source in result.sources
|
||||
]
|
||||
services.last_sources = result.sources
|
||||
|
||||
yield _sse("sources", {"sources": sources})
|
||||
yield _sse("token", {"text": result.text})
|
||||
yield _sse("done", {"citations": result.citations})
|
||||
|
||||
return StreamingResponse(
|
||||
event_stream(),
|
||||
media_type="text/event-stream",
|
||||
headers={
|
||||
"Cache-Control": "no-cache",
|
||||
"Connection": "keep-alive",
|
||||
"X-Accel-Buffering": "no",
|
||||
},
|
||||
)
|
||||
|
||||
@app.get("/api/sources", response_model=SourcesResponse)
|
||||
async def sources() -> SourcesResponse:
|
||||
"""Trechos consultados na última resposta."""
|
||||
return SourcesResponse(
|
||||
sources=[
|
||||
Source(
|
||||
reference=source.reference,
|
||||
text=source.text,
|
||||
book=source.book,
|
||||
chapter=source.chapter,
|
||||
verse=source.verse,
|
||||
page=source.page,
|
||||
file=source.file,
|
||||
score=source.score,
|
||||
)
|
||||
for source in services.last_sources
|
||||
]
|
||||
)
|
||||
|
||||
return app
|
||||
|
||||
|
||||
def _sse(event: str, data: dict) -> str:
|
||||
"""Serializa um evento SSE."""
|
||||
return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
|
||||
|
||||
|
||||
async def _run_initial_index(services: Services) -> None:
|
||||
"""Indexa os PDFs na inicialização, sem bloquear o startup."""
|
||||
await asyncio.to_thread(services.indexer.reindex)
|
||||
|
||||
|
||||
async def _run_reindex(services: Services, lock: asyncio.Lock) -> None:
|
||||
"""Executa a reindexação em background, segurando o lock durante o trabalho."""
|
||||
try:
|
||||
await asyncio.to_thread(services.indexer.reindex, True)
|
||||
finally:
|
||||
if lock.locked():
|
||||
lock.release()
|
||||
|
||||
|
||||
def _build_default_app() -> FastAPI:
|
||||
"""Cria a aplicação padrão, tolerando falha no carregamento do Chroma.
|
||||
|
||||
O Uvicorn importa este módulo no arranque. Se o banco vetorial não puder ser
|
||||
aberto (volume ainda não montado, por exemplo), a aplicação precisa subir e
|
||||
reportar o erro em `/api/status`, não recusar o arranque.
|
||||
"""
|
||||
try:
|
||||
return create_app()
|
||||
except Exception as exc: # pragma: no cover - depende do ambiente
|
||||
logger.error("Falha ao inicializar a aplicação: %s", exc)
|
||||
settings = get_settings()
|
||||
embedder = _build_embedder(settings)
|
||||
store = InMemoryVectorStore(embedder=embedder)
|
||||
provider = build_provider(settings)
|
||||
indexer = Indexer(
|
||||
settings=settings,
|
||||
extractor=PdfExtractor(
|
||||
cache_dir=settings.ocr_cache_dir,
|
||||
dpi=settings.ocr_dpi,
|
||||
ocr_lang=settings.ocr_language,
|
||||
ocr_timeout=settings.ocr_timeout,
|
||||
min_legible_ratio=settings.min_legible_ratio,
|
||||
),
|
||||
parser=BibleParser(),
|
||||
embedder=embedder,
|
||||
store=store,
|
||||
)
|
||||
return create_app(
|
||||
settings=settings,
|
||||
store=store,
|
||||
embedder=embedder,
|
||||
provider=provider,
|
||||
indexer=indexer,
|
||||
)
|
||||
|
||||
|
||||
app = _build_default_app()
|
||||
@@ -0,0 +1,215 @@
|
||||
"""Testes da API FastAPI."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from app.config import Settings # noqa: E402
|
||||
from app.embedding import HashEmbedder # noqa: E402
|
||||
from app.main import create_app # noqa: E402
|
||||
from app.vector_store import InMemoryVectorStore # noqa: E402
|
||||
|
||||
|
||||
class StubProvider:
|
||||
async def stream(self, messages, system):
|
||||
yield "Resposta ancorada em Gênesis 1:1."
|
||||
|
||||
async def aclose(self):
|
||||
return None
|
||||
|
||||
|
||||
class FakeIndexer:
|
||||
"""Indexador controlado para os testes de API.
|
||||
|
||||
`release` permite segurar a reindexação em background, de forma que o teste
|
||||
do 409 seja determinístico.
|
||||
"""
|
||||
|
||||
def __init__(self, block: bool = False):
|
||||
self.calls: list[bool] = []
|
||||
self.release = threading.Event()
|
||||
self.release.set()
|
||||
if block:
|
||||
self.release.clear()
|
||||
|
||||
def reindex(self, force: bool = False):
|
||||
self.calls.append(force)
|
||||
self.release.wait(timeout=10)
|
||||
from app.indexer import IndexStatus
|
||||
|
||||
return IndexStatus(state="ready", pdfs_total=1, pdfs_indexed=1, chunks=42)
|
||||
|
||||
def status(self):
|
||||
from app.indexer import IndexStatus
|
||||
|
||||
return IndexStatus(state="ready", pdfs_total=1, pdfs_indexed=1, chunks=42)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def settings(tmp_path):
|
||||
return Settings(
|
||||
pdf_dir=tmp_path / "pdfs",
|
||||
vector_db_dir=tmp_path / "vector_db",
|
||||
embedding_model="hash-test",
|
||||
llm_provider="stub",
|
||||
cors_origins="http://localhost:8080",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(settings):
|
||||
embedder = HashEmbedder(dimensions=64)
|
||||
store = InMemoryVectorStore(embedder=embedder)
|
||||
app = create_app(
|
||||
settings=settings,
|
||||
store=store,
|
||||
embedder=embedder,
|
||||
provider=StubProvider(),
|
||||
indexer=FakeIndexer(),
|
||||
)
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def test_health_returns_ok(client):
|
||||
response = client.get("/api/health")
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["status"] == "ok"
|
||||
assert body["llm_provider"] == "stub"
|
||||
|
||||
|
||||
def test_status_reports_indexation(client):
|
||||
response = client.get("/api/status")
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["state"] == "ready"
|
||||
assert body["chunks"] == 42
|
||||
|
||||
|
||||
def test_reindex_returns_202(client):
|
||||
response = client.post("/api/reindex")
|
||||
|
||||
assert response.status_code == 202
|
||||
assert response.json()["state"] == "indexing"
|
||||
|
||||
|
||||
def test_reindex_requires_token_when_configured(settings):
|
||||
embedder = HashEmbedder(dimensions=64)
|
||||
store = InMemoryVectorStore(embedder=embedder)
|
||||
settings.reindex_token = "segredo"
|
||||
app = create_app(settings=settings, store=store, embedder=embedder, provider=StubProvider())
|
||||
client = TestClient(app)
|
||||
|
||||
assert client.post("/api/reindex").status_code == 401
|
||||
|
||||
authorized = client.post("/api/reindex", headers={"X-Reindex-Token": "segredo"})
|
||||
assert authorized.status_code == 202
|
||||
|
||||
|
||||
def test_concurrent_reindex_is_rejected_with_409(settings):
|
||||
settings.auto_index = False
|
||||
embedder = HashEmbedder(dimensions=64)
|
||||
store = InMemoryVectorStore(embedder=embedder)
|
||||
indexer = FakeIndexer(block=True)
|
||||
app = create_app(
|
||||
settings=settings,
|
||||
store=store,
|
||||
embedder=embedder,
|
||||
provider=StubProvider(),
|
||||
indexer=indexer,
|
||||
)
|
||||
|
||||
# O context manager mantém um único event loop entre as requisições; sem ele,
|
||||
# o TestClient fecha o loop a cada chamada e a tarefa de background seria
|
||||
# cancelada, liberando o lock.
|
||||
with TestClient(app) as client:
|
||||
try:
|
||||
first = client.post("/api/reindex")
|
||||
assert first.status_code == 202
|
||||
|
||||
second = client.post("/api/reindex")
|
||||
assert second.status_code == 409
|
||||
finally:
|
||||
indexer.release.set()
|
||||
|
||||
|
||||
def test_chat_rejects_empty_message(client):
|
||||
response = client.post("/api/chat", json={"message": ""})
|
||||
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
def test_chat_rejects_message_above_the_limit(client):
|
||||
response = client.post("/api/chat", json={"message": "a" * 5000})
|
||||
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
def test_chat_streams_sources_token_and_done(client):
|
||||
from app.bible_parser import VerseChunk
|
||||
|
||||
embedder = HashEmbedder(dimensions=64)
|
||||
store = client.app.state.services.store
|
||||
chunk = VerseChunk(
|
||||
book="Gênesis",
|
||||
chapter=1,
|
||||
verse=1,
|
||||
page=6,
|
||||
file="biblia.pdf",
|
||||
text="No princípio criou Deus os céus e a terra.",
|
||||
)
|
||||
store.upsert([chunk], embedder.encode([chunk.text]))
|
||||
|
||||
response = client.post(
|
||||
"/api/chat",
|
||||
json={"message": "Quem criou o mundo?", "history": []},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"].startswith("text/event-stream")
|
||||
|
||||
body = response.text
|
||||
assert "event: sources" in body
|
||||
assert "event: token" in body
|
||||
assert "event: done" in body
|
||||
|
||||
sources_block = body.split("event: sources")[1].split("event:")[0]
|
||||
payload = json.loads(sources_block.split("data:")[1].strip())
|
||||
assert payload["sources"][0]["reference"] == "Gênesis 1:1"
|
||||
|
||||
|
||||
def test_chat_without_evidence_returns_the_fixed_message(client):
|
||||
response = client.post("/api/chat", json={"message": "fale sobre ações", "history": []})
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "event: token" in response.text
|
||||
assert "não aborda diretamente" in response.text
|
||||
|
||||
|
||||
def test_cors_is_restricted_to_configured_origin(client):
|
||||
response = client.get("/api/health", headers={"Origin": "http://localhost:8080"})
|
||||
|
||||
assert response.headers.get("access-control-allow-origin") == "http://localhost:8080"
|
||||
|
||||
|
||||
def test_cors_rejects_unknown_origin(client):
|
||||
response = client.get("/api/health", headers={"Origin": "http://malvado.example"})
|
||||
|
||||
assert "access-control-allow-origin" not in response.headers
|
||||
|
||||
|
||||
def test_sources_endpoint_returns_a_list(client):
|
||||
response = client.get("/api/sources")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "sources" in response.json()
|
||||
Reference in new issue
Block a user