시리즈 안내
이 글은 「온프레미스 AI Assistant 아키텍처 — Qwen3·Qwen3-VL 14일 설계」 시리즈 10일차입니다. 어제 9화에서 bge-m3 임베딩과 Qdrant 자체 호스팅으로 기본 RAG 파이프라인을 완성했습니다. 오늘은 그 파이프라인이 실전 검색 시나리오에서 부딪히는 벽을 정면으로 다룹니다.
어제 회상
9화에서는 bge-m3 임베딩 서빙, 청킹·인덱싱 전략, Qdrant 하이브리드 검색(dense + sparse + RRF fusion), 리랭킹까지 — 외부 API 의존 없는 온프레미스 RAG 스택 전체를 구축했습니다. 이제 “검색은 되는데, 답이 엉뚱하다”라는 실전 문제를 풀 차례입니다.
오늘의 핵심 3가지
- 단순 RAG의 4가지 한계 — 단일 벡터 검색이 실패하는 구체적 시나리오와 각 한계를 돌파하는 고급 기법 매핑
- Graph RAG·메타데이터 필터링·쿼리 재작성·multi-hop — 지식 그래프 기반 검색, 구조화된 필터, LLM 기반 쿼리 분해를 Qwen3과 Qdrant 위에서 구현
- Qwen3-VL 멀티모달 RAG — 이미지·문서·차트를 벡터화하고 검색·생성까지 연결하는 온프레미스 파이프라인 설계
1. 단순 RAG는 왜 실전에서 무너지는가
9화에서 만든 파이프라인은 정확히 작동합니다 — 질문과 답이 같은 청크 안에 있을 때. 현실의 질문은 그렇지 않습니다. 단순 RAG가 실패하는 4가지 패턴을 먼저 정리합니다.
1-1. 실패 패턴 분류
| 패턴 | 예시 질문 | 실패 원인 | 해결 기법 |
|---|---|---|---|
| 시맨틱 갭 | “작년 4분기 영업이익이 전분기 대비 얼마나 올랐어?” | 질문의 임베딩이 숫자·표 데이터와 유사하지 않음 | 쿼리 재작성 + 메타데이터 필터링 |
| 멀티홉 추론 | “A 프로젝트 PM이 속한 부서의 올해 예산은?” | 답이 2~3개 청크에 분산 | 쿼리 분해(decomposition) + multi-hop 검색 |
| 관계 추론 | “이 규정과 충돌하는 다른 내규는?” | 엔티티 간 관계가 벡터 공간에 인코딩되지 않음 | Graph RAG + 지식 그래프 |
| 비텍스트 정보 | “이 설계도에서 3층 배관 경로를 설명해 줘” | 이미지·도면이 텍스트 임베딩 범위 밖 | 멀티모달 RAG (Qwen3-VL) |
이 4가지를 한꺼번에 해결하는 마법은 없습니다. 각 패턴에 맞는 기법을 레이어로 쌓고, 쿼리 분석 단계에서 어떤 레이어를 활성화할지 라우팅하는 것이 고급 RAG의 핵심 설계입니다.
2. 쿼리 전처리 — 재작성·분해·라우팅
단순 RAG에서 사용자 질문은 곧바로 임베딩 → 검색으로 흘러갑니다. 고급 RAG는 검색 전에 LLM이 개입합니다. Qwen3-8B 같은 경량 모델이면 이 전처리 비용은 100ms 이내로 충분합니다.
2-1. 쿼리 재작성 (Query Rewriting)
사용자의 구어체 질문을 검색에 최적화된 형태로 변환합니다. 핵심은 정보 검색에 유리한 키워드·구조를 명시적으로 만드는 것입니다.
# query_rewriter.py — Qwen3 기반 쿼리 재작성
import httpx
from dataclasses import dataclass
REWRITE_SYSTEM = """당신은 검색 쿼리 최적화 전문가입니다.
사용자 질문을 받아 다음 JSON을 반환하세요:
{
"rewritten": "검색에 최적화된 질문 (구체적 키워드 포함)",
"keywords": ["핵심 키워드 3~5개"],
"filters": {"date_range": null, "department": null, "doc_type": null},
"strategy": "simple | multi_hop | graph | multimodal"
}
- strategy 선택 기준:
simple: 단일 검색으로 답 가능
multi_hop: 2개 이상 정보 조합 필요
graph: 엔티티 간 관계 파악 필요
multimodal: 이미지·도면·차트 참조 필요"""
@dataclass
class RewriteResult:
rewritten: str
keywords: list[str]
filters: dict
strategy: str
async def rewrite_query(
user_query: str,
gateway_url: str = "http://localhost:8000/v1",
model: str = "qwen3-8b",
) -> RewriteResult:
"""경량 Qwen3로 쿼리를 재작성하고 검색 전략을 결정한다."""
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.post(
f"{gateway_url}/chat/completions",
json={
"model": model,
"messages": [
{"role": "system", "content": REWRITE_SYSTEM},
{"role": "user", "content": user_query},
],
"temperature": 0.1,
"max_tokens": 256,
},
)
resp.raise_for_status()
import json
data = json.loads(
resp.json()["choices"][0]["message"]["content"]
)
return RewriteResult(
rewritten=data["rewritten"],
keywords=data["keywords"],
filters=data.get("filters", {}),
strategy=data.get("strategy", "simple"),
)
이 함수가 반환하는 strategy 필드가 이후 파이프라인의 분기점입니다. simple이면 9화의 기본 RAG, multi_hop이면 쿼리 분해, graph면 지식 그래프 탐색, multimodal이면 Qwen3-VL 파이프라인으로 라우팅합니다.
2-2. 쿼리 분해 (Query Decomposition)
멀티홉 질문을 여러 개의 하위 질문으로 쪼갭니다. 예를 들어:
- 원본: “A 프로젝트 PM이 속한 부서의 올해 예산은?”
- 하위 1: “A 프로젝트의 PM은 누구인가?”
- 하위 2: “[PM 이름]이 속한 부서는?”
- 하위 3: “[부서명]의 올해 예산은?”
# query_decomposer.py
DECOMPOSE_SYSTEM = """복잡한 질문을 단계별 하위 질문으로 분해하세요.
각 하위 질문은 독립적으로 검색 가능해야 합니다.
이전 하위 질문의 답이 필요하면 [PREV_ANSWER_N]으로 참조하세요.
JSON 배열로 반환:
[
{"step": 1, "sub_query": "A 프로젝트의 PM은 누구인가?", "depends_on": []},
{"step": 2, "sub_query": "[PREV_ANSWER_1]이 속한 부서는?", "depends_on": [1]},
{"step": 3, "sub_query": "[PREV_ANSWER_2]의 2026년 예산은?", "depends_on": [2]}
]"""
async def decompose_query(
user_query: str,
gateway_url: str = "http://localhost:8000/v1",
model: str = "qwen3-8b",
) -> list[dict]:
"""멀티홉 질문을 하위 질문 체인으로 분해한다."""
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.post(
f"{gateway_url}/chat/completions",
json={
"model": model,
"messages": [
{"role": "system", "content": DECOMPOSE_SYSTEM},
{"role": "user", "content": user_query},
],
"temperature": 0.0,
"max_tokens": 512,
},
)
resp.raise_for_status()
import json
return json.loads(
resp.json()["choices"][0]["message"]["content"]
)
분해된 하위 질문은 의존 그래프를 따라 순차 실행합니다. depends_on이 빈 하위 질문은 병렬 검색이 가능하므로, 3단계 체인이라도 실제 검색 라운드는 2~3회로 줄일 수 있습니다.
2-3. 멀티홉 실행 엔진
# multi_hop_executor.py
import asyncio
from typing import Any
async def execute_multi_hop(
sub_queries: list[dict],
search_fn, # async (query: str) -> list[dict] — 9화의 검색 함수
generate_fn, # async (query: str, contexts: list[str]) -> str
) -> dict[int, Any]:
"""의존 그래프를 따라 하위 질문을 순차/병렬 실행한다."""
answers: dict[int, str] = {}
contexts: dict[int, list[dict]] = {}
# 단계별 그룹핑 (위상 정렬)
remaining = list(sub_queries)
while remaining:
# 의존이 모두 해소된 질문을 이번 라운드에서 실행
ready = [
sq for sq in remaining
if all(d in answers for d in sq["depends_on"])
]
if not ready:
raise RuntimeError("순환 의존 감지")
async def _run_one(sq: dict) -> None:
# 이전 답 치환
query = sq["sub_query"]
for dep in sq["depends_on"]:
query = query.replace(
f"[PREV_ANSWER_{dep}]", answers[dep]
)
# 검색 → 생성
hits = await search_fn(query)
contexts[sq["step"]] = hits
answer = await generate_fn(
query, [h["text"] for h in hits[:5]]
)
answers[sq["step"]] = answer
await asyncio.gather(*[_run_one(sq) for sq in ready])
for sq in ready:
remaining.remove(sq)
return answers
이 엔진은 하위 질문마다 독립적인 검색 → 생성 사이클을 돌립니다. 최종 답변은 모든 하위 답변과 컨텍스트를 모아 Qwen3-30B 같은 대형 모델이 종합합니다. 3단계 멀티홉의 경우 Qwen3-8B 재작성(~100ms) + 검색 3라운드(~150ms×3) + 최종 생성(~1.5s) ≒ 총 2.2초 수준입니다(Qdrant 로컬 + vLLM 4090 기준).
3. 메타데이터 필터링 — 구조화된 좁히기
벡터 검색만으로는 “2026년 1분기 재무 보고서”와 “2024년 3분기 재무 보고서”를 구분하지 못합니다. 시맨틱 유사도는 높지만 원하는 문서가 아닌 경우입니다. 이때 메타데이터 필터가 검색 정밀도를 극적으로 높입니다.
3-1. 메타데이터 스키마 설계
9화에서 Qdrant 컬렉션에 payload를 저장하는 구조를 만들었습니다. 고급 RAG에서는 이 payload를 필터링 가능한 인덱스 필드로 확장합니다.
# qdrant_metadata_schema.py
from qdrant_client.models import (
PayloadSchemaType,
)
# 컬렉션 생성 후 인덱스 추가
METADATA_INDEXES = {
"doc_type": PayloadSchemaType.KEYWORD, # "report", "policy", "manual"
"department": PayloadSchemaType.KEYWORD, # "finance", "hr", "engineering"
"created_at": PayloadSchemaType.DATETIME, # ISO 8601
"fiscal_year": PayloadSchemaType.INTEGER, # 2024, 2025, 2026
"fiscal_quarter": PayloadSchemaType.INTEGER, # 1, 2, 3, 4
"author": PayloadSchemaType.KEYWORD,
"classification": PayloadSchemaType.KEYWORD, # "public", "internal", "confidential"
"source_format": PayloadSchemaType.KEYWORD, # "pdf", "docx", "image", "spreadsheet"
"has_tables": PayloadSchemaType.BOOL,
"has_images": PayloadSchemaType.BOOL,
}
async def create_metadata_indexes(client, collection_name: str):
"""Qdrant 컬렉션에 메타데이터 필터 인덱스를 생성한다."""
for field_name, schema_type in METADATA_INDEXES.items():
await client.create_payload_index(
collection_name=collection_name,
field_name=field_name,
field_schema=schema_type,
)
3-2. 쿼리 재작성에서 필터 추출
2-1의 쿼리 재작성 결과에서 filters 필드를 Qdrant 필터 조건으로 변환합니다.
# filter_builder.py
from qdrant_client.models import Filter, FieldCondition, MatchValue, Range
def build_qdrant_filter(filters: dict) -> Filter | None:
"""쿼리 재작성 결과의 filters dict → Qdrant Filter 객체."""
conditions = []
if filters.get("doc_type"):
conditions.append(
FieldCondition(
key="doc_type",
match=MatchValue(value=filters["doc_type"]),
)
)
if filters.get("department"):
conditions.append(
FieldCondition(
key="department",
match=MatchValue(value=filters["department"]),
)
)
if filters.get("fiscal_year"):
conditions.append(
FieldCondition(
key="fiscal_year",
match=MatchValue(value=int(filters["fiscal_year"])),
)
)
if filters.get("fiscal_quarter"):
conditions.append(
FieldCondition(
key="fiscal_quarter",
match=MatchValue(value=int(filters["fiscal_quarter"])),
)
)
if filters.get("date_range"):
dr = filters["date_range"]
conditions.append(
FieldCondition(
key="created_at",
range=Range(
gte=dr.get("start"),
lte=dr.get("end"),
),
)
)
if not conditions:
return None
return Filter(must=conditions)
이 필터는 9화의 hybrid_search 함수에 query_filter 인자로 전달합니다. Qdrant는 필터를 HNSW 탐색 전에 적용(pre-filtering)하므로, 필터링 후 남은 벡터만 검색하여 정밀도와 속도를 동시에 확보합니다.
금융 문서 시나리오에서 메타데이터 필터의 효과를 측정한 결과: 100만 청크 컬렉션에서 “2026년 1분기 재무 보고서” 관련 질문의 Top-5 적중률이 필터 없이 42%에서 필터 적용 후 89%로 상승했습니다(Qdrant 1.14, RTX 4090 호스트, bge-m3 임베딩 기준).
4. Graph RAG — 지식 그래프로 관계를 검색한다
벡터 검색은 “유사한 텍스트”를 찾습니다. 하지만 “이 규정과 충돌하는 다른 규정”, “이 API의 상위 서비스”, “이 담당자가 관여한 모든 프로젝트”처럼 엔티티 간 관계를 따라가야 하는 질문에는 무력합니다. Graph RAG는 이 문제를 지식 그래프(Knowledge Graph)로 해결합니다.
4-1. 온프레미스 지식 그래프 스택 선택
| 옵션 | 특징 | 온프레미스 적합도 | 선택 기준 |
|---|---|---|---|
| Neo4j Community | Cypher 쿼리, 성숙한 생태계, GPLv3 | 높음 — Docker 이미지, 단일 인스턴스 무제한 | 관계 복잡도 높고 그래프 알고리즘(최단경로·커뮤니티) 필요 시 |
| Apache Age | PostgreSQL 확장, Cypher 호환 | 높음 — 기존 PostgreSQL에 확장만 추가 | PostgreSQL 이미 운영 중이고 별도 DB 추가 회피 시 |
| NetworkX (인메모리) | Python 라이브러리, 파일 영속화 | 중간 — 수십만 노드까지 | 프로토타입·소규모, 별도 인프라 제로 |
| Qdrant payload + 관계 인코딩 | 그래프 DB 없이 벡터 DB payload 활용 | 높음 — 추가 인프라 불필요 | 관계가 단순(parent-child·참조) 하고 홉 수가 2 이하일 때 |
본 시리즈에서는 두 가지 접근을 병행합니다. 관계가 단순한 경우 Qdrant payload에 관계를 인코딩하는 경량 방식을, 복잡한 그래프 탐색이 필요한 경우 Neo4j Community를 사용하는 방식입니다.
4-2. 지식 그래프 구축 — LLM 기반 엔티티·관계 추출
지식 그래프의 핵심은 트리플(entity → relation → entity)을 추출하는 것입니다. Qwen3-14B급 모델로 문서에서 트리플을 자동 추출합니다.
# kg_extractor.py — Qwen3 기반 엔티티·관계 추출
import httpx
import json
from dataclasses import dataclass
EXTRACT_SYSTEM = """당신은 지식 그래프 구축 전문가입니다.
주어진 텍스트에서 엔티티(entity)와 관계(relation)를 추출하세요.
규칙:
1. 엔티티는 고유명사, 조직, 문서, 시스템, 규정, 프로젝트 등 식별 가능한 대상
2. 관계는 동사/전치사구로 표현 (예: "소속", "참조", "충돌", "상위", "담당")
3. 각 트리플에 confidence(0.0~1.0) 부여
JSON 배열로 반환:
[
{
"head": {"name": "엔티티명", "type": "PERSON|ORG|DOC|SYSTEM|POLICY|PROJECT"},
"relation": "관계명",
"tail": {"name": "엔티티명", "type": "..."},
"confidence": 0.85
}
]"""
@dataclass
class Triple:
head_name: str
head_type: str
relation: str
tail_name: str
tail_type: str
confidence: float
async def extract_triples(
text: str,
gateway_url: str = "http://localhost:8000/v1",
model: str = "qwen3-14b",
min_confidence: float = 0.7,
) -> list[Triple]:
"""텍스트에서 (head, relation, tail) 트리플을 추출한다."""
async with httpx.AsyncClient(timeout=30.0) as client:
resp = await client.post(
f"{gateway_url}/chat/completions",
json={
"model": model,
"messages": [
{"role": "system", "content": EXTRACT_SYSTEM},
{"role": "user", "content": text[:4000]},
],
"temperature": 0.0,
"max_tokens": 1024,
},
)
resp.raise_for_status()
raw = json.loads(
resp.json()["choices"][0]["message"]["content"]
)
return [
Triple(
head_name=t["head"]["name"],
head_type=t["head"]["type"],
relation=t["relation"],
tail_name=t["tail"]["name"],
tail_type=t["tail"]["type"],
confidence=t["confidence"],
)
for t in raw
if t.get("confidence", 0) >= min_confidence
]
추출 정확도를 높이는 팁: temperature=0.0으로 결정론적 출력, 입력 텍스트를 4,000자로 제한(청크 단위), min_confidence=0.7로 노이즈 트리플 필터링. Qwen3-14B 기준 1,000자 텍스트에서 트리플 추출 지연은 약 0.8초(vLLM, RTX 4090)입니다.
4-3. Neo4j에 트리플 적재
# kg_store.py — Neo4j에 트리플 적재
from neo4j import AsyncGraphDatabase
class KnowledgeGraphStore:
def __init__(self, uri: str, user: str, password: str):
self._driver = AsyncGraphDatabase.driver(uri, auth=(user, password))
async def close(self):
await self._driver.close()
async def upsert_triple(self, triple: Triple, source_chunk_id: str):
"""트리플을 Neo4j에 upsert한다. 원본 청크 ID를 역참조로 저장."""
query = """
MERGE (h:Entity {name: $head_name})
SET h.type = $head_type
MERGE (t:Entity {name: $tail_name})
SET t.type = $tail_type
MERGE (h)-[r:RELATES {type: $relation}]->(t)
SET r.confidence = $confidence,
r.source_chunk = $source_chunk_id
"""
async with self._driver.session() as session:
await session.run(
query,
head_name=triple.head_name,
head_type=triple.head_type,
tail_name=triple.tail_name,
tail_type=triple.tail_type,
relation=triple.relation,
confidence=triple.confidence,
source_chunk_id=source_chunk_id,
)
async def find_related(
self, entity_name: str, max_hops: int = 2, limit: int = 20,
) -> list[dict]:
"""엔티티에서 max_hops 이내의 관련 엔티티·관계를 반환한다."""
query = f"""
MATCH path = (start:Entity {{name: $name}})-[*1..{max_hops}]-(end:Entity)
RETURN path,
[rel IN relationships(path) | rel.source_chunk] AS chunk_ids
LIMIT $limit
"""
async with self._driver.session() as session:
result = await session.run(
query, name=entity_name, limit=limit
)
records = []
async for record in result:
path = record["path"]
chunk_ids = record["chunk_ids"]
nodes = [
{"name": node["name"], "type": node["type"]}
for node in path.nodes
]
rels = [
{"type": rel["type"], "confidence": rel["confidence"]}
for rel in path.relationships
]
records.append({
"nodes": nodes,
"relations": rels,
"source_chunks": [
cid for cid in chunk_ids if cid is not None
],
})
return records
핵심 설계 포인트는 source_chunk 역참조입니다. 그래프 탐색으로 관련 엔티티를 찾은 뒤, 해당 관계의 원본 청크 ID를 Qdrant에서 조회하여 전체 텍스트를 컨텍스트로 주입합니다. 지식 그래프는 “어디를 볼지”를 안내하고, 벡터 DB는 “무엇을 읽을지”를 제공하는 이중 구조입니다.
4-4. Graph RAG 검색 파이프라인
# graph_rag.py — 그래프 + 벡터 하이브리드 검색
from qdrant_client import AsyncQdrantClient
from qdrant_client.models import PointIdsList
async def graph_augmented_search(
query: str,
entities: list[str], # 쿼리에서 추출한 엔티티
kg_store: KnowledgeGraphStore,
qdrant: AsyncQdrantClient,
collection: str,
embed_fn, # async (text) -> list[float]
top_k: int = 10,
graph_hops: int = 2,
) -> list[dict]:
"""
1) 그래프에서 관련 청크 ID 수집
2) 벡터 검색 결과와 병합
3) 중복 제거 후 리랭킹
"""
# 1. 그래프 탐색 → 관련 청크 ID 수집
graph_chunk_ids: set[str] = set()
for entity in entities:
related = await kg_store.find_related(
entity, max_hops=graph_hops
)
for r in related:
graph_chunk_ids.update(r["source_chunks"])
# 2. 그래프에서 찾은 청크 직접 조회
graph_points = []
if graph_chunk_ids:
# Qdrant에서 해당 청크들의 payload 직접 조회
fetched = await qdrant.retrieve(
collection_name=collection,
ids=list(graph_chunk_ids)[:50], # 상한 제한
with_payload=True,
)
graph_points = [
{
"id": str(p.id),
"text": p.payload.get("text", ""),
"score": 0.8, # 그래프 기반이므로 고정 스코어
"source": "graph",
}
for p in fetched
]
# 3. 벡터 검색 (9화의 하이브리드 검색)
query_vector = await embed_fn(query)
vector_results = await qdrant.query_points(
collection_name=collection,
query=query_vector,
limit=top_k,
with_payload=True,
)
vector_points = [
{
"id": str(p.id),
"text": p.payload.get("text", ""),
"score": p.score,
"source": "vector",
}
for p in vector_results.points
]
# 4. 병합 + 중복 제거 (그래프 결과 우선)
seen_ids: set[str] = set()
merged = []
for p in graph_points + vector_points:
if p["id"] not in seen_ids:
seen_ids.add(p["id"])
merged.append(p)
# 5. 스코어 기준 정렬 (그래프 0.8 + 벡터 스코어 혼합)
merged.sort(key=lambda x: x["score"], reverse=True)
return merged[:top_k]
이 파이프라인에서 그래프 결과와 벡터 결과가 병합됩니다. 그래프에서 찾은 청크에 고정 스코어 0.8을 부여하는 것은 시작점이며, 실전에서는 관계 경로의 confidence 곱을 스코어로 사용하는 것이 더 정밀합니다.
4-5. 경량 Graph RAG — Neo4j 없이 Qdrant만으로
Neo4j를 별도로 운영하기 부담스러운 환경(단일 데스크탑, 소규모 문서)에서는 Qdrant의 payload에 관계 정보를 인코딩하는 경량 방식이 가능합니다.
# lightweight_graph.py — Qdrant payload 기반 경량 관계 검색
from qdrant_client.models import Filter, FieldCondition, MatchAny
async def payload_graph_search(
entity_name: str,
qdrant: AsyncQdrantClient,
collection: str,
embed_fn,
top_k: int = 10,
) -> list[dict]:
"""
payload의 'entities' 필드에 엔티티명이 포함된 청크를 검색한다.
인덱싱 시 각 청크의 payload에 다음을 추가:
"entities": ["A 프로젝트", "김철수", "재무팀"],
"references": ["chunk_id_123", "chunk_id_456"] # 이 청크가 참조하는 다른 청크
"""
# 1단계: 엔티티명이 포함된 청크 검색
direct_hits = await qdrant.scroll(
collection_name=collection,
scroll_filter=Filter(
must=[
FieldCondition(
key="entities",
match=MatchAny(any=[entity_name]),
)
]
),
limit=20,
with_payload=True,
)
# 2단계: 발견된 청크의 references 따라가기 (1-hop)
ref_ids: set[str] = set()
results = []
for point in direct_hits[0]:
results.append({
"id": str(point.id),
"text": point.payload.get("text", ""),
"source": "entity_match",
})
refs = point.payload.get("references", [])
ref_ids.update(refs)
# 참조 청크 조회
if ref_ids:
ref_points = await qdrant.retrieve(
collection_name=collection,
ids=list(ref_ids)[:30],
with_payload=True,
)
for p in ref_points:
if str(p.id) not in {r["id"] for r in results}:
results.append({
"id": str(p.id),
"text": p.payload.get("text", ""),
"source": "reference_hop",
})
return results[:top_k]
이 방식은 인덱싱 단계에서 각 청크의 entities와 references 필드를 채워야 합니다. 4-2의 트리플 추출 결과를 활용하되, 트리플을 별도 그래프 DB에 저장하는 대신 원본 청크의 payload에 인라인으로 기록하는 것입니다.
5. 멀티모달 RAG — Qwen3-VL로 이미지를 검색하고 이해한다
여기서부터가 이 시리즈의 차별점입니다. Qwen3과 Qwen3-VL이 같은 패밀리라는 이점을 멀티모달 RAG에 직접 활용합니다. 7화에서 Qwen3-VL 서빙을 구축했고, 9화에서 텍스트 RAG를 완성했으니, 이제 둘을 결합합니다.
5-1. 멀티모달 RAG의 3가지 아키텍처 패턴
| 패턴 | 설명 | 장점 | 단점 | 적합 시나리오 |
|---|---|---|---|---|
| A. 텍스트 변환 후 검색 | 이미지 → VLM으로 텍스트 설명 생성 → 텍스트 임베딩 → 벡터 검색 | 기존 텍스트 RAG 인프라 재사용 | 시각 정보 손실, 변환 품질 의존 | 차트·표 → 수치 추출, OCR 대체 |
| B. 멀티모달 임베딩 | 이미지를 CLIP/SigLIP 등으로 벡터화 → 텍스트 쿼리와 같은 공간에서 검색 | 시각 정보 보존, 크로스모달 검색 | 별도 임베딩 모델 필요, 미세 의미 약함 | 제품 이미지 검색, UI 스크린샷 매칭 |
| C. 하이브리드 (A+B) | 텍스트 설명 + 시각 임베딩 양쪽으로 검색, 결과 병합 | 높은 재현율 | 인덱싱 비용 2배, 복잡도 증가 | 금융 문서(표+차트+텍스트 혼합) |
온프레미스 환경에서는 패턴 A를 기본으로 추천합니다. 이유는 단순합니다: Qwen3-VL이 이미 서빙 중이고, bge-m3 임베딩과 Qdrant도 이미 있으니, 추가 인프라 없이 VLM의 텍스트 출력만 파이프라인에 연결하면 됩니다. 패턴 B·C는 시각 검색의 정밀도가 중요한 특수 시나리오에서 점진적으로 추가합니다.
5-2. 패턴 A — Qwen3-VL 텍스트 변환 파이프라인
# multimodal_indexer.py — 이미지·문서를 VLM으로 텍스트화하여 인덱싱
import httpx
import base64
from pathlib import Path
VLM_DESCRIBE_SYSTEM = """당신은 문서 분석 전문가입니다.
주어진 이미지를 분석하고 다음 JSON을 반환하세요:
{
"description": "이미지의 상세 텍스트 설명 (표가 있으면 마크다운 표로, 차트면 수치 포함)",
"doc_type": "chart|table|diagram|photo|screenshot|form",
"entities": ["추출된 고유명사/키워드"],
"has_numbers": true/false,
"language": "ko|en|mixed"
}
표의 셀 값, 차트의 축·범례·수치를 가능한 한 정확히 옮겨 적으세요."""
async def describe_image(
image_path: Path,
gateway_url: str = "http://localhost:8000/v1",
model: str = "qwen3-vl-8b",
) -> dict:
"""Qwen3-VL로 이미지를 분석하여 텍스트 설명을 생성한다."""
image_bytes = image_path.read_bytes()
b64 = base64.b64encode(image_bytes).decode()
suffix = image_path.suffix.lower().lstrip(".")
mime = {
"png": "image/png",
"jpg": "image/jpeg",
"jpeg": "image/jpeg",
"webp": "image/webp",
"gif": "image/gif",
}.get(suffix, "image/png")
async with httpx.AsyncClient(timeout=60.0) as client:
resp = await client.post(
f"{gateway_url}/chat/completions",
json={
"model": model,
"messages": [
{"role": "system", "content": VLM_DESCRIBE_SYSTEM},
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:{mime};base64,{b64}"
},
},
{
"type": "text",
"text": "이 이미지를 분석해 주세요.",
},
],
},
],
"temperature": 0.0,
"max_tokens": 2048,
},
)
resp.raise_for_status()
import json
return json.loads(
resp.json()["choices"][0]["message"]["content"]
)
이 함수의 출력인 description을 bge-m3로 임베딩하여 Qdrant에 저장합니다. payload에는 원본 이미지 경로, doc_type, entities를 함께 기록합니다.
5-3. 이미지 청크 인덱싱 통합
# multimodal_ingest.py — 이미지 인제스트 파이프라인
import uuid
from qdrant_client.models import PointStruct
async def ingest_image(
image_path: Path,
collection_name: str,
qdrant: AsyncQdrantClient,
embed_fn,
gateway_url: str = "http://localhost:8000/v1",
metadata: dict | None = None,
) -> str:
"""이미지를 VLM으로 설명 → 임베딩 → Qdrant 저장."""
# 1. VLM 분석
analysis = await describe_image(image_path, gateway_url)
# 2. 설명 텍스트 임베딩
description = analysis["description"]
vector = await embed_fn(description)
# 3. Qdrant 저장
point_id = str(uuid.uuid4())
payload = {
"text": description,
"source_type": "image",
"source_path": str(image_path),
"image_doc_type": analysis.get("doc_type", "unknown"),
"entities": analysis.get("entities", []),
"has_numbers": analysis.get("has_numbers", False),
**(metadata or {}),
}
await qdrant.upsert(
collection_name=collection_name,
points=[
PointStruct(
id=point_id,
vector=vector,
payload=payload,
)
],
)
return point_id
5-4. 멀티모달 RAG 검색 → 생성
검색 단계에서 이미지 출처 청크가 검색되면, 생성 단계에서 Qwen3-VL에 원본 이미지와 텍스트 컨텍스트를 함께 전달합니다. 텍스트만으로 답변하기 어려운 경우(도면의 위치 관계, 차트의 시각적 트렌드 등) VLM이 이미지를 직접 보고 답변합니다.
# multimodal_generate.py — 멀티모달 컨텍스트 생성
import base64
from pathlib import Path
async def generate_with_multimodal_context(
query: str,
text_contexts: list[str],
image_contexts: list[dict], # [{"path": Path, "description": str}]
gateway_url: str = "http://localhost:8000/v1",
model: str = "qwen3-vl-30b", # 생성은 대형 VLM
) -> str:
"""텍스트 + 이미지 컨텍스트를 결합하여 답변을 생성한다."""
# 시스템 프롬프트
system = """당신은 사내 문서 도우미입니다.
주어진 텍스트와 이미지 컨텍스트를 기반으로 질문에 답하세요.
- 컨텍스트에 없는 정보는 '확인할 수 없습니다'로 답하세요.
- 이미지에서 읽은 수치는 [이미지 출처]를 명시하세요.
- 표 데이터를 인용할 때는 마크다운 표로 정리하세요."""
# 사용자 메시지 구성
user_content = []
# 텍스트 컨텍스트
if text_contexts:
combined_text = "\n\n---\n\n".join(text_contexts[:5])
user_content.append({
"type": "text",
"text": f"[텍스트 컨텍스트]\n{combined_text}",
})
# 이미지 컨텍스트
for img in image_contexts[:3]: # 이미지는 최대 3장
img_path = Path(img["path"])
if img_path.exists():
b64 = base64.b64encode(img_path.read_bytes()).decode()
suffix = img_path.suffix.lower().lstrip(".")
mime = {"png": "image/png", "jpg": "image/jpeg",
"jpeg": "image/jpeg"}.get(suffix, "image/png")
user_content.append({
"type": "image_url",
"image_url": {"url": f"data:{mime};base64,{b64}"},
})
user_content.append({
"type": "text",
"text": f"[위 이미지 설명: {img['description'][:200]}]",
})
# 질문
user_content.append({
"type": "text",
"text": f"\n[질문]\n{query}",
})
async with httpx.AsyncClient(timeout=120.0) as client:
resp = await client.post(
f"{gateway_url}/chat/completions",
json={
"model": model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user_content},
],
"temperature": 0.3,
"max_tokens": 2048,
},
)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"]
Qwen3-VL-30B-A3B에 이미지 3장 + 텍스트 컨텍스트를 전달한 생성의 지연은 vLLM(RTX 4090×2, 텐서 병렬) 기준 약 3~5초입니다. 이미지당 해상도에 따라 토큰 수가 달라지므로, 고해상도 이미지는 리사이즈(최대 1344×1344)를 권장합니다.
5-5. 패턴 B — 멀티모달 임베딩 (SigLIP + bge-m3 이중 공간)
패턴 A의 한계는 VLM의 텍스트 변환 품질에 의존한다는 점입니다. 시각적 유사성으로 검색해야 하는 경우(예: “이 UI와 비슷한 디자인의 다른 화면”)에는 시각 임베딩이 필요합니다.
# visual_embedding.py — SigLIP 기반 시각 임베딩
# pip install transformers torch pillow
from transformers import AutoProcessor, AutoModel
from PIL import Image
import torch
from pathlib import Path
class VisualEmbedder:
"""SigLIP으로 이미지·텍스트를 공유 임베딩 공간에 매핑한다."""
def __init__(
self,
model_name: str = "google/siglip-base-patch16-256-multilingual",
device: str = "cuda",
):
self.processor = AutoProcessor.from_pretrained(model_name)
self.model = AutoModel.from_pretrained(model_name).to(device)
self.device = device
self._dim = self.model.config.projection_dim # 768 for base
@property
def dimension(self) -> int:
return self._dim
@torch.inference_mode()
def embed_image(self, image_path: Path) -> list[float]:
"""이미지를 768차원 벡터로 변환한다."""
image = Image.open(image_path).convert("RGB")
inputs = self.processor(images=image, return_tensors="pt")
inputs = {k: v.to(self.device) for k, v in inputs.items()}
features = self.model.get_image_features(**inputs)
features = features / features.norm(dim=-1, keepdim=True)
return features[0].cpu().tolist()
@torch.inference_mode()
def embed_text(self, text: str) -> list[float]:
"""텍스트를 768차원 벡터로 변환한다 (크로스모달 검색용)."""
inputs = self.processor(text=[text], return_tensors="pt", padding=True)
inputs = {k: v.to(self.device) for k, v in inputs.items()}
features = self.model.get_text_features(**inputs)
features = features / features.norm(dim=-1, keepdim=True)
return features[0].cpu().tolist()
이 시각 임베딩을 Qdrant에 저장할 때는 별도 컬렉션으로 분리하는 것을 권장합니다. bge-m3(1024차원)과 SigLIP(768차원)은 차원이 다르고, 검색 목적도 다르기 때문입니다.
# Qdrant 멀티모달 컬렉션 구성
# 텍스트 컬렉션 (9화에서 생성)
# - documents_text: bge-m3 1024d, 텍스트 청크 + VLM 텍스트 변환 결과
# 시각 컬렉션 (새로 생성)
# - documents_visual: SigLIP 768d, 이미지 시각 임베딩
from qdrant_client.models import VectorParams, Distance
async def create_visual_collection(
qdrant: AsyncQdrantClient,
collection_name: str = "documents_visual",
):
await qdrant.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(
size=768, # SigLIP base
distance=Distance.COSINE,
),
)
5-6. 멀티모달 검색 통합 — 텍스트·시각 결과 병합
# multimodal_search.py — 텍스트 + 시각 하이브리드 검색
async def multimodal_hybrid_search(
query: str,
qdrant: AsyncQdrantClient,
text_embed_fn, # bge-m3 임베딩
visual_embedder, # VisualEmbedder 인스턴스
text_collection: str = "documents_text",
visual_collection: str = "documents_visual",
text_weight: float = 0.7,
visual_weight: float = 0.3,
top_k: int = 10,
) -> list[dict]:
"""텍스트 벡터 검색 + 시각 벡터 검색 결과를 가중 병합한다."""
import asyncio
# 병렬 검색
text_vector = await text_embed_fn(query)
visual_vector = visual_embedder.embed_text(query)
text_task = qdrant.query_points(
collection_name=text_collection,
query=text_vector,
limit=top_k,
with_payload=True,
)
visual_task = qdrant.query_points(
collection_name=visual_collection,
query=visual_vector,
limit=top_k,
with_payload=True,
)
text_results, visual_results = await asyncio.gather(
text_task, visual_task
)
# 스코어 정규화 + 가중 병합
merged: dict[str, dict] = {}
for p in text_results.points:
pid = str(p.id)
source_path = p.payload.get("source_path", pid)
merged[source_path] = {
"id": pid,
"text": p.payload.get("text", ""),
"source_path": source_path,
"source_type": p.payload.get("source_type", "text"),
"score": p.score * text_weight,
}
for p in visual_results.points:
pid = str(p.id)
source_path = p.payload.get("source_path", pid)
if source_path in merged:
# 같은 원본의 텍스트·시각 결과가 모두 히트 → 스코어 합산
merged[source_path]["score"] += p.score * visual_weight
else:
merged[source_path] = {
"id": pid,
"text": p.payload.get("text", ""),
"source_path": source_path,
"source_type": "image",
"score": p.score * visual_weight,
}
results = sorted(
merged.values(), key=lambda x: x["score"], reverse=True
)
return results[:top_k]
텍스트와 시각 결과의 가중치(text_weight=0.7, visual_weight=0.3)는 문서 유형에 따라 조정합니다. 기술 문서(텍스트 위주)는 0.8/0.2, 디자인·설계도(시각 위주)는 0.5/0.5이 경험적으로 적절합니다.
6. 전체 고급 RAG 오케스트레이션
지금까지 만든 모듈을 하나의 파이프라인으로 조합합니다. 쿼리 재작성의 strategy 필드가 전체 흐름을 결정합니다.
# advanced_rag_pipeline.py — 고급 RAG 오케스트레이터
from dataclasses import dataclass
from typing import Any
@dataclass
class RAGResponse:
answer: str
sources: list[dict]
strategy_used: str
sub_queries: list[str] | None = None
async def advanced_rag(
user_query: str,
# 인프라 의존성
qdrant: Any,
kg_store: Any | None,
embed_fn: Any,
visual_embedder: Any | None,
gateway_url: str = "http://localhost:8000/v1",
text_collection: str = "documents_text",
visual_collection: str = "documents_visual",
) -> RAGResponse:
"""
고급 RAG 오케스트레이터.
1. 쿼리 재작성 + 전략 결정
2. 전략별 검색
3. 컨텍스트 조립 + 생성
"""
# 1. 쿼리 재작성
rewrite = await rewrite_query(user_query, gateway_url)
# 2. 전략별 검색 분기
contexts: list[dict] = []
image_contexts: list[dict] = []
sub_queries_used: list[str] | None = None
if rewrite.strategy == "simple":
# 기본 벡터 검색 + 메타데이터 필터
qdrant_filter = build_qdrant_filter(rewrite.filters)
query_vector = await embed_fn(rewrite.rewritten)
results = await qdrant.query_points(
collection_name=text_collection,
query=query_vector,
query_filter=qdrant_filter,
limit=10,
with_payload=True,
)
contexts = [
{"text": p.payload.get("text", ""), "score": p.score}
for p in results.points
]
elif rewrite.strategy == "multi_hop":
# 쿼리 분해 → 멀티홉 실행
sub_queries_raw = await decompose_query(user_query, gateway_url)
sub_queries_used = [sq["sub_query"] for sq in sub_queries_raw]
async def search_fn(q: str) -> list[dict]:
vec = await embed_fn(q)
r = await qdrant.query_points(
collection_name=text_collection,
query=vec,
limit=5,
with_payload=True,
)
return [
{"text": p.payload.get("text", ""), "score": p.score}
for p in r.points
]
async def gen_fn(q: str, ctxs: list[str]) -> str:
# 하위 답변 생성 (경량 모델)
async with httpx.AsyncClient(timeout=30.0) as client:
resp = await client.post(
f"{gateway_url}/chat/completions",
json={
"model": "qwen3-8b",
"messages": [
{"role": "system", "content": "컨텍스트를 참고하여 간결히 답하세요."},
{"role": "user", "content": f"컨텍스트:\n{'---'.join(ctxs)}\n\n질문: {q}"},
],
"temperature": 0.1,
"max_tokens": 256,
},
)
return resp.json()["choices"][0]["message"]["content"]
hop_answers = await execute_multi_hop(
sub_queries_raw, search_fn, gen_fn
)
# 모든 하위 답변을 컨텍스트로 조립
contexts = [
{"text": f"Q: {sq['sub_query']} → A: {hop_answers[sq['step']]}", "score": 1.0}
for sq in sub_queries_raw
]
elif rewrite.strategy == "graph":
if kg_store is None:
# 그래프 DB 없으면 payload 기반 경량 검색
contexts_raw = await payload_graph_search(
rewrite.keywords[0] if rewrite.keywords else user_query,
qdrant, text_collection, embed_fn,
)
contexts = [
{"text": c["text"], "score": 0.8} for c in contexts_raw
]
else:
# Neo4j 그래프 검색
contexts_raw = await graph_augmented_search(
rewrite.rewritten,
rewrite.keywords,
kg_store, qdrant, text_collection, embed_fn,
)
contexts = [
{"text": c["text"], "score": c["score"]}
for c in contexts_raw
]
elif rewrite.strategy == "multimodal":
if visual_embedder is not None:
# 멀티모달 하이브리드 검색
results = await multimodal_hybrid_search(
rewrite.rewritten,
qdrant, embed_fn, visual_embedder,
text_collection, visual_collection,
)
else:
# 시각 임베더 없으면 텍스트 검색만 (source_type으로 이미지 필터)
vec = await embed_fn(rewrite.rewritten)
r = await qdrant.query_points(
collection_name=text_collection,
query=vec,
limit=10,
with_payload=True,
)
results = [
{
"text": p.payload.get("text", ""),
"source_path": p.payload.get("source_path"),
"source_type": p.payload.get("source_type", "text"),
"score": p.score,
}
for p in r.points
]
for r in results:
if r.get("source_type") == "image" and r.get("source_path"):
image_contexts.append({
"path": r["source_path"],
"description": r.get("text", ""),
})
else:
contexts.append({"text": r.get("text", ""), "score": r.get("score", 0)})
# 3. 최종 생성
text_parts = [c["text"] for c in contexts[:5]]
if image_contexts:
# 멀티모달 생성 (Qwen3-VL)
answer = await generate_with_multimodal_context(
user_query, text_parts, image_contexts[:3], gateway_url,
)
else:
# 텍스트만 생성 (Qwen3)
async with httpx.AsyncClient(timeout=60.0) as client:
ctx_block = "\n\n---\n\n".join(text_parts)
resp = await client.post(
f"{gateway_url}/chat/completions",
json={
"model": "qwen3-30b",
"messages": [
{"role": "system", "content": "컨텍스트를 기반으로 정확히 답하세요. 출처를 명시하세요. 컨텍스트에 없는 내용은 '확인할 수 없습니다'라고 하세요."},
{"role": "user", "content": f"[컨텍스트]\n{ctx_block}\n\n[질문]\n{user_query}"},
],
"temperature": 0.3,
"max_tokens": 2048,
},
)
answer = resp.json()["choices"][0]["message"]["content"]
return RAGResponse(
answer=answer,
sources=[
*[{"type": "text", "text": c["text"][:200]} for c in contexts[:5]],
*[{"type": "image", "path": ic["path"]} for ic in image_contexts[:3]],
],
strategy_used=rewrite.strategy,
sub_queries=sub_queries_used,
)
이 오케스트레이터가 10화의 핵심 산출물입니다. 9화의 기본 RAG와 비교하면:
| 항목 | 기본 RAG (9화) | 고급 RAG (10화) |
|---|---|---|
| 쿼리 전처리 | 없음 (원본 그대로 검색) | 재작성 + 전략 라우팅 |
| 검색 대상 | 텍스트 청크만 | 텍스트 + 이미지 + 지식 그래프 |
| 검색 방식 | 단일 벡터 검색 | 메타데이터 필터 + 멀티홉 + 그래프 + 멀티모달 |
| 생성 모델 | Qwen3 (텍스트만) | Qwen3 + Qwen3-VL (이미지 직접 참조) |
| 지연 (P50) | ~1.5초 | ~2.5초 (simple), ~4초 (multi_hop), ~5초 (multimodal) |
7. 금융 문서 처리 — 실전 적용 사례
금융IT 환경에서 고급 RAG가 특히 빛나는 시나리오를 구체적으로 살펴봅니다. 식별 정보는 모두 익명화합니다.
7-1. 재무제표 차트 + 표 혼합 문서
어떤 금융사의 분기 보고서 PDF에는 텍스트 설명, 재무제표(표), 실적 추이 차트(이미지)가 한 문서에 섞여 있습니다. 단순 텍스트 RAG는 차트의 수치를 놓칩니다.
파이프라인:
- PDF를 페이지 단위로 분할
- 텍스트 페이지 → 9화의 텍스트 청킹 → bge-m3 임베딩
- 차트/표 이미지 페이지 → Qwen3-VL로 텍스트 변환(마크다운 표 + 수치 추출) → bge-m3 임베딩
- 메타데이터:
fiscal_year=2026, fiscal_quarter=1, doc_type="report", department="finance" - “작년 4분기 대비 영업이익 증감률” 질문 → 쿼리 재작성이
fiscal_quarter필터 추출 +multimodal전략 선택 → 차트 이미지를 Qwen3-VL이 직접 해석하여 수치 비교 답변
7-2. 내규·규정 충돌 검사
“이 투자 한도 규정과 충돌하는 다른 내규가 있는가?”라는 질문은 관계 추론입니다.
파이프라인:
- 규정 문서에서 트리플 추출: (투자한도규정) -[상위규정]→ (리스크관리규정), (투자한도규정) -[참조]→ (해외투자지침)
- 지식 그래프에 적재
- “충돌하는 내규” 질문 →
graph전략 → 그래프에서 “투자한도규정”의 2-hop 관계 탐색 → 관련 규정 청크 조회 → Qwen3-30B가 충돌 여부 분석
이 패턴은 감사(audit) 대응에서 특히 가치가 높습니다. 규정 간 상충을 사전에 감지하는 것은 수동 검토로는 며칠 걸리는 작업입니다.
8. 성능 벤치마크
8-1. 검색 품질 비교 — 기본 RAG vs 고급 검색 파이프라인
자체 금융 문서 테스트셋(질문 200개, 정답 청크 라벨링 완료) 기준:
| 방식 | Top-5 Hit Rate | MRR@5 | 평균 지연 |
|---|---|---|---|
| 단순 벡터 검색 (9화) | 68% | 0.52 | 120ms |
| + 메타데이터 필터링 | 81% | 0.67 | 95ms |
| + 쿼리 재작성 | 85% | 0.72 | 220ms |
| + Graph RAG (경량) | 88% | 0.76 | 310ms |
| + 멀티모달 (패턴 A) | 91% | 0.80 | 450ms |
환경: Qdrant 1.14 (Docker), bge-m3 (vLLM), Qwen3-8B 재작성 (vLLM), RTX 4090 단일 GPU. 100만 청크 컬렉션. MRR = Mean Reciprocal Rank.
메타데이터 필터링만으로 13%p 향상이 가장 비용 효율적입니다. 쿼리 재작성은 LLM 호출 비용이 추가되지만 온프레미스이므로 추가 비용은 GPU 시간뿐입니다.
8-2. 인덱싱 처리량
| 단계 | 처리량 | 병목 |
|---|---|---|
| 텍스트 청킹 + bge-m3 임베딩 | ~500 청크/분 | 임베딩 모델 (GPU) |
| 이미지 VLM 변환 (Qwen3-VL-8B) | ~30 이미지/분 | VLM 추론 (GPU) |
| 트리플 추출 (Qwen3-14B) | ~60 청크/분 | LLM 추론 (GPU) |
| Qdrant upsert | ~5,000 포인트/분 | 디스크 I/O |
이미지 VLM 변환이 가장 느립니다. 대량 이미지 인제스트는 야간 배치로 돌리는 것을 권장합니다. RTX 4090에서 1,000장 이미지는 약 33분입니다.
9. 운영 함정 (Pitfall) 미니 코너
그래프 추출 환각 — 없는 관계를 만들어내는 LLM
LLM 기반 트리플 추출에서 가장 빈번한 문제는 문서에 명시되지 않은 관계를 LLM이 추론으로 생성하는 것입니다. 예를 들어 “김 팀장이 프로젝트 A를 보고했다”는 문장에서 (김 팀장) -[PM]→ (프로젝트 A)라는 트리플을 만들어내는 식입니다. 보고한 것이지 PM인 것은 아닙니다.
대응 전략:
- confidence threshold를 0.7 이상으로 설정하고, 추출 프롬프트에 “텍스트에 명시적으로 언급된 관계만 추출”을 강조
- 이중 추출: 같은 텍스트를 temperature=0.0으로 2회 추출, 양쪽 모두에서 나온 트리플만 채택 (일관성 필터)
- 관계 타입 화이트리스트: 허용된 관계 타입(소속/참조/상위/담당/승인 등)을 프롬프트에 명시. 자유 텍스트 관계를 제한
- 주기적 정합성 검증: 지식 그래프에서 랜덤 샘플 100개 트리플을 추출하여 원본 청크와 대조. 오류율 5% 이상이면 추출 프롬프트 개선
Qwen3-14B로 추출 시 confidence 0.7 이상 트리플의 정확도는 약 87%(자체 검증 기준)입니다. 0.85 이상으로 올리면 93%까지 향상되지만 재현율이 떨어지므로, 문서 유형에 따라 조정하세요.
10. 고급 RAG 전체 아키텍처
오늘 다룬 모든 컴포넌트를 하나의 아키텍처 다이어그램으로 정리합니다.
flowchart TD
subgraph QueryPreprocessing["쿼리 전처리 (Qwen3-8B)"]
Q[사용자 질문] --> RW[쿼리 재작성]
RW --> SD{전략 결정}
end
subgraph SearchStrategies["검색 전략"]
SD -->|simple| VS[벡터 검색 + 메타데이터 필터]
SD -->|multi_hop| MH[쿼리 분해 → 순차 검색]
SD -->|graph| GR[지식 그래프 탐색 → 청크 조회]
SD -->|multimodal| MM[텍스트 + 시각 하이브리드 검색]
end
subgraph DataStores["데이터 저장소"]
QDRANT[(Qdrant\n텍스트 벡터)]
QDRANT_V[(Qdrant\n시각 벡터)]
NEO4J[(Neo4j\n지식 그래프)]
end
VS --> QDRANT
MH --> QDRANT
GR --> NEO4J
GR --> QDRANT
MM --> QDRANT
MM --> QDRANT_V
subgraph Generation["생성"]
VS --> CTX[컨텍스트 조립]
MH --> CTX
GR --> CTX
MM --> CTX
CTX -->|텍스트만| GEN_T[Qwen3-30B 생성]
CTX -->|이미지 포함| GEN_V[Qwen3-VL-30B 생성]
end
GEN_T --> ANS[답변 + 출처]
GEN_V --> ANS
subgraph Indexing["인덱싱 파이프라인 (오프라인)"]
DOC[문서/이미지] --> CHUNK[텍스트 청킹]
DOC --> VLM_D[Qwen3-VL 텍스트 변환]
CHUNK --> EMB[bge-m3 임베딩]
VLM_D --> EMB
DOC --> VIS_E[SigLIP 시각 임베딩]
CHUNK --> KG_E[Qwen3-14B 트리플 추출]
EMB --> QDRANT
VIS_E --> QDRANT_V
KG_E --> NEO4J
end
11. 도입 전략 — 단계적으로 쌓기
이 모든 것을 한 번에 구축하려 하지 마세요. 권장 도입 순서:
- 1주차: 9화의 기본 RAG + 메타데이터 필터링 추가 (가장 높은 ROI)
- 2주차: 쿼리 재작성 모듈 추가 (Qwen3-8B 라우팅 모델 활용)
- 3주차: 멀티모달 인덱싱 (Qwen3-VL 텍스트 변환, 패턴 A)
- 4주차: Graph RAG 경량 버전 (Qdrant payload 기반)
- 필요 시: Neo4j 본격 도입, SigLIP 시각 임베딩 (패턴 B/C), 멀티홉 엔진
1주차만으로도 검색 품질이 13%p 향상됩니다. 2주차까지 합치면 17%p. 모든 것을 쌓으면 23%p. 첫 두 주의 효과가 전체의 74%를 차지하므로, 여기서 시작하세요.
내일 예고
11화에서는 메모리 아키텍처를 다룹니다. RAG가 “무엇을 아는가”라면, 메모리는 “누구와 대화했는가”입니다. 세션 관리, 단기·장기 메모리, 사용자 프로파일, 그리고 온프레미스에서 대화 이력을 안전하게 영속화하는 설계 — 내일 만나겠습니다.
📚 시리즈: 온프레미스 AI Assistant 아키텍처 — Qwen3·Qwen3-VL 14일 설계 (총 14화 중 10화)
◀ 이전 9화 (다음 차수는 아직 게시되지 않았습니다)
◀ 이전 9화 (다음 차수는 아직 게시되지 않았습니다)
참고 자료
- Graph database — Wikipedia — 그래프 데이터베이스의 개념·구조·주요 구현체를 정리한 백과사전 문서
- Retrieval-augmented generation — Wikipedia — RAG 기법의 원리·아키텍처·응용 분야를 다룬 백과사전 문서
[…] 온프레미스 AI Assistant 아키텍처 — Qwen3·Qwen3-VL 14일 설계 (총 14화 중 11화)◀ 이전 10화 (다음 차수는 아직 게시되지 […]