vLLM에 pdf 넣는법

2025-11-04

vllm 서빙하는 모델 gptoss에 pdf 넣는법

요약 먼저: vLLM으로 서빙 중인 gpt-oss는 “PDF 파일을 직접 넣어 처리”하는 인터페이스가 없습니다. vLLM OpenAI 호환 서버는 텍스트/토큰 스트림을 받는 API이고, gpt-oss는 공개된 가이드 기준 텍스트 전용 모델이라서(이미지·PDF 업로드 파라미터 없음) PDF는 사전 전처리(텍스트 추출→청소→청크→검색/RAG)로 넣어야 합니다. docs.vllm.ai+4

아래는 “PDF를 gpt-oss(vLLM 서버)로 넣는” 가장 흔한 구현 레시피입니다. 단계별로 그대로 따라하면 됩니다.

vLLM 서버로 gpt-oss 띄우기

CUDA 환경에서 예시:

pip install vllm
python -m vllm.entrypoints.openai.api_server \
  --model openai/gpt-oss-20b \
  --max-model-len 32768 \
  --port 8000

서버는 OpenAI 호환 Chat/Completions 엔드포인트를 제공합니다. 기본적으로 파일 업로드 필드는 없고 문자열 프롬프트만 받습니다. docs.vllm.ai

PDF → 텍스트 추출

Python 도구 예: pypdf, pdfminer.six, unstructured, docling 등.
표·머리글·각주가 섞여 있으면 docling/Unstructured 계열이 품질이 더 낫습니다. 추출 후 다음을 꼭 하세요.
• 페이지 헤더/푸터 제거, 공백 정규화, 본문 병합
• 너무 긴 표는 CSV로 따로 저장(필요시)
관련 이슈와 파이프라인 선택은 커뮤니티/라이브러리 이슈를 참고하세요. GitHub

청크 나누기

800~1500자(또는 200~400 토큰) 단위로 자르고, 문단 경계를 최대한 유지합니다.
각 청크에 원문 페이지/섹션 메타데이터를 붙입니다.

임베딩 생성 → 벡터DB 적재

임베딩 모델(예: bge-m3, jina-embeddings, sentence-transformers 등)로 각 청크를 벡터화하고, FAISS/Chroma/PGVector에 넣습니다.
이 단계는 LLM(gpt-oss/vLLM)과 분리된 별도 과정입니다. vLLM 자체는 임베딩을 만들지 않습니다. RAG 패턴 예시는 다수의 vLLM 배포 예제에서도 동일합니다. validatedpatterns.io+1

질의 시 검색(RAG) → 프롬프트 구성 → vLLM에 전달

유저 질문을 벡터 검색으로 top-k 청크와 매칭
아래 System/Assistant 컨텍스트에 “검색 결과”를 넣고, gpt-oss의 권장 포맷(하모니 포맷)을 지키며 요청을 보냅니다. gpt-oss는 OpenAI 호환 서버에서 잘 동작하도록 가이드가 있습니다. cookbook.openai.com+1

실전 코드 스니펫

A. PDF 텍스트 추출/청크(간단 예시)

from pypdf import PdfReader

def extract_pdf_text(path):
    reader = PdfReader(path)
    pages = []
    for i, p in enumerate(reader.pages):
        t = p.extract_text() or ""
        pages.append({"page": i+1, "text": " ".join(t.split())})
    return pages

def chunk_text(text, max_chars=1200, overlap=150):
    out, start = [], 0
    while start < len(text):
        end = min(len(text), start + max_chars)
        out.append(text[start:end])
        start = end - overlap
        if start < 0: start = 0
    return out

pages = extract_pdf_text("sample.pdf")
chunks = []
for page in pages:
    for c in chunk_text(page["text"]):
        chunks.append({"page": page["page"], "content": c})

B. 임베딩→FAISS 적재(예시: sentence-transformers)

from sentence_transformers import SentenceTransformer
import faiss, numpy as np

model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
emb = model.encode([c["content"] for c in chunks], convert_to_numpy=True, normalize_embeddings=True)
index = faiss.IndexFlatIP(emb.shape[1])
index.add(emb)

C. 검색→vLLM(gpt-oss) 호출(OpenAI 호환)

import numpy as np, requests, json

def search(query, topk=5):
    q = model.encode([query], convert_to_numpy=True, normalize_embeddings=True)
    scores, idx = index.search(q, topk)
    ctx = []
    for i in idx[0]:
        ctx.append(f"[p.{chunks[i]['page']}]\n{chunks[i]['content']}")
    return "\n\n".join(ctx)

def ask_vllm(query):
    context = search(query, topk=5)
    messages = [
        {"role":"system","content":"You are a helpful assistant. Use the provided CONTEXT; if missing, say you don't know."},
        {"role":"user","content":f"CONTEXT:\n{context}\n\nQUESTION:\n{query}"}
    ]
    resp = requests.post(
        "http://localhost:8000/v1/chat/completions",
        headers={"Content-Type":"application/json"},
        data=json.dumps({"model":"openai/gpt-oss-20b","messages":messages,"temperature":0.2})
    ).json()
    return resp["choices"][0]["message"]["content"]

print(ask_vllm("이 PDF에서 반품 정책 핵심 요약해줘"))

D. curl로 간단 테스트

curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "openai/gpt-oss-20b",
    "messages": [
      {"role":"system","content":"Answer using the context."},
      {"role":"user","content":"CONTEXT:\n<여기에 검색으로 모은 PDF 청크>\n\nQUESTION:\n요약해줘"}
    ],
    "temperature": 0.2
  }'

자주 묻는 포인트

왜 PDF 파일 자체를 바로 못 넣나?
vLLM의 OpenAI 호환 서버는 파일 업로드/스토리지 기능이 아니라 텍스트 프롬프트 API입니다. 즉, 파일은 애플리케이션 레이어에서 처리해서 텍스트(또는 검색된 컨텍스트)로 바꾼 다음 모델에 넣어야 합니다. docs.vllm.ai
gpt-oss가 PDF나 이미지까지 직접 읽는 멀티모달인가?
공개 가이드 기준 gpt-oss는 텍스트 중심의 오픈 웨이트 패밀리로 소개되었습니다. 멀티모달 입력을 직접 받는 형태가 아니라, 텍스트로 컨텍스트를 주는 방식이 표준입니다. OpenAI+2
“PDF 질의응답” 정밀도를 더 높이려면?
청크 크기/오버랩 최적화, 표/코드 블록 보존, 섹션별 재순위화(re-ranking), “출처 페이지 표시”를 프롬프트에 강제, 그리고 답변 근거를 따로 출력하도록 규칙을 넣으면 품질이 안정됩니다. vLLM 레시피/쿠크북의 gpt-oss 운영 가이드도 함께 보세요. docs.vllm.ai+1

한 줄 정리: vLLM+gpt-oss로 PDF를 “넣는” 것은 PDF를 텍스트로 전처리하고, 검색 기반(RAG)으로 관련 청크를 추려 OpenAI 호환 API에 문자열로 보내는 것입니다. 이 흐름을 코드와 같이 구성하면 바로 운영에 올릴 수 있습니다.

openai agent 라는 패키지가 있는데 여기에 그런 레이어가 포함되어있나?

짧은 결론부터 말하면, “openai agent”로 알려진 OpenAI Agents SDK는 에이전트 오케스트레이션(툴 호출, 세션/메모리, 핸드오프, 가드레일 등)을 제공하지만 PDF 파싱·청크·임베딩·벡터스토어 적재 같은 “문서 RAG 인입 레이어”는 기본 내장되어 있지 않습니다. 다만 OpenAI 모델을 쓸 때는 호스티드 도구(FileSearchTool 등)를 통해 OpenAI Vector Stores에 올려둔 문서를 검색하는 방식은 지원합니다. 자체 vLLM+gpt-oss로 돌릴 때는 그 레이어를 직접 붙여야 합니다. openai.github.io+1

핵심 포인트를 정리하면 아래와 같습니다.

SDK가 해주는 일
에이전트 루프, 함수툴(FunctionTool), 세션/트레이싱, 핸드오프, MCP(Model Context Protocol) 연결 등 “오케스트레이션”이 주역할입니다. PDF를 읽어 텍스트로 만들고 임베딩을 뽑아 벡터DB에 적재하는 파이프라인은 앱 쪽에서 구성해야 합니다. openai.github.io+1
OpenAI 호스티드 도구가 있을 때
OpenAI Responses 모델을 쓰면 FileSearchTool, WebSearchTool, Code Interpreter 같은 “호스티드 툴”을 바로 붙일 수 있습니다. 이 경우 PDF는 OpenAI Vector Stores API로 업로드·처리하고, 에이전트는 FileSearchTool로 검색합니다. SDK 자체가 PDF를 파싱해주진 않고, 벡터스토어도 OpenAI 쪽 리소스를 이용하는 구성입니다. openai.github.io
gpt-oss(+vLLM)와 함께 쓰는 경우
vLLM의 OpenAI-호환 엔드포인트로 에이전트의 모델 클라이언트를 바꿔치기 할 수 있습니다. 다만 이때는 OpenAI 호스티드 도구(FileSearchTool 등)가 그대로 동작하지 않거나 전제가 바뀌므로, 직접 함수툴로 “검색/RAG”를 구현하거나 MCP 서버(자체 문서 검색 서버)를 붙여서 문서 검색을 노출하는 패턴이 일반적입니다. cookbook.openai.com+2
최소 구현 스케치

PDF 인입: pypdf/unstructured로 텍스트 추출 → 청크 → 임베딩(bge/jina/sentence-transformers) → FAISS/Chroma 적재
에이전트 툴: search(query) 같은 Python 함수툴을 정의해 top-k 청크를 반환하도록 구현
에이전트: Agent(tools=[search])로 등록, 모델은 vLLM의 Chat Completions/Responses 호환 엔드포인트로 지정
OpenAI 모델로 바꿀 때만 FileSearchTool을 사용해 OpenAI Vector Store를 조회 가능
이 구성은 SDK 문서의 “Tools”, “Context management”, “MCP” 섹션 흐름과 딱 맞습니다. openai.github.io+2

이름 혼동 주의
Assistants API는 Responses API로 대체되는 흐름이며, Agents SDK는 이 Responses 기반에서 돌아갑니다. 최신 가이드와 레퍼런스를 참고해서 구성하세요. reuters.com+1

정리

포함되어 있는가? 에이전트 오케스트레이션은 포함, “PDF→RAG 인입 레이어”는 미포함. OpenAI 모델일 때만 FileSearchTool로 OpenAI Vector Stores를 조회할 수 있음. vLLM+gpt-oss에서는 직접 함수툴이나 MCP로 붙여야 합니다. openai.github.io+1

원하시면 지금 쓰시는 vLLM 서버 주소와 선호 임베딩/벡터DB를 가정해, Agents SDK용 최소 예제(함수툴 기반 검색 + vLLM 연결) 스니펫을 바로 만들어 드립니다.

openai vector store는 어떤식으로 구성되어있어?

좋아요 — “OpenAI Vector Store”가 대체로 어떻게 구성되어 있는지(=아키텍처, 주요 컴포넌트, API 흐름) 같이 살펴봅시다. (약간 기술적이긴 하지만, 전체 그림을 이해하면 이후 구현할 때 훨씬 수월해요.)

1. 기본 개념

Vector Store는 간단히 말해 “문서/청크(텍스트 단위)를 임베딩 벡터로 변환해 저장하고, 이후 검색 가능하도록 조회하는 저장소”입니다. OpenAI 쪽에서 제공하는 API는 문서를 업로드 → 임베딩/청크화(내부 처리) → 벡터저장소 생성 및 파일 바인딩 → 검색 질의(Query) → 관련 청크 반환하는 흐름을 지원합니다. platform.openai.com+3

즉, 내부적으로는 다음 요소들이 있습니다:

문서 파일(예: PDF, 텍스트) 업로드 및 처리된 파일 배치(file batch)
벡터 변환(임베딩) + 청크(metadata 포함)
저장소(Vector store) 객체: 여러 파일/청크들을 묶음
검색 API: 벡터 유사도 기반 조회 → 반환된 청크와 메타정보

2. OpenAI API 상의 구성 요소

OpenAI 문서를 기준으로 주요 리소스(resource)들이 어떻게 연결되는지요:

Vector Store 리소스 : /v1/vector_stores 엔드포인트로 만들 수 있음. platform.openai.com+1
파일 배치(File Batches) : 업로드된 파일이 청크/벡터화되어 저장되는 단위. /v1/vector_stores/{vector_store_id}/file_batches 같은 구조 있음. platform.openai.com
파일(파일 업로드) : /v1/vector_stores/{vs_id}/files 엔드포인트에서 파일을 추가할 수 있음. platform.openai.com
검색(Search) : 저장된 Vector Store에 대해 “유사벡터 기반 검색”을 요청 가능 (예: 파일 검색, 청크 검색) platform.openai.com+1

3. 흐름 예시로 보기

아래처럼 단계별 구성됩니다:

파일 업로드
- 예: PDF를 업로드 → OpenAI 측에서 자동으로 청크화/임베딩/저장 준비 → 파일 배치 생성
Vector Store 생성
- 저장소 이름, 설명 등의 메타데이터 제공 → 생성
파일을 Vector Store에 바인딩(=파일 추가)
- 생성한 Vector Store ID에 파일 배치/파일을 업로드하는 API 호출
검색 요청
- 사용자가 질의(query)를 보내면, API는 임베딩 생성 → 저장된 벡터들과 유사도 비교 → 관련 청크 반환
- 반환된 청크를 기반으로 응답 생성(LLM 호출 전처리 역할)
LLM 호출(별도)
- 검색된 청크는 LLM(예: gpt-모델)에 컨텍스트로 전달 → 답변 생성

4. 주요 특성 및 한계

파일/문서 메타데이터는 비교적 제한적: 업로드된 파일이 얼마나 긴지(token count, char count) 등의 정보 부족하다는 사용자의 피드백이 있음. Reddit
청크화 + 임베딩 + 저장/관리 과정을 “사용자제어” 하려면 OpenAI 호스티드 기능보다는 자체 벡터DB + 임베딩 워크플로우가 더 유연함
검색 품질은 임베딩 모델(어떤 모델을 썼는가) + 청크 단위/문서 구조 설계에 크게 좌우됨

5. gpt-oss/vLLM 활용 시 고려사항

당신이 vLLM + gpt-oss 조합으로 쓰는 상황이라면, OpenAI Vector Store API를 그대로 의존하기보다는 다음과 같은 변형이 필요할 수 있어요:

OpenAI Vector Store API를 쓰고 싶으면 OpenAI 인증/요금/호스팅 구조 따라야 함
또는 자체 벡터DB(예: FAISS, Milvus 등) + 임베딩 모델 + 검색 함수툴을 만들어서 “유사 기능”을 구현
Agents SDK나 워크플로우에서는 “검색 툴(tool)”이 이 저장소와 연결됨: 사용자가 질의하면 검색툴이 관련 청크를 꺼내오고, LLM에게 문맥으로 제공

필요하시다면, OpenAI Vector Store API를 실제로 호출하는 예제 코드(Python) 만들어 드릴 수 있어요—특히 파일 업로드 → 검색까지 흐름. 그렇게 할까요?

마지막에 말한 검색툴은 자동으로 동작하는건지 llm이 요청하면 동작하는건지 궁금해

아주 핵심을 짚으셨습니다.
OpenAI의 “검색 툴(FileSearchTool)”은 자동으로 동작하지 않습니다.
LLM이 “필요하다고 판단할 때” — 즉, 모델이 툴 호출을 생성하면 — 그때 실행됩니다.

이걸 조금 더 세밀하게 뜯어보면 이렇게 됩니다.

1. 에이전트의 기본 구조

OpenAI Agents SDK(혹은 Responses API)에서 에이전트는 다음 두 가지를 갖습니다.

LLM 모델 : gpt-4o, gpt-4o-mini 등
툴(tool) : 함수 호출(Function Tool), FileSearchTool, WebSearchTool, CodeInterpreter 등

에이전트는 LLM에게 “너는 이런 툴들을 쓸 수 있다”고 알려주고,
대화 중에 LLM이 판단해 특정 툴을 호출할 때만 실제 실행 로직이 수행됩니다.

2. 툴 호출 과정

모델이 내부적으로 이런 구조의 응답을 냅니다.

{
  "type": "function_call",
  "name": "search",
  "arguments": { "query": "계약서 요약" }
}

SDK는 이걸 감지하고 search() 함수 (혹은 FileSearchTool)을 호출합니다.
그 결과를 받아 모델에게 다시 전달하면, 모델은 그 결과를 바탕으로 최종 답변 을 만듭니다.

즉, LLM이 툴 사용을 “결정”하고, 호출은 SDK가 “대리 실행”합니다.

3. 자동성과 자율성의 정도

자동 실행 아님 : 사용자가 “검색을 꼭 하라”고 지정하지 않으면, 모델이 문맥상 필요하다고 판단해야 호출합니다.
프롬프트 설계 가능 : system prompt 에 “파일 내용을 참조할 때는 항상 FileSearchTool을 사용하라”처럼 명시해 주면 호출 확률이 높아집니다.
Agent Loop : SDK가 이 과정을 자동 반복해줍니다. 모델이 툴 호출 → 실행 → 결과 전달 → 다시 모델 호출 … 최종 답변이 완성될 때까지 루프를 돕습니다.

4. 예시로 보는 흐름

사용자: “이 계약서 요약해줘.”

모델은 FileSearchTool이 등록되어 있음을 알고 있음.
LLM이 내부적으로 판단: “파일에서 내용 가져와야 하네.”
LLM → function_call: file_search(query="계약서 요약")
SDK → 실제 Vector Store 검색 수행
검색 결과를 모델에게 다시 입력
모델 → 검색 결과를 요약한 최종 답변 생성

5. vLLM + gpt-oss 조합에서는

vLLM은 OpenAI의 Responses API 내부 오케스트레이션 기능이 없습니다.
따라서 “자동 툴 호출”을 구현하려면 SDK가 필요합니다.
직접 서버나 에이전트 루프를 만들고, 모델의 함수 호출 결과(JSON)를 해석해 수동으로 “search() → 결과 → 모델 재호출”을 구현해야 합니다.

정리하자면:
FileSearchTool은 모델이 필요하다고 판단할 때만 호출되고, SDK가 실행을 중계합니다.
즉, “모델이 요청하면 자동으로 동작한다”는 표현이 가장 정확합니다.

다음 단계로, 원하신다면 vLLM + gpt-oss 조합에서 이 “툴 호출 루프”를 흉내내는 최소 예제(모델 → function_call → search → 재호출)를 만들어 드릴 수 있습니다.

영어로 작성된 문서 rag를 한글로 쿼리할 수 있어?

가능합니다. 방법은 크게 두 가지 계열입니다. 한글 쿼리 → 영어 문서 RAG를 지원하려면 “다국어 임베딩”을 쓰거나 “번역 레이어”를 끼우면 됩니다.

다국어 임베딩으로 직접 검색

핵심: 영어 문서 청크와 한국어 쿼리를 같은 임베딩 공간으로 매핑.
모델 예: bge-m3, e5-multilingual, LaBSE, mContriever 등.
장점: 번역 없이도 개념 맞춤. 고유명사·전문용어에 강함.
단점: 임베딩 품질·차원 수가 커서 메모리↑. 영어 전용 임베딩 대비 약간의 정밀도 손실이 있을 수 있음.

번역 후 검색(translate-then-retrieve)

파이프라인: 한글 쿼리 → 영어로 번역 → 영어 전용 임베딩(예: e5-large, bge-large-en 등)으로 검색 → 답변은 한글로 생성.
장점: 영어 전용 임베딩/리랭커 품질을 그대로 활용 가능.
단점: 번역 오류가 쿼리의 핵심 토픽을 비트는 경우가 있음. 엔티티(제품명, 표기) 주의.

권장 설계(현업 안정판)

인덱싱: 영어 문서를 문단/문장 단위로 청크, 메타데이터(페이지, 섹션, 언어=en) 저장.
임베딩: 다국어 임베딩 인덱스 1개는 반드시 구축. 여력이 되면 영어 전용 인덱스를 추가로 구성해 듀얼 인덱스 검색.
검색 전략: 한글 쿼리에 대해
1. 다국어 인덱스에서 top-k
2. 영어 전용 인덱스(쿼리 번역)에서도 top-k
3. 두 결과를 점수 정규화 후 병합
리랭킹: 다국어 리랭커(bge-reranker-v2-m3 같은 범용 다국어 리랭커)로 상위 50→10 재정렬.
하이브리드: BM25(키워드) + 벡터 점수 가중 평균. 고유명사 매칭 강화.
생성: gpt-oss(vLLM)에 “출처 표시, 영어 인용문은 원문 유지, 최종 답변은 한국어”를 시스템 규칙으로 주입.

간단 파이프라인 예시(핵심 로직만)

인덱싱
1. PDF → 텍스트 추출 → 청크
2. multilingual_embed(chunks) → FAISS/Chroma/PGVector에 저장
3. 선택: english_embed(chunks)로 영어 전용 인덱스도 병행 구축
질의
1. ko_query → multilingual_embed(ko_query)로 검색
2. ko_query → en_query로 번역 → english_embed(en_query)로 검색
3. 결과 병합 → 다국어 리랭킹
4. 상위 n개를 컨텍스트로 vLLM에 넣고 한국어로 답변 생성

프롬프트 팁

시스템 규칙에 “문헌은 영어 원문을 그대로 인용, 설명은 한국어로 작성, 출처 페이지/섹션을 함께 적시”를 명시.
다국어 혼용 문서일 경우 “언어 감지 → 쿼리와 다른 언어라도 동일 개념이면 포함” 같은 지침을 추가.

실무 체크리스트

토크나이저 길이: 청크는 200~400토큰(문단 경계 유지).
메타데이터: page, section, filename, lang. 답변에 근거 표시 용이.
고유명사: 대소문자·하이픈·스네이크케이스 보존. 정규화 과도하게 하지 않기.
캐싱: 빈번한 쿼리 번역/임베딩 결과 캐시.
평가: 한-영 쌍으로 질의–정답 골드셋을 만들어 MRR@n, Hit@n, nDCG로 비교.

vLLM + gpt-oss 연결 포인트

위 검색 결과(top-k 청크)를 user 메시지에 CONTEXT로 넣고 질문은 한국어로 전달.
답변은 한국어로 강제, 필요한 부분만 영어 원문 인용.
함수툴(예: search_docs)을 한 개 제공해 “모델이 필요 시 호출”하도록하면 자동 루프(에이전트 스타일)도 재현 가능.

요약

가능하고, 실전에서는 “다국어 임베딩 + 번역 병행 + 리랭킹”이 안정적입니다. 이미 구축된 영어 코퍼스에 최소 변경으로 한글 질의를 붙일 수 있고, gpt-oss 서빙과도 잘 맞습니다.

아 그리고 아까 search function을 만든다고 했잖아. 그 함수는 모든 RAG 문서를 다 뒤져 아니면 문서가 만약 100개 등록됐다고 치면 가장 관련된 10개만 추려서 10개에 대한 매칭만 해?

짧게 핵심부터: 잘 만든 search 함수는 “전부 뒤지기”가 아니라 보통 2단계로 좁힙니다. 1단계에서 문서 레벨로 상위 N개를 고르고, 2단계에서 그 문서들 안에서 청크 레벨로 상위 K개만 뽑아 LLM에 넣습니다. 보편값은 N≈5~20, K≈3~8, 그리고 최종 컨텍스트 토큰 상한으로 컷을 겁니다.

아키텍처를 단계별로 정리해볼게요.

컬렉션과 인덱스

문서 100개가 등록되어 있다면 우선 전부를 하나의 벡터 인덱스(또는 컬렉션)로 관리합니다.
문서 메타데이터(파일명, 섹션, 날짜, 언어, 권한, 태그)를 함께 저장해 필터링에 활용합니다.
하이브리드 검색 권장: 벡터(의미) + BM25(키워드)를 함께 써서 고유명사·숫자 매칭을 보완합니다.

1단계: 문서 레벨 프리셀렉션

쿼리 임베딩과 문서 대표 벡터를 비교해 상위 N개 문서를 고릅니다.
문서 대표 벡터는 보통 문서 제목+요약(abstract) 또는 상위 몇 개 청크의 평균으로 만듭니다.
필터 예: lang=en, updated_at>2025-01-01, tag in [“policy”, “spec”] 같은 조건을 먼저 걸어 탐색 공간을 줄입니다.
다중 전략 결합: 벡터 상위 N1, BM25 상위 N2를 뽑아 Reciprocal Rank Fusion 같은 방식으로 합칩니다.

2단계: 청크 레벨 정밀 검색

선택된 N개 문서 각각에서 top_k_chunks_per_doc=K로 청크를 뽑습니다.
중복·유사 청크는 Max Marginal Relevance(MMR)로 다양화합니다.
전체 모아서 재랭킹: 다국어 리랭커 또는 cross-encoder로 상위 50→최종 10개로 줄입니다.

컨텍스트 빌더

최종 청크들을 근거 메타데이터와 함께 LLM 컨텍스트로 직조합니다.
토큰 한도(max_context_tokens)에 맞춰 자르고, 가능한 한 문단 경계를 유지합니다.
규칙을 강제: 답변은 한국어, 인용은 원문 유지, 각 근거에 페이지/섹션 표기.

운영 팁

쿼리 라우팅: 분야별 컬렉션이 여러 개면 먼저 라우터(분야 분류기)로 컬렉션을 1~2개로 좁힌 뒤 위 과정을 수행합니다.
캐싱: 잦은 쿼리의 임베딩과 상위 결과를 캐시해 지연을 줄입니다.
임계값: 유사도 스코어가 임계값 미만이면 “근거 없음”으로 처리해 환각을 줄입니다.
권한: 멀티유저라면 문서별 ACL을 메타데이터 필터에 반드시 반영합니다.
실험: N, K, 임계값을 검증셋으로 튜닝합니다. 일반적으로 N을 늘리면 리콜↑, 지연/비용↑.

간단한 의사코드

입력: query, filters, top_docs=N, top_chunks_per_doc=K, max_context_tokens
출력: ranked_chunks

q_emb = embed(query)
doc_candidates = fuse(
vector_search(DOC_INDEX, q_emb, N_vec),
bm25_search(DOC_TEXTS, query, N_bm25)
)[:N]
chunk_pool = []
for d in doc_candidates:
chunk_pool += vector_search(d.CHUNK_INDEX, q_emb, K)
chunk_pool = mmr_dedup(chunk_pool)
ranked = rerank_cross_encoder(query, chunk_pool)
context = pack_by_tokens(ranked, max_context_tokens)
return context, ranked[:final_k]

질문으로 주신 두 선택지를 정리하자면

모든 RAG 문서를 다 뒤지는 일괄 전수검색은 규모가 커질수록 느려지고 비용이 큽니다.
보통은 “가장 관련된 문서 상위 N개”를 추려 그 안에서만 K개 청크를 뽑는 계층적 검색을 씁니다. 100개 문서면 상위 10개 문서 × 문서당 5개 청크 같은 구성이 현실적입니다.

마지막으로 숫자 레시피 예시

top_docs=10, top_chunks_per_doc=5, rerank_top=60→final=12, max_context_tokens=6k
하이브리드 비율: vector:bm25 = 0.7:0.3 가중 평균
임계값: cosine < 0.25면 제외, 문서 최신가중치 e^(−λ·age_days) 적용

이렇게 구성하면 정확도·지연·비용의 균형이 좋아지고, 쿼리 폭이 넓어져도 깔끔하게 확장됩니다.

FunctionTool로 너가 말한 기능을 구현하고 싶어. 대략적인 인터페이스 함수들과 동작을 잡아보자

좋아, FunctionTool로 RAG 검색을 붙일 때 쓸 “대략적 인터페이스”와 동작 흐름을 잡아볼게. vLLM+gpt-oss에서도 그대로 쓰고, OpenAI Agents SDK에도 거의 그대로 맵핑되도록 설계했어.

핵심 구성요소

인덱싱 파이프라인: 파일→텍스트→청크→임베딩→저장
검색 파이프라인: 질의→필터→문서 상위 N→청크 상위 K→재랭크→컨텍스트 패킹
FunctionTool: 모델이 호출하면 위 검색 파이프라인을 실행해 컨텍스트와 근거 메타를 반환

데이터 모델(간단)

ts

type DocMeta = {
  doc_id: string
  title: string
  path?: string
  lang?: 'en'|'ko'|'mixed'
  tags?: string[]
  created_at?: string
  updated_at?: string
  acl?: string[]            // 접근 허용 사용자/역할
}

type Chunk = {
  chunk_id: string
  doc_id: string
  content: string
  page?: number
  section?: string
  tokens?: number
}

type Scored<T> = T & { score: number }

FunctionTool 시그니처(검색 전용)
OpenAI-style JSON 스키마를 따라 설계. 툴 이름은 search_docs.

{
  "name": "search_docs",
  "description": "질문과 필터를 받아 관련 문서와 청크를 검색해 LLM 컨텍스트를 만든다.",
  "parameters": {
    "type": "object",
    "properties": {
      "query": { "type": "string", "description": "사용자 질문(ko/en 가능)" },
      "filters": {
        "type": "object",
        "properties": {
          "langs": { "type": "array", "items": { "type": "string" } },
          "tags":  { "type": "array", "items": { "type": "string" } },
          "doc_ids": { "type": "array", "items": { "type": "string" } },
          "updated_after": { "type": "string", "description": "ISO8601" }
        }
      },
      "top_docs": { "type": "integer", "default": 10, "minimum": 1 },
      "top_chunks_per_doc": { "type": "integer", "default": 5, "minimum": 1 },
      "final_chunks": { "type": "integer", "default": 12, "minimum": 1 },
      "max_context_tokens": { "type": "integer", "default": 6000, "minimum": 512 },
      "use_hybrid": { "type": "boolean", "default": true },
      "use_mmr": { "type": "boolean", "default": true },
      "use_reranker": { "type": "boolean", "default": true },
      "translate_query_to_en": { "type": "boolean", "default": false }
    },
    "required": ["query"]
  }
}

FunctionTool 반환 포맷
LLM이 바로 먹일 수 있게 컨텍스트 텍스트와 근거 리스트를 함께 줌.

ts

type SearchResult = {
  context: string                 // LLM에 바로 넣을 컨텍스트
  citations: Array<{
    doc_id: string
    title?: string
    page?: number
    section?: string
    snippet: string
    score: number
  }>
  diagnostics?: {
    used_docs: Array<Scored<DocMeta>>
    used_chunks: Array<Scored<Chunk>>
    query_lang?: 'ko'|'en'|'unknown'
    translated_query?: string
  }
}

동작 의사코드

def search_docs(query, filters=None, top_docs=10, top_chunks_per_doc=5,
                final_chunks=12, max_context_tokens=6000,
                use_hybrid=True, use_mmr=True, use_reranker=True,
                translate_query_to_en=False) -> SearchResult:

    # 0) 권한/필터링 프레딕트
    candidate_docs = filter_docs(filters)  # lang/tags/doc_ids/updated_after/ACL 반영

    # 1) 쿼리 준비
    lang = detect_lang(query)  # ko/en 추정
    q_for_vec = query
    if translate_query_to_en and lang == 'ko':
        q_for_vec = translate_ko_to_en(query)  # 선택 기능

    # 2) 문서 레벨 프리셀렉션
    doc_scores_vec = vector_search_docs(candidate_docs, q_for_vec, top=top_docs*3)
    doc_scores_kw  = bm25_search_docs(candidate_docs, query, top=top_docs*3) if use_hybrid else []
    doc_candidates = fuse_scores(doc_scores_vec, doc_scores_kw)[:top_docs]

    # 3) 청크 레벨 정밀 검색
    chunk_pool = []
    for doc in doc_candidates:
        chunks_vec = vector_search_chunks(doc.doc_id, q_for_vec, top=top_chunks_per_doc*3)
        chunks_kw  = bm25_search_chunks(doc.doc_id, query, top=top_chunks_per_doc*2) if use_hybrid else []
        chunk_pool.extend(fuse_scores(chunks_vec, chunks_kw))

    if use_mmr:
        chunk_pool = mmr_diversify(chunk_pool, lambda c: c.content, top=final_chunks*4)

    # 4) 재랭킹
    ranked = chunk_pool
    if use_reranker:
        ranked = cross_encoder_rerank(query, chunk_pool)  # 상위 50→정렬

    # 5) 컨텍스트 패킹
    selected = take_until_token_budget(ranked, final_chunks, max_context_tokens)
    context = build_context(selected)  # [Doc:Title p.X] + 본문… 형태

    # 6) 인용/디버그 정보
    citations = [{
        "doc_id": c.doc_id,
        "title": get_title(c.doc_id),
        "page": c.page,
        "section": c.section,
        "snippet": shorten(c.content, 300),
        "score": float(c.score)
    } for c in selected]

    diags = {
        "used_docs": doc_candidates,
        "used_chunks": selected,
        "query_lang": lang,
        "translated_query": q_for_vec if q_for_vec != query else None
    }

    return {"context": context, "citations": citations, "diagnostics": diags}

컨텍스트 빌드 규칙(요약)

문서/페이지/섹션 메타를 명시
청크 사이에 구분자
원문이 영어라도 그대로 두고, 최종 답변 언어는 LLM 시스템 프롬프트로 강제

[Source] Title="Spec ABC", DocID=doc_123, Page=5, Section="3.2"
"Chunk content here..."

[Source] Title="Spec XYZ", DocID=doc_987, Page=12
"Another chunk..."

Agents SDK에 등록 예시

from openai import OpenAI
client = OpenAI()

def search_docs_tool(params):
    return search_docs(**params)

tools = [{
  "type": "function",
  "function": {
    "name": "search_docs",
    "description": "문서/청크 검색",
    "parameters": { ... 위 JSON 스키마 ... }
  }
}]

messages = [
  {"role":"system","content":"답변은 한국어로. 반드시 search_docs를 이용해 근거를 확인하고, 인용을 포함해라."},
  {"role":"user","content":"영어로 된 제품 보증 조건 요약해줘"}
]

resp = client.chat.completions.create(
  model="gpt-4o-mini",
  messages=messages,
  tools=tools
)
# tool_calls 감지 → search_docs_tool 실행 → 결과를 tool 메시지로 첨부 → 모델 재호출

vLLM+gpt-oss 커스텀 루프

vLLM은 툴 호출 오케스트레이션이 없으니, 모델 출력의 function_call(JSON)을 직접 파싱해서 search_docs를 호출한 뒤 결과를 다시 모델에 넣는 루프를 구현하면 된다.
또는 “수동 모드”로, 사용자 질문이 오면 서버에서 search_docs를 먼저 실행해 context를 만들고, 그걸 user 메시지에 붙여 모델을 한 번만 호출하는 방식도 간단하고 튼튼하다.

인덱싱용 보조 툴(선택)
파일 업로드/재빌드/상태 확인을 위해 추가 FunctionTool을 두면 운영이 편하다.

upsert_documents(files, tags, lang_hint)
rebuild_index(scope='all' 'doc_id')
describe_doc(doc_id)
purge_docs(doc_ids)

최소 운영 파라미터의 추천 기본값

top_docs=10
top_chunks_per_doc=5
final_chunks=12
max_context_tokens=6000
use_hybrid=true, use_mmr=true, use_reranker=true
translate_query_to_en=false (다국어 임베딩이면 불필요, 영어 전용 임베딩이면 true)

여기까지가 견고한 골격이야. 다음 단계로는 너희가 쓰는 임베딩/벡터DB를 가정해서 search_docs 내부의 vector_search_*와 bm25, reranker, 패킹 함수를 실코드로 메워 넣으면 바로 붙일 수 있다.

Database를 만들기도 해야해. 이를 위해 문서를 DB에 추가하거나 하는 것도 필요할 것 같은데

좋아, RAG용 Database까지 포함한 “문서 수집·저장·인덱싱·검색” 풀사이클 인터페이스를 잡아보자. 운영을 염두에 두고 설계하면 나중에 안 아파요.

핵심 구조 개요

원본 저장소: 업로드한 파일을 보관하는 Object Storage(S3, GCS, MinIO 등)
메타·텍스트 DB: 문서/청크/태그/ACL/상태를 관리하는 RDB(PostgreSQL 권장)
벡터 인덱스: pgvector 또는 외부 벡터DB(FAISS, Milvus, Chroma, Qdrant)
잡 큐: 임베딩·재인덱싱 같은 비동기 작업용(예: Celery, RQ, Sidekiq, Temporal)

권장 구현 프로파일
PostgreSQL + pgvector 한 방으로 끝내기
장점: 트랜잭션, 조인, 필터링, 권한체크가 깔끔. 운영 단순.
대안: 텍스트는 Postgres, 벡터는 FAISS/Milvus로 분리. 성능 튜닝 자유도↑.

데이터 모델 스키마(Postgres + pgvector)
SQL DDL 예시. 임베딩 차원은 예로 1024를 사용.

sql

create extension if not exists vector;

create table documents (
  doc_id          uuid primary key,
  title           text not null,
  source_uri      text,                 -- S3 키, GCS 경로 등
  mime_type       text,
  lang            text,                 -- 'en','ko','mixed' 등
  tags            text[],
  bytes           bigint,
  pages           int,
  created_at      timestamptz default now(),
  updated_at      timestamptz default now(),
  status          text not null default 'ready',  -- 'ready','processing','error','disabled'
  acl_roles       text[] default array[]::text[], -- 접근 허용 역할
  acl_users       text[] default array[]::text[]  -- 접근 허용 사용자
);

create table chunks (
  chunk_id        uuid primary key,
  doc_id          uuid references documents(doc_id) on delete cascade,
  page            int,
  section         text,
  content         text not null,
  token_count     int,
  created_at      timestamptz default now(),
  updated_at      timestamptz default now()
);

-- 하나의 컬럼에 다국어 임베딩을 저장(권장). 필요하면 영어 전용 추가 컬럼도.
create table chunk_embeddings (
  chunk_id        uuid primary key references chunks(chunk_id) on delete cascade,
  emb_mtext       vector(1024),         -- 다국어 임베딩
  emb_en          vector(1024)          -- 선택: 영어 전용 임베딩
);

-- BM25 대체용 전체검색 인덱스
create index idx_chunks_fts on chunks using gin (to_tsvector('simple', content));

-- 태그/언어/상태/업데이트일 필터링
create index idx_documents_tags on documents using gin (tags);
create index idx_documents_lang on documents (lang);
create index idx_documents_status on documents (status);
create index idx_documents_updated on documents (updated_at);

-- 벡터 인덱스(IVFFlat 예시). 규모 커지면 lists 파라미터 튜닝.
create index idx_chunk_emb_mtext on chunk_embeddings using ivfflat (emb_mtext vector_cosine_ops) with (lists = 200);
create index idx_chunk_emb_en    on chunk_embeddings using ivfflat (emb_en    vector_cosine_ops) with (lists = 200);

스토리지 계층
원문 파일은 RDB에 넣지 말고 Object Storage에 넣고 documents.source_uri로 연결. 썸네일·텍스트 추출 캐시를 두면 속도가 빨라짐.

FunctionTool 인터페이스 설계
검색뿐 아니라 업서트·재인덱싱·관리까지 FunctionTool로 노출하면 LLM 에이전트나 관리용 UI에서 자동화하기 좋다.

문서 업서트
이 함수는 DB와 스토리지를 갱신하고, 비동기 인덱싱 잡을 발행.

{
  "name": "upsert_documents",
  "description": "문서를 DB와 스토리지에 추가/갱신하고 인덱싱 잡을 큐에 넣는다.",
  "parameters": {
    "type": "object",
    "properties": {
      "items": {
        "type": "array",
        "items": {
          "type": "object",
          "properties": {
            "title": { "type": "string" },
            "source_uri": { "type": "string" },
            "mime_type": { "type": "string" },
            "lang_hint": { "type": "string" },
            "tags": { "type": "array", "items": { "type": "string" } },
            "acl_roles": { "type": "array", "items": { "type": "string" } },
            "acl_users": { "type": "array", "items": { "type": "string" } },
            "doc_id": { "type": "string", "description": "갱신 시 지정" }
          },
          "required": ["title", "source_uri"]
        }
      },
      "reindex": { "type": "boolean", "default": true }
    },
    "required": ["items"]
  }
}

동작

source_uri 유효성 확인 및 메타 추출(bytes, pages 등)
documents upsert
상태를 processing으로 바꾸고 인덱싱 잡 enqueue
인덱싱 워커(비동기 잡)
추출→청크→임베딩→저장 파이프라인. 실패 시 documents.status='error'와 메시지 저장.
의사코드

def worker_index_document(doc_id):
    doc = db.get_document(doc_id)
    raw = storage.download(doc.source_uri)
    text, pages = extract_text(raw)             # pypdf/unstructured/docling
    chunks = chunking(text)                     # 200~400 토큰 권장
    db.delete_chunks(doc_id)                    # 전체 재빌드 시
    for ch in chunks:
        chunk_id = uuid4()
        db.insert_chunk(chunk_id, doc_id, ch.page, ch.section, ch.content, ch.tokens)
    # 임베딩
    vecs_m = embed_multilingual([c.content for c in chunks])
    vecs_en = None  # 필요하면 번역 후 또는 영어전용 임베딩
    db.insert_embeddings([ (chunks[i].id, vecs_m[i], vecs_en[i] if vecs_en else None) for i in range(len(chunks)) ])
    db.update_document_status(doc_id, 'ready')

문서 상태 조회/설명

{
  "name": "describe_doc",
  "description": "문서 메타데이터와 인덱싱 상태를 조회한다.",
  "parameters": {
    "type": "object",
    "properties": { "doc_id": { "type": "string" } },
    "required": ["doc_id"]
  }
}

반환은 title, lang, tags, updated_at, status, chunk_count 등.

선택적 재빌드

{
  "name": "rebuild_index",
  "description": "문서 또는 전체 컬렉션의 인덱스를 재생성한다.",
  "parameters": {
    "type": "object",
    "properties": {
      "scope": { "type": "string", "enum": ["all","doc"], "default": "doc" },
      "doc_id": { "type": "string" }
    },
    "required": ["scope"]
  }
}

삭제

{
  "name": "purge_docs",
  "description": "문서를 완전히 삭제한다(청크, 임베딩 포함).",
  "parameters": {
    "type": "object",
    "properties": { "doc_ids": { "type": "array", "items": { "type": "string" } } },
    "required": ["doc_ids"]
  }
}

검색 툴(앞서 정의한 search_docs)
필터에 tags/lang/updated_after/acl 포함. top_docs, top_chunks_per_doc, final_chunks로 폭을 제어.

권한(ACL) 처리
검색 쿼리에서 반드시 다음을 적용.

로그인 유저의 roles, user_id를 받아 documents.acl_roles/acl_users에 매칭
필터 단계에서 미리 제외
감사 로그: 누가 어떤 doc_id를 근거로 보았는지 기록

성능 팁

토큰 예산 우선: final_chunks는 8~16 사이, 총 4~8k 토큰 내에서 패킹
인덱스 튜닝: IVFFlat lists, probes 조정. pgvector의 SET ivfflat.probes=값
하이브리드: to_tsvector(simple)로 키워드 인덱스. 고유명사/숫자 강화
캐싱: 쿼리→상위 doc_ids, chunk_ids 캐시. 임베딩 결과도 메모리 캐시
최신 가중치: score = α·cosine + β·bm25 + γ·recency_boost

간단 API 흐름 예시

클라이언트가 파일 업로드 → 백엔드가 S3에 저장 후 upsert_documents 호출
워커가 인덱싱 완료 → 문서 status=ready
사용자가 질문 → search_docs 호출 → 컨텍스트와 citations 반환
LLM 호출 시 컨텍스트 삽입 → 한국어 답변 생성

서버 사이드 의사코드 스니펫
문서 업서트

def upsert_documents(items, reindex=True):
    results = []
    for it in items:
        doc_id = it.get("doc_id") or uuid4()
        db.upsert_document(
            doc_id=doc_id,
            title=it["title"],
            source_uri=it["source_uri"],
            mime_type=it.get("mime_type"),
            lang=it.get("lang_hint"),
            tags=it.get("tags", []),
            acl_roles=it.get("acl_roles", []),
            acl_users=it.get("acl_users", []),
            status="processing" if reindex else "ready"
        )
        if reindex:
            queue.enqueue("worker_index_document", doc_id=doc_id)
        results.append({"doc_id": str(doc_id), "queued": reindex})
    return {"items": results}

검색 핵심 쿼리(pgvector)

sql

-- 다국어 임베딩으로 top N 청크
select c.chunk_id, c.doc_id, c.page, c.section, c.content,
       1 - (ce.emb_mtext <=> :qvec) as score
from chunk_embeddings ce
join chunks c on c.chunk_id = ce.chunk_id
join documents d on d.doc_id = c.doc_id
where d.status = 'ready'
  and (:lang is null or d.lang = :lang)
  and (:tags is null or d.tags && :tags::text[])
  and (:user_id = any(d.acl_users) or array_length(d.acl_users,1) is null
       or :role && d.acl_roles)
order by ce.emb_mtext <=> :qvec
limit :k;

운영 체크리스트
문서 상태 전이: ready ↔ processing ↔ error
중복 방지: 같은 source_uri+hash로 idempotent 업서트
관찰성: 인덱싱 시간, 청크 수, 임베딩 시간, 검색 지연, 히트율 지표 수집
재현성: 임베딩 모델 버전 저장. 교체 시 롤링 재인덱싱

정리

문서 메타·청크·임베딩을 RDB로 명확히 분리하고, 파일은 Object Storage에 둔다.
FunctionTool로 upsert/rebuild/describe/purge/search를 노출해 LLM이 필요한 순간에만 호출하게 한다.
비동기 워커를 두어 인덱싱을 오프로드한다.
pgvector 하나로도 충분히 단단한 RAG DB를 만들 수 있다.

다음으로 원하시면 이 스키마를 기준으로 실제 Python 저장소 레이어(ORM), 워커(Celery), 그리고 search_docs 내부 SQL까지 꽂아 넣은 미니 레퍼런스를 만들어줄게.

임베딩 등은 그냥 유저의 로컬 파일 시스템에 저장한다고 가정하자.

좋아. “임베딩과 인덱스까지 모두 로컬 파일 시스템에 저장”한다는 가정으로, 풀사이클 인터페이스와 디렉터리 구조, 파일 포맷, FunctionTool 시그니처, 동작 스케치를 잡아줄게. 외부 DB 없이도 단단히 돌아가도록 설계했어. 필요 라이브러리는 pypdf/unstructured, sentence-transformers(or bge/e5), faiss, sqlite3 정도로 가정한다.

설계 요약

스토리지 계층은 전부 로컬 디렉터리. 원본 파일과 파생 산출물(텍스트, 청크, 임베딩, 인덱스)을 규칙적으로 저장.
메타데이터는 SQLite 한 파일로 관리해 조회·필터·ACL가 빠름. 대안으로 JSONL만 써도 되지만 규모가 커지면 느려짐.
벡터 검색은 FAISS를 사용하고 인덱스 파일로 영속화. 키워드 검색은 SQLite FTS5로 대체.
FunctionTool은 업서트, 재인덱싱, 검색, 설명, 삭제까지 한 세트로 노출.

디렉터리 구조 예시
rag_store/
blobs/ 원본 파일 보관
{doc_id}/{filename}
texts/ 텍스트 추출 캐시
{doc_id}.txt
chunks/ 청크 JSONL
{doc_id}.jsonl [{"chunk_id","page","section","content","tokens"}…]
embeddings/ 임베딩 파일
mtext/{doc_id}.npy float32 [num_chunks, dim]
en/{doc_id}.npy 선택: 영어 전용
indices/ 전체 컬렉션용 인덱스
faiss_mtext.index
faiss_en.index
manifest/ 버전·모델 추적
embedding_manifest.json [{"model":"bge-m3","dim":1024,"built_at":…}]
locks/
index.lock
rag_meta.sqlite SQLite 메타DB(문서·청크·ACL·FTS5)

SQLite 스키마 요약
documents(doc_id pk, title, source_path, mime, lang, tags json, bytes, pages, status, acl_users json, acl_roles json, created_at, updated_at, embed_model_mtext, embed_model_en)
chunks(chunk_id pk, doc_id fk, page, section, content, tokens)
fts_chunks(content) USING FTS5(content, content='chunks', content_rowid='rowid')
인덱스: documents(status, updated_at), chunks(doc_id)

FunctionTool 인터페이스

upsert_documents
name: upsert_documents
description: 로컬 파일을 등록하고 인덱싱 잡을 트리거한다.
parameters:
items: [{title, source_path, lang_hint?, tags?, acl_users?, acl_roles?, doc_id?}]
reindex: boolean = true
rebuild_index
name: rebuild_index
description: 전체 또는 특정 문서의 임베딩·FAISS 인덱스를 재생성한다.
parameters:
scope: "all"|"doc"
doc_id?: string
which: "mtext"|"en"|"both" = "mtext"
describe_doc
name: describe_doc
description: 문서 메타·상태·청크 수를 조회한다.
parameters:
doc_id: string
purge_docs
name: purge_docs
description: 문서를 로컬 스토리지와 메타DB, 인덱스에서 완전 삭제한다.
parameters:
doc_ids: string[]
search_docs
name: search_docs
description: 쿼리와 필터로 문서 상위 N → 청크 상위 K → 재랭크 후 컨텍스트 빌드
parameters:
query: string
filters?: { langs?: string[], tags?: string[], doc_ids?: string[], updated_after?: string }
top_docs?: int = 10
top_chunks_per_doc?: int = 5
final_chunks?: int = 12
max_context_tokens?: int = 6000
use_hybrid?: boolean = true 벡터 + FTS5
use_mmr?: boolean = true
use_reranker?: boolean = true 크로스 인코더가 있으면 사용
translate_query_to_en?: boolean = false

주요 동작 스케치

A) 업서트와 인덱싱

upsert_documents(items):
a. 파일 존재 확인 → rag_store/blobs/{doc_id}/에 복사
b. rag_meta.sqlite의 documents에 upsert, status='processing'
c. 인덱싱 워커 트리거(worker_index_document(doc_id))
worker_index_document(doc_id):
a. 텍스트 추출 → rag_store/texts/{doc_id}.txt
b. 청크 생성 → rag_store/chunks/{doc_id}.jsonl 저장, chunks 테이블 교체 삽입
c. 임베딩 생성 → rag_store/embeddings/mtext/{doc_id}.npy 저장
d. SQLite FTS5 동기화: chunks를 fts_chunks에 삽입
e. documents.status='ready', updated_at=now
f. 글로벌 FAISS 인덱스 갱신: indices/faiss_mtext.index 로드 → add_with_ids → 저장

B) 검색
search_docs(query, …):

쿼리 전처리와 임베딩 준비(필요시 ko→en 번역)
문서 레벨 프리셀렉션
a. FTS5 문서 후보: SELECT doc_id FROM fts_chunks WHERE fts_chunks MATCH ? LIMIT N_bm25
b. FAISS 후보: 쿼리 임베딩으로 topN 청크 검색 → doc_id 집계 상위 N_vec
c. 두 결과를 RRF로 합쳐 상위 top_docs 도출
청크 레벨 정밀 검색
a. 선택 문서별로 top_chunks_per_doc씩 FAISS 검색
b. 필요시 FTS5 결과도 병합
c. MMR로 다양화, 크로스 인코더 재랭크
컨텍스트 패킹
a. final_chunks 개수와 max_context_tokens 안에서 패킹
b. citations에 doc_id, title, page, section, score 포함
반환: {context, citations, diagnostics}

C) 삭제와 재인덱싱
purge_docs: 파일 디렉터리, 텍스트, 청크 jsonl, npy 제거 → SQLite에서 문서·청크 삭제 → FAISS에서 해당 ID 삭제 후 저장
rebuild_index: scope='all'이면 임베딩과 FAISS를 제로부터 리빌드, which로 mtext/en 선택

파일 포맷과 저장 방식
임베딩 저장: NumPy .npy, dtype=float32, shape=[num_chunks, dim]
FAISS 인덱스: IVF_FLAT 또는 HNSW. index.reset(); index.add_with_ids(vecs, global_ids) 후 write_index
글로벌 ID: 64-bit 정수. 권장 매핑은 hash64("{doc_id}:{chunk_id}")로 만들고 SQLite에 gid 컬럼으로 저장
청크 JSONL: 한 줄에 한 청크. 대용량 append/replace가 쉬워 운영 간단

간단 코드 스니펫들

임베딩 npy 저장/로드
import numpy as np

def save_embeddings_npy(path, matrix):
path.parent.mkdir(parents=True, exist_ok=True)
np.save(path, matrix.astype(np.float32))

def load_embeddings_npy(path):
return np.load(path).astype(np.float32)

FAISS 인덱스 로드/저장/갱신
import faiss
from pathlib import Path

def load_faiss(path, dim):
if Path(path).exists():
return faiss.read_index(str(path))
index = faiss.index_factory(dim, "IVF4096,Flat", faiss.METRIC_INNER_PRODUCT)
quantizer = faiss.IndexFlatIP(dim)
index_ivf = faiss.IndexIVFFlat(quantizer, dim, 4096, faiss.METRIC_INNER_PRODUCT)
return index if isinstance(index, faiss.Index) else index_ivf

def save_faiss(index, path):
Path(path).parent.mkdir(parents=True, exist_ok=True)
faiss.write_index(index, str(path))

def add_embeddings(index, vecs, ids):
faiss.normalize_L2(vecs)
if hasattr(index, "is_trained") and not index.is_trained:
index.train(vecs)
index.add_with_ids(vecs, ids)

SQLite FTS5 삽입
import sqlite3, json

def upsert_chunks_sqlite(db_path, doc_id, chunks):
conn = sqlite3.connect(db_path)
c = conn.cursor()
c.execute("DELETE FROM chunks WHERE doc_id=?", (doc_id,))
for ch in chunks:
c.execute("INSERT INTO chunks(chunk_id,doc_id,page,section,content,tokens) VALUES(?,?,?,?,?,?)",
(ch["chunk_id"], doc_id, ch.get("page"), ch.get("section"), ch["content"], ch.get("tokens",0)))
c.execute("INSERT INTO fts_chunks(rowid, content) SELECT rowid, content FROM chunks WHERE doc_id=?", (doc_id,))
conn.commit()
conn.close()

검색 파이프라인 핵심
def search_pipeline(query, top_docs=10, top_chunks_per_doc=5, final_chunks=12):

1) 쿼리 임베딩

q_vec = embed_mtext([query])[0].astype("float32")
faiss.normalize_L2(q_vec.reshape(1,-1))

2) 글로벌 청크에서 topN 후보

D, I = faiss_index.search(q_vec.reshape(1,-1), 200)
chunk_candidates = gather_chunks_by_ids(I[0], D[0]) # SQLite에서 메타 로드

3) 문서 집계 → 상위 top_docs

doc_scores = aggregate_by_doc(chunk_candidates)
doc_top = take_top(doc_scores, top_docs)

4) 문서별 재검색

refined = []
for d in doc_top:
refined += search_chunks_in_doc(d.doc_id, q_vec, top_chunks_per_doc)
refined = mmr_diversify(refined, k=final_chunks*4)
ranked = rerank_cross_encoder(query, refined)[:final_chunks]

5) 컨텍스트 빌드

context, citations = build_context_and_citations(ranked)
return {"context": context, "citations": citations}

운영·안전 포인트
락킹: indices/index.lock 파일로 파일 잠금 후 인덱스 저장.
원자성: 인덱스 저장은 tmp에 쓴 뒤 rename으로 교체.
버전관리: manifest/embedding_manifest.json에 모델명, 차원, 토큰라이저 버전 기록.
권한: documents.acl_users/acl_roles를 검색 전 필터에 반영.
복구: 청크 JSONL → SQLite 재구축, npy → FAISS 재빌드가 가능하도록 유지.

추천 기본값
embedding 모델: bge-m3, dim=1024
청크 크기: 200~400 토큰, overlap 15%
faiss nprobe: 8~32, IVF lists: 4096부터 시작
검색 폭: top_docs=10, top_chunks_per_doc=5, final_chunks=12, max_context_tokens=6000
하이브리드: use_hybrid=true, FTS5 가중치 0.3, 벡터 0.7

정리
로컬 파일 시스템만으로도 문서 등록부터 인덱싱, 검색까지 충분히 안정적인 RAG 스토어를 만들 수 있다. 핵심은 규칙적인 디렉터리 구조, SQLite+FTS5로의 메타·키워드 처리, FAISS 인덱스의 안전 저장, 그리고 FunctionTool 세트를 통한 일관된 오케스트레이션이다. 다음 단계로 원하면 이 스펙을 바로 실행 가능한 최소 Python 패키지 구조와 CLI까지 포함해 뼈대 코드를 만들어줄 수 있다.

유저가 사용할 땐 어떤 순서로 해야해?

다음 순서대로 쓰면 됩니다. 한 번 익히면 흐름이 늘 같습니다: 등록 → 인덱싱 → 검색 → 답변.

최초 1회 설정

로컬 디렉터리 준비: rag_store/ 하위에 blobs, texts, chunks, embeddings, indices, manifest, locks와 rag_meta.sqlite 생성.
임베딩/인덱스 초기화: embedding 모델 선택(bge-m3 등), FAISS 인덱스 파일 생성(indices/faiss_mtext.index).

문서 등록

파일을 준비하고 upsert_documents 함수(또는 API/CLI)를 호출합니다.
예시(Python/FunctionTool 호출 파라미터)
{
"items": [
{
"title": "spec_A.pdf",
"source_path": "/path/spec_A.pdf",
"lang_hint": "en",
"tags": ["spec","v1"]
}
],
"reindex": true
}
내부 동작: 파일을 blobs/{doc_id}/로 복사하고 documents 테이블에 기록, status='processing', 인덱싱 잡을 큐에 넣음.

인덱싱 자동 처리

워커가 텍스트 추출 → 청크 → 임베딩(npy) 저장 → FAISS 인덱스 갱신 → FTS5 업데이트 → status='ready'로 전환.
필요하면 describe_doc(doc_id)로 상태 확인. 청크 수, 페이지 수, 임베딩 모델 버전 등을 바로 확인 가능.

검색과 질의응답

사용자가 질문을 입력하면, 애플리케이션이 먼저 search_docs를 호출해 컨텍스트를 확보합니다.
권장 파라미터 예:
{
"query": "반품 정책 핵심이 뭐야?",
"filters": { "langs": ["en"], "tags": ["policy"] },
"top_docs": 10,
"top_chunks_per_doc": 5,
"final_chunks": 12,
"use_hybrid": true,
"use_mmr": true,
"use_reranker": true,
"max_context_tokens": 6000
}
search_docs가 반환한 context와 citations를 LLM에 넣어 답변 생성.
한글 질의로 영어 문서를 찾으려면 다국어 임베딩 또는 translate_query_to_en=true를 사용.

운영 중 문서 추가/갱신

새 파일 추가: upsert_documents(items=[…], reindex=true).
기존 문서 교체: 같은 doc_id로 upsert하거나 source_path를 업데이트 후 reindex.
여러 파일 배치 등록도 동일. 상태가 ‘ready’가 되면 바로 검색에 포함됩니다.

재인덱싱과 청소

임베딩 모델 교체나 청크 규칙 변경 시 rebuild_index로 전체 또는 특정 문서를 재빌드.
{
"scope": "all",
"which": "mtext"
}
문서 삭제는 purge_docs로 처리. 로컬 파일, 텍스트, 청크, 임베딩, 메타DB, FAISS 참조까지 정리.

권한과 필터링

사용자별로 roles/user_id를 애플리케이션이 전달하고, search_docs가 ACL을 반영해 필터링.
태그, 언어, 수정일(updated_after)로 검색 범위를 줄여 지연을 안정화.

장애/품질 관리 팁

status='error'일 때 describe_doc로 에러 메시지 확인 후 재시도.
검색 품질 저하 시 이 순서로 점검: 청크 크기(200~400 토큰), 임베딩 모델 일치, 하이브리드 검색 활성화, 리랭커 적용, MMR 다양화, 최신 가중치 적용.
지연이 느리면 top_docs, top_chunks_per_doc, final_chunks를 줄이고, FAISS nprobe 튜닝.

사용자 경험 관점의 최소 흐름(진짜 실무용 압축본)

파일 드래그·드롭 또는 업로드 → upsert_documents(reindex=true)
잠시 후 준비 완료 → describe_doc로 ready 확인
질문 입력 → 백엔드가 search_docs 호출 → context 수집
LLM 호출 → 한국어 답변과 출처 표시
추가 파일이 생기면 1로 돌아감

API/CLI 예시 한 줄씩

등록: cli upsert –path /docs/spec.pdf –title "spec" –lang en –tag spec –reindex
상태: cli describe –doc-id
검색: cli search –q "refund policy" –lang en –top-docs 10 –top-chunks 5
재빌드: cli rebuild –scope all
삭제: cli purge –doc-ids ,

이 흐름을 앱에 녹이면 최종 사용자는 “문서 업로드 → 질문”만 알면 되고, 나머지 전처리·검색·컨텍스트 패킹은 백엔드가 FunctionTool로 수행합니다. 다음 단계로 원하면, 위 순서를 그대로 수행하는 REST 라우트 또는 간단한 CLI 스켈레톤을 만들어 줄게.