ApeRAG

1. Components

1.1. Core Services

API Server (ApeRAG)

• 역할: 메인 웹 API 서버 (FastAPI)
• 포트: 8000
• 기능: REST API 엔드포인트, 인증, 요청 처리

Frontend (aperag-frontend)

• 역할: 웹 UI (Next.js)
• 포트: 3000
• 기능: 사용자 인터페이스, 문서 관리, 채팅

1.2. Background Processing

Celery Beat (aperag-celerybeat)

• 역할: 작업 스케줄러
• 기능: 정기 작업 스케줄링 (1분마다)

Celery Workers (aperag-celeryworker)

• 역할: 백그라운드 작업 실행자
• 기능:
  • 문서 파싱 (parse_document_task)
  • 인덱스 생성 (create_index_task)
  • 임베딩 생성 (Vector 인덱싱)
  • 요약 생성 (Summary 인덱싱)
  • 비전 처리 (Vision 인덱싱)

Flower (aperag-flower)

• 역할: Celery 모니터링 대시보드
• 포트: 5555
• 기능:
  • 워커 상태 모니터링
  • 태스크 진행 상황 추적
  • 성능 메트릭 확인

1.3. Data Storage

PostgreSQL (aperag-postgres)

• 역할: 메인 데이터베이스
• 포트: 5432
• 기능: 사용자 데이터, 컬렉션, 문서 메타데이터

Redis (aperag-redis)

• 역할: 캐시 및 메시지 브로커
• 포트: 6379
• 기능: Celery 브로커, 세션 캐시, 임시 데이터

Qdrant (aperag-qdrant)

• 역할: 벡터 데이터베이스
• 포트: 6333
• 기능: 임베딩 벡터 저장, 유사도 검색

Elasticsearch (aperag-es)

• 역할: 전문 검색 엔진
• 포트: 9200
• 기능: 전체 텍스트 검색, 인덱싱

Neo4j (aperag-neo4j)

• 역할: 그래프 데이터베이스
• 포트: 7474 (HTTP), 7687 (Bolt)
• 기능:
  • 그래프 인덱싱 (Graph Index)
  • 문서 간 관계 추적
  • 지식 그래프 구축

2. Functions

2.1. Parsing

• 메인 모듈: /aperag/docparser/
• 핵심 파일: doc_parser.py, base.py

def get_default_config() -> list["ParserConfig"]:
    return [
        ParserConfig(name=MinerUParser.name, enabled=False),     # 기본적으로 비활성화 
        ParserConfig(name=DocRayParser.name, enabled=True),      # 활성화
        ParserConfig(name=ImageParser.name, enabled=True),       # 활성화  
        ParserConfig(name=AudioParser.name, enabled=True),       # 활성화
        ParserConfig(name=MarkItDownParser.name, enabled=True),  # 활성화
    ]

2.1.1. Parser List

MinerUParser (1st Priority)

지원 확장자: .pdf, .doc, .docx, .ppt, .pptx, .png, .jpg, .jpeg

• 기본적으로 비활성화 -> 유료
• MinerU API 토큰이 있을 때만 활성화
• 가장 정교한 파싱 (표, 수식 등 복잡한 구조)

DocRayParser (2nd Priority)

지원 확장자: .pdf, .docx, .doc, .pptx, .ppt

• 복잡한 레이아웃 문서 전용
• MarkItDown과 겹치는 형식이지만 더 정교한 파싱
• 따로 Docker service 해서 사용 가능 (Optional) -> Manual 참고

ImageParser (3rd Priority)

지원 확장자: .jpg, .jpeg, .png, .bmp, .tiff, .tif

• 이미지에서 텍스트 추출 전용
• PaddleOCR 서비스 사용

AudioParser (4th Priority)

지원 확장자: .mp3, .mp4, .mpeg, .mpga, .m4a, .wav, .webm, .ogg, .flac

• 음성 파일 전사 전용
• Whisper ASR 서비스 사용

MarkItDownParser (5th Priority)

지원 확장자: .txt, .text, .md, .markdown, .html, .htm, .ipynb, .pdf, .docx, .doc, .xlsx, .xls, .pptx, .ppt, .epub

• 가장 광범위한 지원
• 기본 파서로 거의 모든 문서 형식 처리
• LibreOffice(soffice)를 사용해 구형 Office 문서(.doc, .ppt)를 현대 형식으로 변환

2.1.2. Parser Selection Logic

def parse_file(self, path: Path, metadata: dict[str, Any] = {}, **kwargs) -> list[Part]:
    extension = path.suffix
    last_err = None
    for parser_name in self.parsing_order:  # 설정된 순서대로 시도
        parser = self.parsers[parser_name]
        if not self._parser_accept(parser_name, extension):  # 확장자 지원 확인
            continue
        try:
            return parser.parse_file(path, metadata, **kwargs)
        except FallbackError as e:  # 실패하면 다음 파서로
            last_err = e
    raise ValueError(f'No parser can handle file with extension "{extension}"')

1. 전문 파서 우선: 특정 형식에 최적화된 파서 먼저 시도
2. 폴백 메커니즘: 실패 시 다음 파서로 자동 전환
3. 범용 파서 보장: MarkItDown이 마지막 안전망 역할

PDF File Example

1. MinerUParser 시도 → 실패 시 FallbackError
2. DocRayParser 시도 → 성공하면 결과 반환
3. ImageParser → PDF 지원 안함, 건너뛰기
4. AudioParser → PDF 지원 안함, 건너뛰기
5. MarkItDownParser → DocRayParser가 성공했으므로 실행 안됨

2.2. Chunking

• 메인 모듈: /aperag/docparser/
• 핵심 파일: chunking.py
• 파라미터 설정: /aperag/config.py, .env

def rechunk(parts: list[Part], chunk_size: int, chunk_overlap: int, tokenizer: Callable[[str], List[int]]) -> list[Part]:
    rechunker = Rechunker(chunk_size, chunk_overlap, tokenizer)
    return rechunker(parts)

2.2.1. Chunking 구성 요소

Rechunker

역할: Part 객체들을 의미적 단위로 재구성

• 입력: Part 객체 리스트, chunk_size=400, chunk_overlap=20, tokenizer=cl100k_base
• 출력: 재구성된 Part 객체 리스트
• 핵심 기능: 제목 계층구조 보존, 토큰 크기 제어

Group (그룹 데이터 구조)

@dataclass
class Group:
    title_level: int    # 제목 레벨 (H1=1, H2=2, ...)
    title: str          # 제목 텍스트
    items: list[Part]   # 해당 그룹의 Part들
    tokens: int | None  # 토큰 수 (캐싱용)

SimpleSemanticSplitter

 역할: 큰 텍스트를 의미 단위로 분할

• 계층적 구분자: 문단 → 줄바꿈 → 문장 → 구분자 → 공백 순
• 다국어 지원: 중국어, 영어 구분자 지원
• 재귀적 분할: 최적 크기까지 반복 분할

2.2.2. Workflow

1단계: 그룹화 (to_groups)

def _to_groups(self, parts: list[Part]) -> list[Group]:
    # Part들을 제목 기반으로 그룹화
    # - TitlePart가 있으면 새 그룹 시작
    # - 중첩된 제목은 새 그룹을 만들지 않음
    # - 일반 텍스트는 현재 그룹에 추가

• 그룹화 규칙: 
  • 제목(TitlePart)이 있으면 새 그룹 시작
  • 중첩된 제목은 새 그룹을 만들지 않음
  • 일반 텍스트는 현재 그룹에 추가

2단계: 연속 제목 그룹 병합 (merge_consecutive_title_groups)

def _merge_consecutive_title_groups(self, groups: list[Group]) -> list[Group]:
    # 연속된 제목들을 하나의 그룹으로 병합
    # 예: H2 → H3 → H3 → H2 구조를 하나의 그룹으로

• 병합 규칙: 
  • 연속된 제목들을 하나의 그룹으로 병합
  • 계층구조 유지: 상위 제목이 하위 제목을 포함
  • 내용 그룹 포함: 제목 그룹 다음의 내용도 함께 병합

3단계: 재청킹 (rechunk)

def _rechunk(self, groups: list[Group]) -> list[Part]:
    # 그룹들을 토큰 크기에 맞춰 재구성
    # - 제목 계층구조 보존
    # - 토큰 크기 제어
    # - 오버랩 처리

• 재청킹 규칙:
  • 제목 스택 관리: 현재 제목 경로 추적
  • 병합 가능성 확인: 토큰 크기와 계층구조 고려
  • 오버랩 처리: 청크 간 연속성 보장
  • 분할 필요시 SimpleSemanticSplitter 사용

LEVELED_SEPARATORS = [
    ["\n\n"],           # 문단 구분 (최우선)
    ["\n"],             # 줄바꿈
    ["。", "！", "？"],  # 중국어 문장 종료
    ['.', '!', '?'],    # 영어 문장 종료
    ["；", "，", "、"],  # 중국어 구분자
    [';', ','],         # 영어 구분자
    ["》", "）", "】", "」", "'", '"'],  # 괄호, 따옴표
    ['"', '>', ')', ']', '}', "'", '"'],  # 영어 괄호
    [" ", "\t"],        # 공백 (최후 수단)
]

def _recursive_split(self, s: str, chunk_size: int, chunk_overlap: int, level: int) -> list[str]:
    # 1. 크기 확인: chunk_size 이하면 반환
    # 2. 구분자 시도: 현재 레벨의 구분자로 분할
    # 3. 재귀 분할: 각 청크를 더 작게 분할
    # 4. 병합: 작은 청크들을 적절히 병합
    # 5. 오버랩: 청크 간 겹침 처리

2.3. Embedding

• 메인 모듈: /aperag/llm/embed/
• 핵심 파일: embedding_service.py, embedding_utils.py, base_embedding.py

def create_embeddings_and_store(
    parts: List[Part],                    # 청킹된 Part 객체들
    vector_store_adaptor: VectorStoreConnectorAdaptor,
    embedding_model: Embeddings,
    chunk_size: int = None,               # 기본값: 400토큰
    chunk_overlap: int = None,            # 기본값: 20토큰
    tokenizer=None,                       # 기본값: cl100k_base
) -> List[str]:  # 벡터 스토어 ID 리스트

2.4. Indexing

• 메인 모듈: /aperag/index/
• 핵심 파일: manager.py, base.py, 각 인덱스 타입별 파일

class IndexType(Enum):
    """Index type enumeration"""
    VECTOR = "VECTOR"      # 벡터 인덱스
    FULLTEXT = "FULLTEXT"  # 전문 검색 인덱스
    GRAPH = "GRAPH"        # 그래프 인덱스
    SUMMARY = "SUMMARY"    # 요약 인덱스
    VISION = "VISION"      # 비전 인덱스

2.4.1. Index Type

VectorIndexer (벡터 인덱스)

역할: 임베딩 벡터 기반 유사도 검색

• 데이터베이스: Qdrant
• 검색 방식: 코사인 유사도, 유클리드 거리
• 장점: 의미적 유사성 검색, 다국어 지원
• 활성화: 항상 활성화 (기본 인덱스)

class VectorIndexer(BaseIndexer):
    def create_index(self, document_id: str, content: str, doc_parts: List[Any], collection, **kwargs) -> IndexResult:
        # 1. 임베딩 서비스 가져오기
        embedding_service, embedding_dim = get_collection_embedding_service_sync(collection)
        
        # 2. 벡터 스토어 연결
        vector_store_adaptor = get_vector_db_connector(collection)
        
        # 3. 임베딩 생성 및 저장
        vector_ids = create_embeddings_and_store(
            parts=doc_parts,
            vector_store_adaptor=vector_store_adaptor,
            embedding_model=embedding_service,
            chunk_size=settings.chunk_size,
            chunk_overlap=settings.chunk_overlap_size,
            tokenizer=get_default_tokenizer(),
        )
        
        return IndexResult(success=True, index_type=IndexType.VECTOR, data={"vector_ids": vector_ids})

FulltextIndexer (전문 검색 인덱스)

역할: 키워드 기반 텍스트 검색

• 데이터베이스: Elasticsearch
• 검색 방식: BM25 알고리즘, 부울 검색
• 장점: 정확한 키워드 매칭, 빠른 검색
• 활성화: Elasticsearch 설정 시 활성화

class FulltextIndexer(BaseIndexer):
    def create_index(self, document_id: str, content: str, doc_parts: List[Any], collection, **kwargs) -> IndexResult:
        # 1. Elasticsearch 클라이언트 생성
        es_client = AsyncElasticsearch(self.es_host, **_create_es_client_config())
        
        # 2. 인덱스 이름 생성
        index_name = generate_fulltext_index_name(collection.id)
        
        # 3. 청킹 및 문서 저장
        tokenizer = get_default_tokenizer()
        chunked_parts = rechunk(doc_parts, settings.chunk_size, settings.chunk_overlap_size, tokenizer)
        
        # 4. Elasticsearch에 문서 저장
        for part in chunked_parts:
            doc = {
                "content": part.content,
                "metadata": part.metadata,
                "document_id": document_id,
                "collection_id": collection.id,
            }
            await es_client.index(index=index_name, body=doc)
        
        return IndexResult(success=True, index_type=IndexType.FULLTEXT)

GraphIndexer (그래프 인덱스)

역할: 엔티티 관계 기반 검색

• 데이터베이스: Neo4j
• 검색 방식: 그래프 탐색, 관계 추론
• 장점: 복잡한 관계 이해, 추론 검색
• 활성화: enable_knowledge_graph 설정 시 활성화

class GraphIndexer(AsyncIndexer):
    def is_enabled(self, collection) -> bool:
        config = parseCollectionConfig(collection.config)
        return config.enable_knowledge_graph or False
    
    async def create_index_async(self, document_id: str, content: str, doc_parts: List[Any], collection, **kwargs) -> IndexResult:
        # 1. LightRAG 매니저 가져오기
        lightrag_manager = get_lightrag_manager(collection)
        
        # 2. 문서를 그래프로 변환
        graph_result = await lightrag_manager.ainsert(content)
        
        # 3. 그래프 데이터베이스에 저장
        # - 엔티티 추출
        # - 관계 생성
        # - 그래프 구조 저장
        
        return IndexResult(success=True, index_type=IndexType.GRAPH, data={"graph_nodes": graph_result})

SummaryIndexer (요약 인덱스)

역할: 문서 요약 기반 검색

• 데이터베이스: 벡터 데이터베이스 (요약 임베딩)
• 검색 방식: 요약 임베딩 유사도 검색
• 장점: 문서 전체 맥락 이해, 빠른 개요 검색
• 활성화: LLM 서비스 설정 시 활성화

class SummaryIndexer(BaseIndexer):
    def create_index(self, document_id: str, content: str, doc_parts: List[Any], collection, **kwargs) -> IndexResult:
        # 1. Map-Reduce 전략으로 요약 생성
        # 1.1 청킹 (Map 단계)
        chunked_parts = rechunk(doc_parts, settings.chunk_size, settings.chunk_overlap_size, tokenizer)
        
        # 1.2 각 청크 요약 (Map 단계)
        chunk_summaries = []
        for part in chunked_parts:
            summary = completion_service.complete(
                prompt=f"다음 텍스트를 요약해주세요:\n\n{part.content}",
                max_tokens=200
            )
            chunk_summaries.append(summary)
        
        # 1.3 전체 요약 생성 (Reduce 단계)
        combined_summary = "\n".join(chunk_summaries)
        final_summary = completion_service.complete(
            prompt=f"다음 요약들을 종합하여 최종 요약을 생성해주세요:\n\n{combined_summary}",
            max_tokens=500
        )
        
        # 2. 요약을 임베딩으로 변환
        embedding_service, _ = get_collection_embedding_service_sync(collection)
        summary_vector = embedding_service.embed_query(final_summary)
        
        # 3. 벡터 데이터베이스에 저장
        vector_store_adaptor = get_vector_db_connector(collection)
        summary_node = TextNode(text=final_summary, embedding=summary_vector)
        vector_ids = vector_store_adaptor.connector.store.add([summary_node])
        
        return IndexResult(success=True, index_type=IndexType.SUMMARY, data={"summary_vector_id": vector_ids[0]})

VisionIndexer (비전 인덱스)

역할: 이미지 내용 기반 검색

• 데이터베이스: 벡터 데이터베이스 (이미지 임베딩)
• 검색 방식: 이미지 임베딩 유사도 검색
• 장점: 이미지 내용 이해, 멀티모달 검색
• 활성화: 멀티모달 임베딩 모델 설정 시 활성화

# Path A: Pure Vision Embedding
if embedding_svc.is_multimodal():
    try:
        nodes: List[TextNode] = []
        image_uris = []
        for part in image_parts:
            # 1. 이미지를 base64로 인코딩
            b64_image = base64.b64encode(part.data).decode("utf-8")
            mime_type = part.mime_type or "image/png"
            data_uri = f"data:{mime_type};base64,{b64_image}"
            image_uris.append(data_uri)
            
            # 2. 메타데이터 설정
            metadata = part.metadata.copy()
            metadata["indexer"] = "vision"
            metadata["index_method"] = "multimodal_embedding"
            nodes.append(TextNode(text="", metadata=metadata))

        # 3. 멀티모달 임베딩 생성
        vectors = embedding_svc.embed_documents(image_uris)
        for i, node in enumerate(nodes):
            node.embedding = vectors[i]

        # 4. 벡터 스토어에 저장
        ctx_ids = vector_store_adaptor.connector.store.add(nodes)
        
# Path B: Vision-to-Text
if completion_svc and completion_svc.is_vision_model():
    try:
        text_nodes: List[TextNode] = []
        for part in image_parts:
            # 1. 이미지를 base64로 인코딩
            b64_image = base64.b64encode(part.data).decode("utf-8")
            mime_type = part.mime_type or "image/png"
            data_uri = f"data:{mime_type};base64,{b64_image}"

            # 2. 상세한 프롬프트로 이미지 분석
            prompt = """Analyze the provided image and extract its content with high fidelity...
            1. Overall Summary
            2. Detailed Text Extraction
            3. Chart/Graph Analysis
            4. Object and Scene Recognition"""

            # 3. Vision 모델로 텍스트 생성
            description = completion_svc.generate(
                history=[], 
                prompt=prompt, 
                images=[data_uri]
            )

            # 4. 생성된 텍스트를 임베딩
            metadata["index_method"] = "vision_to_text"
            text_nodes.append(TextNode(text=description, metadata=metadata))

        # 5. 텍스트 임베딩 생성 및 저장
        vectors = embedding_svc.embed_documents([node.get_content() for node in text_nodes])
        for i, node in enumerate(text_nodes):
            node.embedding = vectors[i]
        
        ctx_ids = vector_store_adaptor.connector.store.add(text_nodes)

2.4.2 Workflow

# 1. 문서 파싱 및 청킹
parsed_data = parse_document_content(document, collection)
content, doc_parts, local_doc = parsed_data

# 2. 각 인덱스 타입별 생성
indexers = {
    IndexType.VECTOR: VectorIndexer(),
    IndexType.FULLTEXT: FulltextIndexer(),
    IndexType.GRAPH: GraphIndexer(),
    IndexType.SUMMARY: SummaryIndexer(),
    IndexType.VISION: VisionIndexer(),
}

results = []
for index_type, indexer in indexers.items():
    if indexer.is_enabled(collection):
        try:
            result = indexer.create_index(document_id, content, doc_parts, collection)
            results.append(result)
        except Exception as e:
            logger.error(f"Failed to create {index_type} index: {e}")
            results.append(IndexResult(success=False, index_type=index_type, error=str(e)))

2.4.3 Hybird Search

# 1. 검색 노드들 구성
nodes = {}
edges = []
merge_node_id = "merge"

# 2. 각 인덱스 타입별 검색 노드 생성
if data.vector_search:
    nodes["vector_search"] = NodeInstance(...)
    edges.append(Edge(source="vector_search", target=merge_node_id))

if data.fulltext_search:
    nodes["fulltext_search"] = NodeInstance(...)
    edges.append(Edge(source="fulltext_search", target=merge_node_id))

if data.graph_search:
    nodes["graph_search"] = NodeInstance(...)
    edges.append(Edge(source="graph_search", target=merge_node_id))

if data.summary_search:
    nodes["summary_search"] = NodeInstance(...)
    edges.append(Edge(source="summary_search", target=merge_node_id))

if data.vision_search:
    nodes["vision_search"] = NodeInstance(...)
    edges.append(Edge(source="vision_search", target=merge_node_id))

# 3. Merge 노드 (결과 통합)
nodes[merge_node_id] = NodeInstance(
    id=merge_node_id,
    type="merge",
    input_values={
        "merge_strategy": "union",      # 합집합 전략
        "deduplicate": True,            # 중복 제거
        "vector_search_docs": "{{ nodes.vector_search.output.docs }}",
        "fulltext_search_docs": "{{ nodes.fulltext_search.output.docs }}",
        "graph_search_docs": "{{ nodes.graph_search.output.docs }}",
        "summary_search_docs": "{{ nodes.summary_search.output.docs }}",
        "vision_search_docs": "{{ nodes.vision_search.output.docs }}",
    }
)

# 4. Rerank 노드 (재랭킹)
nodes["rerank"] = NodeInstance(
    id="rerank",
    type="rerank",
    input_values={
        "use_rerank_service": use_rerank_service,
        "model": model,
        "model_service_provider": model_service_provider,
        "custom_llm_provider": custom_llm_provider,
        "docs": "{{ nodes.merge.output.docs }}",
    }
)
edges.append(Edge(source=merge_node_id, target="rerank"))

2.5 Reranking