AI Agents · 강의

다중 문서 질의응답 에이전트

문서 모음을 색인하고 모든 문서에 걸쳐 질문에 답합니다.

다중 문서 질의응답 에이전트은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

다중 문서 질의응답 개요

다중 문서 질의응답 에이전트는 N개 문서 모음에서 관련 콘텐츠를 검색하고, 답변을 종합하며, 각 주장을 출처에 연결해 질문에 답합니다.

단일 문서 질의응답과 달리 다중 문서 에이전트는 여러 출처에 걸친 정보 충돌을 처리하고, 질문과 가장 관련성이 높은 문서가 무엇인지 추론해야 합니다.

여러 문서 색인하기

질문에 답하기 전에 모든 문서를 색인해야 합니다. 즉, 문서를 구문 분석하고 청크로 나누고 임베딩한 후 벡터 데이터베이스에 저장해야 합니다. 각 청크에는 원본 문서로 연결되는 메타데이터가 함께 저장됩니다.

import chromadb
from chromadb.utils import embedding_functions
import os

client = chromadb.PersistentClient(path='./doc_index')
ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv('OPENAI_API_KEY'),
    model_name='text-embedding-3-small'
)
collection = client.get_or_create_collection('documents', embedding_function=ef)

def index_document(doc_id, doc_path, doc_title):
    # Parse and chunk
    chunks = pdf_to_chunks(doc_path, chunk_size=800, overlap=150)
    for i, chunk in enumerate(chunks):
        chunk_id = f'{doc_id}_chunk_{i}'
        collection.add(
            ids=[chunk_id],
            documents=[chunk['text']],
            metadatas=[{
                'doc_id': doc_id,
                'title': doc_title,
                'page': chunk['page'],
                'source_file': doc_path
            }]
        )
    print(f'Indexed {len(chunks)} chunks from: {doc_title}')

프롬프트에 출처 표시하기

검색한 청크를 LLM에 전달할 때 각 청크에 문서 출처를 표시하세요. 그러면 LLM이 답변에서 번호가 매겨진 참조를 사용해 출처를 인용할 수 있습니다.

def format_chunks_for_prompt(chunks, max_chars=4000):
    sections = []
    used_chars = 0

    for i, chunk in enumerate(chunks, 1):
        meta = chunk['metadata']
        header = f"[Source {i}: {meta['title']}, page {meta.get('page', '?')}]"
        content = chunk['text'][:600]
        entry = f'{header}\n{content}'

        if used_chars + len(entry) > max_chars:
            break

        sections.append(entry)
        used_chars += len(entry)

    return '\n\n'.join(sections)

QA_PROMPT = '''Answer the question based on the provided document excerpts.
Cite sources as [Source N]. If sources conflict, mention both views.

{context}

Question: {question}
Answer:'''

def answer_question(question):
    chunks = retrieve_relevant_chunks(question, n_results=6)
    context = format_chunks_for_prompt(chunks)
    return llm_call(QA_PROMPT.format(context=context, question=question))

문서 간 추론

일부 질문은 일치하는 청크 하나를 찾는 것만으로는 부족하고 여러 문서의 정보를 종합해야 합니다. 예를 들어 "세 계약 중 위약 조항이 가장 낮은 계약은 무엇입니까?"와 같은 질문입니다.

두 단계 접근 방식을 사용하세요. 각 문서에서 관련 청크를 검색한 다음 LLM에 문서 간 내용을 비교하고 종합하도록 요청하세요.

def cross_document_compare(question, doc_ids):
    # Retrieve best chunks per document
    per_doc_chunks = {}
    for doc_id in doc_ids:
        results = collection.query(
            query_texts=[question],
            n_results=3,
            where={'doc_id': {'$eq': doc_id}}  # filter by document
        )
        if results['documents'][0]:
            per_doc_chunks[doc_id] = results['documents'][0]

    # Format with document labels
    context_parts = []
    for doc_id, texts in per_doc_chunks.items():
        doc_label = f'Document {doc_id}'
        combined = ' '.join(texts[:2])[:600]
        context_parts.append(f'== {doc_label} ==\n{combined}')

    comparison_context = '\n\n'.join(context_parts)
    return llm_call(f'Compare these documents to answer: {question}\n\n{comparison_context}')

충돌하는 정보 처리하기

서로 다른 문서에서 충돌하는 사실을 제시할 수 있습니다. 한 계약서에는 대금 지급 기한이 30일이라고 되어 있지만 다른 계약서에는 60일이라고 되어 있을 수 있습니다. 에이전트는 한쪽을 조용히 선택하는 대신 이러한 충돌을 감지하고 사용자에게 드러내야 합니다.

CONFLICT_PROMPT = '''You are analyzing multiple document sources.
Some may contain conflicting information.

For each factual claim you make:
1. Cite the source document
2. If another source contradicts it, explicitly note the conflict
3. Indicate which source you believe is more authoritative, if possible

Document excerpts:
{context}

Question: {question}

Answer (with conflict notes where applicable):'''

def answer_with_conflict_detection(question):
    chunks = retrieve_relevant_chunks(question, n_results=8)
    context = format_chunks_for_prompt(chunks)
    return llm_call(CONFLICT_PROMPT.format(
        context=context, question=question
    ))

출처 인용 생성하기

답변을 생성한 후 어떤 출처 문서가 인용되었는지 추출하고 구조화된 목록으로 반환하세요. 이를 통해 사용자가 검증을 위해 원본 문서를 쉽게 찾을 수 있습니다.

import re

def extract_citations(answer_text, chunks):
    # Find all [Source N] references in the answer
    cited_nums = set(int(m) for m in re.findall(r'\[Source (\d+)\]', answer_text))

    citations = []
    for num in sorted(cited_nums):
        idx = num - 1
        if idx < len(chunks):
            meta = chunks[idx]['metadata']
            citations.append({
                'source_num': num,
                'title':     meta.get('title', 'Unknown'),
                'page':      meta.get('page', 'N/A'),
                'file':      meta.get('source_file', '')
            })

    return citations

def answer_with_citations(question):
    chunks = retrieve_above_threshold(question)
    context = format_chunks_for_prompt(chunks)
    answer = llm_call(QA_PROMPT.format(context=context, question=question))
    citations = extract_citations(answer, chunks)
    return {'answer': answer, 'citations': citations}

Cross-Encoder로 재순위화하기

초기 벡터 검색은 이중 인코더를 사용하므로 빠르지만 근사적입니다. 교차 인코더는 각 (질의, 청크) 쌍을 함께 평가해 상위 결과의 순위를 다시 매깁니다. 더 정확하지만 느립니다. 이 2단계 접근 방식은 최종 답변의 품질을 향상합니다.

# pip install sentence-transformers
from sentence_transformers import CrossEncoder

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def rerank_chunks(question, chunks, top_k=4):
    # Score each chunk against the question
    pairs = [(question, c['text']) for c in chunks]
    scores = reranker.predict(pairs)

    # Attach scores and re-sort
    scored_chunks = list(zip(scores, chunks))
    scored_chunks.sort(key=lambda x: x[0], reverse=True)

    top_chunks = [chunk for _, chunk in scored_chunks[:top_k]]
    print(f'Re-ranked {len(chunks)} chunks -> kept top {top_k}')
    return top_chunks

def answer_with_reranking(question):
    # Retrieve more initially
    initial_chunks = retrieve_relevant_chunks(question, n_results=12)
    # Re-rank for precision
    top_chunks = rerank_chunks(question, initial_chunks, top_k=4)
    context = format_chunks_for_prompt(top_chunks)
    return llm_call(QA_PROMPT.format(context=context, question=question))

문서 수준 메타데이터 필터링

사용자가 특정 문서나 날짜 범위를 지정하면 임베딩 검색 전에 메타데이터 수준에서 필터링하세요. 이렇게 하면 관련 없는 문서가 검색 결과에 섞이는 것을 방지할 수 있습니다.

def retrieve_filtered(question, filters=None, n_results=8):
    query_kwargs = {
        'query_texts': [question],
        'n_results': n_results,
        'include': ['documents', 'metadatas', 'distances']
    }

    # ChromaDB metadata filters
    # Example: {'doc_id': 'contract_2024', 'year': {'$gte': 2023}}
    if filters:
        query_kwargs['where'] = filters

    results = collection.query(**query_kwargs)
    return [
        {'text': t, 'metadata': m, 'relevance': 1 - d}
        for t, m, d in zip(
            results['documents'][0],
            results['metadatas'][0],
            results['distances'][0]
        )
    ]

# Example usage
chunks = retrieve_filtered(
    'What are the payment terms?',
    filters={'doc_id': {'$in': ['contract_a', 'contract_b']}}
)

새 문서로 색인 업데이트하기

문서 모음은 시간이 지나면서 변합니다. 새 파일이 추가되고 기존 파일이 수정됩니다. 색인 처리 과정은 증분 업데이트를 지원해야 합니다. 새 문서를 추가하고, 수정된 문서를 다시 색인하며, 삭제된 문서를 제거하세요.

import os
import hashlib

# Track indexed documents by file hash
index_registry = {}  # {filepath: {hash, doc_id, indexed_at}}

def file_hash(filepath):
    with open(filepath, 'rb') as f:
        return hashlib.md5(f.read()).hexdigest()

def index_if_new_or_changed(filepath, title):
    fhash = file_hash(filepath)
    existing = index_registry.get(filepath)

    if existing and existing['hash'] == fhash:
        print(f'Skipping unchanged: {title}')
        return existing['doc_id']

    if existing:
        # Remove old chunks from vector store
        collection.delete(where={'doc_id': {'': existing['doc_id']}})
        print(f'Re-indexing updated: {title}')
    else:
        print(f'Indexing new: {title}')

    doc_id = hashlib.md5(filepath.encode()).hexdigest()[:8]
    index_document(doc_id, filepath, title)
    index_registry[filepath] = {'hash': fhash, 'doc_id': doc_id}
    return doc_id

def sync_document_directory(directory):
    pdf_files = [f for f in os.listdir(directory) if f.endswith('.pdf')]
    for fname in pdf_files:
        fpath = os.path.join(directory, fname)
        title = fname.replace('.pdf', '').replace('_', ' ').title()
        index_if_new_or_changed(fpath, title)
    print(f'Sync complete: {len(pdf_files)} files processed')

지식 확인

검색 증강 생성을 사용하는 다중 문서 질의응답 시스템에서 관련성 임계값은 무엇을 방지합니까?

복습: 다중 문서 질의응답 에이전트

다중 문서 질의응답의 과정은 다음과 같습니다. 모든 문서를 색인(구문 분석 → 청크로 나누기 → 임베딩 → 메타데이터와 함께 저장) → 모든 문서에서 관련 청크 검색 → 출처 레이블을 붙여 형식화 → 인용과 함께 답변 종합.

고급 기법으로는 비교 질문을 위한 문서 간 비교, 충돌 감지를 유도하는 프롬프트, 관련성 임계값 필터링, 정밀도를 높이는 교차 인코더 재순위화, 특정 문서나 날짜 범위로 질의 범위를 제한하는 메타데이터 필터링이 있습니다.

자주 묻는 질문