Merge pull request #20 from 8ocket/develop

Develop

Author: sugarghost

app/api/sessions.py

@@ -2,7 +2,7 @@
 import uuid
 from typing import Annotated
 
-from fastapi import APIRouter, Depends, Request
+from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import StreamingResponse
 from sqlalchemy.orm import Session
 
@@ -11,7 +11,7 @@
 from app.exceptions import MindLogError
 from app.limiter import limiter
 from app.schemas.session import MessageCreateRequest
-from app.services import chat_service
+from app.services import chat_service, session_service
 from app.utils.sse import format_sse_done, format_sse_error, stream_sse_events
 
 logger = logging.getLogger(__name__)
@@ -82,7 +82,31 @@ async def generate():
         finally:
             yield format_sse_done()
 
-    return StreamingResponse(generate(), media_type="text/event-stream")
+    return StreamingResponse(
+        generate(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+            "Connection": "keep-alive",
+        },
+    )
+
+
+@router.delete("/{session_id}", status_code=204)
+async def delete_session(
+    session_id: uuid.UUID,
+    db: DbSession,
+):
+    """
+    세션 삭제 시 AI 소유 데이터 정리.
+
+    Java BE가 counseling_sessions 및 cascade 테이블 삭제 완료 후 호출한다.
+    AI 서비스는 embeddings 테이블의 해당 session_id 레코드를 삭제한다.
+    """
+    deleted = session_service.delete_session_data(session_id, db)
+    if not deleted:
+        raise HTTPException(status_code=404, detail="해당 세션의 AI 데이터를 찾을 수 없습니다.")
 
 
 @router.post("/{session_id}/finalize")
@@ -126,4 +150,12 @@ async def generate():
         finally:
             yield format_sse_done()
 
-    return StreamingResponse(generate(), media_type="text/event-stream")
+    return StreamingResponse(
+        generate(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+            "Connection": "keep-alive",
+        },
+    )

app/models/knowledge.py

@@ -165,6 +165,12 @@ class Embedding(Base):
         nullable=False,
         comment="원본_ID",
     )
+    session_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True),
+        nullable=True,
+        index=True,
+        comment="세션_ID (삭제용 메타데이터, FK 없음)",
+    )
     user_id: Mapped[uuid.UUID] = mapped_column(
         UUID(as_uuid=True),
         ForeignKey("users.user_id", ondelete="CASCADE"),

app/models/session.py

@@ -52,7 +52,6 @@ class CounselingSession(Base):
     )
     summary_id: Mapped[uuid.UUID | None] = mapped_column(
         UUID(as_uuid=True),
-        ForeignKey("session_summaries.summary_id"),
         nullable=True,
         comment="세션_요약_ID",
     )
@@ -83,7 +82,6 @@ class CounselingSession(Base):
         back_populates="session",
         uselist=False,
         cascade="all, delete-orphan",
-        foreign_keys="[CounselingSession.summary_id]",
     )
     emotion_extractions: Mapped[list[EmotionExtraction]] = relationship(
         back_populates="session", cascade="all, delete-orphan"

app/services/chat_service.py

@@ -23,6 +23,8 @@
 from sqlalchemy.orm import Session
 
 from app.config import get_settings
+from langchain_core.messages import AIMessageChunk
+
 from app.graph.builder import get_graph
 from app.models.session import CounselingSession
 from app.services import redis_service
@@ -106,18 +108,25 @@ async def stream_message(
         output_blocked = False
         _t0 = time.monotonic()
         try:
-            async for event in graph.astream_events(initial_state, config=config, version="v2"):
-                if event["event"] == "on_chat_model_stream":
-                    chunk_content = event["data"]["chunk"].content
-                    if chunk_content:
-                        full_chunks.append(chunk_content)
-                        yield "chunk", {"content": chunk_content}
-                elif event["event"] == "on_chat_model_end":
-                    output = event["data"].get("output")
-                    if output and hasattr(output, "response_metadata"):
-                        action = output.response_metadata.get("amazon-bedrock-guardrailAction")
-                        if action == "INTERVENED":
-                            output_blocked = True
+            last_token = None
+            async for chunk in graph.astream(
+                initial_state, config=config, stream_mode="messages", version="v2"
+            ):
+                if chunk["type"] != "messages":
+                    continue
+                token, _metadata = chunk["data"]
+                if not isinstance(token, AIMessageChunk):
+                    continue
+                last_token = token
+                if token.content:
+                    full_chunks.append(token.content)
+                    yield "chunk", {"content": token.content}
+
+            # Bedrock 가드레일 출력 차단 여부 확인 (마지막 토큰의 response_metadata)
+            if last_token is not None and hasattr(last_token, "response_metadata"):
+                action = last_token.response_metadata.get("amazon-bedrock-guardrailAction")
+                if action == "INTERVENED":
+                    output_blocked = True
         except Exception as e:
             logger.error(
                 "LangGraph 스트리밍 실패 | session_id=%s | error=%s", session_id, e, exc_info=True

app/services/embedding_service.py

@@ -133,6 +133,7 @@ async def store_session_embedding(
     embedding = Embedding(
         source_type="session_context",
         source_id=session_id,
+        session_id=session_id,
         user_id=user_id,
         vector=vector,
         model_version=_settings.bedrock_embedding_model_id,
@@ -151,6 +152,7 @@ async def store_proposition_embeddings(
     *,
     records: list[tuple[SessionProposition, str]],
     user_id: uuid.UUID,
+    session_id: uuid.UUID,
     session_summary_fact: str | None = None,
 ) -> None:
     """
@@ -189,6 +191,7 @@ async def store_proposition_embeddings(
             Embedding(
                 source_type="proposition",
                 source_id=prop.proposition_id,
+                session_id=session_id,
                 user_id=user_id,
                 vector=vector,
                 model_version=_settings.bedrock_embedding_model_id,

app/services/knowledge_service.py

@@ -302,6 +302,7 @@ async def extract_and_store_propositions(
         db,
         records=records,
         user_id=user_id,
+        session_id=session_id,
         session_summary_fact=session_summary_fact,
     )
 

app/services/session_service.py

@@ -0,0 +1,43 @@
+import logging
+import uuid
+
+from sqlalchemy.orm import Session
+
+from app.models.knowledge import Embedding
+
+logger = logging.getLogger(__name__)
+
+
+def delete_session_data(session_id: uuid.UUID, db: Session) -> bool:
+    """
+    세션 삭제 시 AI 서비스가 자체 관리하는 데이터를 정리한다.
+
+    호출 시점: Java BE가 counseling_sessions 및 cascade 테이블을 삭제한 이후.
+
+    삭제 대상:
+      - embeddings (source_type='session_context', source_type='proposition')
+        → Embedding.session_id 컬럼으로 일괄 삭제
+
+    처리하지 않는 항목:
+      - session_messages, session_propositions 등: Java BE cascade로 이미 삭제됨
+      - knowledge_relations.session_id: ondelete="SET NULL"으로 자동 null화
+      - knowledge_entities: 사용자 지식 누적 데이터이므로 유지
+      - Redis rag:{user_id}:* 캐시: TTL=180초 자연 만료
+
+    Returns:
+        True  — 임베딩 레코드가 존재해 삭제 성공
+        False — 해당 session_id의 임베딩 레코드 없음 (404 반환용)
+    """
+    deleted_count = (
+        db.query(Embedding)
+        .filter(Embedding.session_id == session_id)
+        .delete(synchronize_session=False)
+    )
+    db.commit()
+
+    if deleted_count > 0:
+        logger.info("세션 임베딩 삭제 완료 | session_id=%s | count=%d", session_id, deleted_count)
+    else:
+        logger.warning("삭제할 임베딩 없음 | session_id=%s", session_id)
+
+    return deleted_count > 0

app/utils/sse.py

@@ -4,6 +4,7 @@
 API 레이어의 중복 SSE 포맷팅 코드를 한 곳으로 통합한다.
 """
 
+import asyncio
 import json
 from collections.abc import AsyncIterator
 
@@ -50,3 +51,4 @@ async def stream_sse_events(
             yield format_sse(sse_event_name, {"content": event_data.get("content", "")})
         else:
             yield format_sse(sse_event_name, event_data)
+        await asyncio.sleep(0)  # OS TCP 버퍼 flush 기회 제공

migrations/versions/l2m3n4o5p6q7_add_session_id_to_embeddings.py

@@ -0,0 +1,35 @@
+"""add_session_id_to_embeddings
+
+embeddings 테이블에 session_id 컬럼 추가.
+Java BE 선삭제 후 AI DELETE API 호출 구조에서,
+session_propositions가 이미 삭제된 시점에도
+embeddings를 session_id로 직접 삭제할 수 있도록 메타데이터 컬럼 추가.
+
+Revision ID: l2m3n4o5p6q7
+Revises: be83bf3c61d9
+Create Date: 2026-04-14 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.postgresql import UUID
+
+revision: str = "l2m3n4o5p6q7"
+down_revision: Union[str, None] = "3488b28d150f"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "embeddings",
+        sa.Column("session_id", UUID(as_uuid=True), nullable=True, comment="세션_ID (삭제용 메타데이터, FK 없음)"),
+    )
+    op.create_index("ix_embeddings_session_id", "embeddings", ["session_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_embeddings_session_id", table_name="embeddings")
+    op.drop_column("embeddings", "session_id")