Embedding RAGの限界とPageIndex：ベクターレスRAGを自社サイトで試した【LangGraph実装付き】

https://neosophie.com/
├── /ja/blog/
│   ├── /ja/blog/20260226-japanese-asr-benchmark
│   ├── /ja/blog/20260305-vibevoice-quantization
│   └── /ja/blog/tags/
│       ├── /ja/blog/tags/asr
│       └── /ja/blog/tags/diarization
├── /products
└── /contact

page_by_path = {
    "/": {"title": "Neosophie", "content": "..."},
    "/products": {"title": "Products", "content": "..."},
    "/ja/blog/20260226-japanese-asr-benchmark": {"title": "日本語ASRモデル比較", "content": "..."},
    "/ja/blog/tags/asr": {"title": "タグ: asr", "content": "..."},
    # ...
}

# url_tree.py
def _ancestor_paths(path: str) -> list[str]:
    """
    "/docs/api/auth" → ["/docs/api/auth", "/docs/api", "/docs", "/"]
    """
    result: list[str] = [path]
    current = path
    while True:
        if current == "/":
            break
        parent = current.rsplit("/", 1)[0] or "/"
        result.append(parent)
        current = parent
    return result

# 全クロール済みURLに対して祖先パスを収集
all_paths: set[str] = set()
for path in list(page_by_path):
    all_paths.update(_ancestor_paths(path))

from dataclasses import dataclass, field

@dataclass
class UrlTreeNode:
    segment: str          # パスの末尾セグメント（例: "auth"）
    path: str             # フルパス（例: "/docs/api/auth"）
    url: str              # 完全URL
    title: str            # ページタイトル（クロール済みのみ）
    crawled: bool         # 実際に取得したページかどうか
    children: list["UrlTreeNode"] = field(default_factory=list)

nodes: dict[str, UrlTreeNode] = {}
for path in sorted(all_paths):
    page = page_by_path.get(path)
    crawled = page is not None
    segment = path.rsplit("/", 1)[-1] or "/"
    nodes[path] = UrlTreeNode(
        segment=segment,
        path=path,
        url=f"https://example.com{path}",
        title=page.get("title", "") if page else "",
        crawled=crawled,
    )

roots: list[UrlTreeNode] = []
for path, node in nodes.items():
    if path == "/":
        roots.append(node)
        continue
    parent_path = path.rsplit("/", 1)[0] or "/"
    parent = nodes.get(parent_path)
    if parent is not None:
        parent.children.append(node)
    else:
        roots.append(node)  # 親がなければルート扱い

def _collapse_single_child_intermediates(
    nodes: list[UrlTreeNode],
) -> list[UrlTreeNode]:
    result = []
    for node in nodes:
        # 再帰的に子を先に処理
        node.children = _collapse_single_child_intermediates(node.children)
        # 未クロール かつ 子が1つだけ → このノードをスキップ
        if not node.crawled and len(node.children) == 1:
            result.append(node.children[0])
        else:
            result.append(node)
    return result

Before:
/
└── ja (未クロール、子が1つ)
    └── blog (クロール済み)

After:
/
└── blog  ← /ja が消えて直結

import re

_CURRENT_PAGE_PATTERN = re.compile(
    r"このページ|今のページ|今見ている|ここ[にのはをが]|要約|まとめ|説明して",
)

def _node_classify(self, state: dict) -> dict:
    question = state.get("question", "")
    current_url = state.get("context_info", {}).get("current_url")
    # ユーザーが特定ページを開いている かつ パターンマッチ
    if current_url and self._CURRENT_PAGE_PATTERN.search(question):
        return {"question_type": "current_page"}
    return {"question_type": "general"}

def _node_fetch_current_page(self, state: dict) -> dict:
    current_url = state.get("context_info", {}).get("current_url", "")
    index = self._load_page_index()
    pages_by_url = {p["url"]: p for p in index.get("pages", [])}
    page = pages_by_url.get(current_url)
    if page:
        return {"selected_pages": [page]}
    # フォールバック：先頭4件
    return {"selected_pages": list(pages_by_url.values())[:4]}

def _node_build_subtrees(self, state: dict) -> dict:
    index = self._load_page_index()
    url_tree = index.get("url_tree", [])
    pages_by_url = {p["url"]: p for p in index.get("pages", [])}
    seen_paths: set[str] = set()
    subtrees: list[dict] = []

    # ルート直下の子を優先的に追加（重要度が高いため）
    for root in url_tree:
        for child in root.get("children", []):
            path = child.get("path", "")
            if path not in seen_paths:
                seen_paths.add(path)
                leaf_pages = self._collect_all_leaf_pages(child, pages_by_url)
                subtrees.append({
                    "path": path,
                    "url": child.get("url", ""),
                    "title": child.get("title", ""),
                    "leaf_pages": leaf_pages,
                })

    # さらに深い非リーフノードも追加（重複はseen_pathsで除外）
    self._collect_non_leaf_subtrees(url_tree, pages_by_url, subtrees, seen_paths)
    return {"subtrees": subtrees}

@staticmethod
def _collect_all_leaf_pages(
    node: dict,
    pages_by_url: dict,
) -> list[dict]:
    """サブツリーのリーフページを再帰的に収集"""
    children = node.get("children", [])
    if not children:
        # 末端ノード → そのページを返す
        page = pages_by_url.get(node.get("url", ""))
        return [page] if page else []
    # 中間ノード → 子を再帰して集約
    result = []
    for child in children:
        result.extend(
            RagAgentService._collect_all_leaf_pages(child, pages_by_url)
        )
    return result

def _node_select_subtrees(self, state: dict) -> dict:
    question = state.get("question", "")
    subtrees = state.get("subtrees", [])

    # TOC形式のテキスト組み立て
    toc_lines = []
    for i, st in enumerate(subtrees, start=1):
        preview = [p.get("title", "") for p in st["leaf_pages"][:6]]
        titles_str = ", ".join(preview)
        if len(st["leaf_pages"]) > 6:
            titles_str += f" 他{len(st['leaf_pages']) - 6}件"
        toc_lines.append(
            f"id={i} | section={st['title']} | url={st['url']} | "
            f"ページ数={len(st['leaf_pages'])} | 含まれるページ: {titles_str}"
        )

    system_prompt = (
        "あなたは検索プランナーです。"
        "サイト構造から質問に最も関連するセクションIDを最大2件選んでください。"
        "必ずJSON配列のみで返してください。例: [1]"
    )
    human_prompt = f"質問: {question}\n\nサイト構造:\n" + "\n".join(toc_lines)

    response = self._llm.invoke(
        [SystemMessage(content=system_prompt), HumanMessage(content=human_prompt)]
    )
    selected_ids = self._extract_id_list(response.content)  # "[1, 3]" → [1, 3]

    selected_paths = [
        st["path"]
        for i, st in enumerate(subtrees, start=1)
        if i in set(selected_ids)
    ]
    return {"selected_subtree_paths": selected_paths}

def _node_select_pages(self, state: dict) -> dict:
    question = state.get("question", "")
    leaf_candidates = state.get("leaf_candidates", [])

    # ページ目次の組み立て（headingsを使ってページ内容を要約）
    page_lines = []
    for i, page in enumerate(leaf_candidates, start=1):
        headings = page.get("headings", [])
        headings_str = ", ".join(headings[:5])
        page_lines.append(
            f"id={i} | title={page.get('title', '')} "
            f"| url={page.get('url', '')} | headings={headings_str}"
        )

    system_prompt = (
        "あなたは検索プランナーです。"
        "目次候補から質問に最も関連するページIDを最大4件選んでください。"
        "必ずJSON配列のみで返してください。例: [3, 8]"
    )
    human_prompt = f"質問: {question}\n\n目次候補:\n" + "\n".join(page_lines)

    response = self._llm.invoke(
        [SystemMessage(content=system_prompt), HumanMessage(content=human_prompt)]
    )
    selected_ids = self._extract_id_list(response.content)

    selected_pages = [
        page
        for i, page in enumerate(leaf_candidates, start=1)
        if i in set(selected_ids)
    ]
    return {"selected_pages": selected_pages}

def _node_answer(self, state: dict) -> dict:
    question = state.get("question", "")
    selected_pages = state.get("selected_pages", [])
    context_info = state.get("context_info", {})

    # コンテキスト文字列の組み立て（最大14,000文字）
    context_parts = []
    total_chars = 0
    for i, page in enumerate(selected_pages, start=1):
        content = page.get("content", "")[:3500]  # 1ページあたり上限
        entry = (
            f"[{i}] title={page.get('title', '')}\n"
            f"url={page.get('url', '')}\n"
            f"headings={page.get('headings', '')}\n"
            f"content={content}"
        )
        total_chars += len(entry)
        if total_chars > 14000:
            break
        context_parts.append(entry)

    context = "\n\n".join(context_parts)

    system_prompt = (
        "あなたはWebサイト案内AIです。"
        "与えられた文脈だけを根拠に日本語で回答してください。"
        "文脈が不足する場合は不足を明示してください。"
        "表は絶対に使わないこと。情報の整理には箇条書きを優先してください。"
        "最後に参照URLを箇条書きで示してください。"
    )
    human_prompt = (
        f"ユーザーコンテキスト:\n"
        f"- ユーザーが閲覧中のURL: {context_info.get('current_url', 'N/A')}\n"
        f"- 現在時刻: {context_info.get('timestamp', 'N/A')}\n\n"
        f"文脈:\n{context}\n\n"
        f"質問: {question}"
    )

    response = self._llm.invoke(
        [SystemMessage(content=system_prompt), HumanMessage(content=human_prompt)]
    )
    return {"answer": response.content}

id=1 | section=Contact | Neosophie | url=https://neosophie.com/contact | ページ数=1
id=2 | section=ブログ | Neosophie | url=https://neosophie.com/ja/blog | ページ数=17 | 含まれるページ: AIエンジニアが毎日チェック...他11件
id=4 | section=Products | Neosophie | url=https://neosophie.com/products | ページ数=1 | 含まれるページ: Products | Neosophie

質問: ASRに関するブログは何がある？

サイト構造:
id=1 | section=Contact | url=https://neosophie.com/contact | ページ数=1 | ...
id=2 | section=ブログ  | url=https://neosophie.com/ja/blog | ページ数=17 | 含まれるページ: AIエンジニアが毎日チェック..., 【2026年】NeMo・VibeVoice..., ...

質問: ASRに関するブログは何がある？

目次候補:
id=20 | title=AIエンジニアが毎日チェックする情報収集サイト17選 | headings=...
id=21 | title=【2026年最新】日本語ASRモデル比較：Whisper・Qwen3・Voxtral・ReazonSpeech | headings=要約, テスト環境...
id=6  | title=VibeVoice-ASRを量子化してDERはどう変わるか？ | headings=要約, はじめに...
id=11 | title=タグ: asr | headings=# asr, VibeVoice-ASR..., 【2026年最新】日本語ASR...
...（17件）

文脈:
[1] title=【2026年最新】日本語ASRモデル比較 ...
    url=https://neosophie.com/ja/blog/20260226-japanese-asr-benchmark
    content=8種類の日本語音声認識（ASR）モデルを同一条件でベンチマーク...

[2] title=VibeVoice-ASRを量子化してDERはどう変わるか？ ...
...