[Feat/#72] 내부 API 동기화와 서버 안 주기 인덱스 갱신

.env.example

@@ -24,6 +24,11 @@ GOOGLE_CLOUD_LOCATION=global
 
 GEMINI_MODEL=gemini-3.5-flash-lite
 
+# ─── 인덱스 주기 갱신 (refresh.py) ───────────────────────────
+# 공지·행사를 백엔드 내부 API에서 다시 받아 인덱스를 바꿔 끼우는 주기(초).
+# BACKEND_BASE_URL과 INTERNAL_TOKEN이 둘 다 있을 때만 켜진다. 로컬 개발·평가는 비워 두면 꺼진다.
+# INDEX_REFRESH_SECONDS=600
+
 # dev면 확인용 데모 페이지(GET /)가 열린다. 배포할 때는 prod
 ENV=dev
 

AGENTS.md

@@ -133,7 +133,8 @@ prompts/
 └── system_v2.md          현재 시스템 프롬프트 (코드 하드코딩 금지)
 eval/
 └── golden.jsonl          평가 골든셋
-sync.py                   백엔드 공지·행사 동기화
+sync.py                   백엔드 공지·행사·아카이빙·열린피드백 동기화 (내부 API /v1/ai/**)
+refresh.py                서버 안 주기 인덱스 갱신 (동기화 → 빌드 → 검사 → 교체)
 chunk.py                  청킹 + 메타데이터
 build_index.py            인덱스 빌드
 search.py                 검색
@@ -243,7 +244,7 @@ ID로 하나 꺼내는 것은 DB가 잘하지만, 수백 건 중 관련된 것
 - 백엔드와 같은 내부망에 둔다. 인터넷에 직접 노출하지 않는다.
 - `INTERNAL_TOKEN`을 백엔드의 `STREAM_AI_INTERNAL_TOKEN`과 같은 값으로 넣는다. `ENV=prod`인데 비어 있으면 서버가 뜨지 않는다.
 - `.env`는 서버에서 직접 만든다. 저장소에 올리지 않는다.
-- 인덱스 갱신 스케줄을 건다. 갱신이 멈추면 공지 수정이 반영되지 않는다.
+- 인덱스 갱신은 서버 안에서 10분마다 자동으로 돈다(`refresh.py`). `BACKEND_BASE_URL`과 `INTERNAL_TOKEN`이 없으면 꺼진다. 꺼지면 공지 수정이 반영되지 않는다.
 
 ## Git 컨벤션
 

app.py

@@ -1,7 +1,9 @@
+import asyncio
 import json
 import logging
 import os
 import secrets
+from contextlib import asynccontextmanager
 from datetime import datetime
 from pathlib import Path
 from typing import Literal
@@ -13,6 +15,8 @@
 from pydantic import BaseModel, Field
 
 import llm
+import refresh
+import sync
 from search import search
 from tools import TOOLS
 
@@ -40,7 +44,22 @@
         "ENV=prod 인데 환경변수 INTERNAL_TOKEN이 없습니다. 백엔드와 같은 값을 넣으세요."
     )
 
-app = FastAPI()
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """서버가 켜질 때 인덱스 주기 갱신을 띄우고, 꺼질 때 멈춘다(refresh.py 참고).
+
+    lifespan: 서버가 켜질 때와 꺼질 때 한 번씩 실행되는 자리. `yield` 앞이 켜질 때, 뒤가 꺼질 때다.
+    백엔드 주소나 내부 토큰이 없으면(로컬 개발·평가) 갱신을 켜지 않고 이미지에 든 인덱스로 답한다.
+    """
+    task = asyncio.create_task(refresh.run_forever()) if sync.is_configured() else None
+    if task is None:
+        logger.warning("인덱스 주기 갱신 꺼짐: BACKEND_BASE_URL 또는 INTERNAL_TOKEN이 없습니다. 빌드 때 만든 인덱스로 답합니다")
+    yield
+    if task is not None:
+        task.cancel()
+
+
+app = FastAPI(lifespan=lifespan)
 
 
 class Msg(BaseModel):

docs/guides/deploy.md

@@ -73,6 +73,17 @@ curl -s -o /dev/null -w "%{http_code}\n" https://ai.cs.kookmin.ac.kr/v1/models
 | `STREAM_AI_BASE_URL` | `http://<Coolify가 정한 AI 서버 컨테이너 이름>:8000` |
 | `STREAM_AI_INTERNAL_TOKEN` | 2번의 `INTERNAL_TOKEN`과 같은 값 |
 
+### 5. 인덱스 주기 갱신
+
+공지·행사·아카이빙·열린피드백은 서버 안에서 **10분마다** 백엔드 내부 API(`/v1/ai/**`, stream-server #312)에서 다시 받아 검색 인덱스를 바꿔 끼운다(`refresh.py`, #72).
+이미지를 다시 빌드하는 게 아니다. 컨테이너 안 `index/` 폴더에 파일만 새로 쓴다.
+
+- `BACKEND_BASE_URL`과 `INTERNAL_TOKEN`이 있으면 자동으로 켜진다. 둘 중 하나라도 없으면 꺼지고 이미지 빌드 때 만든 인덱스(시드 문서)로 답한다
+- 백엔드는 같은 `INTERNAL_TOKEN` 값(백엔드 쪽 `STREAM_AI_INTERNAL_TOKEN`)으로 `/v1/ai/**`를 확인한다. 새 비밀값은 없다
+- 주기를 바꾸려면 `INDEX_REFRESH_SECONDS`(기본 600)를 넣는다
+- 재배포·재시작하면 컨테이너 안 인덱스가 사라지지만, 켜지자마자 1회 돌아 몇 초 안에 최신으로 돌아온다
+- 갱신이 실패하거나 받은 문서가 0건·절반 아래로 줄면 **기존 인덱스를 그대로 쓴다**
+
 ## 배포 후 확인
 
 1. Coolify 로그에 `Uvicorn running on http://0.0.0.0:8000`이 보인다
@@ -93,6 +104,16 @@ curl -s -o /dev/null -w "%{http_code}\n" https://ai.cs.kookmin.ac.kr/v1/models
 | 백엔드 로그에 연결 실패 | 같은 네트워크가 아니거나 `STREAM_AI_BASE_URL`의 컨테이너 이름이 틀렸다 |
 | Coolify 빌드 실패 | Coolify 빌드 로그를 본다. 로컬에서 `docker build .`가 되는지 먼저 확인한다 |
 
+## 인덱스 갱신 막히는 곳
+
+| 증상 | 원인 |
+| --- | --- |
+| 로그에 `인덱스 주기 갱신 꺼짐` | `BACKEND_BASE_URL` 또는 `INTERNAL_TOKEN`이 없다 |
+| 로그에 `인덱스 갱신 실패` + `401` | 백엔드 `STREAM_AI_INTERNAL_TOKEN`과 AI 서버 `INTERNAL_TOKEN`이 다르다 |
+| 로그에 `인덱스 갱신 실패` + `404` | 백엔드에 #312(`/v1/ai/**`)가 아직 배포되지 않았다 |
+| 로그에 `인덱스 갱신 건너뜀: ... 크게 줄었습니다` | 받은 문서가 직전의 절반 아래다. 백엔드 오류인지, 실제로 많이 지웠는지 확인한다. 실제로 지운 거라면 서버를 재시작하면 새 기준으로 받아들인다 |
+| 새 공지를 올렸는데 10분이 지나도 모름 | 위 로그를 본다. 갱신 성공 로그는 INFO라 지금 로깅 설정으로는 보이지 않는다 |
+
 ## 되돌리기
 
 Coolify의 Deployments 화면에서 이전 배포를 골라 다시 띄운다.

refresh.py

@@ -0,0 +1,129 @@
+"""검색 인덱스를 서버 안에서 주기적으로 다시 만든다.
+
+왜 필요한가: 인덱스는 이미지를 빌드할 때 한 번 만들어진다(`Dockerfile`의 `build_index.py`).
+서버는 그걸 메모리에 들고 있어서, 학생회가 공지를 올려도 재배포하기 전에는 챗봇이 모른다.
+그래서 서버가 켜져 있는 동안 10분마다 백엔드에서 다시 받아 인덱스를 바꿔 끼운다.
+**이미지를 다시 만드는 게 아니다.** 컨테이너 안의 `index/` 폴더에 파일만 새로 쓴다.
+
+한 번 갱신하는 순서 (AGENTS.md "인덱스 갱신"):
+    1. 동기화    sync.py가 백엔드 내부 API에서 공지·행사·아카이빙·열린피드백을 받는다
+    2. 빌드      시드 문서·FAQ와 합쳐 새 폴더(index/build_YYYYMMDD_HHMMSS/)에 인덱스를 만든다
+    3. 검사      받은 문서가 0건이거나 직전보다 절반 아래로 줄면 바꾸지 않는다
+    4. 교체      검색이 새 인덱스를 쓰게 하고, index/current 링크를 새 폴더로 바꾼다
+    5. 정리      이전 빌드는 몇 개만 남기고 지운다
+
+어느 단계에서든 실패하면 기존 인덱스를 그대로 쓴다. 백엔드가 잠깐 죽어도 챗봇은
+마지막으로 성공한 인덱스로 계속 답한다.
+
+컨테이너 안 파일이라 재배포·재시작하면 사라진다. 그래서 서버가 켜지자마자 한 번 바로 돌고,
+그 전 몇 초 동안은 이미지에 든 인덱스(시드 문서)로 답한다.
+"""
+
+import asyncio
+import json
+import logging
+import os
+import shutil
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+import build_index
+import search
+import sync
+
+logger = logging.getLogger(__name__)
+
+INDEX_ROOT = "index"
+CURRENT_LINK = os.path.join(INDEX_ROOT, "current")
+BUILD_PREFIX = "build_"
+
+# 몇 초마다 다시 만들지. 공지가 수백 건이라 한 번에 몇 초면 끝난다.
+INTERVAL_SECONDS = int(os.getenv("INDEX_REFRESH_SECONDS", "600"))
+
+# 이전 빌드를 몇 개 남길지(AGENTS.md: 2~3개). 문제가 생겼을 때 직전 인덱스와 비교해 볼 수 있다.
+KEEP_BUILDS = 3
+
+# 받은 문서가 직전의 이 비율 아래로 줄면 이상하다고 보고 바꾸지 않는다.
+# 백엔드 오류로 일부만 오는 경우를 막는다. 정말로 공지를 많이 지웠다면 경고 로그를 보고 사람이 판단한다.
+MIN_RATIO = 0.5
+
+# 직전에 받은 문서 수. 서버가 켜진 뒤 첫 갱신에는 비교할 값이 없어 0건만 아니면 받아들인다.
+_last_synced_count: int | None = None
+
+
+def refresh_once() -> bool:
+    """한 번 갱신한다. 새 인덱스로 바꿨으면 True, 검사에 걸려 그대로 뒀으면 False.
+
+    동기화·빌드·읽기에서 난 예외는 밖으로 던진다. 부르는 쪽(`run_forever`)이 로그만 남기고
+    기존 인덱스를 그대로 둔다.
+    """
+    global _last_synced_count
+
+    synced = sync.fetch_documents()
+    if not synced:
+        logger.warning("인덱스 갱신 건너뜀: 백엔드에서 받은 문서가 0건입니다")
+        return False
+    if _last_synced_count and len(synced) < _last_synced_count * MIN_RATIO:
+        logger.warning("인덱스 갱신 건너뜀: 받은 문서가 %d건 → %d건으로 크게 줄었습니다",
+                       _last_synced_count, len(synced))
+        return False
+
+    # 시드 문서·FAQ는 학생회가 올리는 글이 아니라 우리가 직접 쓴 고정 안내라 매번 같이 넣는다.
+    # 예전 가짜 데이터(test_data의 공지·행사)는 넣지 않는다. 백엔드 실데이터가 그 자리를 대신한다.
+    docs = build_index.load_seed() + build_index.load_faq() + synced
+    index = build_index.build(docs)
+
+    build_dir = os.path.join(INDEX_ROOT, BUILD_PREFIX + datetime.now(ZoneInfo("Asia/Seoul")).strftime("%Y%m%d_%H%M%S"))
+    os.makedirs(build_dir, exist_ok=True)
+    path = os.path.join(build_dir, "bm25.json")
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(index, f, ensure_ascii=False)
+
+    # 검색이 새 인덱스를 쓰게 한다. 파일을 다 읽은 뒤에 바뀌므로 질문 처리 중에 반쯤 바뀌는 일은 없다.
+    search.use_index(path)
+    _point_current(build_dir)
+    _remove_old_builds()
+
+    _last_synced_count = len(synced)
+    logger.info("인덱스 갱신: 문서 %d건 → 청크 %d개 (%s)", index["doc_count"], index["chunk_count"], build_dir)
+    return True
+
+
+def _point_current(build_dir: str) -> None:
+    """index/current 링크를 새 빌드 폴더로 바꾼다.
+
+    임시 이름으로 링크를 먼저 만들고 `os.replace`로 한 번에 덮어쓴다. 기존 링크를 지우고 새로 만들면
+    그 사이 아주 잠깐 링크가 없는 순간이 생긴다. `os.replace`는 그런 순간이 없다(원자적 교체).
+    """
+    temp = CURRENT_LINK + ".tmp"
+    if os.path.lexists(temp):
+        os.remove(temp)
+    # 링크 안에는 index/ 기준 상대 경로를 적는다. 폴더째 옮겨도 깨지지 않는다.
+    os.symlink(os.path.basename(build_dir), temp)
+    os.replace(temp, CURRENT_LINK)
+
+
+def _remove_old_builds() -> None:
+    """가장 최근 KEEP_BUILDS개만 남기고 이전 빌드 폴더를 지운다. 폴더 이름이 시각이라 이름순이 곧 시간순이다."""
+    builds = sorted(
+        name for name in os.listdir(INDEX_ROOT)
+        if name.startswith(BUILD_PREFIX) and os.path.isdir(os.path.join(INDEX_ROOT, name))
+    )
+    for name in builds[:-KEEP_BUILDS]:
+        shutil.rmtree(os.path.join(INDEX_ROOT, name), ignore_errors=True)
+
+
+async def run_forever() -> None:
+    """서버가 켜져 있는 동안 INTERVAL_SECONDS마다 갱신한다. `app.py`가 서버 시작 때 띄운다.
+
+    갱신 자체는 파일을 쓰고 계산하는 일이라 별도 스레드(`asyncio.to_thread`)에서 돌린다.
+    그래야 갱신하는 몇 초 동안에도 학생 질문이 막히지 않는다.
+    """
+    logger.info("인덱스 주기 갱신 시작: %d초마다", INTERVAL_SECONDS)
+    while True:
+        try:
+            await asyncio.to_thread(refresh_once)
+        except Exception:
+            # 상세 원인은 로그에만 남기고 다음 주기에 다시 시도한다. 기존 인덱스는 그대로다.
+            logger.exception("인덱스 갱신 실패. 기존 인덱스를 그대로 씁니다")
+        await asyncio.sleep(INTERVAL_SECONDS)

search.py

@@ -158,6 +158,17 @@ def _index() -> dict:
     return _INDEX
 
 
+def use_index(path: str) -> None:
+    """검색에 쓸 인덱스를 다른 파일로 바꿔 끼운다. `refresh.py`가 새 인덱스를 만든 뒤 부른다.
+
+    새 파일을 끝까지 읽고 계산까지 마친 뒤에 한 번에 바꾼다. 그래서 바꾸는 도중에 들어온 질문도
+    옛 인덱스나 새 인덱스 중 하나로 온전히 검색된다. 반쯤 바뀐 인덱스를 보는 일은 없다.
+    파일이 깨져 읽다가 실패하면 예외가 나고 기존 인덱스는 그대로 남는다.
+    """
+    global _INDEX
+    _INDEX = load_index(path)
+
+
 def score(query: str, index: dict | None = None) -> list[tuple[int, float]]:
     """질문에 대한 청크별 BM25 점수를 (청크 위치, 점수)로 돌려준다.
 

sync.py

@@ -0,0 +1,132 @@
+"""백엔드 내부 API에서 공지·행사·아카이빙·열린피드백을 받아 인덱싱할 문서 모양으로 바꾼다.
+
+이 모듈은 받아서 바꾸는 일만 한다. 인덱스를 만들고 바꿔 끼우는 일은 `refresh.py`가 한다.
+
+부르는 곳은 백엔드의 AI 서버 전용 조회 API `/v1/ai/**`다(stream-server #312).
+학생 JWT가 아니라 내부 토큰(`X-Internal-Token`)으로 부른다. 이 작업은 학생 질문 없이 혼자 돌기 때문이다.
+백엔드가 학생 화면과 같은 규칙으로 걸러서 주므로(삭제·미게시 제외), 여기서 다시 거르지 않는다.
+
+돌려주는 모양은 `build_index.py`의 로더(`load_notices`, `load_events`)와 같다.
+`{"text": 본문, "meta": {문서ID, 출처URL, 제목, 게시일, 유효기간, 문서유형}}`
+문서ID 접두어(notice-, event-, archive-, feedback-)도 같게 맞췄다. 골든셋의 근거문서ID를 다시 맞출 때 이 이름을 쓴다.
+
+담당: AGENTS.md상 승빈 담당 파일이다. 내부 API 연결은 이상현이 먼저 만들었다(2026-11-09 합의).
+"""
+
+import logging
+import os
+
+import httpx
+from dotenv import load_dotenv
+
+load_dotenv()
+
+logger = logging.getLogger(__name__)
+
+# 백엔드 내부 주소(예: http://<백엔드 컨테이너 이름>:8080). tools.py의 재고 조회와 같은 값을 쓴다.
+BACKEND_BASE_URL = os.getenv("BACKEND_BASE_URL", "").rstrip("/")
+# 백엔드 → AI 서버 방향에 쓰는 값과 같다. 백엔드도 같은 값으로 /v1/ai/**를 확인한다.
+INTERNAL_TOKEN = os.getenv("INTERNAL_TOKEN", "")
+
+# 네 번 부르는 동안 하나라도 이 시간을 넘기면 이번 동기화는 실패로 보고, 기존 인덱스를 그대로 쓴다.
+TIMEOUT_SECONDS = 10
+
+
+def is_configured() -> bool:
+    """동기화에 필요한 값이 다 있는지. 없으면 동기화를 켜지 않는다(로컬 개발·평가)."""
+    return bool(BACKEND_BASE_URL and INTERNAL_TOKEN)
+
+
+def _date(value: str | None) -> str:
+    """'2026-10-16T10:00:00'이든 '2026-10-16'이든 날짜 부분만 남긴다. 없으면 빈 문자열."""
+    return (value or "")[:10]
+
+
+def _datetime(value: str | None) -> str:
+    """'2026-11-20T14:00:00' → '2026-11-20 14:00'. 사람이 읽는 본문에 넣을 때 쓴다."""
+    return (value or "")[:16].replace("T", " ")
+
+
+def _get(client: httpx.Client, path: str) -> list[dict]:
+    """내부 API 하나를 불러 data 목록을 돌려준다. 실패하면 예외를 던진다.
+
+    한 종류라도 실패하면 이번 동기화 전체를 실패로 본다. 공지만 빠진 인덱스로 바꿔 끼우면
+    챗봇이 갑자기 공지를 하나도 모르게 되기 때문이다. 실패하면 `refresh.py`가 기존 인덱스를 그대로 둔다.
+    """
+    response = client.get(f"{BACKEND_BASE_URL}{path}")
+    response.raise_for_status()
+    body = response.json()
+    if not body.get("success"):
+        raise RuntimeError(f"내부 API 실패: {path} code={body.get('code')}")
+    return body.get("data") or []
+
+
+def _notice(n: dict) -> dict:
+    return {"text": n["content"], "meta": {
+        "문서ID": f"notice-{n['noticeId']}", "제목": n["title"],
+        "게시일": _date(n.get("createdAt")), "유효기간": "",
+        "출처URL": "", "문서유형": "공지"}}
+
+
+def _event(e: dict) -> dict:
+    """행사 소개. 날짜·장소도 본문에 넣는다 — "해커톤 언제야", "어디서 해" 같은 질문이 이 문서로 답된다.
+
+    모집 인원·남은 자리처럼 계속 바뀌는 값은 백엔드가 주지 않는다(질문할 때 따로 조회한다).
+    행사에는 작성일이 없어 게시일은 신청 시작일(없으면 행사 시작일)로 둔다. 기존 로더와 같다.
+    """
+    lines = [e["title"], e.get("description") or ""]
+    if e.get("applyStartAt") or e.get("applyEndAt"):
+        lines.append(f"신청 기간: {_datetime(e.get('applyStartAt'))} ~ {_datetime(e.get('applyEndAt'))}")
+    if e.get("eventStartAt") or e.get("eventEndAt"):
+        lines.append(f"행사 일시: {_datetime(e.get('eventStartAt'))} ~ {_datetime(e.get('eventEndAt'))}")
+    if e.get("place"):
+        lines.append(f"장소: {e['place']}")
+    return {"text": "\n".join(lines), "meta": {
+        "문서ID": f"event-{e['eventId']}", "제목": e["title"],
+        "게시일": _date(e.get("applyStartAt") or e.get("eventStartAt")),
+        "유효기간": _date(e.get("eventEndAt")),
+        "출처URL": "", "문서유형": "행사"}}
+
+
+def _archive(a: dict) -> dict:
+    """지난 활동 기록. 아카이빙에는 작성일이 없어 게시일은 활동 시작일로 둔다."""
+    lines = [a.get("content") or ""]
+    if a.get("startDate"):
+        lines.append(f"활동 기간: {a['startDate']} ~ {a.get('endDate') or ''}")
+    if a.get("location"):
+        lines.append(f"장소: {a['location']}")
+    if a.get("departmentName"):
+        lines.append(f"담당: {a['departmentName']}")
+    return {"text": "\n".join(lines), "meta": {
+        "문서ID": f"archive-{a['archiveId']}", "제목": a["title"],
+        "게시일": _date(a.get("startDate")), "유효기간": _date(a.get("endDate")),
+        "출처URL": "", "문서유형": "아카이빙"}}
+
+
+def _feedback(f: dict) -> dict:
+    """질문 문장을 본문에 함께 넣는다. 학생이 묻는 말로도 검색되게 하려는 것이다(기존 로더와 같다)."""
+    return {"text": f"질문: {f['question']}\n답변: {f['answer']}", "meta": {
+        "문서ID": f"feedback-{f['feedbackId']}", "제목": f["question"],
+        "게시일": _date(f.get("answeredAt")), "유효기간": "",
+        "출처URL": "", "문서유형": "열린피드백"}}
+
+
+def fetch_documents() -> list[dict]:
+    """공지·행사·아카이빙·열린피드백 전체를 받아 인덱싱 문서 목록으로 돌려준다.
+
+    실패하면 예외를 던진다(이유는 `_get` 참고). 토큰은 로그에 남기지 않는다.
+    """
+    if not is_configured():
+        raise RuntimeError("BACKEND_BASE_URL 또는 INTERNAL_TOKEN이 없어 동기화할 수 없습니다")
+
+    headers = {"X-Internal-Token": INTERNAL_TOKEN}
+    with httpx.Client(headers=headers, timeout=TIMEOUT_SECONDS) as client:
+        notices = _get(client, "/v1/ai/notices")
+        events = _get(client, "/v1/ai/events")
+        archives = _get(client, "/v1/ai/archives")
+        feedbacks = _get(client, "/v1/ai/feedbacks")
+
+    logger.info("동기화: 공지 %d, 행사 %d, 아카이빙 %d, 열린피드백 %d",
+                len(notices), len(events), len(archives), len(feedbacks))
+    return ([_notice(n) for n in notices] + [_event(e) for e in events]
+            + [_archive(a) for a in archives] + [_feedback(f) for f in feedbacks])