Classificazione e filtraggio dei risultati di ricerca

Perché il ranking e il filtraggio sono importanti

Valutazione della rilevanza con 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]

Valutazione della rilevanza con gli embedding

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)

Ranking ibrido: BM25 + embedding

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)

Deduplicazione per URL

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'])

Valutazione della qualità del dominio

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}")

Filtraggio dei risultati di bassa qualità

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')

Troncamento dei risultati per il budget di contesto

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')

Formattazione dei risultati per il prompt dell'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}'
)

Memorizzazione nella cache dei risultati di ricerca

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

Pipeline completa di filtraggio e ranking

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)

Verifica delle conoscenze

Riepilogo: ranking e filtraggio dei risultati di ricerca