main.py — lifespan으로 시작/종료 훅 묶기

@asynccontextmanager
async def lifespan(app: FastAPI):
    global _kafka_task
    await ensure_collections()
    if settings.kafka_enabled:
        _kafka_task = asyncio.create_task(run_consumer())
    yield
    if _kafka_task:
        _kafka_task.cancel()

FastAPI의 구식 @app.on_event("startup") 대신 lifespan 컨텍스트 매니저를 쓴 이유는 이게 최신 권장 방식이기도 하고, yield 전/후로 시작/종료 로직을 한 함수 안에 같이 두면 "이 리소스가 언제 시작해서 언제 정리되는지"가 한눈에 보이기 때문이다.

_kafka_task.cancel()만 하고 await로 태스크가 실제로 끝나길 기다리지 않는다 — aiokafka의 consumer.stop()이 finally 블록에서 호출되긴 하지만, cancel 직후 프로세스가 바로 종료되는 상황(로컬 개발 중 Ctrl+C)에서는 정리가 끝까지 실행되지 않을 수 있다.

routers/health.py

@router.get("/health", response_model=HealthResponse)
async def health() -> HealthResponse:
    return HealthResponse(status="ok", ai_enabled=settings.ai_enabled, qdrant_connected=await is_connected(), kafka_consumer_running=kafka_consumer.is_running)

qdrant_connected는 매 호출마다 실제로 Qdrant에 get_collections()를 날려 확인하는 라이브 체크이고, kafka_consumer_running은 consumer.py의 전역 플래그를 읽는 캐싱된 상태다 — 이 비대칭은 의도한 것: Qdrant는 매번 물어봐도 비용이 작은 반면, Kafka 컸슈머가 "진짜 살아있는지"를 매번 확인하려면 별도의 핑 메커니즘이 필요해서 지금은 "루프에 진입했는가"만 본다.

routers/ingest.py

if filename.lower().endswith(".pdf"):
    reader = PdfReader(io.BytesIO(raw))
    text = "\n".join(page.extract_text() or "" for page in reader.pages)
else:
    text = raw.decode("utf-8", errors="ignore")

errors="ignore"로 디코딩 에러를 조용히 무시하게 한 건 "일부 글자가 깨지더라도 문서 적재 자체는 실패하지 않게" 하기 위한 실용적 선택이다 — 데이터 품질보다 가용성을 우선한 트레이드오프라, EUC-KR 등 다른 인코딩이 섞여 들어오면 조용히 깨진 텍스트가 임베딩될 수 있다.

routers/chat.py — SSE 스트리밍을 스레드와 큐로 다리 놓기

Gemini SDK의 스트리밍 생성(generate_stream)은 동기 제네레이터다. 이걸 async 핸들러 안에서 그냥 돌리면 SDK가 네트워크 응답을 기다리는 동안 이벤트 루프 전체가 블로킹된다.

queue: asyncio.Queue[str | None] = asyncio.Queue()
loop = asyncio.get_running_loop()

def produce() -> None:
    try:
        for token in generate_stream(prompt):
            loop.call_soon_threadsafe(queue.put_nowait, token)
    finally:
        loop.call_soon_threadsafe(queue.put_nowait, None)

loop.run_in_executor(None, produce)

while True:
    token = await queue.get()
    if token is None:
        break
    yield f"data: {json.dumps({'token': token})}\n\n"

동기 제네레이터를 별도 스레드(run_in_executor)에서 통째로 소비시키고, 스레드가 받은 토큰을 asyncio.Queue에 넣어서 메인 이벤트 루프로 넘긴다. queue.put_nowait을 직접 호출하지 않고 loop.call_soon_threadsafe로 감싼 이유는 asyncio.Queue가 스레드 안전하지 않기 때문이다.

SSE 스펙의 event: 필드로 토큰 스트림과 메타데이터(라우팅 결과, 출처, 종료 신호)를 구분했다 — 클라이언트가 이벤트 이름별로 다른 핸들러를 붙일 수 있게.