Spaces:
Sleeping
Sleeping
File size: 3,214 Bytes
2ae7490 94fcd9b 2ae7490 057c21e 2ae7490 057c21e 94fcd9b 2ae7490 94fcd9b 2ae7490 94fcd9b 2ae7490 94fcd9b 2ae7490 94fcd9b 2ae7490 94fcd9b 2ae7490 94fcd9b 2ae7490 94fcd9b 2ae7490 94fcd9b 2ae7490 94fcd9b 2ae7490 94fcd9b 2ae7490 94fcd9b 057c21e 2ae7490 057c21e 2ae7490 057c21e 2ae7490 057c21e 2ae7490 057c21e 2ae7490 057c21e 2ae7490 057c21e 2ae7490 057c21e 2ae7490 057c21e 2ae7490 057c21e 2ae7490 057c21e 94fcd9b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | """
QA endpoint for grant analyst queries.
Clean implementation using the QA service layer with validated schemas.
"""
import asyncio
import json
from typing import Dict, Any
from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from uuid import uuid4
from src.logging.logger import get_logger
from analyzer.models import QARequest, QAChunk
from analyzer.qa_service import stream_answer, answer_question as qa_answer
logger = get_logger()
router = APIRouter(prefix="/qa", tags=["qa"])
@router.post("/")
async def qa_endpoint(request: QARequest) -> Dict[str, Any]:
"""
Answer a grant-related question (non-streaming).
Uses the QA service layer with validated models and prompt injection hardening.
Args:
request: QARequest with query and optional filters
Returns:
Dict with answer, citations, and metadata
Raises:
HTTPException: On service failures
"""
# Ensure session ID is set
if not request.session_id:
request.session_id = str(uuid4())
try:
# Use service layer
result = qa_answer(request)
# Log interaction
logger.info(
f"QA request processed: session={request.session_id}, "
f"latency={result['latency_ms']}ms, success={result['success']}"
)
return result
except Exception as e:
logger.error(f"QA endpoint error: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
@router.post("/stream")
async def qa_stream_endpoint(request: QARequest) -> StreamingResponse:
"""
Answer a grant-related question with Server-Sent Events streaming.
Streams the LLM response as it generates for better UX.
Args:
request: QARequest with query and optional filters
Returns:
StreamingResponse with SSE-formatted NDJSON stream
Raises:
HTTPException: On service failures
"""
# Ensure session ID is set
if not request.session_id:
request.session_id = str(uuid4())
async def event_stream():
"""Generate SSE event stream."""
try:
# Stream from service layer
for chunk in stream_answer(request):
# Serialize chunk to NDJSON
chunk_json = chunk.model_dump_json()
yield f"data: {chunk_json}\n\n"
# Allow other tasks to run
await asyncio.sleep(0)
except Exception as e:
logger.error(f"Stream error: {e}", exc_info=True)
# Send error chunk
error_chunk = QAChunk(type="error", error=str(e))
yield f"data: {error_chunk.model_dump_json()}\n\n"
return StreamingResponse(
event_stream(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no" # Disable nginx buffering
}
)
@router.get("/test")
async def test_qa() -> Dict[str, str]:
"""Test endpoint to verify QA service is running."""
return {
"status": "ok",
"message": "QA service is running",
"endpoint": "/qa"
}
|