https://tech.aptpod.co.jp/entry/2025/12/15/100000

はじめに：SA（ソリューションアーキテクト）の「マニュアル迷子」問題

こんにちは、奥山です。

「aptpod Advent Calendar 2025」の12月15日の記事です。

普段は intdash のソリューションアーキテクト（SA）として、お客様への技術提案やアーキテクチャ設計を行っています。

SAという仕事柄、製品の仕様やAPIの詳細について即答を求められることが多いのですが、intdashのマニュアルやSDKリファレンスは膨大です。 正直なところ、すべてを脳内にインデックスするのは人間の限界を超えています。「あれ、あの設定値の上限いくつだっけ？」「このAPIのエラーコードの意味は？」と、日々PDFやWebマニュアルの海を彷徨うこともしばしば。

そこで、「マニュアルの内容を全部知っているAI（RAG）」 を構築し、自分の「相棒」にすることにしました。

今回は、RAG（Retrieval-Augmented Generation）構築において最も重要であり、かつ泥臭い部分である「データ収集からベクトル化までのパイプライン」の実装について紹介します。

www.aptpod.co.jp

RAGは「前処理」で決まる

RAGを作ろうとすると、ついLLMのプロンプトや検索アルゴリズムに目が行きがちですが、実運用で使える精度を出すには「いかに高品質なデータをVector DBに入れるか」が勝負です。

単にHTMLをテキスト化して突っ込むだけでは、以下のような壁にぶつかります。

重複地獄: URLパラメータ違いの同一ページが大量に検索に引っかかる。
構造崩壊: コードブロックやテーブルがただの文字列になり、意味不明になる。
APIエラー: トークン数制限やレートリミットで処理が止まる。

これらを解決するために構築した、Pythonによるデータパイプラインを紹介します。

アーキテクチャ概要

今回作成したパイプラインは以下の4ステップ構成です。

Crawler: マニュアルサイトを巡回し、HTMLを収集（URL正規化付き）。
Cleaner: HTMLからノイズを除去し、Markdownライクなテキストへ変換。
Chunker: LLMのトークン数に基づいてテキストを分割。
Embedder: OpenAI APIでベクトル化し、ChromaDBへ格納（リトライ処理付き）。

実装の「こだわり」ポイント

ここからは、実際のコード（抜粋）を見ながら、それぞれの工程での工夫点を紹介します。

クローラー：URL正規化で重複を排除する (`fetch_urls.py`)

Webマニュアルには、?state=... や #section-1 といった、コンテンツそのものには影響しないパラメータやフラグメントが付くことがあります。これらをそのまま収集すると、同じ内容のドキュメントが大量にDBに入ってしまいます。

そこで、収集前にURLを正規化する関数を実装しました。

# fetch_urls.py より抜粋
def normalize_url(url: str, params_to_remove: set[str] = {'state'}) -> str:
    """
    指定されたクエリパラメータを除去し、残りのパラメータをソートしてURLを正規化する。
    フラグメントも除去する。
    """
    parsed = urlparse(url)
    query_params = parse_qs(parsed.query)
    # 指定されたパラメータを除去
    filtered_params = {k: v for k, v in query_params.items() if k not in params_to_remove}
    # 残りのパラメータをキーでソートして一貫性を保つ
    sorted_params = sorted(filtered_params.items())
    new_query = urlencode(sorted_params, doseq=True)
    # URLを再構築（フラグメントなし）
    normalized_parsed = parsed._replace(query=new_query, fragment='')
    return urlunparse(normalized_parsed)

これにより、無駄なAPIコストを削減し、検索結果の多様性を確保しています。

クリーニング：コードとテーブルを守る「プレースホルダー」 (`html_to_text.py`)

BeautifulSoupなどで単純に get_text() すると、表組み（Table）やソースコード（Pre/Code）の構造が崩れてしまい、LLMが理解できないテキストになってしまいます。

そこで、「重要な要素を一度プレースホルダーに置換して退避させ、テキスト化後に復元する」 という処理を入れました。

# html_to_text.py より抜粋
def clean_html(html: str) -> str:
    # ... (省略) ...
    
    # --- コードブロック(<pre>)をプレースホルダーに置換 ---
    for i, pre_tag in enumerate(target_node.find_all("pre")):
        placeholder = f"__PRE_BLOCK_{uuid.uuid4()}__"
        code_text = pre_tag.get_text()
        # Markdownのコードブロック形式で保存
        block_placeholders[placeholder] = f"\n```\n{code_text.strip()}\n```\n"
        pre_tag.replace_with(NavigableString(placeholder))

    # ... (テキスト抽出処理) ...

    # --- プレースホルダーを元の整形済みテキストに戻す ---
    final_text = clean_text_with_placeholders
    for placeholder, replacement in block_placeholders.items():
        final_text = final_text.replace(placeholder, replacement)
        
    return final_text

このひと手間により、SDKのサンプルコードや仕様表が綺麗なMarkdown形式で保存され、RAGの回答精度が劇的に向上します。

チャンキング：文字数ではなく「トークン数」で切る (`chunk_text.py`)

OpenAIのEmbeddingモデル（text-embedding-3-small）には、入力トークン数の上限（8191 tokens）があります。日本語は英語に比べて文字数とトークン数の乖離が大きいため、単純な文字数カットでは上限オーバーのエラーが出るリスクがあります。

ここではtiktokenライブラリを使用し、正確なトークン数ベースでチャンク分割を行っています。

# chunk_text.py より抜粋
import tiktoken

tokenizer = tiktoken.encoding_for_model("text-embedding-3-small")

def get_token_count(text: str) -> int:
    return len(tokenizer.encode(text))

# この関数を使って、TARGET_TOKEN_COUNT (例: 1000) に収まるように分割

また、文脈が途切れないよう OVERLAP_TOKEN_COUNT = 100 程度のオーバーラップ（重複区間）を持たせて分割しています。

ベクトル化：APIレート制限との戦い (`vectorize_and_store.py`)

数万件のチャンクを一気にベクトル化しようとすると、確実に OpenAI API の Rate Limit (429 Too Many Requests) に引っかかります。これに対処するため、Exponential Backoff（指数バックオフ） アルゴリズムによるリトライ処理を実装しました。

# vectorize_and_store.py より抜粋
def get_embeddings_with_retry(texts: list[str], model: str) -> list[list[float]] | None:
    retries = 0
    backoff_time = 1.0 # 初期待機時間(秒)
    
    while retries < MAX_RETRIES:
        try:
            response = openai_client.embeddings.create(input=texts, model=model)
            return [e.embedding for e in response.data]
        except RateLimitError as e:
            print(f"Rate limit exceeded. Retrying in {backoff_time} seconds...")
            time.sleep(backoff_time + random.uniform(0, 0.1)) # Jitterを追加
            backoff_time *= 2 # 待機時間を倍にする
        # ...

これにより、夜間にスクリプトを放置しておいても、エラーで止まることなく最後まで処理が完了するようになりました。

構築結果

このパイプラインで生成実行した結果、ローカルの output/chroma_db ディレクトリに ChromaDB が構築されました。

確認用スクリプト (check.py) で中身を見てみると…

$ python check.py
Checking ChromaDB at: .../output/chroma_db
Available collections: ['intdash_docs_embeddings']
Collection 'intdash_docs_embeddings' contains 15432 documents

SDKのリファレンスからユーザーガイドまで、約1万5千件の知識チャンクが格納されたデータベースが完成しました！

実際にベクトルサーバーを検索してみる

以下は、構築したRAGシステムの検索をテストした際の結果です。「intdashとは」という質問を投げかけ、ベクトルデータベース（ChromaDB）から関連性の高い情報を5件取得しています。 Chunk1が最も関連性が高く、Chunk2.3.4と連なっていきます。また、出力時はURLも添付することで生成AIなどがURLからフェッチすることも想定して添付するようにしています。

$ python3 qa.py --query "intdashとは"
Loading vector store from: /workspaces/intdash-doc/output/chroma_db
Vector store loaded successfully.
Searching for top 5 chunks related to: 'intdashとは'
Found 5 chunks.
type = <class 'list'>
docs = [生データ]

--- Retrieved Chunks ---

--- Chunk 1 ---
Content:
intdashとは
intdashは、さまざまなIoTデータを低遅延に伝送するハイパフォーマンスなデータ伝送ミドルウェアです。
独自開発のプロトコルにより、帯域の細いモバイル回線でも確実かつリアルタイムにデータを伝送します。
産業用バスを流れる秒間数百〜数千点にもなる大量データにも対応しています。
intdashにより、自動車、ロボット、産業機械、人などの間で大量のデータを相互に低遅延で伝送できます。
多数のセンサーから収集した多様なデータを統一的に管理できるため、センサーフュージョン処理（複数のセンサーから得られたデータを統合して単一のセンサーでは得られない情報を得る）にも活用できます。
また、intdashが収集した時系列データは、活用しやすい形式でサーバーに保存されます。
ユーザーは、過去のデータをダッシュボードで可視化したり、解析用プログラムで処理したり、機械学習用のデータとして使用したりと、データを自在に活用することができます。
Source URL: https://xxxx.xxxx.xx/xxxxxxx/docs/2024R1L/ja/introduction/what-is-intdash.html
Original File: manual_docs_2024R1L_ja_introduction_what-is-intdash.txt
Chunk Index: 1
Token Count: 426

--- Chunk 2 ---
Content:
intdash-agentd help
intdash-agentdコマンドの使い方を表示します。
利用方法


intdash-agentd help

Source URL: https://xxxx.xxxx.xx/xxxxxxx/docs/2024R1L/ja/agent2/reference/cli/intdash-agentd/agentd-help.html
Original File: manual_docs_2024R1L_ja_agent2_reference_cli_intdash-agentd_agentd-help.txt
Chunk Index: 1
Token Count: 35

--- Chunk 3 ---
Content:
intdash-agent-tool help


intdash-agent-tool help


intdash-agent-toolコマンドの使い方を表示します。
Source URL: https://xxxx.xxxx.xx/xxxxxxx/docs/2024R1L/ja/agent2/reference/cli/intdash-agent-tool/agent-tool-help.html
Original File: manual_docs_2024R1L_ja_agent2_reference_cli_intdash-agent-tool_agent-tool-help.txt
Chunk Index: 1
Token Count: 32

--- Chunk 4 ---
Content:
intdash-agentctl help
intdash-agentctlコマンドの使い方を表示します。
利用方法


intdash-agentctl help

Source URL: https://xxxx.xxxx.xx/xxxxxxx/docs/2024R1L/ja/agent2/reference/cli/intdash-agentctl/help.html
Original File: manual_docs_2024R1L_ja_agent2_reference_cli_intdash-agentctl_help.txt
Chunk Index: 1
Token Count: 35

--- Chunk 5 ---
Content:
intdashサーバー構築・運用ガイド
本セクションでは、intdash API、intdashのフロントエンドアプリケーション、およびVisual M2M Data Visualizerの構築・運用方法についての情報を提供します。
対象読者
本サイトは、ネットワーク管理やサーバー管理についての基礎知識と、使用する構築方法についての知識（例: AMIを使用して構築する場合はAWSについての知識）を持っている方を対象としています。
Source URL: https://xxxx.xxxx.xx/xxxxxxx/docs/2024R1L/ja/server-setup-and-operation/index.html
Original File: manual_docs_2024R1L_ja_server-setup-and-operation_index.txt
Chunk Index: 1
Token Count: 180

まとめ

今回は、社内ドキュメントをRAG化するためのデータパイプラインの実装について紹介しました。世の中には LlamaIndex などの便利なフレームワークもありますが、独自のマニュアル構造に合わせて前処理を細かく制御し、精度を極限まで高めたい場合、このようにPythonでスクリプトを組むアプローチは非常に有効です。さらに、このRAGをMCP*1(Model Context Protocol)サーバーを立ち上げて、Claude Codeなどから対話させるとさらに嬉しさ倍増します。みなさんもチャレンジしてみてください。

*1:ClaudeなどのAIアシスタントと外部ツールを接続するための標準規格