10장: 실전 프로젝트 — 온디바이스 AI 시스템 구축

# 프로젝트 디렉터리 생성
mkdir -p local-doc-qa && cd local-doc-qa
python -m venv .venv
source .venv/bin/activate
 
# 핵심 의존성 설치
pip install \
  chromadb==0.5.23 \
  sentence-transformers==3.3.1 \
  pymupdf==1.25.1 \
  requests==2.32.3 \
  rich==13.9.4
 
# llama.cpp 서버 빌드 (Metal 지원)
git clone https://github.com/ggerganov/llama.cpp.git
cd llama.cpp
cmake -B build -DLLAMA_METAL=ON
cmake --build build --config Release -j
cd ..

# LLM 모델 다운로드 (Llama 3.2 7B Q4_K_M, 약 4.1GB)
huggingface-cli download \
  bartowski/Llama-3.2-7B-GGUF \
  Llama-3.2-7B-Q4_K_M.gguf \
  --local-dir models/
 
# 임베딩 모델은 sentence-transformers가 자동 다운로드
# nomic-embed-text-v1.5 (약 280MB)

import fitz  # PyMuPDF
from dataclasses import dataclass
 
 
@dataclass
class TextChunk:
    text: str
    source: str
    page_number: int
    chunk_index: int
 
 
def extract_text_from_pdf(pdf_path: str) -> list[tuple[int, str]]:
    """PDF에서 페이지별 텍스트를 추출합니다."""
    doc = fitz.open(pdf_path)
    pages = []
    for page_num, page in enumerate(doc):
        text = page.get_text("text")
        if text.strip():
            pages.append((page_num + 1, text))
    doc.close()
    return pages
 
 
def split_into_chunks(
    pages: list[tuple[int, str]],
    source: str,
    chunk_size: int = 512,
    overlap: int = 64,
) -> list[TextChunk]:
    """페이지 텍스트를 고정 크기 청크로 분할합니다."""
    chunks: list[TextChunk] = []
    chunk_index = 0
 
    for page_num, text in pages:
        words = text.split()
        start = 0
 
        while start < len(words):
            end = start + chunk_size
            chunk_text = " ".join(words[start:end])
 
            if len(chunk_text.strip()) > 50:
                chunks.append(TextChunk(
                    text=chunk_text,
                    source=source,
                    page_number=page_num,
                    chunk_index=chunk_index,
                ))
                chunk_index += 1
 
            start += chunk_size - overlap
 
    return chunks

import chromadb
from chromadb.config import Settings
from sentence_transformers import SentenceTransformer
from document_ingester import TextChunk
 
 
class LocalVectorStore:
    def __init__(
        self,
        persist_dir: str = "./vector_db",
        embedding_model: str = "nomic-ai/nomic-embed-text-v1.5",
    ):
        self.embedder = SentenceTransformer(
            embedding_model,
            trust_remote_code=True,
        )
        self.client = chromadb.PersistentClient(
            path=persist_dir,
            settings=Settings(anonymized_telemetry=False),
        )
        self.collection = self.client.get_or_create_collection(
            name="documents",
            metadata={"hnsw:space": "cosine"},
        )
 
    def add_chunks(self, chunks: list[TextChunk]) -> int:
        """텍스트 청크를 벡터화하여 저장합니다."""
        texts = [chunk.text for chunk in chunks]
        embeddings = self.embedder.encode(
            texts,
            show_progress_bar=True,
            normalize_embeddings=True,
        ).tolist()
 
        ids = [f"{chunk.source}_{chunk.chunk_index}" for chunk in chunks]
        metadatas = [
            {
                "source": chunk.source,
                "page": chunk.page_number,
                "chunk_index": chunk.chunk_index,
            }
            for chunk in chunks
        ]
 
        self.collection.add(
            ids=ids,
            embeddings=embeddings,
            documents=texts,
            metadatas=metadatas,
        )
        return len(chunks)
 
    def search(
        self,
        query: str,
        n_results: int = 5,
    ) -> list[dict]:
        """쿼리와 유사한 청크를 검색합니다."""
        query_embedding = self.embedder.encode(
            [query],
            normalize_embeddings=True,
        ).tolist()
 
        results = self.collection.query(
            query_embeddings=query_embedding,
            n_results=n_results,
        )
 
        return [
            {
                "text": doc,
                "source": meta["source"],
                "page": meta["page"],
                "distance": dist,
            }
            for doc, meta, dist in zip(
                results["documents"][0],
                results["metadatas"][0],
                results["distances"][0],
            )
        ]

import requests
from vector_store import LocalVectorStore
 
 
SYSTEM_PROMPT = """당신은 문서 분석 전문가입니다. 
주어진 컨텍스트만을 기반으로 질문에 답변하세요.
컨텍스트에 없는 내용은 "해당 정보를 문서에서 찾을 수 없습니다"라고 답변하세요.
답변은 한국어로 작성하세요."""
 
 
class LocalQAEngine:
    def __init__(
        self,
        vector_store: LocalVectorStore,
        llm_url: str = "http://localhost:8080",
    ):
        self.vector_store = vector_store
        self.llm_url = llm_url
 
    def build_prompt(
        self,
        question: str,
        contexts: list[dict],
    ) -> str:
        """검색 결과를 포함한 프롬프트를 구성합니다."""
        context_text = "\n\n---\n\n".join(
            f"[출처: {ctx['source']}, {ctx['page']}페이지]\n{ctx['text']}"
            for ctx in contexts
        )
 
        return (
            f"<|system|>\n{SYSTEM_PROMPT}\n"
            f"<|user|>\n"
            f"다음 문서 내용을 참고하여 질문에 답변하세요.\n\n"
            f"## 문서 내용\n{context_text}\n\n"
            f"## 질문\n{question}\n"
            f"<|assistant|>\n"
        )
 
    def ask(
        self,
        question: str,
        n_contexts: int = 5,
    ) -> dict:
        """질문에 대한 답변을 생성합니다."""
        contexts = self.vector_store.search(question, n_results=n_contexts)
 
        prompt = self.build_prompt(question, contexts)
 
        response = requests.post(
            f"{self.llm_url}/completion",
            json={
                "prompt": prompt,
                "n_predict": 1024,
                "temperature": 0.3,
                "top_p": 0.9,
                "stop": ["<|user|>", "<|system|>"],
            },
        )
 
        result = response.json()
 
        return {
            "answer": result["content"].strip(),
            "sources": [
                {"source": ctx["source"], "page": ctx["page"]}
                for ctx in contexts
            ],
            "tokens_generated": result.get("tokens_predicted", 0),
            "generation_time_ms": result.get("timings", {}).get(
                "predicted_ms", 0
            ),
        }

import sys
from pathlib import Path
from rich.console import Console
from rich.panel import Panel
from document_ingester import extract_text_from_pdf, split_into_chunks
from vector_store import LocalVectorStore
from qa_engine import LocalQAEngine
 
console = Console()
 
 
def ingest_documents(pdf_dir: str, store: LocalVectorStore) -> int:
    """디렉터리 내 모든 PDF를 벡터 DB에 수집합니다."""
    total_chunks = 0
    pdf_files = list(Path(pdf_dir).glob("*.pdf"))
 
    for pdf_path in pdf_files:
        console.print(f"  처리 중: {pdf_path.name}")
        pages = extract_text_from_pdf(str(pdf_path))
        chunks = split_into_chunks(pages, source=pdf_path.name)
        count = store.add_chunks(chunks)
        total_chunks += count
        console.print(f"  -> {count}개 청크 저장 완료")
 
    return total_chunks
 
 
def main():
    store = LocalVectorStore(persist_dir="./local_vector_db")
    engine = LocalQAEngine(vector_store=store)
 
    if len(sys.argv) > 1 and sys.argv[1] == "ingest":
        pdf_dir = sys.argv[2] if len(sys.argv) > 2 else "./documents"
        console.print(Panel("문서 수집 시작", style="bold"))
        total = ingest_documents(pdf_dir, store)
        console.print(f"총 {total}개 청크 수집 완료")
        return
 
    console.print(Panel("문서 Q&A 시스템 (종료: quit)", style="bold"))
 
    while True:
        question = console.input("\n질문: ")
        if question.lower() in ("quit", "exit", "q"):
            break
 
        with console.status("답변 생성 중..."):
            result = engine.ask(question)
 
        console.print(f"\n{result['answer']}")
        console.print(
            f"\n참조: {', '.join(f\"{s['source']} {s['page']}p\" for s in result['sources'])}",
            style="dim",
        )
 
 
if __name__ == "__main__":
    main()

pip install pyinstaller
pyinstaller --onefile --name local-doc-qa main.py
# dist/local-doc-qa 실행 파일 생성