docs: .env.example 전면 업데이트 — 누락된 설정값 모두 추가

기존에 빠져 있던 항목 추가:
- LLM: ENABLE_THINKING, THINK_VERBOSE
- Qdrant: QDRANT_URL, QDRANT_COLLECTION
- Embedding: EMBEDDING_MODEL_ID, EMBEDDING_DEVICE
- RAG: RAG_TOP_K, RAG_VERBOSE, RAG_SHOW_SOURCES, LANGGRAPH_VERBOSE
- Semantic Chunker: SEMANTIC_BREAKPOINT_THRESHOLD_TYPE, SEMANTIC_BUFFER_SIZE
- Reranker: RERANKER_ENABLED, RERANKER_MODEL_ID, RERANKER_FETCH_K
- Voice: WHISPER_MODEL_SIZE, TTS_VOICE
- Vision: VISION_ENABLED, VISION_MODEL_ID, VISION_MAX_TOKENS
- IDEA-1: CONV_RAG_ENABLED
- IDEA-2: TELEGRAM_BOT_TOKEN, TELEGRAM_USER_MAP
- IDEA-5: CRAG_ENABLED
- IDEA-8: GRAPH_ENABLED

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

.env.example

@@ -1,17 +1,88 @@
-# LLM 모델 설정
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 율봇 환경 설정 예시 (.env.example)
+# 실제 사용 시 .env로 복사 후 값을 채워주세요.
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+# ── LLM ──────────────────────────────────────────
 MODEL_ID=mlx-community/Qwen3-8B-4bit
 MAX_TOKENS=1024
-MAX_HISTORY_TURNS=30
-COMPACT_THRESHOLD=40
+MAX_HISTORY_TURNS=10        # 메모리에 유지할 최대 대화 턴 수
+COMPACT_THRESHOLD=20        # 이 턴 초과 시 오래된 대화를 LLM으로 자동 요약
+ENABLE_THINKING=true        # Thinking 모드 활성화 (Qwen3 지원)
+THINK_VERBOSE=false         # true 시 UI에 thinking 토큰 스트리밍
 
-# MySQL 설정 (미설정 시 DB 기능 비활성화)
+# ── MySQL ─────────────────────────────────────────
+# 미설정(DB_USER 빈 값) 시 DB 기능 전체 비활성화 (인메모리 모드)
 DB_HOST=localhost
 DB_PORT=3306
 DB_NAME=youlbot
 DB_USER=
 DB_PASSWORD=
 
-# LangSmith 트레이싱 (Phase 7) — https://smith.langchain.com 에서 API 키 발급
+# ── Qdrant ────────────────────────────────────────
+QDRANT_URL=http://localhost:6333
+QDRANT_COLLECTION=youlbot_docs
+
+# ── Embedding ────────────────────────────────────
+EMBEDDING_MODEL_ID=BAAI/bge-m3
+EMBEDDING_DEVICE=mps        # mps (Apple Silicon) | cpu | cuda
+
+# ── RAG 검색 ─────────────────────────────────────
+RAG_TOP_K=3                 # 최종 반환할 문서 청크 수
+RAG_VERBOSE=false           # true 시 검색 쿼리·청크 내용 출력
+RAG_SHOW_SOURCES=false      # true 시 답변 아래 출처(파일명·페이지) 표시
+LANGGRAPH_VERBOSE=false     # true 시 LangGraph 노드 전환 로그 출력
+
+# ── Semantic Chunker (Phase 13) ───────────────────
+SEMANTIC_BREAKPOINT_THRESHOLD_TYPE=percentile   # percentile | standard_deviation | interquartile | gradient
+SEMANTIC_BUFFER_SIZE=1      # 인접 문장 묶음 크기 (1=단일 문장)
+
+# ── Reranker (Phase 13-B) ────────────────────────
+RERANKER_ENABLED=false
+RERANKER_MODEL_ID=cross-encoder/mmarco-mMiniLMv2-L12-H384-v1  # 한국어 지원 다국어 모델
+RERANKER_FETCH_K=10         # rerank 전 후보 수 (RAG_TOP_K보다 커야 함)
+
+# ── Hybrid Search (Phase 18) — BM25 + Vector ─────
+# 활성화 후 기존 문서는 재수집 필요
+HYBRID_SEARCH_ENABLED=false
+SPARSE_MODEL_ID=Qdrant/bm25
+
+# ── Query Rewriting (Phase 19) ───────────────────
+QUERY_REWRITE_ENABLED=false
+
+# ── CRAG — 검색 결과 없을 때 web_search 자동 fallback (IDEA-5) ──
+CRAG_ENABLED=false
+
+# ── 대화 기반 자동 RAG 인덱싱 (IDEA-1) ───────────
+# 응답 완료 후 LLM이 유용한 정보 판단 → Qdrant 자동 저장 (background task)
+CONV_RAG_ENABLED=false
+
+# ── 지식 그래프 / GraphRAG (IDEA-8) ──────────────
+# add_relation / query_entity 도구 활성화 + 시스템 프롬프트 자동 주입
+GRAPH_ENABLED=false
+
+# ── REST API (Phase 22) ───────────────────────────
+# 빈 값이면 인증 없음 (개발 모드)
+API_TOKEN=
+
+# ── LangSmith 트레이싱 (Phase 7) ─────────────────
+# https://smith.langchain.com 에서 API 키 발급
 LANGCHAIN_TRACING_V2=false
 LANGCHAIN_API_KEY=
 LANGCHAIN_PROJECT=youlbot
+
+# ── 음성 인터페이스 (Phase 14) ────────────────────
+WHISPER_MODEL_SIZE=small    # tiny | base | small | medium | large
+TTS_VOICE=Yuna              # macOS say 명령어 한국어 음성 (Yuna | Siri 등)
+
+# ── 멀티모달 이미지 이해 (Phase 17) ──────────────
+VISION_ENABLED=false
+VISION_MODEL_ID=mlx-community/Qwen2.5-VL-7B-Instruct-4bit
+VISION_MAX_TOKENS=512
+
+# ── 스마트 알림 / Telegram push (IDEA-2) ─────────
+# BotFather에서 봇 생성 후 토큰 입력
+TELEGRAM_BOT_TOKEN=
+# user_id → Telegram numeric chat_id 매핑 (JSON 형식)
+# 각 가족의 Telegram ID는 봇에 /start 전송 후 로그에서 확인
+TELEGRAM_USER_MAP={"아록": "", "근혜": "", "도율": "", "하율": ""}

api.py

@@ -0,0 +1,209 @@
+"""율봇 REST API — Phase 22.
+
+실행:
+    uvicorn api:app --host 0.0.0.0 --port 8000
+
+클라이언트 예시:
+    import httpx, json
+    headers = {"Authorization": "Bearer YOUR_TOKEN"}
+    with httpx.Client() as c:
+        with c.stream("POST", "http://localhost:8000/chat",
+                      json={"message": "안녕", "user_id": "홍길동"},
+                      headers=headers, timeout=120) as r:
+            for line in r.iter_lines():
+                if not line.startswith("data: "):
+                    continue
+                payload = json.loads(line[6:])
+                if isinstance(payload, dict) and payload.get("__done"):
+                    break  # run_id = payload["run_id"]
+                print(payload, end="", flush=True)
+"""
+import json
+import os
+import tempfile
+from contextlib import asynccontextmanager
+
+from dotenv import load_dotenv
+load_dotenv()
+
+from fastapi import Depends, FastAPI, File, Header, HTTPException, UploadFile
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel
+
+from container import Container
+from services.agent.agent_service import AgentService
+
+_container = Container()
+_container.db_service().connect()
+_container.db_service().init_schema()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    scheduler = _container.scheduler_service()
+    scheduler.start()
+    yield
+    scheduler.shutdown()
+
+
+app = FastAPI(title="율봇 API", version="1.0", lifespan=lifespan)
+
+_cfg = _container.config()
+_agent_cache: dict[str, AgentService] = {}
+
+# Vision 모델 — VISION_ENABLED=true 시 lazy 초기화
+_vision_model = _container.vision_model() if _cfg.vision_enabled else None
+
+
+def _get_agent(user_id: str) -> AgentService:
+    if user_id not in _agent_cache:
+        _agent_cache[user_id] = AgentService(
+            chat_model=_container.chat_model(),
+            retriever_service=_container.retriever_service(),
+            system_prompt=_cfg.system_prompt,
+            rag_verbose=_cfg.rag_verbose,
+            rag_show_sources=_cfg.rag_show_sources,
+            langgraph_verbose=_cfg.langgraph_verbose,
+            think_verbose=_cfg.think_verbose,
+            query_rewrite_enabled=_cfg.query_rewrite_enabled,
+            user_profile_repository=_container.user_profile_repository(),
+            conversation_repository=_container.conversation_repository(),
+            reminder_repository=_container.reminder_repository(),
+            ingestion_service=_container.ingestion_service() if _cfg.conv_rag_enabled else None,
+            crag_enabled=_cfg.crag_enabled,
+            conv_rag_enabled=_cfg.conv_rag_enabled,
+            graph_service=_container.graph_service() if _cfg.graph_enabled else None,
+            user_id=user_id,
+        )
+        if _vision_model:
+            _agent_cache[user_id].set_vision_model(_vision_model)
+    return _agent_cache[user_id]
+
+
+def _auth(authorization: str = Header(default="")):
+    """API_TOKEN 설정 시 Bearer 토큰 검증. 미설정 시 인증 스킵(개발 모드)."""
+    token = _cfg.api_token
+    if token and authorization != f"Bearer {token}":
+        raise HTTPException(status_code=401, detail="Unauthorized")
+
+
+# ── 요청/응답 모델 ────────────────────────────────────────────
+
+
+class ChatRequest(BaseModel):
+    message: str
+    user_id: str = "default"
+    show_thinking: bool = False
+    image_base64: str | None = None  # base64 인코딩된 이미지 (선택)
+
+
+class FeedbackRequest(BaseModel):
+    user_id: str = "default"
+    user_msg: str
+    asst_msg: str
+    rating: int
+    run_id: str | None = None
+
+
+# ── 엔드포인트 ────────────────────────────────────────────────
+
+
+@app.get("/health")
+async def health():
+    return {"status": "ok"}
+
+
+@app.post("/chat")
+async def chat(req: ChatRequest, _=Depends(_auth)):
+    """SSE 스트리밍 응답. 각 라인: `data: <JSON 토큰>\n\n`, 종료: `data: [DONE]\n\n`"""
+    agent = _get_agent(req.user_id)
+
+    # 이미지 base64 → 임시 파일 저장
+    image_path: str | None = None
+    tmp_path: str | None = None
+    if req.image_base64 and _vision_model:
+        import base64
+        img_bytes = base64.b64decode(req.image_base64)
+        suffix = ".jpg"
+        if img_bytes[:4] == b"\x89PNG":
+            suffix = ".png"
+        elif img_bytes[:4] == b"GIF8":
+            suffix = ".gif"
+        tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False, dir="/tmp", prefix="youlbot_img_")
+        tmp.write(img_bytes)
+        tmp.close()
+        image_path = tmp.name
+        tmp_path = tmp.name
+
+    async def generate():
+        try:
+            async for token in agent.stream_response(
+                req.message, show_thinking=req.show_thinking, image_path=image_path
+            ):
+                yield f"data: {json.dumps(token, ensure_ascii=False)}\n\n"
+            yield f"data: {json.dumps({'__done': True, 'run_id': agent.last_run_id}, ensure_ascii=False)}\n\n"
+        finally:
+            if tmp_path and os.path.exists(tmp_path):
+                os.unlink(tmp_path)
+
+    return StreamingResponse(generate(), media_type="text/event-stream")
+
+
+@app.post("/feedback")
+async def save_feedback(req: FeedbackRequest, _=Depends(_auth)):
+    """👍/👎 피드백 저장. LangSmith 트레이싱 활성화 시 자동 연동."""
+    _container.feedback_repository().save_feedback(
+        req.user_id, req.user_msg, req.asst_msg, req.rating, req.run_id
+    )
+    if req.run_id and os.getenv("LANGCHAIN_TRACING_V2") == "true":
+        try:
+            from langsmith import Client
+            Client().create_feedback(run_id=req.run_id, key="user_feedback", score=req.rating)
+        except Exception:
+            pass
+    return {"saved": True}
+
+
+@app.post("/reset")
+async def reset(user_id: str = "default", _=Depends(_auth)):
+    """대화 이력 초기화."""
+    if user_id in _agent_cache:
+        _agent_cache[user_id].reset()
+    return {"reset": True, "user_id": user_id}
+
+
+@app.post("/ingest")
+async def ingest(file: UploadFile = File(...), _=Depends(_auth)):
+    """PDF 또는 TXT 파일을 업로드해 벡터DB에 수집."""
+    suffix = os.path.splitext(file.filename or "")[1] or ".bin"
+    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as f:
+        f.write(await file.read())
+        tmp_path = f.name
+    try:
+        count = _container.ingestion_service().ingest([tmp_path])
+        return {"chunks": count, "filename": file.filename}
+    finally:
+        os.unlink(tmp_path)
+
+
+@app.get("/documents")
+async def list_documents(_=Depends(_auth)):
+    """등록된 문서 경로 목록 반환."""
+    return {"documents": _container.retriever_service().list_documents()}
+
+
+@app.delete("/documents/{source:path}")
+async def delete_document(source: str, _=Depends(_auth)):
+    """source 경로에 해당하는 모든 청크 삭제."""
+    _container.retriever_service().delete_document(source)
+    return {"deleted": source}
+
+
+@app.get("/reminders/{user_id}")
+async def list_reminders(user_id: str, days_ahead: int = 30, _=Depends(_auth)):
+    """user_id의 예정 알림 목록 반환 (기본 30일 이내)."""
+    items = _container.reminder_repository().get_upcoming(user_id, days_ahead=days_ahead)
+    return {"reminders": [
+        {"id": r["id"], "remind_date": str(r["remind_date"]), "message": r["message"]}
+        for r in items
+    ]}

app.py

@@ -71,6 +71,7 @@ def _get_agent(user_id: str) -> AgentService:
             rag_show_sources=_cfg.rag_show_sources,
             langgraph_verbose=_cfg.langgraph_verbose,
             think_verbose=_cfg.think_verbose,
+            query_rewrite_enabled=_cfg.query_rewrite_enabled,
             user_profile_repository=container.user_profile_repository(),
             conversation_repository=container.conversation_repository(),
             user_id=user_id,

config.py

@@ -41,6 +41,16 @@ class Config(BaseSettings):
     reranker_enabled: bool = False
     reranker_model_id: str = "cross-encoder/mmarco-mMiniLMv2-L12-H384-v1"  # 한국어 지원 다국어 모델
     reranker_fetch_k: int = 10  # rerank 전 벡터 검색 후보 수 (rag_top_k보다 커야 함)
+
+    # Hybrid Search (Phase 18) — BM25 + Vector
+    hybrid_search_enabled: bool = False
+    sparse_model_id: str = "Qdrant/bm25"  # fastembed sparse 모델 (언어 무관 BM25)
+
+    # Query Rewriting (Phase 19) — 구어체 질문을 검색 최적화 쿼리로 변환
+    query_rewrite_enabled: bool = False
+
+    # REST API (Phase 22) — 빈 문자열이면 인증 스킵 (개발 모드)
+    api_token: str = ""
     rag_verbose: bool = False
     rag_show_sources: bool = False
     langgraph_verbose: bool = False
@@ -49,7 +59,28 @@ class Config(BaseSettings):
     whisper_model_size: str = "small"
     tts_voice: str = "Yuna"  # macOS say 명령어 한국어 음성
 
-    system_prompt: str = """모든 응답과 내부 사고 과정을 반드시 한국어로 작성하세요.
+    # Vision (Phase 17)
+    vision_enabled: bool = False
+    vision_model_id: str = "mlx-community/Qwen2.5-VL-7B-Instruct-4bit"
+    vision_max_tokens: int = 512
+
+    # 지식 그래프 (IDEA-8) — GraphRAG
+    graph_enabled: bool = False
+
+    # CRAG — 검색 결과 없을 때 web_search 자동 fallback (IDEA-5)
+    crag_enabled: bool = False
+
+    # 대화 기반 자동 RAG 인덱싱 (IDEA-1)
+    conv_rag_enabled: bool = False
+
+    # Scheduler / Telegram 알림 (IDEA-2)
+    # TELEGRAM_BOT_TOKEN: BotFather에서 발급받은 봇 토큰
+    # TELEGRAM_USER_MAP: JSON 형식으로 user_id → Telegram chat_id 매핑
+    #   예) TELEGRAM_USER_MAP={"아록": "123456789", "근혜": "987654321"}
+    telegram_bot_token: str = ""
+    telegram_user_map: str = "{}"
+
+    system_prompt: str = """모든 사고 과정(thinking)과 답변은 반드시 한국어로만 작성하세요. 영어 사용 절대 금지.
 
 당신의 이름은 '율봇'입니다. 친절하고 따뜻한 한국어 상담 도우미입니다.
 육아와 금융 두 분야를 전문으로 합니다.

container.py

@@ -10,14 +10,19 @@ from services.db.mysql_service import DatabaseService
 from services.db.conversation_repository import ConversationRepository
 from services.db.user_profile_repository import UserProfileRepository
 from services.db.feedback_repository import FeedbackRepository
+from services.db.reminder_repository import ReminderRepository
+from services.scheduler_service import SchedulerService
+from services.knowledge.graph_service import GraphService
 from services.ui.cli_service import CliUiService
 from services.events.event_bus import EventBus
 from services.events.handlers import StreamTokenHandler, StreamEndHandler
 from langchain_huggingface import HuggingFaceEmbeddings
+from langchain_qdrant import FastEmbedSparse
 from services.rag.ingestion_service import IngestionService
 from services.rag.rerank_service import RerankService
 from services.rag.retriever_service import RetrieverService
 from services.agent.agent_service import AgentService
+from services.model.mlx_vision_model import MlxVisionModel
 
 
 class Container(containers.DeclarativeContainer):
@@ -67,6 +72,23 @@ class Container(containers.DeclarativeContainer):
         db=db_service,
     )
 
+    reminder_repository = providers.Singleton(
+        ReminderRepository,
+        db=db_service,
+    )
+
+    scheduler_service = providers.Singleton(
+        SchedulerService,
+        reminder_repo=reminder_repository,
+        bot_token=providers.Callable(lambda c: c.telegram_bot_token, config),
+        user_map_json=providers.Callable(lambda c: c.telegram_user_map, config),
+    )
+
+    graph_service = providers.Singleton(
+        GraphService,
+        db=db_service,
+    )
+
     history_service = providers.Factory(
         HistoryService,
         system_prompt=providers.Callable(lambda c: c.system_prompt, config),
@@ -96,6 +118,16 @@ class Container(containers.DeclarativeContainer):
         model_kwargs=providers.Callable(lambda c: {"device": c.embedding_device}, config),
     )
 
+    reranker = providers.Callable(
+        lambda c: RerankService(c.reranker_model_id) if c.reranker_enabled else None,
+        config,
+    )
+
+    sparse_embeddings = providers.Singleton(
+        lambda c: FastEmbedSparse(model_name=c.sparse_model_id) if c.hybrid_search_enabled else None,
+        config,
+    )
+
     ingestion_service = providers.Singleton(
         IngestionService,
         embeddings=embeddings,
@@ -105,11 +137,7 @@ class Container(containers.DeclarativeContainer):
             lambda c: c.semantic_breakpoint_threshold_type, config
         ),
         buffer_size=providers.Callable(lambda c: c.semantic_buffer_size, config),
-    )
-
-    reranker = providers.Callable(
-        lambda c: RerankService(c.reranker_model_id) if c.reranker_enabled else None,
-        config,
+        sparse_embeddings=sparse_embeddings,
     )
 
     retriever_service = providers.Singleton(
@@ -120,6 +148,14 @@ class Container(containers.DeclarativeContainer):
         top_k=providers.Callable(lambda c: c.rag_top_k, config),
         reranker=reranker,
         rerank_fetch_k=providers.Callable(lambda c: c.reranker_fetch_k, config),
+        sparse_embeddings=sparse_embeddings,
+    )
+
+    # Phase 17 — Vision Model (lazy load)
+    vision_model = providers.Singleton(
+        MlxVisionModel,
+        model_id=providers.Callable(lambda c: c.vision_model_id, config),
+        max_tokens=providers.Callable(lambda c: c.vision_max_tokens, config),
     )
 
     # Phase 3 — LangGraph Agent
@@ -132,6 +168,7 @@ class Container(containers.DeclarativeContainer):
         rag_show_sources=providers.Callable(lambda c: c.rag_show_sources, config),
         langgraph_verbose=providers.Callable(lambda c: c.langgraph_verbose, config),
         think_verbose=providers.Callable(lambda c: c.think_verbose, config),
+        query_rewrite_enabled=providers.Callable(lambda c: c.query_rewrite_enabled, config),
         user_profile_repository=user_profile_repository,
         conversation_repository=conversation_repository,
     )

docs/01-plan/features/phase17-multimodal.plan.md

@@ -0,0 +1,202 @@
+---
+template: plan
+version: 1.3
+feature: phase17-multimodal
+date: 2026-06-02
+author: sal
+project: youlbot
+status: Draft
+---
+
+# phase17-multimodal Planning Document
+
+> **Summary**: analyze_image 도구 방식으로 이미지 이해 기능을 추가한다.
+> Qwen3-8B가 대화를 유지하고, 이미지 첨부 시 Qwen2.5-VL-7B를 도구로 호출해 설명을 얻은 뒤 답변한다.
+>
+> **Project**: youlbot
+> **Author**: sal
+> **Date**: 2026-06-02
+> **Status**: Draft
+
+---
+
+## Executive Summary
+
+| Perspective | Content |
+|-------------|---------|
+| **Problem** | 이유식 사진·금융 서류 등 이미지를 텍스트로만 처리하는 현재 한계 |
+| **Solution** | Qwen2.5-VL-7B를 `analyze_image` LangChain 도구로 래핑, Qwen3-8B가 필요 시 자동 호출 |
+| **Function/UX Effect** | 채팅창에 이미지 첨부 → 자동 분석 → 육아·금융 상담으로 자연스럽게 연결 |
+| **Core Value** | 텍스트 추론 품질(Qwen3-8B)을 유지하면서 이미지 이해 기능 추가 |
+
+---
+
+## Context Anchor
+
+| Key | Value |
+|-----|-------|
+| **WHY** | 손이 자유롭지 않은 육아 상황에서 사진 한 장으로 재료 분석·서류 해석이 가능해야 함 |
+| **WHO** | 아록(주 사용자) — 이유식 사진, 건강보험 서류, 접종 기록지 등 촬영 후 질문 |
+| **RISK** | 16GB 메모리에서 두 모델 동시 로드 시 OOM 가능 → Vision 모델 lazy load로 완화 |
+| **SUCCESS** | 이미지 첨부 → analyze_image 도구 자동 호출 → 설명이 대화 히스토리에 남아 후속 질문 가능 |
+| **SCOPE** | 이미지 분석 + 채팅 연동. 동영상·실시간 캡처는 제외 |
+
+---
+
+## 1. Overview
+
+### 1.1 Purpose
+사진을 첨부하면 `analyze_image` 도구가 Qwen2.5-VL-7B를 호출해 이미지 설명을 생성하고,
+Qwen3-8B가 그 설명을 컨텍스트로 삼아 육아·금융 상담 답변을 제공한다.
+
+### 1.2 모델 분담
+
+| 모델 | 역할 | 메모리 |
+|------|------|--------|
+| Qwen3-8B-4bit | 대화·추론·도구 결정 (항상 로드) | ~5GB |
+| Qwen2.5-VL-7B-Instruct-4bit | 이미지 분석 (lazy load) | ~5GB |
+| 합계 | — | ~10GB / 16GB 사용 가능 |
+
+---
+
+## 2. Scope
+
+### 2.1 In Scope
+- `mlx-vlm` 패키지로 Vision 모델 로드 및 추론
+- `analyze_image(image_path, prompt)` LangChain 도구 구현
+- AgentService: 요청에 이미지 있을 때 도구 동적 주입
+- API(`/chat`): 이미지 파일 업로드 지원 (multipart form)
+- WebUI: 채팅 입력창에 이미지 첨부 버튼 추가
+- Telegram: 사진 메시지 수신 → 이미지 다운로드 → API 전달
+
+### 2.2 Out of Scope
+- 동영상 분석
+- 이미지 생성(text-to-image)
+- 실시간 카메라 입력
+
+---
+
+## 3. Architecture — C방식 (analyze_image 도구)
+
+```
+사용자
+  │  텍스트 + 이미지(선택)
+  ▼
+API /chat  (multipart form)
+  │  image → /tmp/youlbot_img_xxx.jpg 저장
+  │  image_path → AgentService.stream_response(message, image_path=...)
+  ▼
+AgentService
+  │  image_path 있을 때: analyze_image 도구를 tools 목록에 동적 추가
+  │  image_path를 도구 클로저로 바인딩
+  ▼
+LangGraph ReAct
+  │  Qwen3-8B가 이미지 관련 질문 감지 → analyze_image() 자동 호출
+  ▼
+analyze_image 도구
+  │  mlx_vision_model.analyze(image_path, prompt)
+  ▼
+MlxVisionModel (Qwen2.5-VL-7B, lazy load)
+  │  이미지 설명 텍스트 반환
+  ▼
+LangGraph
+  │  설명이 ToolMessage로 대화 히스토리에 저장
+  ▼
+Qwen3-8B  →  최종 답변 생성
+```
+
+**핵심 특성:**
+- Vision 모델은 처음 analyze_image 호출 시 로드 (이후 캐시)
+- 이미지 설명이 대화 히스토리에 남아 후속 질문("그 재료로 이유식 만들어줘") 가능
+- 이미지 없는 메시지는 기존과 완전히 동일하게 동작
+
+---
+
+## 4. 변경 파일 목록
+
+### 신규 생성
+| 파일 | 설명 |
+|------|------|
+| `services/model/mlx_vision_model.py` | MlxVisionModel 클래스 (mlx-vlm 래퍼, lazy load) |
+
+### 수정
+| 파일 | 변경 내용 |
+|------|----------|
+| `config.py` | `vision_enabled: bool`, `vision_model_id: str` 추가 |
+| `container.py` | `vision_model` Singleton 프로바이더 추가 |
+| `services/agent/tools.py` | `make_vision_tool(vision_model, image_path)` 추가 |
+| `services/agent/agent_service.py` | `stream_response(image_path=None)` 파라미터 추가, 도구 동적 주입 |
+| `api.py` | `/chat` → multipart form으로 변경, 이미지 temp 저장 |
+| `youlbot-webui/api_client.py` | `chat(image_path=None)` 파라미터 추가, multipart 전송 |
+| `youlbot-webui/app.py` | 채팅 입력 영역에 이미지 업로드 컴포넌트 추가 |
+
+---
+
+## 5. 주요 구현 세부사항
+
+### 5.1 MlxVisionModel
+```python
+class MlxVisionModel:
+    def __init__(self, model_id: str): ...
+    
+    def analyze(self, image_path: str, prompt: str = "이 이미지를 한국어로 자세히 설명해줘.") -> str:
+        # 첫 호출 시 lazy load
+        # mlx_vlm.generate() 호출
+        # 한국어 설명 반환
+```
+
+### 5.2 make_vision_tool
+```python
+def make_vision_tool(vision_model, image_path: str):
+    @tool
+    def analyze_image(prompt: str = "이 이미지를 설명해줘") -> str:
+        """현재 첨부된 이미지를 분석한다."""
+        return vision_model.analyze(image_path, prompt)
+    return analyze_image
+```
+
+### 5.3 API /chat 변경
+- JSON Body → `multipart/form-data`
+- 필드: `message`, `user_id`, `show_thinking`, `image` (optional file)
+- 이미지를 `/tmp/youlbot_img_{uuid}.{ext}`에 저장 후 agent에 전달
+- 응답 완료 후 temp 파일 삭제
+
+### 5.4 WebUI 변경
+- `gr.Image(type="filepath", ...)` 컴포넌트 채팅 입력 영역에 추가
+- 이미지 첨부 시 api_client.chat()에 image_path 전달
+- 전송 후 이미지 초기화
+
+---
+
+## 6. 환경 설정
+
+```env
+# .env 추가
+VISION_ENABLED=true
+VISION_MODEL_ID=mlx-community/Qwen2.5-VL-7B-Instruct-4bit
+```
+
+```bash
+# 패키지 설치
+pip install mlx-vlm
+```
+
+---
+
+## 7. 위험 요소 및 대응
+
+| 위험 | 대응 |
+|------|------|
+| 16GB에서 두 모델 동시 OOM | Vision 모델 lazy load + 미사용 시 unload 옵션 제공 |
+| mlx-vlm API 변경 가능성 | MlxVisionModel로 캡슐화해 교체 용이하게 |
+| Telegram 이미지 전달 복잡성 | Phase 17-B로 분리, 우선 WebUI만 구현 |
+| 이미지 temp 파일 누적 | 응답 완료 후 즉시 삭제 |
+
+---
+
+## 8. 성공 기준
+
+- [ ] 이미지 첨부 시 `analyze_image` 도구가 자동 호출되어 설명 생성
+- [ ] "이 사진에서 뭐가 보여?" 후속 질문이 히스토리 기반으로 동작
+- [ ] 이미지 없는 일반 질문은 기존과 동일하게 Qwen3-8B로 처리
+- [ ] 16GB 환경에서 OOM 없이 동작

docs/ROADMAP.md

@@ -6,13 +6,15 @@
 |------|------|
 | LLM | Qwen3-8B-4bit (MLX, Apple Silicon) |
 | Agent | LangGraph ReAct + Tool Calling + Thinking 모드 |
+| Scheduler | asyncio task 기반 알림 스케줄러 — D-7/D-1/D-0 Telegram push (`SchedulerService`) |
 | RAG | Qdrant + BAAI/bge-m3 임베딩 + Semantic Chunking (`SemanticChunker`) + Reranker (BAAI/bge-reranker-v2-m3) |
-| Tools | `search_documents`, `web_search`, `get_current_date`, `remember_user_info`, `recall_user_info` (5개) |
+| Tools | `search_documents`, `web_search`, `get_current_date`, `remember_user_info`, `recall_user_info`, `set_reminder`, `list_reminders` (7개) |
 | Feedback | Gradio 👍/👎 → `td_feedback` DB 저장 + LangSmith `create_feedback()` 연동 |
 | UI | CLI + Gradio Web UI + 음성 입력(STT)/출력(TTS) |
 | Memory | LangGraph MemorySaver (세션 내) + MySQL 대화 저장 + 장기 사용자 프로필 |
 | Tracing | LangSmith 트레이싱 |
-| Streaming | 비동기 토큰 스트리밍 + `<think>` 블록 파싱 |
+| Streaming | 비동기 토큰 스트리밍 + 타입별 이벤트 분리 (`__meta` / `__thinking` / `__status`) |
+| 사고 과정 UI | 스트리밍 중 현재 줄 실시간 표시 → 완료 후 접기/펼치기 (`<details>`) |
 | History Compact | 대화 20턴 초과 시 오래된 절반을 LLM으로 자동 요약 (`CompactService`) |
 | 나이 계산 | 시스템 프롬프트에 오늘 날짜 주입 + 한국 나이/만 나이 자동 계산 |
 
@@ -32,6 +34,17 @@ DB 스키마(`td_conversations.user_id`, `td_user_profile.user_id`)는 `_migrate
 ### ✅ 버그 4 — 나이 계산 오류 (수정 완료)
 LLM이 훈련 데이터 기준 연도로 나이를 계산하는 문제. `AgentService.call_model()`에서 매 호출 시 시스템 프롬프트 앞에 `오늘 날짜: {date.today().isoformat()}`를 주입. 프로필에서 생년월일/생년 값을 파싱해 한국 나이(현재연도-출생연도+1)와 만 나이(생일 기준 정확 계산)를 자동 계산해 시스템 프롬프트에 포함.
 
+### ✅ 버그 6 — TTS가 진행 메시지까지 읽는 문제 (수정 완료)
+
+`stream_response()`가 `[LangGraph → agent: ...]`, `문서 검색 중...` 등 진행 메시지와 실제 답변을 동일한 plain string으로 yield해 TTS가 전부 읽던 문제.
+
+- `stream_response()` yield 타입 분리: 답변 → `plain str`, 진행/thinking/출처 → `{"__meta": str}` dict
+- thinking 토큰은 별도 `{"__thinking": str}` key 사용
+- `call_model` 시작 직후 `writer({"__start": True})` emit → `{"__status": label}` 변환으로 LLM 추론 전 즉각 피드백
+- `api.py`: `json.dumps(token)` 이 dict/str 모두 처리하므로 변경 없음
+- WebUI `respond()`: `tts_text` 누적 변수 분리, `__meta`·`__thinking` 토큰 제외 후 TTS 전달
+- Telegram `bot.py`: `__meta`·`__thinking` 토큰 skip
+
 ### ✅ 버그 5 — 사고 과정(thinking) 체크박스 무효 (수정 완료)
 ON/OFF와 무관하게 사고 과정이 표시되지 않던 버그.
 - `call_model` 내부에서 `get_stream_writer()`로 thinking 토큰을 custom 이벤트로 emit → 답변 앞에 먼저 스트리밍
@@ -184,41 +197,252 @@ turns = conversation_repository.load_turns_after(self._conv_id, None, limit=10)
 
 ---
 
-## Phase 18 — Hybrid Search (BM25 + Vector) ★★☆
+## ✅ Phase 18 — Hybrid Search (BM25 + Vector) ★★☆
 
 **배경**: 한국어 질문에서 고유명사·전문용어가 포함된 경우 의미 검색(Dense)만으로는 recall이 떨어진다. BM25 키워드 검색과 결합(Hybrid)하면 보완이 가능하다.
 
-**구현 방식**:
-- Qdrant의 Sparse Vector 지원 활용 (`FastEmbedSparseEmbeddings` 또는 BM42)
-- 인덱싱 시 dense + sparse 두 벡터 동시 저장
-- 검색 시 `RRF(Reciprocal Rank Fusion)`로 결과 통합
-- `IngestionService`, `RetrieverService` 양쪽 수정 필요
+**구현 내용**:
+- `FastEmbedSparse(model_name="Qdrant/bm25")` — 언어 무관 BM25 sparse 임베딩 (`fastembed` 패키지)
+- `IngestionService`: `HYBRID_SEARCH_ENABLED=true` 시 dense + sparse 동시 저장 (`RetrievalMode.HYBRID`)
+- `RetrieverService`: hybrid 스토어로 검색 → Qdrant 내장 RRF로 결과 통합; sparse vector 미설정 컬렉션은 dense로 자동 폴백
+- `_ensure_collection_schema()`: hybrid 전환 시 스키마 불일치 컬렉션 자동 재생성 (기존 문서 재수집 필요)
+- `.env` `HYBRID_SEARCH_ENABLED=true`로 활성화, 활성화 후 기존 문서 재수집 필요
+
+| 설정 | 기본값 | 설명 |
+|------|--------|------|
+| `HYBRID_SEARCH_ENABLED` | `false` | `true`로 설정 시 활성화 |
+| `SPARSE_MODEL_ID` | `Qdrant/bm25` | fastembed sparse 모델 (첫 실행 시 자동 다운로드) |
 
 **난이도**: 중간 | **임팩트**: 높음 (키워드 포함 질문 recall 대폭 향상)
 
 ---
 
-## Phase 19 — Query Rewriting ★☆☆
+## ✅ Phase 19 — Query Rewriting ★☆☆
 
 **배경**: 사용자 구어체 질문("아이가 밥을 안 먹어요")은 벡터 검색에 최적화되어 있지 않다. LLM이 검색 전에 질문을 재작성하면 관련 문서 검색 확률이 높아진다.
 
-**구현 방식**:
-- LangGraph에 `query_rewrite` 노드 추가 (agent → query_rewrite → tools 순서)
-- 또는 `search_documents` 도구 내부에서 rewrite 후 검색
-- 프롬프트: "다음 질문을 문서 검색에 최적화된 키워드 중심 문장으로 변환하세요"
+**구현 내용**:
+- LangGraph 그래프에 `query_rewrite` 노드 추가 — `agent → query_rewrite → tools` 순서
+- `search_documents` 호출 시에만 작동하는 조건부 라우팅 (`route_after_agent`): 다른 도구 호출이나 tool 없음 케이스는 그대로 통과
+- 구어체 → 키워드 중심 쿼리로 변환 + 대명사·지시어를 구체적 명칭으로 해소 (이전 대화 2턴 컨텍스트 활용)
+- `tools_condition` 제거 → 커스텀 `route_after_agent` 함수로 대체
+- 변환 결과를 custom stream 이벤트로 emit → `RAG_VERBOSE=true` 시 `쿼리 최적화: "원본" → "최적화"` 출력
+- `.env` `QUERY_REWRITE_ENABLED=true`로 활성화
 
 **난이도**: 하 | **임팩트**: 중간 (구어체 질문 검색 품질 향상)
 
 ---
 
-## Phase 20 — RAG 품질 자동 평가 (RAGAS) ★☆☆
+## ✅ Phase 21 — Telegram Bot ★★☆
+
+**배경**: Gradio Web UI는 브라우저에서만 사용 가능. 텔레그램으로 이동 중에도 율봇과 대화하고 싶음.
+
+**구현 방식**: youlbot REST API(Phase 22) 호출 — `youlbot-telegram/` 별도 프로젝트로 분리.
+
+```
+youlbot-telegram/
+├── bot.py        ← Application (python-telegram-bot >= 20.0, async)
+│   ├── /start, /reset CommandHandler
+│   └── MessageHandler → api_client.chat() → edit_message_text() (타이핑 효과)
+├── api_client.py ← httpx 기반 REST API 클라이언트 (chat/reset)
+├── .env          ← TELEGRAM_BOT_TOKEN, YOULBOT_API_URL, 유저 ID 매핑
+└── requirements.txt
+```
+
+**구현 내용**:
+- `python-telegram-bot>=20.0` (asyncio 기반)
+- `youlbot-telegram/bot.py` — 새 진입점 (`python bot.py`로 실행)
+- `/start` — 환영 메시지 + 매핑된 youlbot 사용자 이름 표시
+- `/reset` — `api_client.reset(user_id)` 호출로 대화 이력 초기화
+- 일반 메시지 → `api_client.chat()` SSE 스트리밍 → 0.6초 간격 실시간 편집
+- Telegram numeric ID → youlbot user_id `.env` 매핑 (`USER_아록_TELEGRAM_ID` 등)
+- 미등록 사용자에게 Telegram ID 안내 메시지 표시
+
+**실행 방법**:
+```bash
+cd youlbot-telegram
+python bot.py
+```
+
+**난이도**: 중간 | **임팩트**: 높음 (모바일·이동 중 접근)
+
+---
+
+## ✅ Phase 22 — REST API (FastAPI) ★★☆
+
+**배경**: 다른 Python 스크립트나 원격 서버에서 율봇을 호출하려면 HTTP API가 필요하다.  
+Telegram Bot을 별도 프로젝트로 분리해 이 API를 호출하는 구조로 사용 가능.
+
+**구현 내용**:
+- `api.py` — FastAPI 앱, `uvicorn api:app --host 0.0.0.0 --port 8000`으로 실행
+- SSE(`text/event-stream`) 스트리밍: 각 라인 `data: <JSON 토큰>\n\n`, 종료 `data: [DONE]\n\n`
+- Bearer Token 인증 (`.env` `API_TOKEN` 설정; 빈 값이면 개발 모드 무인증)
+- `user_id` 파라미터로 멀티유저 지원 (기존 DB·메모리 구조 그대로 재사용)
+
+| 엔드포인트 | 설명 |
+|-----------|------|
+| `GET  /health` | 헬스체크 |
+| `POST /chat` | SSE 스트리밍 대화 (`message`, `user_id`, `show_thinking`) |
+| `POST /reset` | 대화 이력 초기화 (`user_id`) |
+| `POST /ingest` | PDF/TXT 파일 업로드 → 벡터DB 수집 |
+| `GET  /documents` | 등록 문서 목록 |
+| `DELETE /documents/{source}` | 문서 삭제 |
+
+**클라이언트 예시 (별도 Telegram 봇 프로젝트)**:
+```python
+import httpx, json
+
+API_URL = "http://192.168.10.x:8000"
+HEADERS = {"Authorization": "Bearer YOUR_TOKEN"}
+
+async def ask_youlbot(message: str, user_id: str) -> str:
+    full = ""
+    async with httpx.AsyncClient(timeout=120) as client:
+        async with client.stream("POST", f"{API_URL}/chat",
+                                 json={"message": message, "user_id": user_id},
+                                 headers=HEADERS) as r:
+            async for line in r.aiter_lines():
+                if line.startswith("data: ") and line != "data: [DONE]":
+                    full += json.loads(line[6:])
+    return full
+```
+
+**난이도**: 중간 | **임팩트**: 높음 (확장성·외부 연동)
+
+---
+
+## ✅ Phase 23 — WebUI 분리 (youlbot-webui 별도 프로젝트) ★★☆
+
+**배경**: 현재 `app.py`(Gradio)는 `container.py`를 직접 import해 서비스를 사용한다.
+REST API(Phase 22)를 완성했으므로, WebUI를 독립 프로젝트로 분리해 API만 호출하도록 변경한다.
+분리 후 youlbot은 순수 백엔드(API 서버)로만 동작하며, Telegram Bot과 WebUI가 모두 같은 API를 공유한다.
+
+**구현 내용**:
+
+**① youlbot/api.py 보완**
+- `POST /feedback` 엔드포인트 추가 (FeedbackRepository 노출 + LangSmith 연동)
+- `/chat` SSE 마지막 이벤트에 `run_id` 포함 → 피드백 연결 가능
+  ```
+  data: {"__done": true, "run_id": "uuid"}
+  ```
+
+**② 신규 프로젝트 youlbot-webui/**
+```
+youlbot-webui/
+├── app.py           ← Gradio UI (REST API 호출 방식으로 재작성)
+├── api_client.py    ← httpx 기반 API 클라이언트 (chat/reset/ingest/documents/feedback)
+├── .env             ← YOULBOT_API_URL, YOULBOT_API_TOKEN
+├── .env.example
+└── requirements.txt ← gradio, httpx, python-dotenv, openai-whisper
+```
+
+| 기존 app.py (container 직접 사용) | 변경 후 (API 클라이언트) |
+|---|---|
+| `container.ingestion_service()` | `api_client.ingest(path)` |
+| `agent.stream_response()` | `api_client.chat(msg, user_id)` |
+| `retriever.list_documents()` | `api_client.list_documents()` |
+| `feedback_repo.save_feedback()` | `api_client.save_feedback(...)` |
+| STT (Whisper) | 변경 없음 — WebUI 로컬 실행 유지 |
+| TTS (macOS say) | 변경 없음 — WebUI 로컬 실행 유지 |
+
+**실행 방법**:
+```bash
+# 백엔드
+cd youlbot && uvicorn api:app --host 0.0.0.0 --port 8000
+
+# WebUI (별도 터미널, 별도 프로젝트)
+cd youlbot-webui && python app.py
+```
+
+기존 `youlbot/app.py`는 레거시 직접 실행 옵션으로 보존.
+
+**난이도**: 중간 | **임팩트**: 높음 (백엔드/프론트엔드 완전 분리, 다중 클라이언트 지원)
+
+---
+
+## ✅ Phase 24 — 사고 과정 UI 분리 & 실시간 피드백 ★★☆
+
+**배경**: 사고 과정(thinking)·진행 로그가 답변과 섞여 출력되고, 10초 동안 아무 피드백 없이 대기하는 UX 문제.
+
+**구현 내용**:
+
+**① 스트리밍 토큰 타입 분리 (`youlbot/services/agent/agent_service.py`)**
+- 답변: `yield str` (기존 그대로)
+- 진행 메시지(`[LangGraph → ...]`, `문서 검색 중...` 등): `yield {"__meta": str}`
+- 사고 과정 내용: `yield {"__thinking": str}`
+- LLM 추론 시작 즉시: `writer({"__start": True})` → `yield {"__status": label}` 으로 변환
+
+**② 사고 과정 전용 박스 (`youlbot-webui/app.py`)**
+
+| 단계 | 표시 방식 | 비고 |
+|------|-----------|------|
+| 전송 즉시 | `🤔 질문을 분석하고 있습니다...` | 단순 div, LLM 추론 전 즉각 표시 |
+| 스트리밍 중 | `🤔 분석 중...` + 현재 줄 | plain `<div>`, 새 줄 도착 시 이전 줄 교체 |
+| 진행 로그 | `🤔 분석 중...` + 로그 메시지 | `__meta` 토큰 전체를 표시 |
+| 완료 | `💭 분석 완료 ▶` | `<details>` 로 전환, 클릭 시 전체 내용 펼침 |
+
+- 스트리밍 중 `<details>` 미사용 → 내용 업데이트 시 닫힘 현상 없음
+- TTS는 순수 답변 토큰만 읽음 (`__meta`·`__thinking` 제외)
+- 챗봇에는 답변만 표시 (진행 메시지 숨김)
+- `show_thinking` 체크박스 기본값 ON으로 변경
+
+**③ 멀티클라이언트 대응**
+- `youlbot-telegram/bot.py`: `__meta`·`__thinking` 토큰 skip → 순수 답변만 스트리밍
+- `asyncio.get_event_loop().run_until_complete()` → `asyncio.run()` 전체 교체 (AnyIO 워커 스레드 호환)
+
+**난이도**: 중간 | **임팩트**: 높음 (UX 대폭 개선)
+
+---
+
+## ✅ Phase 25 — RAG 출처 전용 접기/펼치기 박스
+
+**배경**: RAG 검색 출처가 사고 과정(thinking)과 같은 `__meta` 토큰으로 섞여 "💭 분석 완료" 박스 안에 표시되던 문제.
+
+**구현 내용**:
+- `agent_service.py`: 출처를 `{"__meta": "..."}` 개별 토큰 대신 `{"__sources": [{filename, page}, ...]}` 단일 토큰으로 yield
+- `youlbot-webui/app.py`:
+  - `_sources_html()` 헬퍼 추가 — `<details>` 기반 접기/펼치기
+  - chatbot 바로 아래 `source_box = gr.HTML()` 컴포넌트 추가
+  - `respond()`에서 `__sources` 토큰 처리 → 답변 완료 후 "📄 출처 (N개)" 박스 표시
+- `youlbot-telegram/bot.py`: `__sources` 토큰 skip 처리 추가
+
+**난이도**: 하 | **임팩트**: 중간 (UX 개선 — 출처와 사고 과정 분리)
+
+---
+
+## ✅ Phase 20 — RAG 품질 자동 평가 (RAGAS) ★☆☆
 
 **배경**: 청킹 전략·검색 파라미터·Reranker 변경 시 답변 품질이 실제로 나아졌는지 수치로 확인할 방법이 없다.
 
-**구현 방식**:
-- `ragas` 라이브러리로 Faithfulness·Answer Relevancy·Context Recall 자동 측정
-- 테스트 질문-정답 셋을 `eval/` 디렉터리에 관리
-- 설정 변경 후 `python eval/run_ragas.py`로 비교 가능
+**구현 내용**:
+
+```
+eval/
+├── dataset.jsonl      ← 평가용 Q&A 쌍 (질문·정답 — 필요 시 수정)
+├── run_ragas.py       ← 평가 실행 스크립트
+├── requirements.txt   ← ragas==0.2.9, datasets, langchain-google-vertexai
+└── results/           ← report_YYYYMMDD_HHMMSS.{csv,json} 저장
+```
+
+**평가 지표**:
+
+| 지표 | 설명 |
+|------|------|
+| `faithfulness` | 답변이 검색 컨텍스트에 충실한가 (환각 탐지) |
+| `answer_relevancy` | 답변이 질문에 얼마나 관련 있는가 |
+| `context_recall` | 컨텍스트가 정답에 필요한 정보를 포함하는가 |
+| `context_precision` | 검색된 컨텍스트 중 실제 유용한 비율 |
+
+**평가 LLM 우선순위**: OpenAI GPT-4o-mini > Anthropic Claude Haiku > 로컬 Qwen3
+
+**실행 방법**:
+```bash
+# API 서버 실행 후
+python eval/run_ragas.py
+python eval/run_ragas.py --dataset eval/dataset.jsonl --api http://localhost:8000
+```
+
+**호환성 처리**: ragas 0.2가 langchain-community 0.4+에서 `ChatVertexAI` 임포트 실패하는 문제를 런타임 shim으로 우회.
 
 **난이도**: 중간 | **임팩트**: 중간 (장기 품질 관리 기반)
 
@@ -256,13 +480,27 @@ docker-compose.yml
 
 ---
 
-## Phase 17 — 멀티모달 이미지 이해 ★☆☆
+## ✅ Phase 17 — 멀티모달 이미지 이해 ★☆☆
 
 **배경**: 이유식 사진 → 재료 분석, 금융 서류 사진 → 내용 해석 등.
 
-**제약**: Qwen3-14B는 이미지 미지원 → `mlx-community/Qwen2.5-VL-7B-Instruct-4bit` 교체 필요.
+**구현 방식**: Dual-model C방식 — analyze_image 도구
 
-**난이도**: 높음 | **임팩트**: 높음 (장기 과제)
+| 모델 | 역할 |
+|------|------|
+| Qwen3-8B-4bit | 대화·추론 (항상 로드) |
+| Qwen2.5-VL-7B-Instruct-4bit | 이미지 분석 (lazy load) |
+
+- `services/model/mlx_vision_model.py` — MlxVisionModel (mlx-vlm 래퍼, lazy load)
+- `services/agent/tools.py` — `make_vision_tool(vision_model, image_path)` 추가
+- `agent_service.py` — `stream_response(image_path=None)`, config 경유 vision tool 동적 주입
+- `api.py` — `image_base64` 필드 추가, temp 파일 저장 후 응답 완료 시 삭제
+- `youlbot-webui` — `image_input` 컴포넌트 추가, ChatService.chat(image_path=) 연결
+- `.env` — `VISION_ENABLED=true`, `VISION_MODEL_ID` 설정
+
+**실행 방법**: API 서버 재시작 후 WebUI 이미지 첨부 버튼으로 사진 전송
+
+**난이도**: 높음 | **임팩트**: 높음
 
 ---
 
@@ -271,8 +509,8 @@ docker-compose.yml
 ```
 단기 (1~2주)                  중기 (1개월)              장기
 ────────────────────────     ──────────────────────    ──────────────────
-Phase 18 Hybrid Search    →  Phase 15 (모델선택)    →  Phase 16 (Docker)
-Phase 19 Query Rewriting  →  Phase 20 (RAGAS 평가)  →  Phase 17 (멀티모달)
+Phase 20 RAGAS 평가       →  Phase 15 (모델선택)    →  Phase 16 (Docker)
+                                                    →  Phase 17 (멀티모달)
 ```
 
 ### 우선순위 매트릭스
@@ -284,6 +522,7 @@ Phase 19 Query Rewriting  →  Phase 20 (RAGAS 평가)  →  Phase 17 (멀티모
 | 버그 3 단일 사용자 | ✅ 완료 | — | — | — |
 | 버그 4 나이 계산 오류 | ✅ 완료 | — | — | — |
 | 버그 5 thinking 체크박스 무효 | ✅ 완료 | — | — | — |
+| 버그 6 TTS 메타 토큰 혼재 | ✅ 완료 | — | — | — |
 | Phase 4 Web UI | ✅ 완료 | — | — | — |
 | Phase 5 장기 사용자 메모리 | ✅ 완료 | — | — | — |
 | Phase 6 웹 검색 | ✅ 완료 | — | — | — |
@@ -295,9 +534,178 @@ Phase 19 Query Rewriting  →  Phase 20 (RAGAS 평가)  →  Phase 17 (멀티모
 | Phase 13 Semantic Chunker | ✅ 완료 | — | — | — |
 | Phase 14 음성 인터페이스 | ✅ 완료 | — | — | — |
 | Phase 13-B Reranker | ✅ 완료 | — | — | — |
-| Phase 18 Hybrid Search | 🔲 신규 | 중간 | 높음 | ⭐ 1순위 |
-| Phase 19 Query Rewriting | 🔲 신규 | 하 | 중간 | 3순위 |
+| Phase 18 Hybrid Search | ✅ 완료 | — | — | — |
+| Phase 19 Query Rewriting | ✅ 완료 | — | — | — |
+| Phase 21 Telegram Bot | ✅ 완료 | — | — | — |
+| Phase 22 REST API | ✅ 완료 | — | — | — |
+| Phase 23 WebUI 분리 | ✅ 완료 | — | — | — |
+| Phase 24 사고 과정 UI 분리 | ✅ 완료 | — | — | — |
+| Phase 25 RAG 출처 전용 박스 | ✅ 완료 | — | — | — |
+| Phase 20 RAGAS 평가 | ✅ 완료 | — | — | — |
 | Phase 15 모델 선택 | 🔲 미완 | 중간 | 중간 | 4순위 |
-| Phase 20 RAGAS 평가 | 🔲 신규 | 중간 | 중간 | 5순위 |
-| Phase 16 Docker | 🔲 미완 | 높음 | 중간 | 6순위 |
-| Phase 17 멀티모달 | 🔲 미완 | 높음 | 높음 | 7순위 |
+| Phase 16 Docker | 🔲 미완 | 높음 | 중간 | 5순위 |
+| Phase 17 멀티모달 | ✅ 완료 | — | — | — |
+
+---
+
+## 💡 IDEA — 신규 개선 아이디어
+
+### 단기 — 빠르게 임팩트 큰 것
+
+#### ✅ IDEA-1. 대화 기반 자동 RAG 업데이트
+
+**배경**: 현재 문서 업로드만 RAG에 들어간다. 중요 대화 내용 자체를 자동으로 벡터DB에 추가하면 사용할수록 지식이 쌓이는 시스템이 된다.
+
+**구현 내용**:
+- `IngestionService.store_text(text, metadata)` — 단일 텍스트 직접 저장 (semantic chunking 없이)
+- `AgentService._maybe_index_conversation()` — 응답 완료 후 LLM이 유용한 정보 판단 → 요약 → Qdrant 저장 (asyncio background task)
+- `source="conversation"`, `user_id`, `timestamp` 메타데이터로 문서 RAG와 구분
+- `.env` `CONV_RAG_ENABLED=true`로 활성화 (기본 비활성)
+
+**난이도**: 하 | **임팩트**: 높음 (지식 자동 축적)
+
+---
+
+#### ✅ IDEA-2. 스마트 알림 & 일정 연동
+
+**배경**: 예방접종, 약 먹을 시간, 병원 예약 등 날짜 기반 알림이 없다.
+
+**구현 내용**:
+- `td_reminders` 테이블 (user_id, remind_date, message, sent_d0/d1/d7)
+- `set_reminder(remind_date, message)` + `list_reminders()` 도구 — LangGraph 에이전트 자동 호출
+- `SchedulerService` — asyncio task 기반 60초 간격 체크 → D-7/D-1/D-0 Telegram push
+- FastAPI `lifespan`으로 앱 시작/종료 시 스케줄러 자동 관리
+- `GET /reminders/{user_id}` API 엔드포인트 추가
+- `.env` 설정: `TELEGRAM_BOT_TOKEN`, `TELEGRAM_USER_MAP={"아록":"123456"}`
+
+**난이도**: 중간 | **임팩트**: 매우 높음 (육아 핵심 시나리오)
+
+---
+
+#### IDEA-3. 대화 요약 일간 리포트
+
+**배경**: 하루에 어떤 질문을 했고 어떤 결정을 내렸는지 돌아볼 방법이 없다.
+
+**구현 방향**:
+- 매일 자정 cron → 당일 대화 요약 생성 (`CompactService` 재활용)
+- Telegram으로 사용자별 요약 발송
+- `/chat` API + APScheduler로 구현 가능 (새 인프라 불필요)
+
+**난이도**: 하 | **임팩트**: 중간
+
+---
+
+#### IDEA-4. 텔레그램 그룹 채팅 지원
+
+**배경**: 현재 1:1 채팅만 지원한다. 가족 그룹에서 `@율봇` 멘션으로 함께 사용하고 싶다.
+
+**구현 방향**:
+- 그룹 메시지에서 `@율봇` 멘션 감지 → 발신자 Telegram ID로 `user_id` 매핑
+- 그룹 공용 컨텍스트(`user_id="family"`) 옵션
+- `python-telegram-bot` 기존 코드에 그룹 핸들러 추가
+
+**난이도**: 하 | **임팩트**: 높음 (가족 공동 사용)
+
+---
+
+### 중기 — RAG/에이전트 품질 향상
+
+#### ✅ IDEA-5. Agentic RAG — 자기 교정 검색 (CRAG)
+
+**배경**: 현재 `query_rewrite → search_documents` 1회로 끝난다. 검색 결과가 부족하면 재시도나 웹 검색 fallback이 없다.
+
+**구현 내용**:
+- `AgentState(TypedDict)` — `messages` + `crag_fallback_used` 커스텀 상태
+- `crag_check` LangGraph 노드 — `search_documents` 결과가 비었으면 동일 쿼리로 `web_search` AIMessage 자동 주입
+- `route_after_crag` — fallback AIMessage 있으면 tools 재실행, 없으면 agent로 복귀
+- 그래프: `tools → crag_check → route_after_crag → {tools, agent}`
+- 무한 루프 방지: `crag_fallback_used` 플래그로 1회만 fallback
+- `.env` `CRAG_ENABLED=true`로 활성화 (기본 비활성)
+
+**난이도**: 중간 | **임팩트**: 높음 (검색 실패 케이스 대폭 감소)
+
+---
+
+#### IDEA-6. 영수증/가계부 OCR
+
+**배경**: `analyze_image` 도구가 이미 있다. 영수증 사진에서 지출을 자동 기록하면 가계 관리가 가능해진다.
+
+**구현 방향**:
+- `analyze_image` → 금액·항목·날짜 추출 → MySQL `td_expenses` 저장
+- `get_monthly_expenses(month)` 도구 추가 → "이번 달 식비 얼마야?" 대응
+- 카테고리 자동 분류 (식비/의료비/교육비 등)
+
+**난이도**: 중간 | **임팩트**: 높음 (가계 관리 시나리오)
+
+---
+
+#### ✅ IDEA-7. RAG 파라미터 자동 튜닝 (Auto-Eval Loop)
+
+**배경**: RAGAS 평가 인프라는 있는데, 파라미터 변경 효과를 수동으로 비교해야 한다.
+
+**구현 내용**:
+- `eval/auto_tune.py` — API 서버 없이 `RetrieverService` 직접 사용, 파라미터 조합별 `context_precision` + `context_recall` 비교
+- 기본 조합 4개: `baseline(3/10)`, `top_k_5(5/15)`, `top_k_2(2/6)`, `fetch_k_20(3/20)`
+- 평균 점수 기준 최적 조합 추천 + `.env` 설정값 안내
+- `eval/results/tune_YYYYMMDD.json` 저장
+- 실행: `python eval/auto_tune.py [--dataset eval/dataset.jsonl]`
+
+**난이도**: 중간 | **임팩트**: 중간 (장기 품질 자동 관리)
+
+---
+
+### 장기 — 구조적 확장
+
+#### ✅ IDEA-8. GraphRAG / 지식 그래프
+
+**배경**: `td_user_profile`이 flat key-value라 엔티티 간 관계 추론이 불가능하다.
+
+**구현 내용**:
+- `td_knowledge_graph` 테이블 — (user_id, subject, relation, object) 트리플 영구 저장
+- `GraphService` — NetworkX `MultiDiGraph` 인메모리 캐시 + MySQL 영속화
+- `add_relation(subject, relation, obj)` 도구 — 관계 저장
+  예: `도율 -[알레르기]→ 복숭아`, `아록 -[자녀]→ 도율`
+- `query_entity(entity)` 도구 — 출발/도착 방향 모든 관계 조회
+- `call_model`에 저장된 그래프 요약을 시스템 프롬프트에 자동 주입
+- `.env` `GRAPH_ENABLED=true`로 활성화 (기본 비활성)
+
+**사용 예시**:
+```
+사용자: "도율이 복숭아 알레르기가 있어"
+→ add_relation("도율", "알레르기", "복숭아")
+
+사용자: "도율이 먹으면 안 되는 음식은?"
+→ query_entity("도율") → "도율 -[알레르기]→ 복숭아"
+```
+
+**난이도**: 높음 | **임팩트**: 높음 (메모리 추론 능력 대폭 향상)
+
+---
+
+#### IDEA-9. PWA / 모바일 Web UI
+
+**배경**: Gradio는 모바일 UX가 좋지 않다. 네이티브 앱처럼 설치하고 카메라 접근도 원활해야 한다.
+
+**구현 방향**:
+- `youlbot-webui`를 Next.js + shadcn/ui PWA로 재작성
+- 홈 화면 설치, 오프라인 캐시, 네이티브 카메라 접근
+- 기존 REST API 그대로 재사용 (백엔드 변경 없음)
+- STT는 Web Speech API로 대체 (브라우저 내장)
+
+**난이도**: 높음 | **임팩트**: 높음 (모바일 UX 대폭 개선)
+
+---
+
+### IDEA 우선순위 매트릭스
+
+| IDEA | 설명 | 난이도 | 임팩트 | 추천 순위 |
+|------|------|--------|--------|-----------|
+| IDEA-2 스마트 알림 | ✅ asyncio 스케줄러 + Telegram push | 중간 | 매우 높음 | — |
+| IDEA-4 텔레그램 그룹 채팅 | 기존 Bot 코드 확장 | 하 | 높음 | 1순위 |
+| IDEA-3 일간 리포트 | CompactService 재활용 + SchedulerService | 하 | 중간 | 2순위 |
+| IDEA-1 대화 기반 RAG | ✅ asyncio background + Qdrant 저장 | 하 | 높음 | — |
+| IDEA-5 CRAG | ✅ crag_check LangGraph 노드 | 중간 | 높음 | — |
+| IDEA-7 Auto-Eval | ✅ eval/auto_tune.py | 중간 | 중간 | — |
+| IDEA-6 영수증 OCR | analyze_image 재활용 | 중간 | 높음 | 1순위 |
+| IDEA-8 GraphRAG | ✅ NetworkX + MySQL + 2개 도구 | 높음 | 높음 | — |
+| IDEA-9 PWA WebUI | 프론트엔드 재작성 | 높음 | 높음 | 8순위 |

eval/auto_tune.py

@@ -0,0 +1,191 @@
+"""RAG 파라미터 자동 튜닝 스크립트 (IDEA-7)
+
+API 서버 없이 RetrieverService를 직접 사용해 파라미터 조합별 context 품질을 비교한다.
+평가 지표: context_precision, context_recall (RAGAS)
+
+실행:
+    python eval/auto_tune.py [--dataset eval/dataset.jsonl]
+
+출력:
+    eval/results/tune_YYYYMMDD_HHMMSS.json — 조합별 점수 및 추천 파라미터
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from datetime import datetime
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+os.chdir(ROOT)
+
+from dotenv import load_dotenv
+load_dotenv(ROOT / ".env")
+
+# ── Compatibility shim (run_ragas.py 동일) ─────────────────────────────────────
+try:
+    import langchain_community.chat_models.vertexai  # noqa: F401
+except ModuleNotFoundError:
+    try:
+        from langchain_google_vertexai import ChatVertexAI as _CV
+        _stub = type(sys)("langchain_community.chat_models.vertexai")
+        _stub.ChatVertexAI = _CV
+        sys.modules["langchain_community.chat_models.vertexai"] = _stub
+    except ImportError:
+        _stub = type(sys)("langchain_community.chat_models.vertexai")
+        _stub.ChatVertexAI = object
+        sys.modules["langchain_community.chat_models.vertexai"] = _stub
+
+from ragas import evaluate
+from ragas.metrics import context_precision, context_recall
+from ragas.embeddings import LangchainEmbeddingsWrapper
+from ragas.llms import LangchainLLMWrapper
+from datasets import Dataset
+from ragas.run_config import RunConfig
+
+from container import Container
+from services.rag.retriever_service import RetrieverService
+
+_container = Container()
+_container.db_service().connect()
+_container.db_service().init_schema()
+_cfg = _container.config()
+
+# ── 튜닝 대상 파라미터 조합 ────────────────────────────────────────────────────
+
+VARIANTS = [
+    {"name": "baseline",    "top_k": 3, "rerank_fetch_k": 10},
+    {"name": "top_k_5",     "top_k": 5, "rerank_fetch_k": 15},
+    {"name": "top_k_2",     "top_k": 2, "rerank_fetch_k":  6},
+    {"name": "fetch_k_20",  "top_k": 3, "rerank_fetch_k": 20},
+]
+
+
+def _build_retriever(top_k: int, rerank_fetch_k: int) -> RetrieverService:
+    return RetrieverService(
+        embeddings=_container.embeddings(),
+        qdrant_url=_cfg.qdrant_url,
+        collection_name=_cfg.qdrant_collection,
+        top_k=top_k,
+        reranker=_container.reranker() if _cfg.reranker_enabled else None,
+        rerank_fetch_k=rerank_fetch_k,
+        sparse_embeddings=_container.sparse_embeddings() if _cfg.hybrid_search_enabled else None,
+    )
+
+
+def _build_evaluator():
+    if os.getenv("OPENAI_API_KEY"):
+        from langchain_openai import ChatOpenAI
+        print("[AutoTune] 평가 LLM: OpenAI GPT-4o-mini")
+        return LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini", temperature=0))
+    if os.getenv("ANTHROPIC_API_KEY"):
+        from langchain_anthropic import ChatAnthropic
+        print("[AutoTune] 평가 LLM: Claude Haiku")
+        return LangchainLLMWrapper(ChatAnthropic(model="claude-haiku-4-5-20251001", temperature=0))
+    print("[AutoTune] 평가 LLM: 로컬 Qwen3")
+    return LangchainLLMWrapper(_container.chat_model())
+
+
+def run(dataset_path: str) -> None:
+    samples = []
+    with open(dataset_path, encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if line:
+                samples.append(json.loads(line))
+
+    if not samples:
+        print(f"[오류] 데이터셋이 비어 있습니다: {dataset_path}")
+        sys.exit(1)
+
+    print(f"[AutoTune] 파라미터 튜닝 시작 — {len(samples)}개 질문, {len(VARIANTS)}개 조합\n")
+
+    llm = _build_evaluator()
+    emb = LangchainEmbeddingsWrapper(_container.embeddings())
+    run_cfg = RunConfig(timeout=300, max_retries=1, max_workers=1)
+
+    results = []
+
+    for variant in VARIANTS:
+        name = variant["name"]
+        print(f"── {name} (top_k={variant['top_k']}, fetch_k={variant['rerank_fetch_k']}) ──")
+        retriever = _build_retriever(variant["top_k"], variant["rerank_fetch_k"])
+
+        questions, ground_truths, contexts = [], [], []
+        for s in samples:
+            q = s["question"]
+            docs = retriever.search(q)
+            contexts.append([d.page_content for d in docs])
+            questions.append(q)
+            ground_truths.append(s["ground_truth"])
+            print(f"  [{q[:40]}] → {len(docs)}개 청크")
+
+        ds = Dataset.from_dict({
+            "question": questions,
+            "contexts": contexts,
+            "ground_truth": ground_truths,
+        })
+
+        result = evaluate(
+            ds,
+            metrics=[context_precision, context_recall],
+            llm=llm,
+            embeddings=emb,
+            run_config=run_cfg,
+            raise_exceptions=False,
+        )
+        df = result.to_pandas()
+
+        def _score(col: str) -> float | None:
+            if col not in df.columns:
+                return None
+            val = df[col].dropna().mean()
+            return float(val) if val == val else None
+
+        scores = {
+            "context_precision": _score("context_precision"),
+            "context_recall":    _score("context_recall"),
+        }
+        avg = sum(v for v in scores.values() if v is not None) / max(
+            sum(1 for v in scores.values() if v is not None), 1
+        )
+        results.append({**variant, "scores": scores, "avg": avg})
+        print(f"  precision={scores['context_precision']}, recall={scores['context_recall']}, avg={avg:.3f}\n")
+
+    # ── 결과 출력 ──────────────────────────────────────────────────────────────
+    best = max(results, key=lambda r: r["avg"])
+
+    print("=" * 60)
+    print("AutoTune 결과")
+    print("=" * 60)
+    header = f"{'조합':<14} {'precision':>10} {'recall':>10} {'avg':>8}"
+    print(header)
+    print("-" * 60)
+    for r in sorted(results, key=lambda x: x["avg"], reverse=True):
+        marker = " ★" if r["name"] == best["name"] else ""
+        prec = f"{r['scores']['context_precision']:.3f}" if r['scores']['context_precision'] else "N/A"
+        rec  = f"{r['scores']['context_recall']:.3f}"    if r['scores']['context_recall']    else "N/A"
+        print(f"  {r['name']:<12} {prec:>10} {rec:>10} {r['avg']:>8.3f}{marker}")
+    print("=" * 60)
+    print(f"\n추천: top_k={best['top_k']}, rerank_fetch_k={best['rerank_fetch_k']} ({best['name']})")
+    print(f"  .env에 RAG_TOP_K={best['top_k']}, RERANKER_FETCH_K={best['rerank_fetch_k']} 설정\n")
+
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    results_dir = ROOT / "eval" / "results"
+    results_dir.mkdir(exist_ok=True)
+    out = results_dir / f"tune_{ts}.json"
+    out.write_text(
+        json.dumps({"timestamp": ts, "best": best, "all": results}, ensure_ascii=False, indent=2),
+        encoding="utf-8",
+    )
+    print(f"JSON 저장: {out}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="RAG 파라미터 자동 튜닝")
+    parser.add_argument("--dataset", default=str(ROOT / "eval" / "dataset.jsonl"))
+    args = parser.parse_args()
+    run(args.dataset)

eval/dataset.jsonl

@@ -0,0 +1,5 @@
+{"question": "부모의 반응성이 아동의 인지 발달에 어떤 영향을 미치나요?", "ground_truth": "부모의 민감한 반응성은 아동의 인지 발달에 긍정적인 영향을 미친다. 특히 영유아기에 부모가 아동의 신호에 적절히 반응할 때 아동의 탐색 행동과 학습 능력이 향상된다."}
+{"question": "아동의 사회성 발달을 돕기 위해 부모가 할 수 있는 것은?", "ground_truth": "부모는 일관된 애착 관계를 형성하고 긍정적인 상호작용 모델을 보여줌으로써 아동의 사회성 발달을 지원할 수 있다."}
+{"question": "영유아기 발달 평가 방법에는 어떤 것들이 있나요?", "ground_truth": "영유아기 발달 평가는 표준화된 발달 검사, 관찰 기반 평가, 부모 보고 척도 등 다양한 방법을 통해 이루어진다."}
+{"question": "논리수학 지능 발달에 영향을 미치는 요인은?", "ground_truth": "논리수학 지능 발달에는 부모의 상호작용 방식, 탐색 기회 제공, 문제 해결 경험 등이 영향을 미친다."}
+{"question": "어머니의 반응성과 아동 언어 발달의 관계는?", "ground_truth": "어머니의 반응성은 아동의 언어 발달에 긍정적 영향을 미치며, 어머니가 아동의 발화에 민감하게 반응할수록 어휘 습득과 언어 이해 능력이 향상된다."}

eval/requirements.txt

@@ -0,0 +1,3 @@
+ragas==0.2.9
+datasets>=2.14.0
+langchain-google-vertexai>=2.0.0

eval/run_ragas.py

@@ -0,0 +1,257 @@
+"""youlbot RAGAS 평가 스크립트 (Phase 20)
+
+실행:
+    cd /path/to/youlbot
+    python eval/run_ragas.py [--dataset eval/dataset.jsonl] [--api http://localhost:8000]
+
+결과:
+    eval/results/report_YYYYMMDD_HHMMSS.csv
+
+사전 조건:
+    - youlbot API 서버 실행 중 (uvicorn api:app --port 8000)
+    - Qdrant + MySQL 접근 가능
+    - .env에 API_TOKEN, RAG_SHOW_SOURCES=true 설정
+
+평가 지표:
+    - faithfulness       : 답변이 검색 컨텍스트에 충실한가 (환각 탐지)
+    - answer_relevancy   : 답변이 질문에 얼마나 관련 있는가
+    - context_recall     : 컨텍스트가 정답에 필요한 정보를 포함하는가
+    - context_precision  : 검색된 컨텍스트 중 실제 유용한 비율
+
+참고:
+    평가에 로컬 LLM(Qwen3)을 사용하므로 결과 신뢰도는 모델 크기에 의존합니다.
+    더 정확한 평가를 원하면 OPENAI_API_KEY 또는 ANTHROPIC_API_KEY를 설정하세요.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import sys
+from datetime import datetime
+from pathlib import Path
+
+# ── Compatibility shim ───────────────────────────────────────────────────────
+# ragas 0.2.x imports langchain_community.chat_models.vertexai which was
+# removed in langchain-community 0.4+. Re-export from langchain-google-vertexai.
+try:
+    import langchain_community.chat_models.vertexai  # noqa: F401
+except ModuleNotFoundError:
+    try:
+        from langchain_google_vertexai import ChatVertexAI as _CV
+        _stub = type(sys)("langchain_community.chat_models.vertexai")
+        _stub.ChatVertexAI = _CV
+        sys.modules["langchain_community.chat_models.vertexai"] = _stub
+    except ImportError:
+        # vertexai not available — inject an empty stub (unused by our eval)
+        _stub = type(sys)("langchain_community.chat_models.vertexai")
+        _stub.ChatVertexAI = object
+        sys.modules["langchain_community.chat_models.vertexai"] = _stub
+
+from ragas import evaluate
+from ragas.metrics import answer_relevancy, context_precision, context_recall, faithfulness
+from ragas.embeddings import LangchainEmbeddingsWrapper
+from ragas.llms import LangchainLLMWrapper
+from datasets import Dataset
+
+# ── Project path ─────────────────────────────────────────────────────────────
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+os.chdir(ROOT)  # .env 읽기 위해 프로젝트 루트로 이동
+
+from container import Container  # noqa: E402 (after sys.path setup)
+
+_container = Container()
+_container.db_service().connect()
+_container.db_service().init_schema()
+
+
+# ── Answer collection via API ────────────────────────────────────────────────
+
+async def _collect_answer(api_url: str, token: str, message: str) -> str:
+    """youlbot /chat SSE 스트림에서 순수 답변 텍스트만 수집."""
+    import httpx
+
+    headers = {"Authorization": f"Bearer {token}"} if token else {}
+    parts: list[str] = []
+    async with httpx.AsyncClient(timeout=180) as client:
+        async with client.stream(
+            "POST",
+            f"{api_url}/chat",
+            json={"message": message, "user_id": "eval", "show_thinking": False},
+            headers=headers,
+        ) as resp:
+            resp.raise_for_status()
+            async for line in resp.aiter_lines():
+                if not line.startswith("data: "):
+                    continue
+                payload = json.loads(line[6:])
+                if isinstance(payload, str):
+                    parts.append(payload)
+                elif isinstance(payload, dict) and payload.get("__done"):
+                    await resp.aclose()
+                    break
+    return "".join(parts)
+
+
+def collect_answer(api_url: str, token: str, message: str) -> str:
+    return asyncio.run(_collect_answer(api_url, token, message))
+
+
+# ── Evaluator LLM 선택 ────────────────────────────────────────────────────────
+
+def _build_evaluator_llm():
+    """평가용 LLM: OpenAI > Anthropic > 로컬 MLX 순으로 시도."""
+    if os.getenv("OPENAI_API_KEY"):
+        from langchain_openai import ChatOpenAI
+        print("[RAGAS] 평가 LLM: OpenAI GPT-4o-mini")
+        return LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini", temperature=0))
+
+    if os.getenv("ANTHROPIC_API_KEY"):
+        from langchain_anthropic import ChatAnthropic
+        print("[RAGAS] 평가 LLM: Anthropic Claude Haiku")
+        return LangchainLLMWrapper(
+            ChatAnthropic(model="claude-haiku-4-5-20251001", temperature=0)
+        )
+
+    print("[RAGAS] 평가 LLM: 로컬 Qwen3 (신뢰도 제한적)")
+    return LangchainLLMWrapper(_container.chat_model())
+
+
+def _build_evaluator_embeddings():
+    return LangchainEmbeddingsWrapper(_container.embeddings())
+
+
+# ── Main ──────────────────────────────────────────────────────────────────────
+
+def run(dataset_path: str, api_url: str, api_token: str) -> None:
+    # 1. 데이터셋 로드
+    samples = []
+    with open(dataset_path, encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if line:
+                samples.append(json.loads(line))
+
+    if not samples:
+        print(f"[오류] 데이터셋이 비어 있습니다: {dataset_path}")
+        sys.exit(1)
+
+    print(f"[RAGAS] 평가 시작 — {len(samples)}개 질문, API: {api_url}")
+
+    # 2. RetrieverService 초기화
+    retriever = _container.retriever_service()
+
+    # 3. 질문별 context + answer 수집
+    questions: list[str] = []
+    answers: list[str] = []
+    contexts: list[list[str]] = []
+    ground_truths: list[str] = []
+
+    for i, sample in enumerate(samples, 1):
+        q = sample["question"]
+        gt = sample["ground_truth"]
+        print(f"\n[{i}/{len(samples)}] {q[:50]}...")
+
+        docs = retriever.search(q)
+        ctxs = [doc.page_content for doc in docs]
+        print(f"  컨텍스트: {len(ctxs)}개 청크")
+
+        answer = collect_answer(api_url, api_token, q)
+        print(f"  답변: {len(answer)}자")
+
+        questions.append(q)
+        answers.append(answer)
+        contexts.append(ctxs)
+        ground_truths.append(gt)
+
+    # 4. RAGAS Dataset
+    ds = Dataset.from_dict(
+        {
+            "question": questions,
+            "answer": answers,
+            "contexts": contexts,
+            "ground_truth": ground_truths,
+        }
+    )
+
+    # 5. 평가 실행
+    llm = _build_evaluator_llm()
+    emb = _build_evaluator_embeddings()
+
+    # 로컬 LLM은 응답이 느리므로 타임아웃을 충분히 크게, 병렬 작업 수를 줄임
+    from ragas.run_config import RunConfig
+    run_cfg = RunConfig(timeout=600, max_retries=1, max_workers=1)
+
+    print("\n[RAGAS] 지표 계산 중... (로컬 LLM 사용 시 수 분 소요)")
+    result = evaluate(
+        ds,
+        metrics=[faithfulness, answer_relevancy, context_recall, context_precision],
+        llm=llm,
+        embeddings=emb,
+        run_config=run_cfg,
+        raise_exceptions=False,
+    )
+
+    # 6. 결과 출력 및 저장
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    results_dir = ROOT / "eval" / "results"
+    results_dir.mkdir(exist_ok=True)
+    out_csv = results_dir / f"report_{ts}.csv"
+    out_json = results_dir / f"report_{ts}.json"
+
+    df = result.to_pandas()
+    df.to_csv(out_csv, index=False, encoding="utf-8-sig")
+
+    # 점수 추출: to_pandas() 컬럼 평균으로 안전하게 계산 (타임아웃 시 NaN 처리)
+    def _score(col: str) -> float | None:
+        if col not in df.columns:
+            return None
+        val = df[col].dropna().mean()
+        return float(val) if not (val != val) else None  # NaN 체크
+
+    summary = {
+        "timestamp": ts,
+        "dataset": dataset_path,
+        "n_samples": len(samples),
+        "scores": {
+            "faithfulness": _score("faithfulness"),
+            "answer_relevancy": _score("answer_relevancy"),
+            "context_recall": _score("context_recall"),
+            "context_precision": _score("context_precision"),
+        },
+    }
+    out_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    print(f"\n{'='*55}")
+    print("RAGAS 평가 결과")
+    print("="*55)
+    for k, v in summary["scores"].items():
+        bar = "█" * int((v or 0) * 20) if v is not None else ""
+        score_str = f"{v:.3f}" if v is not None else "N/A"
+        print(f"  {k:<22} {score_str}  {bar}")
+    print("="*55)
+    print(f"CSV : {out_csv}")
+    print(f"JSON: {out_json}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="youlbot RAGAS 평가")
+    parser.add_argument(
+        "--dataset",
+        default=str(ROOT / "eval" / "dataset.jsonl"),
+        help="평가 데이터셋 경로 (기본: eval/dataset.jsonl)",
+    )
+    parser.add_argument(
+        "--api",
+        default=os.getenv("YOULBOT_API_URL", "http://localhost:8000"),
+        help="youlbot API URL (기본: http://localhost:8000)",
+    )
+    args = parser.parse_args()
+
+    from dotenv import load_dotenv
+    load_dotenv(ROOT / ".env")
+    api_token = os.getenv("API_TOKEN", "")
+
+    run(args.dataset, args.api, api_token)

requirements.txt

@@ -12,6 +12,12 @@ langchain-qdrant>=0.2.0
 sentence-transformers>=3.0.0
 qdrant-client>=1.9.0
 pdfplumber>=0.11.0
+# Phase 18 — Hybrid Search (BM25 sparse vectors)
+fastembed>=0.3.0
+# Phase 22 — REST API
+fastapi>=0.100.0
+uvicorn[standard]>=0.23.0
+python-multipart>=0.0.7
 # Phase 3 — Agent orchestration
 langgraph>=1.0.0
 # Phase 4 — Web UI
@@ -20,3 +26,5 @@ gradio>=4.0.0
 duckduckgo-search>=6.0.0
 # Phase 14 — 음성 인터페이스 (STT)
 openai-whisper>=20231117
+# IDEA-8 — 지식 그래프 (GraphRAG)
+networkx>=3.0

services/agent/agent_service.py

@@ -1,16 +1,23 @@
+import asyncio
 import os
 import time
 import uuid
-from typing import AsyncIterator
+from typing import Annotated, AsyncIterator, TypedDict
 
 from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, SystemMessage
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.memory import MemorySaver
 from langgraph.config import get_stream_writer
-from langgraph.graph import START, MessagesState, StateGraph
-from langgraph.prebuilt import ToolNode, tools_condition
+from langgraph.graph import END, START, MessagesState, StateGraph, add_messages
+from langgraph.prebuilt import ToolNode
 
-from services.agent.tools import get_current_date, make_memory_tools, make_retriever_tool, make_search_tool, web_search
+
+class AgentState(TypedDict):
+    messages: Annotated[list, add_messages]
+    crag_fallback_used: bool
+
+from services.agent.tools import get_current_date, make_memory_tools, make_reminder_tools, make_retriever_tool, make_search_tool, make_vision_tool, web_search
+from services.agent.graph_tools import make_graph_tools
 
 
 class AgentService:
@@ -28,8 +35,14 @@ class AgentService:
         rag_show_sources: bool = False,
         langgraph_verbose: bool = False,
         think_verbose: bool = False,
+        query_rewrite_enabled: bool = False,
         user_profile_repository=None,
         conversation_repository=None,
+        reminder_repository=None,
+        ingestion_service=None,
+        crag_enabled: bool = False,
+        conv_rag_enabled: bool = False,
+        graph_service=None,
         user_id: str = "default",
     ):
         self._system_prompt = system_prompt
@@ -37,6 +50,7 @@ class AgentService:
         self._rag_show_sources = rag_show_sources
         self._langgraph_verbose = langgraph_verbose
         self._think_verbose = think_verbose
+        self._query_rewrite_enabled = query_rewrite_enabled
         self._source_buffer: list[dict] = []
         self._thread_id = "default"
         self._profile_repo = user_profile_repository
@@ -45,6 +59,10 @@ class AgentService:
         self._pending_history: list = []
         self._user_id = user_id
         self._last_run_id: str | None = None
+        self._ingestion_service = ingestion_service
+        self._crag_enabled = crag_enabled
+        self._conv_rag_enabled = conv_rag_enabled
+        self._graph_service = graph_service
 
         if conversation_repository:
             try:
@@ -69,15 +87,44 @@ class AgentService:
             search_tool = make_search_tool(retriever_service, self._source_buffer)
         else:
             search_tool = make_retriever_tool(retriever_service)
-        tools = [search_tool, web_search, get_current_date]
+        self._base_tools = [search_tool, web_search, get_current_date]
         if user_profile_repository is not None:
             remember_tool, recall_tool = make_memory_tools(user_profile_repository, user_id)
-            tools += [remember_tool, recall_tool]
-        llm_with_tools = chat_model.bind_tools(tools)
+            self._base_tools += [remember_tool, recall_tool]
+        if reminder_repository is not None:
+            set_reminder_tool, list_reminders_tool = make_reminder_tools(reminder_repository, user_id)
+            self._base_tools += [set_reminder_tool, list_reminders_tool]
+        if graph_service is not None:
+            add_relation_tool, query_entity_tool = make_graph_tools(graph_service, user_id)
+            self._base_tools += [add_relation_tool, query_entity_tool]
+        self._vision_model = None  # set via set_vision_model()
+        self._llm_with_tools = chat_model.bind_tools(self._base_tools)
+        self._chat_model = chat_model
 
         async def call_model(state: MessagesState, config: RunnableConfig) -> dict:
             from datetime import date
-            system_content = f"오늘 날짜: {date.today().isoformat()}\n\n" + self._system_prompt
+            system_content = (
+                "【언어 규칙】모든 사고 과정(thinking)과 답변을 반드시 한국어로 작성하세요. "
+                "영어 사용 금지. Think in Korean only.\n\n"
+                f"오늘 날짜: {date.today().isoformat()}\n\n"
+                + self._system_prompt
+            )
+            if self._graph_service:
+                graph_summary = self._graph_service.get_summary(self._user_id)
+                if graph_summary:
+                    system_content += (
+                        "\n\n## 지식 그래프 (저장된 관계 정보)\n"
+                        + graph_summary
+                        + "\n\n**지식 그래프 사용 규칙**: 가족·사물 간 관계 정보(알레르기, "
+                        "가족 관계, 선호도, 질환 등)는 add_relation으로 저장하고, "
+                        "특정 인물 정보 조회 시 query_entity를 먼저 호출하세요."
+                    )
+                else:
+                    system_content += (
+                        "\n\n**지식 그래프 사용 규칙**: 가족·사물 간 관계 정보(알레르기, "
+                        "가족 관계, 선호도, 질환 등)를 언급하면 add_relation으로 저장하세요."
+                    )
+
             if self._profile_repo:
                 profile = self._profile_repo.get_all(self._user_id)
                 if profile:
@@ -113,9 +160,19 @@ class AgentService:
                 writer = get_stream_writer()
             except Exception:
                 writer = None
-            # 체크박스 값을 모델의 enable_thinking으로 전달 (런타임 오버라이드)
-            show_thinking = config.get("configurable", {}).get("show_thinking", False)
-            _llm = llm_with_tools.bind(enable_thinking=show_thinking) if show_thinking != chat_model.enable_thinking else llm_with_tools
+            # LLM 추론 시작 직전에 즉시 신호 emit — UI에 "분석 중" 표시
+            if writer:
+                writer({"__start": True})
+            # 이미지 첨부 시 vision tool 동적 추가 (요청별로 독립적으로 바인딩)
+            cfg = config.get("configurable", {})
+            show_thinking = cfg.get("show_thinking", False)
+            image_path = cfg.get("image_path")
+            if image_path and self._vision_model:
+                tools_for_req = self._base_tools + [make_vision_tool(self._vision_model, image_path)]
+                _llm_base = self._chat_model.bind_tools(tools_for_req)
+            else:
+                _llm_base = self._llm_with_tools
+            _llm = _llm_base.bind(enable_thinking=show_thinking) if show_thinking != chat_model.enable_thinking else _llm_base
             async for chunk in _llm.astream(msgs, config):
                 t = chunk.additional_kwargs.get("thinking", "")
                 if t:
@@ -133,11 +190,135 @@ class AgentService:
                 additional_kwargs=extra,
             )]}
 
-        builder = StateGraph(MessagesState)
+        async def query_rewrite_node(state: MessagesState, config: RunnableConfig) -> dict:
+            last_msg = state["messages"][-1]
+            if not (hasattr(last_msg, "tool_calls") and last_msg.tool_calls):
+                return {}
+
+            # 최근 사용자 메시지 2개를 컨텍스트로 활용 (대명사·지시어 해소)
+            recent_human = [m.content for m in state["messages"][:-1]
+                            if isinstance(m, HumanMessage)][-2:]
+            ctx = ("\n\n이전 대화 컨텍스트:\n" + "\n".join(f"- {m}" for m in recent_human)
+                   if recent_human else "")
+
+            try:
+                writer = get_stream_writer()
+            except Exception:
+                writer = None
+
+            _rewrite_llm = chat_model.bind(enable_thinking=False)
+            new_tool_calls = []
+            for tc in last_msg.tool_calls:
+                if tc["name"] == "search_documents":
+                    original = tc["args"].get("query", "")
+                    prompt = (
+                        f"다음 구어체 질문을 문서 검색에 최적화된 키워드 중심 문장으로 변환하세요.{ctx}\n\n"
+                        f"규칙:\n"
+                        f"- 핵심 개념과 전문용어를 포함하세요\n"
+                        f"- 대명사(이것, 그것, 그 논문 등)는 구체적인 명칭으로 교체하세요\n"
+                        f"- 변환된 질문만 한 문장으로 출력하세요. 부가 설명 없이 질문만 출력하세요\n\n"
+                        f"원본 질문: {original}\n최적화된 질문:"
+                    )
+                    try:
+                        result = await _rewrite_llm.ainvoke([HumanMessage(content=prompt)])
+                        rewritten = result.content.strip()
+                    except Exception as e:
+                        print(f"[QueryRewrite] 실패: {e}")
+                        rewritten = original
+                    if rewritten and rewritten != original:
+                        new_tool_calls.append({**tc, "args": {**tc["args"], "query": rewritten}})
+                        if writer:
+                            writer({"__query_rewrite": {"original": original, "rewritten": rewritten}})
+                    else:
+                        new_tool_calls.append(tc)
+                else:
+                    new_tool_calls.append(tc)
+
+            if not last_msg.id:
+                return {}
+            new_msg = AIMessage(
+                id=last_msg.id,
+                content=last_msg.content,
+                tool_calls=new_tool_calls,
+                additional_kwargs=last_msg.additional_kwargs,
+            )
+            return {"messages": [new_msg]}
+
+        async def crag_check_node(state: AgentState) -> dict:
+            """검색 결과 없을 때 web_search 자동 fallback 주입 (CRAG)."""
+            if state.get("crag_fallback_used", False):
+                return {}
+
+            messages = state["messages"]
+            # 마지막 search_documents 결과 탐색
+            last_search_msg = None
+            for msg in reversed(messages):
+                if hasattr(msg, "name") and msg.name == "search_documents":
+                    last_search_msg = msg
+                    break
+
+            if not last_search_msg or "관련 문서를 찾을 수 없습니다" not in last_search_msg.content:
+                return {}
+
+            # 해당 ToolMessage의 tool_call_id로 원본 AIMessage에서 검색 쿼리 추출
+            tool_call_id = getattr(last_search_msg, "tool_call_id", None)
+            query = ""
+            for msg in reversed(messages):
+                if isinstance(msg, AIMessage) and msg.tool_calls:
+                    for tc in msg.tool_calls:
+                        if tc.get("id") == tool_call_id:
+                            query = tc.get("args", {}).get("query", "")
+                            break
+                if query:
+                    break
+
+            if not query:
+                return {}
+
+            fallback_msg = AIMessage(
+                content="",
+                tool_calls=[{
+                    "id": str(uuid.uuid4()),
+                    "name": "web_search",
+                    "args": {"query": query},
+                    "type": "tool_call",
+                }],
+            )
+            try:
+                writer = get_stream_writer()
+                writer({"__meta": f'\n[CRAG] 문서 없음 → 웹 검색으로 전환... ("{query}")\n'})
+            except Exception:
+                pass
+            return {"messages": [fallback_msg], "crag_fallback_used": True}
+
+        def route_after_crag(state: AgentState) -> str:
+            last_msg = state["messages"][-1] if state["messages"] else None
+            if (isinstance(last_msg, AIMessage) and last_msg.tool_calls
+                    and state.get("crag_fallback_used", False)):
+                return "tools"
+            return "agent"
+
+        def route_after_agent(state: AgentState) -> str:
+            last_msg = state["messages"][-1]
+            if not (hasattr(last_msg, "tool_calls") and last_msg.tool_calls):
+                return END
+            if self._query_rewrite_enabled:
+                if any(tc["name"] == "search_documents" for tc in last_msg.tool_calls):
+                    return "query_rewrite"
+            return "tools"
+
+        builder = StateGraph(AgentState)
         builder.add_node("agent", call_model)
-        builder.add_node("tools", ToolNode(tools))
+        builder.add_node("query_rewrite", query_rewrite_node)
+        builder.add_node("tools", ToolNode(self._base_tools))
         builder.add_edge(START, "agent")
-        builder.add_conditional_edges("agent", tools_condition)
+        builder.add_conditional_edges("agent", route_after_agent)
+        builder.add_edge("query_rewrite", "tools")
+        if crag_enabled:
+            builder.add_node("crag_check", crag_check_node)
+            builder.add_edge("tools", "crag_check")
+            builder.add_conditional_edges("crag_check", route_after_crag)
+        else:
             builder.add_edge("tools", "agent")
 
         self._agent = builder.compile(checkpointer=MemorySaver())
@@ -146,15 +327,30 @@ class AgentService:
     def last_run_id(self) -> str | None:
         return self._last_run_id
 
-    def _make_config(self, show_thinking: bool = False) -> dict:
-        return {"configurable": {"thread_id": self._thread_id, "show_thinking": show_thinking}}
+    def set_vision_model(self, vision_model) -> None:
+        self._vision_model = vision_model
 
-    async def stream_response(self, user_input: str, show_thinking: bool | None = None) -> AsyncIterator[str]:
-        """사용자 입력을 받아 응답 토큰을 순서대로 yield한다."""
+    def _make_config(self, show_thinking: bool = False, image_path: str | None = None) -> dict:
+        cfg: dict = {"thread_id": self._thread_id, "show_thinking": show_thinking}
+        if image_path:
+            cfg["image_path"] = image_path
+        return {"configurable": cfg}
+
+    async def stream_response(
+        self,
+        user_input: str,
+        show_thinking: bool | None = None,
+        image_path: str | None = None,
+    ) -> AsyncIterator[str | dict]:
+        """사용자 입력을 받아 응답 토큰을 순서대로 yield한다.
+
+        실제 답변: plain str
+        진행/thinking/출처 메타데이터: {"__meta": str}  ← 소비자가 TTS 등에서 필터링 가능
+        """
         _think_verbose = show_thinking if show_thinking is not None else self._think_verbose
         self._source_buffer.clear()
         run_id = uuid.uuid4()
-        run_config = {**self._make_config(_think_verbose), "run_id": str(run_id)}
+        run_config = {**self._make_config(_think_verbose, image_path=image_path), "run_id": str(run_id)}
 
         # 재시작 후 첫 호출 시 MySQL 이력을 초기 상태에 주입
         if self._pending_history:
@@ -162,7 +358,7 @@ class AgentService:
             self._pending_history = []
         else:
             all_messages = [HumanMessage(content=user_input)]
-        messages = {"messages": all_messages}
+        messages = {"messages": all_messages, "crag_fallback_used": False}
         response_content = ""  # 실제 답변 내용만 누적 (MySQL 저장용)
         pending_tool_calls: dict = {}  # tool_call_id → {name, args}
         prev_node: str = ""
@@ -176,25 +372,31 @@ class AgentService:
         ):
             mode, data = stream_event
 
-            # ── custom 이벤트 — call_model writer가 emit한 thinking 토큰 ──
+            # ── custom 이벤트 ────────────────────────────────────────────
             if mode == "custom":
+                if isinstance(data, dict) and "__start" in data:
+                    # call_model 시작 즉시 emit — LLM 추론 전에 상태 표시
+                    label = "검색 결과를 분석하고 있습니다..." if prev_node == "tools" else "질문을 분석하고 있습니다..."
+                    yield {"__status": label}
+                    continue
+                if isinstance(data, dict) and "__query_rewrite" in data:
+                    info = data["__query_rewrite"]
+                    if lg or self._rag_verbose:
+                        yield {"__meta": f'\n쿼리 최적화: "{info["original"]}" → "{info["rewritten"]}"\n'}
+                    continue
                 if isinstance(data, dict) and "__thinking" in data:
                     # thinking 첫 토큰 도착 시 agent 레이블 + prev_node 갱신
                     if "agent" != prev_node:
-                        if thinking_open:
-                            yield "\n[/사고 과정]\n"
                         thinking_open = False
                         content_started = False
                         if lg:
                             elapsed = time.perf_counter() - start_time
                             label = "agent: 검색 결과 반영 중" if prev_node == "tools" else "agent: 질문 분석 중"
-                            yield f"\n[LangGraph → {label}] ({elapsed:.2f}s)\n"
+                            yield {"__meta": f"\n[LangGraph → {label}] ({elapsed:.2f}s)\n"}
                         prev_node = "agent"
                     if _think_verbose:
-                        if not thinking_open:
-                            yield "\n[사고 과정]\n"
                         thinking_open = True
-                        yield data["__thinking"]
+                        yield {"__thinking": data["__thinking"]}
                 continue
 
             # ── messages 이벤트 ──────────────────────────────────────
@@ -204,44 +406,39 @@ class AgentService:
             # ── 노드 전환 시 플래그 리셋 + 레이블 출력 ──────────────
             # (agent 레이블은 custom 이벤트 핸들러에서 이미 처리될 수 있으므로 중복 방지)
             if node != prev_node:
-                if thinking_open:
-                    yield "\n[/사고 과정]\n"
                 thinking_open = False
                 content_started = False
                 if lg:
+                    elapsed = time.perf_counter() - start_time
                     if node == "agent":
-                        elapsed = time.perf_counter() - start_time
                         label = "agent: 검색 결과 반영 중" if prev_node == "tools" else "agent: 질문 분석 중"
-                        yield f"\n[LangGraph → {label}] ({elapsed:.2f}s)\n"
+                        yield {"__meta": f"\n[LangGraph → {label}] ({elapsed:.2f}s)\n"}
+                    elif node == "query_rewrite":
+                        yield {"__meta": f"\n[LangGraph → query_rewrite: 쿼리 최적화 중] ({elapsed:.2f}s)\n"}
                     elif node == "tools":
-                        elapsed = time.perf_counter() - start_time
-                        yield f"\n[LangGraph → tools: 도구 실행 중] ({elapsed:.2f}s)\n"
+                        yield {"__meta": f"\n[LangGraph → tools: 도구 실행 중] ({elapsed:.2f}s)\n"}
                 prev_node = node
 
             # ── agent 노드 — AIMessageChunk만 처리 (중복 방지) ──────
             if node == "agent" and isinstance(chunk, AIMessageChunk):
                 if chunk.tool_calls:
-                    if thinking_open:
-                        yield "\n[/사고 과정]\n"
                     thinking_open = False
                     for tc in chunk.tool_calls:
                         pending_tool_calls[tc["id"]] = tc
                         if tc.get("name") == "search_documents":
                             query = tc.get("args", {}).get("query", "")
-                            yield f'\n문서 검색 중... ("{query}")\n' if query else "\n문서 검색 중...\n"
+                            yield {"__meta": f'\n문서 검색 중... ("{query}")\n'} if query else {"__meta": "\n문서 검색 중...\n"}
                         elif tc.get("name") == "web_search":
                             query = tc.get("args", {}).get("query", "")
-                            yield f'\n웹 검색 중... ("{query}")\n' if query else "\n웹 검색 중...\n"
+                            yield {"__meta": f'\n웹 검색 중... ("{query}")\n'} if query else {"__meta": "\n웹 검색 중...\n"}
                         elif lg:
                             args_str = ", ".join(f'{k}="{v}"' for k, v in tc["args"].items())
-                            yield f"  [tool_call: {tc['name']}({args_str})]\n"
+                            yield {"__meta": f"  [tool_call: {tc['name']}({args_str})]\n"}
 
                 elif chunk.content:
-                    if thinking_open:
-                        yield "\n[/사고 과정]\n"
                     thinking_open = False
                     if lg and not content_started:
-                        yield "\n[LangGraph → agent: 최종 답변 생성]\n\n"
+                        yield {"__meta": "\n[LangGraph → agent: 최종 답변 생성]\n\n"}
                         content_started = True
                     response_content += chunk.content
                     yield chunk.content
@@ -252,12 +449,10 @@ class AgentService:
                 if not content_started and not thinking_open:
                     thinking = chunk.additional_kwargs.get("thinking", "")
                     if thinking and _think_verbose:
-                        yield "\n[사고 과정]\n"
-                        yield thinking
-                        yield "\n[/사고 과정]\n"
+                        yield {"__thinking": thinking}
                     if chunk.content:
                         if lg:
-                            yield "\n[LangGraph → agent: 최종 답변 생성]\n\n"
+                            yield {"__meta": "\n[LangGraph → agent: 최종 답변 생성]\n\n"}
                         response_content += chunk.content
                         yield chunk.content
 
@@ -265,26 +460,24 @@ class AgentService:
             elif node == "tools" and hasattr(chunk, "name") and chunk.name == "search_documents":
                 if lg:
                     result_lines = [b for b in chunk.content.split("\n\n") if b.strip()]
-                    yield f"  [결과: {len(result_lines)}개 문서 반환 → agent 복귀]\n"
+                    yield {"__meta": f"  [결과: {len(result_lines)}개 문서 반환 → agent 복귀]\n"}
 
                 if self._rag_verbose:
                     tc = pending_tool_calls.get(chunk.tool_call_id, {})
                     query = tc.get("args", {}).get("query", "")
-                    yield f'\n[문서 검색: "{query}"]\n'
+                    yield {"__meta": f'\n[문서 검색: "{query}"]\n'}
                     for block in chunk.content.split("\n\n"):
                         if block.strip():
                             preview = block.strip().replace("\n", " ")[:80]
-                            yield f"  → {preview}\n"
-                    yield "\n"
+                            yield {"__meta": f"  → {preview}\n"}
+                    yield {"__meta": "\n"}
 
             elif node == "tools" and hasattr(chunk, "name") and chunk.name == "web_search":
                 if lg:
                     result_lines = [b for b in chunk.content.split("\n\n") if b.strip()]
-                    yield f"  [웹 검색 결과: {len(result_lines)}건 → agent 복귀]\n"
-
-        if thinking_open:
-            yield "\n[/사고 과정]\n"
+                    yield {"__meta": f"  [웹 검색 결과: {len(result_lines)}건 → agent 복귀]\n"}
 
+        thinking_open = False
         self._last_run_id = str(run_id)
 
         # 대화 내용을 MySQL에 저장
@@ -295,12 +488,53 @@ class AgentService:
             except Exception as e:
                 print(f"[Agent] 대화 저장 실패: {e}")
 
+        # 대화 내용을 RAG에 비동기 인덱싱 (IDEA-1)
+        if self._conv_rag_enabled and self._ingestion_service and response_content:
+            asyncio.create_task(self._maybe_index_conversation(user_input, response_content))
+
         if self._rag_show_sources and self._source_buffer:
-            yield "\n\n[참고 문서]\n"
+            sources = []
             for src in self._source_buffer:
-                filename = os.path.basename(src["source"])
-                page = f" {src['page']}페이지" if "page" in src else ""
-                yield f"- {filename}{page}\n"
+                entry = {"filename": os.path.basename(src["source"])}
+                if "page" in src:
+                    entry["page"] = src["page"]
+                sources.append(entry)
+            yield {"__sources": sources}
+
+    async def _maybe_index_conversation(self, user_input: str, response: str) -> None:
+        """대화 내용이 RAG에 저장할 만한 정보를 포함하면 Qdrant에 비동기 인덱싱."""
+        if len(response) < 80:
+            return
+
+        prompt = (
+            "다음 대화에서 육아·금융·건강 등 나중에 검색할 만한 유용한 정보가 있으면 "
+            "핵심만 2~4문장으로 간결하게 요약하세요. "
+            "단순 인사, 날짜 확인, 수치 계산은 '없음'이라고만 답하세요.\n\n"
+            f"질문: {user_input}\n"
+            f"답변: {response[:600]}\n\n"
+            "요약 (또는 '없음'):"
+        )
+        try:
+            result = await self._chat_model.bind(enable_thinking=False).ainvoke(
+                [HumanMessage(content=prompt)]
+            )
+            summary = result.content.strip()
+            if not summary or summary == "없음" or len(summary) < 20:
+                return
+            from datetime import datetime
+            metadata = {
+                "source": "conversation",
+                "user_id": self._user_id,
+                "question": user_input[:100],
+                "timestamp": datetime.now().isoformat(),
+            }
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(
+                None, self._ingestion_service.store_text, summary, metadata
+            )
+            print(f"[ConvRAG] 인덱싱 완료: {summary[:60]}...")
+        except Exception as e:
+            print(f"[ConvRAG] 인덱싱 실패: {e}")
 
     def reset(self) -> None:
         """새 thread_id로 대화 히스토리를 초기화한다."""

services/agent/graph_tools.py

@@ -0,0 +1,24 @@
+from langchain_core.tools import tool
+
+
+def make_graph_tools(graph_service, user_id: str = "default"):
+    """지식 그래프 저장/조회 Tool 쌍을 반환한다."""
+
+    @tool
+    def add_relation(subject: str, relation: str, obj: str) -> str:
+        """가족 구성원이나 사물 사이의 관계를 지식 그래프에 저장합니다.
+        알레르기·가족 관계·선호도·질환·특기 등 관계형 정보를 저장할 때 사용하세요.
+        예:
+          subject='도율', relation='알레르기', obj='복숭아'
+          subject='아록', relation='자녀',    obj='도율'
+          subject='근혜', relation='직업',    obj='간호사'
+          subject='하율', relation='좋아하는음식', obj='바나나'"""
+        return graph_service.add_relation(subject, relation, obj, user_id)
+
+    @tool
+    def query_entity(entity: str) -> str:
+        """특정 인물이나 사물에 대해 저장된 모든 관계 정보를 조회합니다.
+        예: entity='도율' → 도율의 알레르기, 나이, 부모, 좋아하는 것 등 모든 알려진 관계"""
+        return graph_service.query_entity(entity, user_id)
+
+    return add_relation, query_entity

services/agent/tools.py

@@ -1,8 +1,19 @@
-from datetime import date
+from datetime import date, datetime
 
 from langchain_core.tools import tool
 
 
+def make_vision_tool(vision_model, image_path: str):
+    """현재 요청에 첨부된 이미지를 분석하는 도구."""
+
+    @tool
+    def analyze_image(prompt: str = "이 이미지를 한국어로 자세히 설명해줘.") -> str:
+        """첨부된 이미지를 분석한다. 이미지 속 음식, 문서, 사람, 사물 등을 파악할 때 사용하세요."""
+        return vision_model.analyze(image_path, prompt)
+
+    return analyze_image
+
+
 @tool
 def get_current_date() -> str:
     """오늘 날짜를 반환합니다. 나이 계산, 날짜 비교 등 현재 날짜가 필요할 때 반드시 먼저 호출하세요."""
@@ -63,6 +74,35 @@ def make_memory_tools(profile_repo, user_id: str = "default"):
     return remember_user_info, recall_user_info
 
 
+def make_reminder_tools(reminder_repo, user_id: str = "default"):
+    """알림 등록/조회 Tool 쌍을 반환한다."""
+
+    @tool
+    def set_reminder(remind_date: str, message: str) -> str:
+        """특정 날짜에 텔레그램으로 알림을 보냅니다.
+        예방접종, 병원 예약, 기념일 등 기억해야 할 날짜를 등록하세요.
+        - remind_date: 알림 날짜 (YYYY-MM-DD 형식). 날짜를 모르면 get_current_date를 먼저 호출하세요.
+        - message: 알림 내용 (구체적으로 작성)
+        등록 시 D-7(7일 전), D-1(하루 전), D-0(당일) 세 번 알림이 발송됩니다."""
+        try:
+            parsed = datetime.strptime(remind_date, "%Y-%m-%d").date()
+        except ValueError:
+            return f"날짜 형식이 잘못되었습니다. YYYY-MM-DD 형식으로 입력해 주세요. (예: 2026-07-01)"
+        reminder_repo.add(user_id, parsed, message)
+        return f"알림이 등록되었습니다. {remind_date}에 '{message}' 알림을 보내드릴게요."
+
+    @tool
+    def list_reminders() -> str:
+        """등록된 예정 알림 목록을 조회합니다. (향후 30일 이내)"""
+        items = reminder_repo.get_upcoming(user_id, days_ahead=30)
+        if not items:
+            return "등록된 예정 알림이 없습니다."
+        lines = [f"- {r['remind_date']}: {r['message']}" for r in items]
+        return "등록된 알림 목록:\n" + "\n".join(lines)
+
+    return set_reminder, list_reminders
+
+
 def make_search_tool(retriever_service, source_buffer: list | None = None):
     """RetrieverService를 클로저로 감싼 문서 검색 Tool을 반환합니다.
 

services/db/mysql_service.py

@@ -110,6 +110,29 @@ class DatabaseService:
                 created_at DATETIME DEFAULT CURRENT_TIMESTAMP
             )
         """)
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS td_knowledge_graph (
+                id INT AUTO_INCREMENT PRIMARY KEY,
+                user_id VARCHAR(50) NOT NULL,
+                subject VARCHAR(200) NOT NULL,
+                relation VARCHAR(100) NOT NULL,
+                object VARCHAR(200) NOT NULL,
+                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                INDEX idx_user_subject (user_id, subject(80))
+            )
+        """)
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS td_reminders (
+                id INT AUTO_INCREMENT PRIMARY KEY,
+                user_id VARCHAR(50) NOT NULL,
+                remind_date DATE NOT NULL,
+                message TEXT NOT NULL,
+                sent_d0 TINYINT(1) NOT NULL DEFAULT 0,
+                sent_d1 TINYINT(1) NOT NULL DEFAULT 0,
+                sent_d7 TINYINT(1) NOT NULL DEFAULT 0,
+                created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+            )
+        """)
         conn.commit()
         self._migrate_schema(conn)
 

services/db/reminder_repository.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+from datetime import date, timedelta
+
+from services.db.mysql_service import DatabaseService
+
+
+class ReminderRepository:
+    """td_reminders 테이블을 통한 알림 저장소."""
+
+    def __init__(self, db: DatabaseService):
+        self._db = db
+
+    def add(self, user_id: str, remind_date: date, message: str) -> int:
+        return self._db.execute_write(
+            "INSERT INTO td_reminders (user_id, remind_date, message) VALUES (%s, %s, %s)",
+            (user_id, remind_date.isoformat(), message),
+        )
+
+    def get_upcoming(self, user_id: str, days_ahead: int = 30) -> list[dict]:
+        today = date.today()
+        limit = today + timedelta(days=days_ahead)
+        return self._db.execute(
+            """SELECT id, remind_date, message
+               FROM td_reminders
+               WHERE user_id = %s AND remind_date >= %s AND remind_date <= %s
+               ORDER BY remind_date""",
+            (user_id, today.isoformat(), limit.isoformat()),
+        )
+
+    def get_due(self, today: date) -> list[dict]:
+        """D-0(당일), D-1(내일), D-7(7일 후) 미발송 알림 반환."""
+        d1 = today + timedelta(days=1)
+        d7 = today + timedelta(days=7)
+        return self._db.execute(
+            """SELECT *,
+                CASE
+                    WHEN remind_date = %s THEN 'd0'
+                    WHEN remind_date = %s THEN 'd1'
+                    WHEN remind_date = %s THEN 'd7'
+                END AS notify_type
+               FROM td_reminders
+               WHERE (remind_date = %s AND sent_d0 = 0)
+                  OR (remind_date = %s AND sent_d1 = 0)
+                  OR (remind_date = %s AND sent_d7 = 0)""",
+            (
+                today.isoformat(), d1.isoformat(), d7.isoformat(),
+                today.isoformat(), d1.isoformat(), d7.isoformat(),
+            ),
+        )
+
+    def mark_sent(self, reminder_id: int, notify_type: str) -> None:
+        col = {"d0": "sent_d0", "d1": "sent_d1", "d7": "sent_d7"}.get(notify_type)
+        if col:
+            self._db.execute_write(
+                f"UPDATE td_reminders SET {col} = 1 WHERE id = %s",
+                (reminder_id,),
+            )

services/knowledge/graph_service.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import networkx as nx
+
+from services.db.mysql_service import DatabaseService
+
+
+class GraphService:
+    """NetworkX 기반 지식 그래프.
+
+    관계 트리플(subject, relation, object)을 MySQL에 영구 저장하고
+    메모리에 로드해 빠른 그래프 쿼리를 제공한다.
+    """
+
+    def __init__(self, db: DatabaseService):
+        self._db = db
+        self._graphs: dict[str, nx.MultiDiGraph] = {}
+
+    def _load(self, user_id: str) -> nx.MultiDiGraph:
+        g = nx.MultiDiGraph()
+        rows = self._db.execute(
+            "SELECT subject, relation, object FROM td_knowledge_graph WHERE user_id = %s",
+            (user_id,),
+        )
+        for row in rows:
+            g.add_edge(row["subject"], row["object"], relation=row["relation"])
+        return g
+
+    def _graph(self, user_id: str) -> nx.MultiDiGraph:
+        if user_id not in self._graphs:
+            self._graphs[user_id] = self._load(user_id)
+        return self._graphs[user_id]
+
+    def _edge_exists(self, g: nx.MultiDiGraph, subject: str, relation: str, obj: str) -> bool:
+        return any(
+            d.get("relation") == relation and target == obj
+            for _, target, d in g.out_edges(subject, data=True)
+        )
+
+    def add_relation(self, subject: str, relation: str, obj: str, user_id: str) -> str:
+        """관계 트리플을 저장한다. 동일 트리플이 존재하면 스킵."""
+        g = self._graph(user_id)
+        if self._edge_exists(g, subject, relation, obj):
+            return f"이미 저장된 관계입니다: {subject} -[{relation}]→ {obj}"
+
+        rows = self._db.execute(
+            "SELECT id FROM td_knowledge_graph "
+            "WHERE user_id=%s AND subject=%s AND relation=%s AND object=%s",
+            (user_id, subject, relation, obj),
+        )
+        if not rows:
+            self._db.execute_write(
+                "INSERT INTO td_knowledge_graph (user_id, subject, relation, object) "
+                "VALUES (%s, %s, %s, %s)",
+                (user_id, subject, relation, obj),
+            )
+        g.add_edge(subject, obj, relation=relation)
+        return f"'{subject} -[{relation}]→ {obj}' 관계를 저장했습니다."
+
+    def query_entity(self, entity: str, user_id: str) -> str:
+        """엔티티에 연결된 모든 관계를 반환한다 (출발/도착 방향 모두)."""
+        g = self._graph(user_id)
+        if entity not in g:
+            return f"'{entity}'에 대해 저장된 정보가 없습니다."
+        lines = []
+        for _, target, data in g.out_edges(entity, data=True):
+            lines.append(f"  {entity} -[{data['relation']}]→ {target}")
+        for source, _, data in g.in_edges(entity, data=True):
+            lines.append(f"  {source} -[{data['relation']}]→ {entity}")
+        if not lines:
+            return f"'{entity}'에 대해 저장된 정보가 없습니다."
+        return f"'{entity}' 관련 정보:\n" + "\n".join(lines)
+
+    def get_summary(self, user_id: str) -> str:
+        """시스템 프롬프트 주입용 전체 관계 요약. 없으면 빈 문자열."""
+        g = self._graph(user_id)
+        if not g.edges:
+            return ""
+        return "\n".join(
+            f"  {s} -[{d['relation']}]→ {t}"
+            for s, t, d in g.edges(data=True)
+        )

services/model/mlx_vision_model.py

@@ -0,0 +1,48 @@
+"""Qwen2.5-VL (mlx-vlm) 기반 이미지 분석 서비스.
+
+첫 analyze() 호출 시 모델을 lazy load해 메모리를 아낀다.
+"""
+from __future__ import annotations
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_PROMPT = "이 이미지를 한국어로 자세히 설명해줘. 사람, 음식, 문서 등 보이는 것을 빠짐없이 설명해."
+
+
+class MlxVisionModel:
+    def __init__(self, model_id: str, max_tokens: int = 512) -> None:
+        self._model_id = model_id
+        self._max_tokens = max_tokens
+        self._model = None
+        self._processor = None
+
+    def _load(self) -> None:
+        if self._model is not None:
+            return
+        logger.info("Vision 모델 로딩 중: %s", self._model_id)
+        from mlx_vlm import load
+        self._model, self._processor = load(self._model_id)
+        logger.info("Vision 모델 로딩 완료")
+
+    def analyze(self, image_path: str, prompt: str = _DEFAULT_PROMPT) -> str:
+        """이미지를 분석해 한국어 설명을 반환한다."""
+        self._load()
+        from mlx_vlm import generate
+        from mlx_vlm.prompt_utils import apply_chat_template
+        from mlx_vlm.utils import load_config
+
+        config = load_config(self._model_id)
+        formatted_prompt = apply_chat_template(
+            self._processor, config, prompt, num_images=1
+        )
+        result = generate(
+            self._model,
+            self._processor,
+            image=image_path,
+            prompt=formatted_prompt,
+            max_tokens=self._max_tokens,
+            verbose=False,
+        )
+        return result if isinstance(result, str) else str(result)

services/rag/ingestion_service.py

@@ -1,6 +1,6 @@
 from langchain_community.document_loaders import PDFPlumberLoader, TextLoader
 from langchain_experimental.text_splitter import SemanticChunker
-from langchain_qdrant import QdrantVectorStore
+from langchain_qdrant import QdrantVectorStore, RetrievalMode
 from qdrant_client import QdrantClient
 from qdrant_client.models import Filter, FieldCondition, MatchValue, FilterSelector
 
@@ -15,10 +15,12 @@ class IngestionService:
         collection_name: str,
         breakpoint_threshold_type: str = "percentile",
         buffer_size: int = 1,
+        sparse_embeddings=None,
     ):
         self._embeddings = embeddings
         self._qdrant_url = qdrant_url
         self._collection_name = collection_name
+        self._sparse_embeddings = sparse_embeddings
         self._splitter = SemanticChunker(
             embeddings=embeddings,
             breakpoint_threshold_type=breakpoint_threshold_type,
@@ -26,6 +28,18 @@ class IngestionService:
         )
         self._client = QdrantClient(url=qdrant_url)
 
+    def _ensure_collection_schema(self) -> None:
+        """Hybrid 모드 전환 시 컬렉션에 sparse vector 설정이 없으면 삭제해 재생성을 유도한다."""
+        if not self._sparse_embeddings:
+            return
+        try:
+            info = self._client.get_collection(self._collection_name)
+            if not info.config.params.sparse_vectors:
+                print(f"[Hybrid] '{self._collection_name}' 컬렉션에 sparse vector 설정이 없어 재생성합니다.")
+                self._client.delete_collection(self._collection_name)
+        except Exception:
+            pass  # 컬렉션 미존재 시 무시
+
     def _delete_by_source(self, source_path: str) -> None:
         """같은 파일 경로로 저장된 기존 청크를 모두 삭제한다."""
         try:
@@ -45,7 +59,23 @@ class IngestionService:
         except Exception:
             pass  # 컬렉션이 없을 때(최초 수집) 무시
 
+    def store_text(self, text: str, metadata: dict) -> None:
+        """단일 텍스트를 Qdrant에 직접 저장 (semantic chunking 없이)."""
+        from langchain_core.documents import Document
+        doc = Document(page_content=text, metadata=metadata)
+        kwargs = dict(
+            documents=[doc],
+            embedding=self._embeddings,
+            url=self._qdrant_url,
+            collection_name=self._collection_name,
+        )
+        if self._sparse_embeddings:
+            kwargs["sparse_embedding"] = self._sparse_embeddings
+            kwargs["retrieval_mode"] = RetrievalMode.HYBRID
+        QdrantVectorStore.from_documents(**kwargs)
+
     def ingest(self, file_paths: list[str]) -> int:
+        self._ensure_collection_schema()
         docs = []
         for path in file_paths:
             self._delete_by_source(path)
@@ -53,10 +83,14 @@ class IngestionService:
             docs.extend(loader.load())
 
         chunks = self._splitter.split_documents(docs)
-        QdrantVectorStore.from_documents(
+        kwargs = dict(
             documents=chunks,
             embedding=self._embeddings,
             url=self._qdrant_url,
             collection_name=self._collection_name,
         )
+        if self._sparse_embeddings:
+            kwargs["sparse_embedding"] = self._sparse_embeddings
+            kwargs["retrieval_mode"] = RetrievalMode.HYBRID
+        QdrantVectorStore.from_documents(**kwargs)
         return len(chunks)

services/rag/retriever_service.py

@@ -1,5 +1,5 @@
 from langchain_core.documents import Document
-from langchain_qdrant import QdrantVectorStore
+from langchain_qdrant import QdrantVectorStore, RetrievalMode
 from qdrant_client import QdrantClient
 from qdrant_client.models import Filter, FieldCondition, MatchValue, FilterSelector
 
@@ -15,24 +15,47 @@ class RetrieverService:
         top_k: int,
         reranker=None,
         rerank_fetch_k: int = 10,
+        sparse_embeddings=None,
     ):
         self._client = QdrantClient(url=qdrant_url)
         self._collection_name = collection_name
-        self._store = QdrantVectorStore(
+        self._top_k = top_k
+        self._reranker = reranker
+        self._rerank_fetch_k = rerank_fetch_k
+        self._sparse_embeddings = sparse_embeddings
+
+        # Dense-only store — hybrid 실패 시 폴백으로도 사용
+        self._dense_store = QdrantVectorStore(
             client=self._client,
             collection_name=collection_name,
             embedding=embeddings,
         )
-        self._top_k = top_k
-        self._reranker = reranker
-        self._rerank_fetch_k = rerank_fetch_k
+
+        if sparse_embeddings:
+            self._store = QdrantVectorStore(
+                client=self._client,
+                collection_name=collection_name,
+                embedding=embeddings,
+                sparse_embedding=sparse_embeddings,
+                retrieval_mode=RetrievalMode.HYBRID,
+            )
+        else:
+            self._store = self._dense_store
 
     def as_retriever(self):
         return self._store.as_retriever(search_kwargs={"k": self._top_k})
 
     def search(self, query: str) -> list[Document]:
         fetch_k = self._rerank_fetch_k if self._reranker else self._top_k
+        try:
             docs = self._store.similarity_search(query, k=fetch_k)
+        except Exception as e:
+            if self._sparse_embeddings:
+                # 컬렉션에 sparse vector 없음 → dense 폴백 (재수집 필요)
+                print(f"[Hybrid] 검색 실패, dense 폴백 (문서 재수집 필요): {e}")
+                docs = self._dense_store.similarity_search(query, k=fetch_k)
+            else:
+                raise
         if self._reranker:
             docs = self._reranker.rerank(query, docs, top_k=self._top_k)
         return docs

services/scheduler_service.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+import asyncio
+import json
+import logging
+import urllib.error
+import urllib.parse
+import urllib.request
+from datetime import date
+
+logger = logging.getLogger(__name__)
+
+_NOTIFY_PREFIX = {
+    "d0": "🔔 오늘 일정",
+    "d1": "📅 내일 일정",
+    "d7": "📆 7일 후 일정",
+}
+
+
+class SchedulerService:
+    """asyncio 태스크 기반 알림 스케줄러.
+
+    매 60초마다 td_reminders를 확인해 D-7/D-1/D-0 Telegram 알림을 발송한다.
+    TELEGRAM_BOT_TOKEN이 비어 있으면 발송 없이 로그만 출력한다.
+    """
+
+    def __init__(self, reminder_repo, bot_token: str, user_map_json: str):
+        self._repo = reminder_repo
+        self._token = bot_token
+        try:
+            self._user_map: dict[str, str] = json.loads(user_map_json) if user_map_json else {}
+        except Exception:
+            self._user_map = {}
+        self._task: asyncio.Task | None = None
+
+    def start(self) -> None:
+        self._task = asyncio.create_task(self._loop())
+        logger.info("[Scheduler] 알림 스케줄러 시작 (60초 간격)")
+
+    def shutdown(self) -> None:
+        if self._task:
+            self._task.cancel()
+            logger.info("[Scheduler] 알림 스케줄러 종료")
+
+    async def _loop(self) -> None:
+        while True:
+            try:
+                await self._check_reminders()
+            except Exception as e:
+                logger.error(f"[Scheduler] 알림 확인 중 오류: {e}")
+            await asyncio.sleep(60)
+
+    async def _check_reminders(self) -> None:
+        today = date.today()
+        reminders = self._repo.get_due(today)
+        for r in reminders:
+            notify_type = r.get("notify_type")
+            if not notify_type:
+                continue
+            text = self._format_message(r["message"], notify_type, r["remind_date"])
+            chat_id = self._user_map.get(r["user_id"])
+            if self._token and chat_id:
+                try:
+                    self._send_telegram(chat_id, text)
+                except Exception as e:
+                    logger.error(f"[Scheduler] Telegram 발송 실패 (user={r['user_id']}): {e}")
+                    continue
+            else:
+                logger.info(f"[Scheduler] 알림(Telegram 미설정): {text}")
+            try:
+                self._repo.mark_sent(r["id"], notify_type)
+            except Exception as e:
+                logger.error(f"[Scheduler] mark_sent 실패: {e}")
+
+    def _format_message(self, message: str, notify_type: str, remind_date) -> str:
+        prefix = _NOTIFY_PREFIX.get(notify_type, "알림")
+        return f"[율봇 알림] {prefix}\n날짜: {remind_date}\n{message}"
+
+    def _send_telegram(self, chat_id: str, text: str) -> None:
+        url = f"https://api.telegram.org/bot{self._token}/sendMessage"
+        data = urllib.parse.urlencode({"chat_id": chat_id, "text": text}).encode()
+        req = urllib.request.Request(url, data=data, method="POST")
+        with urllib.request.urlopen(req, timeout=10):
+            pass