Phase 8은 이를 위해 달려왔습니다. 토큰화 (Tokenization). 임베딩 (Embeddings). 트랜스포머 (Transformers). BERT. GPT. HuggingFace. 미세 조정 (Fine-tuning). 벡터 검색 (Vector search). RAG. 챗봇 아키텍처 (Chatbot architecture). OpenAI API. Claude API. 14개의 기초 포스트가 있었습니다. 이제 여러분은 진짜 결과물을 출시합니다. 이번 캡스톤 프로젝트에서는 DocuMind를 구축합니다. DocuMind는 사용자가 어떤 PDF나 텍스트 문서를 업로드하더라도 그에 대해 지능적인 대화를 나눌 수 있게 해주는 AI 연구 보조 도구입니다. 이 도구는 관련 구절을 검색하고, 출처를 인용하며, 대화 문맥을 유지하고, 공유 가능한 웹 애플리케이션으로 배포됩니다. 지금까지 만들어진 시스템 중 가장 복잡한 시스템은 아닙니다. 하지만 Phase 8의 모든 개념을 요구할 만큼 충분히 복잡하면서도, 하나의 포스트 안에서 완전히 구축할 수 있을 만큼 충분히 단순합니다.
여러분이 구축하게 될 것
DocuMind: AI 연구 보조 도구
주요 기능:
✓ PDF 및 텍텍스트 파일 업로드
✓ 자동 청킹 (Chunking) 및 임베딩 (Embedding)
✓ 업로드된 문서에 대한 시맨틱 검색 (Semantic search)
✓ 메모리를 활용한 멀티턴 대화 (Multi-turn conversation)
✓ 모든 답변에 대한 출처 인용
✓ 다중 문서 지원
✓ 배포 가능한 Streamlit 앱
기술 스택:
- sentence-transformers (임베딩)
- chromadb (벡터 저장소)
- anthropic / openai (LLM)
- streamlit (UI)
- pypdf (PDF 파싱)
- langchain text splitter (청킹)
프로젝트 구조
documind/
├── app.py # Streamlit 애플리케이션 진입점
├── config.py # 설정 및 상수
├── document_processor.py # PDF/텍스트 로딩, 청킹, 임베딩
├── knowledge_base.py # ChromaDB 벡터 저장소 래퍼 (wrapper)
├── conversation.py # 멀티턴 대화 관리
├── llm_client.py # 교체 가능한 LLM 백엔드 (Claude 또는 OpenAI)
├── rag_engine.py # 검색 증강 생성 (RAG) 파이프라인
├── requirements.txt # 모든 의존성
└── README.md # 설정 및 사용 지침
config.py
config.py
from dataclasses import dataclass
@dataclass
class Config :
# 임베딩 모델 (Embedding model)
EMBED_MODEL : str = " all-MiniLM-L6-v2 "
EMBED_DIM : int = 384
# 청킹 (Chunking)
CHUNK_SIZE : int = 512
CHUNK_OVERLAP : int = 64
# 검색 (Retrieval)
TOP_K_RETRIEVE : int = 5
TOP_K_RERANK : int = 3
MIN_SCORE : float = 0.25
# 대화 (Conversation)
MAX_HISTORY_TURNS : int =
8 # LLM LLM_PROVIDER : str = " anthropic " # "anthropic" or "openai" ANTHROPIC_MODEL : str = " claude-3-5-haiku-20241022 " OPENAI_MODEL : str = " gpt-4o-mini " MAX_TOKENS : int = 800 TEMPERATURE : float = 0.2 # ChromaDB CHROMA_PATH : str = " ./chroma_db " COLLECTION_NAME : str = " documind " CONFIG = Config () document_processor.py # document_processor.py import re from pathlib import Path from typing import List , Dict from dataclasses import dataclass @dataclass class Chunk : text : str source : str page : int chunk_idx : int char_start : int def load_text_file ( filepath : str ) -> str : with open ( filepath , " r " , encoding = " utf-8 " , errors = " replace " ) as f : return f . read () def load_pdf ( filepath : str ) -> List [ Dict ]: """ {text, page} 딕셔너리 리스트를 반환합니다. """ try : import pypdf pages = [] with open ( filepath , " rb " ) as f : reader = pypdf . PdfReader ( f ) for i , page in enumerate ( reader . pages ): text = page . extract_text () or "" if text . strip (): pages . append ({ " text " : text , " page " : i + 1 }) return pages except ImportError : raise ImportError ( " 설치: pip install pypdf " ) def clean_text ( text : str ) -> str : text = re . sub ( r " \s+ " , " " , text ) text = re . sub ( r " (\n\s*){3,} " , " \n\n " , text ) return text . strip () def chunk_text ( text : str , source : str , page : int = 1 , chunk_size : int = 512 , overlap : int = 64 ) -> List [ Chunk ]: """ 오버랩을 사용한 재귀 문자 분할입니다. """ text = clean_text ( text ) words = text . split () chunks = [] idx = 0 step = chunk_size - overlap while idx < len ( words ): chunk_words = words [ idx : idx + chunk_size ] chunk_text = " " . join ( chunk_words ) if len ( chunk_text . strip ()) > 50 : chunks .
append(Chunk(text=chunk_text, source=source, page=page, chunk_idx=len(chunks), char_start=idx)) idx += step
return chunks
def process_document(filepath: str, chunk_size: int = 512, overlap: int = 64) -> List[Chunk]: """ Load and chunk any supported document type. """
path = Path(filepath)
all_chunks = []
if path.suffix.lower() == ".pdf":
pages = load_pdf(filepath)
for page_data in pages:
chunks = chunk_text(page_data["text"], source=path.name, page=page_data["page"], chunk_size=chunk_size, overlap=overlap)
all_chunks.extend(chunks)
elif path.suffix.lower() in [".txt", ".md", ".rst"]:
text = load_text_file(filepath)
chunks = chunk_text(text, source=path.name, page=1, chunk_size=chunk_size, overlap=overlap)
all_chunks.extend(chunks)
else:
raise ValueError(f"Unsupported file type: {path.suffix}")
print(f"Processed {path.name}: {len(all_chunks)} chunks")
return all_chunks
knowledge_base.py # knowledge_base.py import uuid from typing import List, Dict, Optional from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings from document_processor import Chunk class KnowledgeBase :
def init(self, embed_model: str, persist_dir: str, collection_name: str):
self.embedder = SentenceTransformer(embed_model)
self.client = chromadb.PersistentClient(path=persist_dir)
self.collection = self.client.get_or_create_collection(name=collection_name, metadata={'hnsw:space': 'cosine'})
self._source_cache = {}
def add_chunks(self, chunks: List[Chunk]) -> int:
if not chunks: return 0
texts = [c.text for c in chunks]
embeddings = self.embedder.encode(texts, show_progress_bar=True).tolist()
ids = [str(uuid.uuid4()) for _ in chunks]
metadatas = [{ "source": c.source, "page": c.page, "chunk_idx": c.chunk_idx} for c in chunks]
chunk_idx } for c in chunks ] self . collection . add ( ids = ids , embeddings = embeddings , documents = texts , metadatas = metadatas ) for chunk in chunks : self . _source_cache [ chunk . source ] = True return len ( chunks ) def retrieve ( self , query : str , top_k : int = 5 , min_score : float = 0.25 , filter_source : Optional [ str ] = None ) -> List [ Dict ]: query_emb = self . embedder . encode ([ query ]). tolist () where = { " source " : filter_source } if filter_source else None results = self . collection . query ( query_embeddings = query_emb , n_results = top_k , where = where , include = [ " documents " , " metadatas " , " distances " ] ) retrieved = [] for text , meta , dist in zip ( results [ " documents " ][ 0 ], results [ " metadatas " ][ 0 ], results [ " distances " ][ 0 ] ): score = 1 - dist if score >= min_score : retrieved . append ({ " text " : text , " source " : meta [ " source " ], " page " : meta [ " page " ], " score " : round ( score , 4 ) }) return retrieved def get_sources ( self ) -> List [ str ]: return list ( self . _source_cache . keys ()) def document_count ( self ) -> int : return self . collection . count () def clear ( self ): self . client . delete_collection ( self . collection . name ) self . collection = self . client . get_or_create_collection ( name = self . collection . name , metadata = { " hnsw:space " : " cosine " } ) self . _source_cache = {} llm_client.py # llm_client.py import os from abc import ABC , abstractmethod from typing import List , Dict class LLMClient ( ABC ): @abstractmethod def complete ( self , system : str , messages : List [ Dict ], max_tokens : int = 800 , temperature : float = 0.2 ) -> str : pass class ClaudeClient ( LLMClient ): def init ( self , model : str = " claude-3-5-haiku-20241022 " ): import anthropic self . client = anthropic . Anthropic ( api_key = os . environ . get ( " ANTHROPIC_API_KEY " )) self .
모델을 정의합니다. complete(self, system, messages, max_tokens=800, temperature=0.2)는 응답을 생성하는 함수입니다. response = self.client.messages.create(model=self.model, max_tokens=max_tokens, temperature=temperature, system=system, messages=messages)를 호출하고, 반환된 객체에서 response.content[0].text를 가져옵니다.
클래스 OpenAIClient (LLMClient)는 다음과 같습니다:
def __init__(self, model: str = "gpt-4o-mini"):
import openai
self.client = openai.OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
self.model = model
def complete(self, system, messages, max_tokens=800, temperature=0.2):
all_msgs = [{"role": "system", "content": system}] + messages
response = self.client.chat.completions.create(
model=self.model,
messages=all_msgs,
max_tokens=max_tokens,
temperature=temperature
)
return response.choices[0].message.content
get_llm_client(provider: str = "anthropic", **kwargs) -> LLMClient는 다음과 같습니다:
def get_llm_client(provider: str = "anthropic", **kwargs) -> LLMClient:
if provider == "anthropic":
return ClaudeClient(**kwargs)
elif provider == "openai":
return OpenAIClient(**kwargs)
raise ValueError(f"Unknown provider: {provider}")
rag_engine.py 파일의 내용은 다음과 같습니다:
from typing import List, Dict, Tuple
from knowledge_base import KnowledgeBase
from llm_client import LLMClient
SYSTEM_PROMPT = """
You are DocuMind, an AI research assistant. Your job is to answer questions based ONLY on the provided document context. Rules:
- Use only information from the provided context
- Cite sources using [1], [2] etc. at the end of relevant sentences
- If the answer is not in the context, say exactly: " I couldn't find information about this in the uploaded documents. "
- Be precise and concise
- If multiple documents are relevant, synthesize them coherently
- Format responses with markdown when helpful
"""
def build_context_block(retrieved: List[Dict]) -> str:
parts = []
for i, doc in enumerate(retrieved, 1):
parts.
append(f" [ {i} ] Source: {doc['source']} (page {doc['page']} , " f" relevance: {doc['score']:.0%} )
{doc['text']} " ) return "
" .join(parts)
def build_rag_message(user_question: str, retrieved: List[Dict]) -> str:
context = build_context_block(retrieved)
return (f" Context from documents:\n\n{context}\n\n" f" Question: {user_question} ")
class RAGEngine:
def __init__(self, kb: KnowledgeBase, llm: LLMClient, top_k: int = 5, min_score: float = 0.25):
self.kb = kb
self.llm = llm
self.top_k = top_k
self.min_score = min_score
def answer(self, question: str, history: List[Dict] = None) -> Tuple[str, List[Dict]]:
""" Returns (answer_text, retrieved_docs)
history: previous conversation messages in API format """
retrieved = self.kb.retrieve(question, top_k=self.top_k, min_score=self.min_score)
messages = list(history or [])
if retrieved:
rag_content = build_rag_message(question, retrieved)
messages.append({"role": "user", "content": rag_content})
else:
messages.append({"role": "user", "content": question})
answer = self.llm.complete(system=SYSTEM_PROMPT, messages=messages, max_tokens=800, temperature=0.2)
return answer, retrieved
conversation.py
from dataclasses import dataclass, field
from typing import List, Dict
import time
@dataclass
class Turn:
question: str
answer: str
sources: List[str]
timestamp: float = field(default_factory=time.time)
class Conversation:
def __init__(self, max_turns: int = 8):
self.turns: List[Turn] = []
self.max_turns: int = max_turns
def add(self, question: str, answer: str, sources: List[str]):
self.turns.append(Turn(question, answer, sources))
def get_api_history(self) -> List[Dict]:
""" Return last max_turns as API message format. """
recent = self.turns[-self.max_turns:]
max_turns :] messages = [] for turn in recent : messages.append({"role": "user", "content": turn.question})
messages.append({"role": "assistant", "content": turn.answer})
return messages
def clear(self):
self.turns = []
def __len__(self):
return len(self.turns)
app.py — The Streamlit Interface
import streamlit as st
import tempfile
import os
from pathlib import Path
from config import CONFIG
from document_processor import process_document
from knowledge_base import KnowledgeBase
from llm_client import get_llm_client
from rag_engine import RAGEngine
from conversation import Conversation
st.set_page_config(page_title="DocuMind — AI Research Assistant", page_icon="📚", layout="wide")
@st.cache_resource
def init_system():
kb = KnowledgeBase(embed_model=CONFIG.EMBED_MODEL, persist_dir=CONFIG.CHROMA_PATH, collection_name=CONFIG.COLLECTION
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94