Zenn
👋

生成AIを活用した類似文書検索の実装と課題

2025/03/26に公開
1

生成AI技術の進化とともに、類似文書検索(Semantic Search)の領域でも革新的なアプローチが可能になってきました。本記事では、生成AIを活用した類似文書検索システムの構築経験と、その過程で直面した課題、特にドキュメント分割とベクトル化における問題について詳しく解説します。

目次

  1. 類似文書検索と生成AI
  2. 類似文書検索パイプラインの構築
  3. ドキュメント分割の課題
  4. 実装上の解決策
  5. 実務での学び

類似文書検索と生成AI

従来の検索システムがキーワードマッチングに依存していたのに対し、生成AIの登場により「意味」に基づく検索が現実的になりました。これはエンベディング(テキストを数値ベクトルに変換する技術)の発展によるものです。しかし、実際の実装では様々な課題に直面します。

類似文書検索パイプラインの構築

初期の課題

最初に直面した大きな課題は埋め込みベクトル(エンベディング)の品質でした。OpenAIのtext-embedding-ada-002を使用していましたが、専門的な技術文書では思ったような類似性スコアが得られませんでした。特定のドメイン知識を含む文書では、汎用的なエンベディングモデルでは微妙なニュアンスを捉えきれなかったのです。

また、長文ドキュメントの扱いも課題でした。4,000トークンを超える技術文書を効果的に分割(チャンキング)する方法に悩みました。単純な文字数や段落での分割では、意味的なまとまりが壊れてしまうことが多かったです。

パイプラインの構築プロセス

1. ドキュメント前処理

def preprocess_document(document):
    # ヘッダー、フッター、不要なメタデータを除去
    cleaned_doc = remove_headers_footers(document)
    
    # HTMLタグや特殊文字の処理
    text_only = clean_html_and_special_chars(cleaned_doc)
    
    # 言語検出と言語固有の前処理
    lang = detect_language(text_only)
    if lang == 'ja':
        # 日本語特有の処理(例:全角・半角の正規化)
        text_only = normalize_japanese_text(text_only)
        
    return text_only

2. 意味を考慮したチャンキング戦略

最も苦労したのがこの部分でした。初めは単純な文字数ベースの分割を試みましたが、重要な文脈が失われる問題が発生しました。

def semantic_chunking(document, max_tokens=1000):
    # 段落や見出しなどの構造情報を活用
    sections = identify_document_sections(document)
    
    chunks = []
    current_chunk = ""
    current_token_count = 0
    
    for section in sections:
        # トークン数を推定
        section_tokens = estimate_tokens(section)
        
        if current_token_count + section_tokens <= max_tokens:
            # 現在のチャンクに追加
            current_chunk += section
            current_token_count += section_tokens
        else:
            # 現在のチャンクを保存し、新しいチャンクを開始
            if current_chunk:
                chunks.append(current_chunk)
            current_chunk = section
            current_token_count = section_tokens
    
    # 最後のチャンクを追加
    if current_chunk:
        chunks.append(current_chunk)
        
    return chunks

最終的には、セクションヘッダーや段落構造を認識し、意味的なまとまりを保持するカスタムチャンキングロジックを開発しました。

3. エンベディングの最適化

初期のエンベディングモデルでは満足な結果が得られなかったため、さまざまな改善を試みました:

  1. ドメイン適応: 特定分野の文書でファインチューニングされたエンベディングモデルを探索
  2. ハイブリッドアプローチ: BM25などのレキシカルな検索手法と、エンベディングベースの意味的検索を組み合わせ
  3. プロンプトエンジニアリング: エンベディング生成前にテキストにコンテキスト情報を追加
def create_hybrid_embeddings(chunks):
    # エンベディングの生成
    embeddings = []
    for chunk in chunks:
        # コンテキスト強化プロンプト
        enhanced_chunk = f"この文書は技術マニュアルの一部です。内容: {chunk}"
        
        # エンベディング生成
        embedding = embedding_model.embed(enhanced_chunk)
        
        # BM25スコアを計算用のトークン化
        tokenized_chunk = tokenize_for_bm25(chunk)
        
        embeddings.append({
            "text": chunk,
            "vector": embedding,
            "tokens": tokenized_chunk
        })
    
    return embeddings

4. ベクトルデータベースの選定と最適化

ベクトルデータベースには初めFaissを使用していましたが、メタデータフィルタリングやスケーラビリティの課題から、最終的にはPineconeに移行しました。

def setup_vector_database(embeddings, metadata):
    # Pineconeインデックスの設定
    dimension = len(embeddings[0]["vector"])
    pinecone.create_index("document-search", dimension=dimension)
    
    # バッチ処理でベクトルをアップロード
    batch_size = 100
    for i in range(0, len(embeddings), batch_size):
        batch = embeddings[i:i+batch_size]
        vectors = []
        
        for j, item in enumerate(batch):
            vector_id = f"doc_{i+j}"
            vectors.append({
                "id": vector_id,
                "values": item["vector"],
                "metadata": {
                    "text": item["text"],
                    "source": metadata[i+j]["source"],
                    "section": metadata[i+j]["section"],
                    "created_at": metadata[i+j]["created_at"]
                }
            })
        
        pinecone.upsert(vectors=vectors, namespace="technical-docs")

5. 検索結果のリランキングと後処理

初期の検索結果ではノイズが多く、関連性の低いドキュメントも含まれていました。この問題を解決するために、LLMを使用したリランキングメカニズムを実装しました:

def rerank_search_results(query, initial_results, top_k=5):
    reranked_results = []
    
    # LLMを用いたリランキング
    for result in initial_results:
        # クエリとチャンク間の関連性をスコアリング
        prompt = f"""
        クエリ: {query}
        ドキュメント: {result['text']}
        
        上記のクエリに対して、このドキュメントの関連性を0から10の数値で評価してください。
        関連性の評価基準:
        - クエリの意図への直接的な回答を含むか
        - 重要なコンセプトやキーワードが一致するか
        - 情報の具体性と有用性
        
        評価: 
        """
        
        # LLMから関連性スコアを取得
        response = llm_model.generate(prompt)
        relevance_score = extract_numeric_score(response)
        
        reranked_results.append({
            "text": result["text"],
            "metadata": result["metadata"],
            "original_score": result["score"],
            "relevance_score": relevance_score
        })
    
    # 再評価されたスコアでソート
    reranked_results.sort(key=lambda x: x["relevance_score"], reverse=True)
    
    return reranked_results[:top_k]

直面した実装上の課題

1. レイテンシとコスト最適化

エンベディング生成とLLM呼び出しのコストが想定以上に高くなりました。これに対処するため:

  • エンベディングのキャッシュ層を導入
  • バッチ処理の最適化
  • 初期フィルタリングにBM25を使用し、高コストのLLMリランキングは上位結果のみに適用

2. 多言語対応の難しさ

日本語と英語の両方のドキュメントを扱う必要がありましたが、言語間でエンベディングの品質が均一ではありませんでした。言語検出と言語固有の前処理ステップを追加することで改善しました。

3. 評価指標の設定

類似文書検索の精度を測るため、次の評価指標を設定しました:

  • Precision@k: 上位k件の結果のうち、関連性のあるドキュメントの割合
  • Mean Reciprocal Rank (MRR): 最初の関連ドキュメントのランク位置の逆数の平均
  • ユーザーフィードバックによる定性評価
def evaluate_search_quality(test_queries, ground_truth, search_results):
    metrics = {
        "precision_at_3": 0,
        "precision_at_5": 0,
        "mrr": 0
    }
    
    for query_id, query in enumerate(test_queries):
        relevant_docs = ground_truth[query_id]
        results = search_results[query_id]
        
        # Precision@kの計算
        p_at_3 = sum([1 for r in results[:3] if r["id"] in relevant_docs]) / 3
        p_at_5 = sum([1 for r in results[:5] if r["id"] in relevant_docs]) / 5
        
        # MRRの計算
        for i, r in enumerate(results):
            if r["id"] in relevant_docs:
                mrr = 1 / (i + 1)
                break
        else:
            mrr = 0
            
        metrics["precision_at_3"] += p_at_3
        metrics["precision_at_5"] += p_at_5
        metrics["mrr"] += mrr
    
    # 平均値の計算
    for key in metrics:
        metrics[key] /= len(test_queries)
        
    return metrics

ドキュメント分割の課題

ここで大きな疑問が生じます。「ドキュメントを分割してベクトル化した場合、そのドキュメントは複数のベクトルで表現されることになり、問題ではないか?」という点です。

確かに、ドキュメントを分割してベクトル化すると、一つのドキュメントが複数のベクトルで表現されることになります。これは類似文書検索の実装において重要な課題の一つです。

分割による表現の分散問題

例えば、10ページのレポートを5つのチャンクに分割すると、5つの異なるベクトルが生成されます。これにより次のような問題が生じます:

  1. ドキュメント全体の一貫性の喪失: 各チャンクは元のドキュメントの一部分のみを表現するため、全体の文脈が分断されます

  2. 検索結果の断片化: 検索時に同じドキュメントの複数のチャンクがヒットし、結果が重複して見えることがあります

  3. 関連性スコアの分散: ドキュメント全体の関連性が複数のベクトルに分散され、正確な評価が難しくなります

実装上の解決策

この問題に対処するため、以下のような解決策を実装しました:

1. メタデータによるグループ化

def process_document(document_id, document_text):
    chunks = semantic_chunking(document_text)
    
    for i, chunk in enumerate(chunks):
        # 各チャンクに元のドキュメントIDを関連付ける
        embedding = generate_embedding(chunk)
        
        metadata = {
            "document_id": document_id,
            "chunk_id": i,
            "total_chunks": len(chunks),
            "title": extract_document_title(document_text),
            "chunk_position": f"{i+1}/{len(chunks)}"
        }
        
        store_vector(embedding, metadata, chunk)

2. ドキュメントレベルでの集約

def search_and_aggregate(query, top_k=5):
    # クエリのエンベディングを生成
    query_embedding = generate_embedding(query)
    
    # チャンクレベルでの検索結果を取得(多めに)
    chunk_results = vector_db.search(query_embedding, top_k=top_k*3)
    
    # ドキュメントIDでグループ化
    document_scores = {}
    document_chunks = {}
    
    for result in chunk_results:
        doc_id = result["metadata"]["document_id"]
        
        # スコアの集約(最大値、平均値、重み付けなど様々な戦略がある)
        if doc_id not in document_scores:
            document_scores[doc_id] = result["score"]
            document_chunks[doc_id] = [result]
        else:
            # 例: 最高スコアを採用
            document_scores[doc_id] = max(document_scores[doc_id], result["score"])
            document_chunks[doc_id].append(result)
    
    # ドキュメントレベルでスコアに基づいてソート
    sorted_docs = sorted(document_scores.items(), key=lambda x: x[1], reverse=True)
    
    # 上位のドキュメントとその関連チャンクを返す
    results = []
    for doc_id, score in sorted_docs[:top_k]:
        doc_chunks = document_chunks[doc_id]
        results.append({
            "document_id": doc_id,
            "score": score,
            "chunks": doc_chunks,
            "title": doc_chunks[0]["metadata"]["title"]
        })
    
    return results

3. 階層的エンベディングアプローチ

より洗練された方法として、ドキュメントの階層的表現を作成することもあります:

def create_hierarchical_embeddings(document):
    # ドキュメント全体の要約を生成
    summary = generate_summary(document)
    summary_embedding = generate_embedding(summary)
    
    # 詳細なチャンクへの分割とエンベディング
    chunks = semantic_chunking(document)
    chunk_embeddings = [generate_embedding(chunk) for chunk in chunks]
    
    # 階層的に保存
    doc_id = store_document_embedding(summary_embedding, document_metadata)
    for i, (chunk, embedding) in enumerate(zip(chunks, chunk_embeddings)):
        store_chunk_embedding(embedding, {
            "document_id": doc_id,
            "chunk_id": i,
            "chunk_text": chunk
        })
        
    return doc_id

検索時には、まずドキュメントレベルで関連性を評価し、次に詳細なチャンクレベルで検索することができます。

実務での学び

実際の運用では、「一つのドキュメントが複数のベクトルで表現される」という特性を積極的に活用することもあります:

  1. 精度とカバレッジのバランス: 分割することで、ドキュメント内の特定のセクションに関する質問に対しても的確に回答できます

  2. ユーザーインターフェースでの工夫: 検索結果表示時に「同じドキュメントからの他の関連セクション」などとして表示

  3. 再ランキングの活用: LLMを使って、同じドキュメントから複数のチャンクがヒットした場合に適切に統合・要約する

さらに、実務を通じて学んだ重要なポイントは以下の通りです:

  1. チャンキングは芸術: ドキュメントの分割方法が検索品質に大きく影響します。一律なサイズ分割ではなく、文書構造を考慮したチャンキングが重要です。

  2. ハイブリッドアプローチの威力: 純粋な意味的検索だけでなく、従来のレキシカル検索との組み合わせが優れた結果をもたらしました。

  3. フィードバックループの重要性: ユーザーからのフィードバックを継続的に収集し、システムを改良することが成功への鍵でした。

  4. メタデータの活用: 文書の作成日、カテゴリ、著者などのメタデータをフィルタリングに活用することで、検索精度が飛躍的に向上しました。

まとめ

生成AIを用いた類似文書検索の実装は、多くの技術的チャレンジを伴いますが、適切なアプローチで大きな価値を生み出せることを実感しています。特にドキュメント分割とベクトル表現の課題は、システム設計において慎重に検討すべき重要な要素です。

ドキュメントをチャンクに分割してベクトル化することで生じる「一つのドキュメントが複数のベクトルで表現される」という特性は、適切に管理すれば、むしろ検索の柔軟性と精度を高める利点にもなり得ます。メタデータの活用、ドキュメントレベルでの集約、階層的エンベディングなどの手法を組み合わせることで、より精度の高い類似文書検索システムを構築することができるでしょう。

1

Discussion

ログインするとコメントできます