omp
68861b502c
Causa do 'Provedor de LLM indisponível' no ambiente com Ollama local foi externa: OLLAMA_BASE_URL exportada no shell (http://localhost:11434/v1) vence o .env na interpolação do docker-compose, e localhost dentro do container é o próprio container. Nada no app indicava a URL em uso. Dois defeitos de código que tornavam isso indetectável — e um deles produziria a mesma mensagem mesmo com a URL correta: - _BaseProvider criava o httpx.AsyncClient com o atributo de classe (120 s) e as subclasses atribuíam self.timeout depois. O httpx congela o timeout na construção, então os 300 s pedidos pelo Ollama nunca valeram: um primeiro atendimento com modelo frio era cortado em 120 s com exatamente a mensagem 'provedor indisponível'. O timeout agora é passado ao super(). - LLMProviderError de conexão não citava o endereço tentado; o log dizia só 'All connection attempts failed'. Passa a incluir a endpoint, com credenciais user:pass@ redigidas. /api/health expõe llm_endpoint para comparação direta com o .env. README e .env.example documentam a precedência shell > .env, o fato de /v1 ser a superfície OpenAI-compatível (o adaptador usa /api/chat nativo) e a latência medida do primeiro atendimento.
373 lines
13 KiB
Python
373 lines
13 KiB
Python
"""API HTTP do Pastor Online.
|
|
|
|
Rotas:
|
|
|
|
* `GET /api/health` — verificação de saúde
|
|
* `GET /api/status` — estado da indexação (fontes lidas e vetores indexados)
|
|
* `POST /api/reindex` — força a reindexação do corpus
|
|
* `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 .chat import ChatService
|
|
from .config import Settings, get_settings
|
|
from .embedding import Embedder, HashEmbedder, SentenceTransformerEmbedder
|
|
from .indexer import Indexer
|
|
from .llm import LLMProvider, LLMProviderError, build_provider
|
|
from .models import (
|
|
ChatRequest,
|
|
HealthResponse,
|
|
ReindexResponse,
|
|
Source,
|
|
SourcesResponse,
|
|
StatusResponse,
|
|
)
|
|
from .retrieval import HybridRetriever
|
|
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]
|
|
retriever: HybridRetriever
|
|
embedder_warning: str | None = None
|
|
|
|
|
|
def _build_embedder(settings: Settings) -> tuple[Embedder, str | None]:
|
|
"""Cria o embedder conforme o modelo configurado.
|
|
|
|
Devolve `(embedder, aviso)`. O aviso existe porque o fallback é silencioso por
|
|
natureza — sem ele, `/api/status` reportaria o nome do modelo configurado
|
|
enquanto a busca roda sem semântica nenhuma, e ninguém perceberia.
|
|
"""
|
|
model = settings.embedding_model.strip().lower()
|
|
if model in ("", "hash", "hash-test", "none"):
|
|
return HashEmbedder(), None
|
|
try:
|
|
return (
|
|
SentenceTransformerEmbedder(
|
|
settings.embedding_model, batch_size=settings.embedding_batch_size
|
|
),
|
|
None,
|
|
)
|
|
except Exception as exc:
|
|
logger.error(
|
|
"Não foi possível carregar o modelo de embedding %s: %s. "
|
|
"Usando embedder determinístico.",
|
|
settings.embedding_model,
|
|
exc,
|
|
)
|
|
return HashEmbedder(), (
|
|
f"O modelo de embedding '{settings.embedding_model}' não carregou ({exc}). "
|
|
"A busca está rodando com um embedder determinístico SEM semântica: "
|
|
"os resultados não são confiáveis. Corrija as dependências e reindexe."
|
|
)
|
|
|
|
|
|
def create_app(
|
|
settings: Settings | None = None,
|
|
store: VectorStore | None = None,
|
|
embedder: Embedder | None = None,
|
|
provider: LLMProvider | None = None,
|
|
indexer: Indexer | None = None,
|
|
embedder_warning: str | None = None,
|
|
) -> FastAPI:
|
|
"""Cria a aplicação FastAPI. As dependências podem ser injetadas nos testes."""
|
|
settings = settings or get_settings()
|
|
|
|
# A ordem importa: o embedder precisa existir antes do store. `VectorStore`
|
|
# default para `HashEmbedder` quando nasce sem um, e herdar o embedder dele
|
|
# faria a aplicação nunca carregar o modelo configurado — a busca passaria a
|
|
# rodar sem semântica enquanto `/api/status` anunciaria o nome do modelo.
|
|
if embedder is None and store is not None:
|
|
embedder = getattr(store, "embedder", None)
|
|
if embedder is None:
|
|
embedder, built_warning = _build_embedder(settings)
|
|
embedder_warning = embedder_warning or built_warning
|
|
if store is None:
|
|
store = VectorStore(
|
|
persist_dir=settings.vector_db_dir,
|
|
collection_name=settings.collection_name,
|
|
embedder=embedder,
|
|
)
|
|
if provider is None:
|
|
provider = build_provider(settings)
|
|
|
|
if indexer is None:
|
|
indexer = Indexer(settings=settings, embedder=embedder, store=store)
|
|
|
|
retriever = HybridRetriever(
|
|
store=store,
|
|
embedder=embedder,
|
|
top_k=settings.top_k,
|
|
min_similarity=settings.min_similarity,
|
|
max_keyword_only=settings.max_keyword_only,
|
|
)
|
|
|
|
chat = ChatService(
|
|
settings=settings,
|
|
store=store,
|
|
embedder=embedder,
|
|
provider=provider,
|
|
retriever=retriever,
|
|
)
|
|
|
|
services = Services(
|
|
settings=settings,
|
|
store=store,
|
|
embedder=embedder,
|
|
provider=provider,
|
|
indexer=indexer,
|
|
chat=chat,
|
|
last_sources=[],
|
|
retriever=retriever,
|
|
embedder_warning=embedder_warning,
|
|
)
|
|
reindex_lock = asyncio.Lock()
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
|
"""Indexa se configurado e sempre reconstrói o índice lexical.
|
|
|
|
O BM25 vive em memória enquanto o Chroma persiste em disco: sem a
|
|
reconstrução no arranque, um restart deixaria a busca híbrida degradada
|
|
em só-vetorial, em silêncio.
|
|
"""
|
|
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,
|
|
# A URL em uso de fato, não a que o .env parece dizer.
|
|
llm_endpoint=getattr(services.provider, "endpoint", None),
|
|
indexing="indexing" if state == "indexing" else "ready",
|
|
)
|
|
|
|
@app.get("/api/status", response_model=StatusResponse)
|
|
async def status() -> StatusResponse:
|
|
"""Estado da indexação: fontes lidas e vetores indexados."""
|
|
current = services.indexer.status()
|
|
errors = list(current.errors)
|
|
if services.embedder_warning:
|
|
errors.insert(0, services.embedder_warning)
|
|
return StatusResponse(
|
|
state=current.state,
|
|
sources_total=current.sources_total,
|
|
sources_indexed=current.sources_indexed,
|
|
chunks=current.chunks,
|
|
model=current.model,
|
|
last_indexed_at=current.last_indexed_at,
|
|
errors=errors,
|
|
verses_indexed=current.verses_indexed,
|
|
verses_total=current.verses_total,
|
|
current_file=current.current_file,
|
|
)
|
|
|
|
@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 do corpus."""
|
|
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,
|
|
"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,
|
|
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 o corpus (se `auto_index`) e reconstrói o índice lexical."""
|
|
if services.settings.auto_index:
|
|
await asyncio.to_thread(services.indexer.reindex)
|
|
await asyncio.to_thread(_rebuild_keyword_index, services)
|
|
|
|
|
|
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)
|
|
await asyncio.to_thread(_rebuild_keyword_index, services)
|
|
finally:
|
|
if lock.locked():
|
|
lock.release()
|
|
|
|
|
|
def _rebuild_keyword_index(services: Services) -> None:
|
|
"""Reconstrói o BM25 a partir do banco vetorial, tolerando falha.
|
|
|
|
Se falhar, a busca continua funcionando no lado vetorial — degradada, mas
|
|
disponível. Silenciar aqui seria esconder o problema, então fica no log.
|
|
"""
|
|
try:
|
|
services.retriever.rebuild()
|
|
except Exception:
|
|
logger.exception(
|
|
"Falha ao reconstruir o índice de palavras-chave; "
|
|
"a busca seguirá apenas vetorial"
|
|
)
|
|
|
|
|
|
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, embedder_warning = _build_embedder(settings)
|
|
store = InMemoryVectorStore(embedder=embedder)
|
|
provider = build_provider(settings)
|
|
indexer = Indexer(settings=settings, embedder=embedder, store=store)
|
|
return create_app(
|
|
settings=settings,
|
|
store=store,
|
|
embedder=embedder,
|
|
provider=provider,
|
|
indexer=indexer,
|
|
embedder_warning=embedder_warning,
|
|
)
|
|
|
|
|
|
app = _build_default_app()
|