AI Agents · レッスン

検索結果のランキングとフィルタリング

Q: 「検索結果のランキングとフィルタリング」で何を学びますか？

関連性のスコアリング、重複除去、コンテキストに最適な結果の選択を学びます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

関連性のスコアリング、重複除去、コンテキストに最適な結果の選択を学びます。

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

BM25による関連性スコアリング

BM25（Best Match 25）は、クエリとのキーワードの重複度に基づいて文書をスコアリングする古典的なテキストランキングアルゴリズムです。語彙マッチング、つまりクエリと文書が同じ単語を共有している場合に効果を発揮します。

pip install rank-bm25でインストールします。

from rank_bm25 import BM25Okapi

def rank_with_bm25(query, results):
    # Tokenize: lowercase and split into words
    tokenized_results = [
        r['content'].lower().split()
        for r in results
    ]
    bm25 = BM25Okapi(tokenized_results)

    query_tokens = query.lower().split()
    scores = bm25.get_scores(query_tokens)

    # Sort results by score descending
    ranked = sorted(
        zip(scores, results),
        key=lambda x: x[0],
        reverse=True
    )
    return [(score, result) for score, result in ranked]

埋め込みによる関連性スコアリング

BM25は完全一致する単語しか検出しません。埋め込みの類似度は意味的な内容を捉えるため、「PythonによるWeb開発」と「DjangoでのWebサイト構築」のように単語が異なっていても、高い類似度になります。

クエリの埋め込みと結果の埋め込みの間でコサイン類似度を使用します。

import numpy as np
import openai
import os

client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

def embed(text):
    resp = client.embeddings.create(
        model='text-embedding-3-small',
        input=text[:8000]
    )
    return np.array(resp.data[0].embedding)

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def rank_by_embedding(query, results):
    q_emb = embed(query)
    scored = []
    for r in results:
        r_emb = embed(r['content'][:1000])
        score = cosine_similarity(q_emb, r_emb)
        scored.append((score, r))
    return sorted(scored, key=lambda x: x[0], reverse=True)

BM25 + 埋め込みによるハイブリッドランキング

BM25と埋め込みによるスコアリングは、関連性の異なる側面を捉えます。ハイブリッドランキングでは、両方のスコアを加重平均で組み合わせ、語彙マッチングと意味マッチングの両方の長所を活かします。

def hybrid_rank(query, results, bm25_weight=0.4, embed_weight=0.6):
    # Get BM25 scores (normalized 0-1)
    bm25_scored = rank_with_bm25(query, results)
    max_bm25 = max(s for s, _ in bm25_scored) or 1
    bm25_norm = {r['url']: s / max_bm25 for s, r in bm25_scored}

    # Get embedding scores
    embed_scored = rank_by_embedding(query, results)
    embed_norm = {r['url']: s for s, r in embed_scored}

    # Combine
    combined = []
    for r in results:
        url = r['url']
        score = (bm25_weight * bm25_norm.get(url, 0) +
                 embed_weight * embed_norm.get(url, 0))
        combined.append((score, r))

    return sorted(combined, key=lambda x: x[0], reverse=True)

URLによる重複除去

検索結果には、同じ記事を複数の配信元が転載したものや、異なるURLパラメーターが付いた同じページなど、ほぼ重複した結果が含まれることがよくあります。LLMに結果を渡す前に、これらを重複除去します。

from urllib.parse import urlparse, urlunparse

def normalize_url(url):
    parsed = urlparse(url)
    # Remove query params and fragment (tracking params, etc.)
    clean = parsed._replace(query='', fragment='')
    return urlunparse(clean).rstrip('/')

def deduplicate_results(results):
    seen_urls = set()
    unique = []
    for r in results:
        url = normalize_url(r.get('url', ''))
        if url not in seen_urls:
            seen_urls.add(url)
            unique.append(r)
    return unique

# Also deduplicate by content similarity (near-duplicate detection)
def deduplicate_by_content(results, min_unique_ratio=0.7):
    unique = [results[0]] if results else []
    for candidate in results[1:]:
        cand_words = set(candidate['content'].lower().split())
        is_duplicate = False
        for kept in unique:
            kept_words = set(kept['content'].lower().split())
            overlap = len(cand_words & kept_words) / max(len(cand_words), 1)
            if overlap > (1 - min_unique_ratio):
                is_duplicate = True
                break
        if not is_duplicate:
            unique.append(candidate)
    return unique

if __name__ == '__main__':
    demo_results = [
        {'url': 'https://example.com/a?utm_source=x'},
        {'url': 'https://example.com/a'},
        {'url': 'https://example.com/b'},
    ]
    unique = deduplicate_results(demo_results)
    print(f'{len(demo_results)} results -> {len(unique)} unique')
    for r in unique:
        print(' -', r['url'])

ドメイン品質スコアリング

docs.python.orgの結果は、無名のブログの結果より信頼性が高くなります。ドメインを複数の品質階層に分けて倍率を割り当て、最終的なランキングに反映します。

DOMAIN_QUALITY = {
    # Tier 1 — authoritative (1.3x boost)
    'docs.python.org': 1.3,
    'developer.mozilla.org': 1.3,
    'arxiv.org': 1.3,
    'github.com': 1.2,
    'stackoverflow.com': 1.2,
    # Tier 2 — good (1.0x, no change)
    # Tier 3 — low quality (penalty)
    'pinterest.com': 0.3,
    'quora.com': 0.5,
    'wikihow.com': 0.6
}

def get_domain_multiplier(url):
    from urllib.parse import urlparse
    domain = urlparse(url).netloc.lower().replace('www.', '')
    return DOMAIN_QUALITY.get(domain, 1.0)  # default: no change

def apply_domain_boost(scored_results):
    boosted = []
    for score, r in scored_results:
        multiplier = get_domain_multiplier(r.get('url', ''))
        boosted.append((score * multiplier, r))
    return sorted(boosted, key=lambda x: x[0], reverse=True)

if __name__ == '__main__':
    scored = [(1.0, {'url': 'https://pinterest.com/x'}), (1.0, {'url': 'https://docs.python.org/x'})]
    for score, r in apply_domain_boost(scored):
        print(f"{r['url']}: boosted score {score:.2f}")

低品質な結果のフィルタリング

ドメインに関係なく、構造上の理由で品質が低い結果もあります。たとえば、有用な情報を含めるには短すぎるもの、ほとんどがナビゲーションテキストで構成されているもの、ログインが必要なページなどです。ランキングの前に、これらを除外します。

MIN_CONTENT_LENGTH = 200  # characters

LOW_QUALITY_SIGNALS = [
    'sign in to view',
    'please log in',
    'subscribe to read',
    '404 not found',
    'access denied',
    'this content is for members only'
]

def is_quality_result(result):
    content = result.get('content', '')

    # Too short
    if len(content) < MIN_CONTENT_LENGTH:
        return False

    # Paywall / access barrier detected
    content_lower = content.lower()
    for signal in LOW_QUALITY_SIGNALS:
        if signal in content_lower:
            return False

    return True

def filter_results(results):
    return [r for r in results if is_quality_result(r)]

if __name__ == '__main__':
    demo_results = [
        {'content': 'Please log in to view this article which has plenty of extra padding text here.'},
        {'content': 'A' * 250},
    ]
    kept = filter_results(demo_results)
    print(f'{len(demo_results)} results -> {len(kept)} passed quality filter')

コンテキスト予算に合わせた結果の切り詰め

フィルタリング後でも、600文字の高品質な結果が5件あれば、合計で3,000文字になります。LLMのコンテキスト予算に収まる結果数を決め、それに応じて切り詰めます。

MAX_CONTEXT_CHARS = 4000
MAX_SNIPPET_CHARS = 600

def truncate_for_context(ranked_results, budget=MAX_CONTEXT_CHARS):
    selected = []
    used_chars = 0

    for score, result in ranked_results:
        content = result.get('content', '')[:MAX_SNIPPET_CHARS]
        entry = f"Source: {result['title']}\nURL: {result['url']}\nContent: {content}"
        entry_len = len(entry)

        if used_chars + entry_len > budget:
            break

        selected.append(result)
        used_chars += entry_len

    return selected

ranked_results = [
    (0.9, {'title': 'Doc A', 'url': 'http://a', 'content': 'x' * 800}),
    (0.7, {'title': 'Doc B', 'url': 'http://b', 'content': 'y' * 800}),
]
budget = 1000
selected = truncate_for_context(ranked_results, budget=budget)
print(f'Selected {len(selected)} results within {budget}-char budget')

LLMプロンプト向けの結果のフォーマット

ランキング、重複除去、切り詰めが完了したら、LLMプロンプト内で結果を番号付きリストとしてフォーマットします。情報源に番号を付けると、モデルは回答内で簡単に引用できます。

def format_results_for_prompt(results):
    lines = ['Here are relevant search results:\n']
    for i, r in enumerate(results, 1):
        lines.append(f'[{i}] {r["title"]}')
        lines.append(f'    URL: {r["url"]}')
        lines.append(f'    {r.get("content", "")[:400]}')
        lines.append('')
    lines.append('Use these sources to answer the question. Cite sources as [1], [2], etc.')
    return '\n'.join(lines)

# Usage in agent prompt
formatted = format_results_for_prompt(selected_results)
response = llm_call(
    system='You are a research assistant.',
    user=f'{formatted}\n\nQuestion: {user_question}'
)

検索結果のキャッシュ

同じクエリが、複数のセッションやエージェントのループで繰り返されることがあります。検索結果を短いTTL（例：1時間）でキャッシュすると、APIコストを削減し、繰り返されるクエリへの応答時間を改善できます。

import hashlib
import time

search_cache = {}  # In production: use Redis or disk cache
CACHE_TTL = 3600  # 1 hour

def cached_search(query, **kwargs):
    cache_key = hashlib.md5(query.encode()).hexdigest()
    entry = search_cache.get(cache_key)

    if entry and (time.time() - entry['ts']) < CACHE_TTL:
        print('Search cache hit')
        return entry['results']

    results = client.search(query=query, **kwargs)
    search_cache[cache_key] = {
        'results': results,
        'ts': time.time()
    }
    return results

完全なフィルタリングとランキングのパイプライン

すべての手順を1つのパイプライン関数にまとめます。取得 → 低品質な結果のフィルタリング → 重複除去 → ランキング → ドメインブーストの適用 → コンテキストに合わせた切り詰め → プロンプト用のフォーマット、という流れです。

def search_and_rank(query, max_context_chars=4000):
    # 1. Fetch
    raw = cached_search(query, max_results=8)
    results = raw.get('results', [])

    # 2. Filter quality
    results = filter_results(results)

    # 3. Deduplicate
    results = deduplicate_results(results)

    # 4. Rank (BM25 fast path — save embedding costs)
    scored = rank_with_bm25(query, results)

    # 5. Domain boost
    scored = apply_domain_boost(scored)

    # 6. Truncate to context budget
    selected = truncate_for_context(scored, budget=max_context_chars)

    # 7. Format
    return format_results_for_prompt(selected)

知識チェック

ハイブリッドランキング（BM25 + 埋め込み）をBM25単独で使用する場合と比べた主な利点は何ですか？

復習：検索結果のランキングとフィルタリング

生の検索結果は、LLMに渡す前に処理する必要があります。パイプラインは、低品質な結果をフィルタリング（短すぎる、ペイウォールで保護されている）→ URLとコンテンツによる重複除去 → BM25や埋め込みの類似度によるランキング → ドメイン品質ブーストの適用 → コンテキスト予算に合わせた切り詰め → 番号付きの情報源としてフォーマット、という流れです。

検索結果をキャッシュすると、繰り返されるクエリのAPIコストを削減できます。プロンプト内の番号付き引用により、LLMは主張を特定の情報源に帰属させられます。

よくある質問