mem0-python-server/mem0core/reranker.py

import httpx

from .config import RERANKER_URL


def rerank_results(query: str, items: list, top_k: int) -> list:
    """Rerank mem0 search results via local reranker, fallback to original order."""
    if not items:
        return items

    documents = [r.get("memory", "") for r in items]
    try:
        response = httpx.post(
            RERANKER_URL,
            json={"query": query, "documents": documents, "top_k": top_k},
            timeout=5.0,
        )
        response.raise_for_status()
        reranked = response.json()["results"]
    except Exception as exc:
        print(f"[reranker] unavailable, skipping: {exc}")
        return items[:top_k]

    text_to_meta = {r.get("memory", ""): r for r in items}
    merged = []
    for item in reranked:
        meta = text_to_meta.get(item["text"])
        if meta:
            merged.append({**meta, "rerank_score": item["score"]})
    return merged