1C용 RAG 시스템: MCP 서버 아키텍처

MCP 서버에서 1C 문서 검색을 위한 다중 계층 RAG 아키텍처

MCP 서버의 documents1c와 metadata1c는 1C 문서와 구성에 대한 정밀 검색을 위한 다중 계층 RAG 시스템을 구현합니다. 각 순위 계층은 청크 컨텍스트부터 벡터 및 전체 텍스트 검색 결합에 이르기까지 특정 과제를 해결합니다. 이 시스템은 PostgreSQL과 pgvector, 캐싱을 위한 Redis, 그리고 1C 용어를 위한 맞춤형 토큰화를 사용합니다.

계층적 벡터 데이터베이스에서 벗어난 이유

Qdrant나 Weaviate와 같은 표준 벡터 저장소는 1C 문서의 동적 특성을 처리하는 데 어려움을 겪습니다. 문서는 컨텍스트 없이는 독립적이지 않습니다: "설정"이라는 제목의 문서는 섹션을 지정해야 하며, Reference.Employees와 같은 객체는 그 속성과 모듈에 대한 설명이 필요합니다.

문서는 자주 업데이트됩니다—새로운 BSP 릴리스, 플랫폼 메서드 등. 벡터 데이터베이스에서 트리를 재구성하는 것은 비용이 많이 듭니다. 해결책: PostgreSQL + pgvector와 각 청크에 내장된 heading_path입니다. 임베딩은 heading_path + text에 대해 생성되며, 청크 텍스트만 저장됩니다.

절감 효과: 별도의 벡터 데이터베이스 대신 article_chunks 테이블에 embedding halfvec(1024) 컬럼을 사용합니다.

계층 1: 제목 컨텍스트가 포함된 청크

마크다운을 청크로 분할할 때 구조를 고려합니다: YAML은 제거되고, 헤더별로 분할되며, 코드 블록은 분리되지 않으며, 220자의 오버랩이 적용됩니다.

def split_markdown_into_chunks(
    text: str,
    max_len: int = 800,
    overlap: int = 220,
    include_heading_path: str = "full_path",
) -> list[str]:
    body = _strip_frontmatter(text)
    sections = _split_by_headers(body)
    chunks: list[str] = []

    for section, heading_path, current_heading in sections:
        context_heading = " > ".join(heading_path)
        section_with_context = f"{context_heading}\n\n{section}"

        if len(section_with_context) <= max_len:
            chunks.append(section_with_context)
        else:
            sub_chunks = _split_section_by_paragraphs(section_with_context, max_len)
            chunks.extend(sub_chunks)

    return _add_overlap(chunks, overlap)

청크 예시:

급여 및 인사 관리 > 급여 계산 > 개인소득세 설정

개인소득세를 올바르게 계산하려면 세율을 지정해야 합니다...

데이터베이스에서 meta는 필터와 링크를 위해 heading_path를 별도로 저장합니다. 임베딩: heading_path + chunk_text.

계층 2: Redis의 임베딩 캐시

임베딩 생성(text-embedding-qwen3-embedding-4b)에는 50–200 ms가 소요됩니다. 반복 쿼리의 경우—TTL이 있는 Redis, 키는 model:dimensions:text의 SHA256입니다.

async def embed_text(text: str) -> list[float]:
    model = _get_embed_model_name()
    dims = _get_embed_dimensions()

    if _cache_enabled():
        cached = await cache_get(text, model, dims)
        if cached is not None:
            return cached

    response = await to_thread(client.embeddings.create, model=model, input=text)
    vector = response.data[0].embedding

    if _cache_enabled():
        await cache_set(text, model, dims, vector, _cache_ttl())

    return vector

캐시는 모델이 변경되면 자동으로 무효화됩니다.

계층 3: halfvec를 사용한 벡터 검색

pgvector의 <=>를 통한 코사인 거리, halfvec(1024) 타입은 메모리를 절약합니다.

SELECT
    ac.id, ac.article_id, ac.chunk_text, ac.meta,
    ac.embedding::halfvec(1024) <=> CAST(:query_embedding AS halfvec(1024)) AS distance
FROM article_chunks ac
INNER JOIN articles a ON ac.article_id = a.id
WHERE a.job_id IN (SELECT id FROM jobs WHERE load_mode IN ('hbk', 'docs'))
AND a.destination = :destination
ORDER BY distance
LIMIT :limit;

점수 = 1 - 거리.

계층 4: RRF를 사용한 순위 병합

벡터 검색은 &OnClientOnServer와 같은 정확한 용어에 대해 약합니다. RRF는 벡터와 FTS 목록을 융합합니다:

@staticmethod
def _rrf_merge(vector_results, fts_results, top_k: int, k: int = 60) -> list[dict]:
    scores: dict[int, float] = {}
    items: dict[int, dict] = {}

    for rank, item in enumerate(vector_results, start=1):
        chunk_id = item["id"]
        scores[chunk_id] = scores.get(chunk_id, 0.0) + 1.0 / (k + rank)
        items[chunk_id] = item

    for rank, item in enumerate(fts_results, start=1):
        chunk_id = item["id"]
        scores[chunk_id] = scores.get(chunk_id, 0.0) + 1.0 / (k + rank)
        if chunk_id not in items:
            items[chunk_id] = item

    sorted_ids = sorted(scores, key=lambda cid: scores[cid], reverse=True)
    # 정규화 및 top_k 반환

HBK 부스트: 필터가 없는 쿼리의 경우, 1C 내장 언어 청크(load_mode='hbk')가 목록의 시작 부분에 추가됩니다.

• 쿼리_임베딩 기준 상위 HBK 청크
• 중복 제외
• 주요 결과 앞에 배치

계층 5: 1C 토큰화를 사용한 BM25 재순위

RRF에서 나온 50–100개의 후보에 대해 BM25가 가중치(벡터 0.6, bm25 0.4)와 함께 적용됩니다. 다음에 효과적입니다:

• 특정 용어: CommonModule, InformationRegister, &OnServer
• 객체 이름: Reference.Nomenclature
• 짧은 기술 쿼리

맞춤형 토큰화:

GUID_RE = re.compile(r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b", re.I)
VERSION_RE = re.compile(r"\bv?8\.\d+(?:\.\d+){1,3}\b", re.I)
DIRECTIVE_RE = re.compile(r"[&#][And-IA-Z][And-YaYoA-Za-z0-9_]*", re.I)

def _tokenize(self, text: str, expand_query: bool = False) -> list[str]:
    text = _norm_text(text)
    out = []

    for g in GUID_RE.findall(text): out.append(g.lower())
    for v in VERSION_RE.findall(text): out.append(v.lower())
    for d in DIRECTIVE_RE.findall(text): out.append(d.lower())

    for t in TOKEN_RE.findall(text):
        if t in RU_STOP: continue
        for p in _split_identifier(t):
            out.append(p.lower())
    return out

CamelCase, snake_case를 분할하고, 지시어와 GUID를 추출합니다.

핵심 요약

• 청크의 heading_path는 문서 트리 없이 컨텍스트를 제공합니다.
• Redis 캐시는 임베딩 지연 시간을 100+ ms에서 0 ms로 줄입니다.
• RRF는 정규화 없이 의미론과 정확한 검색을 결합합니다.
• 1C 토큰화를 사용한 BM25는 관련 기술 용어를 강화합니다.
• halfvec(1024)는 코사인 정확도를 유지하면서 pgvector에서 메모리를 절약합니다.

— Editorial Team