AI Agents · レッスン

複数文書Q&Aエージェント

Q: 「複数文書Q&Aエージェント」で何を学びますか？

文書コーパスをインデックス化し、すべての文書を横断して質問に回答します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

文書コーパスをインデックス化し、すべての文書を横断して質問に回答します。

「複数文書Q&Aエージェント」はCoddyKit上の無料AI Agentsレッスンです。これはレッスン3/4です。下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。

複数文書Q&Aの概要

複数文書Q&Aエージェントは、N個の文書の集合から関連するコンテンツを検索し、回答を統合して、それぞれの主張を出典に対応付けることで質問に答えます。

単一文書のQ&Aとは異なり、複数文書エージェントは、ソース間で情報が矛盾している場合に対応し、質問に最も関連する文書を判断する必要があります。

複数文書のインデックス作成

質問に答える前に、すべての文書をインデックス化する必要があります。つまり、解析、チャンク化、埋め込みの生成を行い、ベクトルデータベースに保存します。各チャンクには、元の文書に戻れるようにソース文書と関連付けるメタデータを付けて保存します。

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))

文書横断の推論

質問によっては、単一の一致するチャンクを見つけるだけでなく、複数の文書から情報を統合する必要があります。例として、「3つの契約書のうち、違約金条項が最も低いのはどれですか。」があります。

2段階のアプローチを使用します。各文書から関連するチャンクを取得し、その後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による再ランキング

初回のベクトル検索では、bi-encoder（高速で近似的）を使用します。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')

理解度チェック

検索拡張生成を使用する複数文書Q&Aシステムで、関連度の閾値は何を防ぐためのものですか。

復習：複数文書Q&Aエージェント

複数文書Q&Aでは、すべての文書をインデックス化（解析 → チャンク化 → 埋め込み → メタデータ付きで保存）→ すべての文書から関連チャンクを検索 → ソースラベル付きで整形 → 引用付きで回答を統合します。

高度なテクニックには、比較を求める質問に対する文書横断の比較、プロンプトによる矛盾検出、関連度の閾値によるフィルタリング、精度を高めるcross-encoderによる再ランキング、特定の文書や日付範囲にクエリを限定するメタデータフィルタリングがあります。

よくある質問