RAGパイプラインのハマりどころ：文字化けと重複登録の実装

以前の記事で、ChromaDB + RabbitMQ + WebSocketによるRAGシステム（consumer.py）の全体設計を紹介した。今回はその「入口」にあたる部分――記事をスクレイピングしてDBに登録する処理――で実際に発生した2つの不具合と、その修正内容を記録として残す。

どちらも派手なバグではないが、放置すると検索結果の質が静かに劣化していくタイプの問題だった。

スクレイピング先の一つ、ibe.tokyoの記事を取り込んだところ、タイトルだけが文字化けして登録された。

原因

requestsはレスポンスヘッダーにcharsetが明記されていない場合、エンコーディングを推測する。この推測がISO-8859-1系に外れると、response.textを取得した時点で既に文字が壊れている。BeautifulSoupに渡す前段階の問題なので、パーサーを変えても直らない。

対処

response.text（str）ではなくresponse.content（bytes）をBeautifulSoupに渡す。バイト列のまま渡すと、内部のUnicodeDammitがHTML中のmeta charset宣言などから正しくエンコーディングを判定してくれる。

# NG: str を渡す（requestsの推測に依存）
soup = BeautifulSoup(response.text, "html.parser")

# OK: bytes を渡す（BeautifulSoup側で正しく判定）
soup = BeautifulSoup(response.content, "html.parser")

title = ""
if soup.find("title"):
    title = soup.find("title").get_text(strip=True)

ChromaDBへの登録関数は、既に登録済みのidをスキップすることで重複登録を防いでいた。これは「新規URLの重複実行」には効くが、「既存記事の本文が更新された場合」には対応できていなかった。

何が起きるか

記事を5チャンクで登録した後、本文を編集して3チャンクに減った場合を考える。IDは{url}#chunk0のように連番で振られるため、更新後に存在しないはずのchunk3・chunk4がDB内に残り続ける。

→ 検索時に古い内容のチャンクがヒットし、更新前の情報を回答に混入させる原因になる。

対処：URL単位で差分を見て、更新時は作り直す

方針はシンプルで、チャンク単位でIDの存在確認をするのをやめ、記事URL単位で「変化があったか」をまとめて判定する。変化があれば、そのURLに紐づく既存チャンクを一度全部削除してから登録し直す。

def add_articles(articles: list[dict]) -> int:
    ids, texts, metadatas = [], [], []

    for article in articles:
        url = article["url"]
        chunks = chunk_text(article["text"])
        new_cids = [f"{url}#chunk{j}" for j in range(len(chunks))]

        # URL単位で既存チャンクを一括取得
        existing = collection.get(where={"url": url})
        existing_ids = existing["ids"]
        existing_docs = dict(zip(existing_ids, existing["documents"]))

        # チャンク数・ID・本文が完全一致なら変更なし
        is_unchanged = (
            existing_ids == new_cids
            and all(existing_docs.get(cid) == t
                    for cid, t in zip(new_cids, chunks))
        )
        if is_unchanged:
            continue

        # 変更ありなら古いチャンクを削除してから作り直す
        if existing_ids:
            collection.delete(ids=existing_ids)

        for j, text_chunk in enumerate(chunks):
            ids.append(new_cids[j])
            texts.append(text_chunk)
            metadatas.append({"url": url, "title": article["title"], "chunk": j})

    if not ids:
        return 0

    embs = embedder.encode(texts, normalize_embeddings=True).tolist()
    collection.upsert(ids=ids, documents=texts,
                       embeddings=embs, metadatas=metadatas)
    return len(ids)

• 文字化けはresponse.contentを渡すだけで解決するケースが多い。まず疑うべきはBeautifulSoupではなくrequests側のエンコーディング推測。
• 重複防止は「IDの存在確認」だけでは不十分で、「内容が変わったかどうか」まで見る必要がある。
• 今回の実装は記事ごとにcollection.getを呼んでいるため、数千記事規模になるとオーバーヘッドが目立つ可能性がある。全url→idsマッピングを事前に1回で取得してメモリ上で処理する方式への切り替えは今後の課題。