0Pricing
AI Agents · 课时

对搜索结果进行排序与筛选

评估相关性、去重，并选择最适合上下文的结果。

对搜索结果进行排序与筛选 是 CoddyKit 上的免费 AI Agents 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

排序和筛选为何重要

搜索 API 会返回 5 到 10 个结果，但它们与智能体任务的相关性、可靠性和实用性并不相同。直接将原始结果提供给 LLM 会浪费上下文令牌，还可能引入噪声或错误信息。

在结果到达 LLM 之前进行排序和筛选，可以提高信噪比。

使用 BM25 进行相关性评分

BM25（最佳匹配 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 网络开发”和“使用 Django 构建网站”的相似度仍然会很高。

使用查询嵌入与结果嵌入之间的余弦相似度。

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)

按网址去重

搜索结果通常包含近似重复项：来自多个转载来源的同一篇文章，或带有不同网址参数的同一个页面。在将结果传递给 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')

根据上下文预算截断结果

即使完成筛选，也可能有 5 个高质量结果，每个包含 600 个字符，总计 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 小时）缓存搜索结果，可以降低接口成本，并缩短重复查询的响应时间。

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

完整的筛选与排序流程

将所有步骤串联到一个流程函数中：获取结果 → 筛除低质量结果 → 去重 → 排序 → 应用域名加权 → 根据上下文截断 → 格式化为提示词。

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 之前需要经过处理。处理流程是：筛除低质量结果（过短、需要付费访问）→ 按网址和内容去重 → 按 BM25 和/或嵌入相似度排序 → 应用域名质量加权 → 截断至上下文预算 → 格式化为编号来源。

缓存搜索结果可以降低重复查询的接口成本。提示词中的编号引用可以让 LLM 将陈述归因于具体来源。

常见问题解答

「对搜索结果进行排序与筛选」课时是免费的吗？

是的 — 「对搜索结果进行排序与筛选」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「对搜索结果进行排序与筛选」这节课中我会学到什么？

评估相关性、去重，并选择最适合上下文的结果。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。

「对搜索结果进行排序与筛选」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 使用 Tavily 和 SerpAPI 搜索
  2. 对搜索结果进行排序与筛选
  3. 深度研究循环模式
  4. 结合网络搜索与 RAG
← 返回 AI Agents