Multi-Signal Retrieval

Copydef get_bm25_params(query: str, *, lemmatized: Optional[str] = None) -> tuple:
    """Get BM25 sigmoid parameters based on query length.
 
    Longer queries tend to have higher raw BM25 scores, so we adjust
    the sigmoid midpoint and steepness accordingly.
 
    Returns:
        (midpoint, steepness) for sigmoid normalization.
    """
    if lemmatized is None:
        from mem0.utils.lemmatization import lemmatize_for_bm25
 
        lemmatized = lemmatize_for_bm25(query)
    num_terms = len(lemmatized.split()) if lemmatized else 1
 
    if num_terms <= 3:
        return 5.0, 0.7
    elif num_terms <= 6:
        return 7.0, 0.6
    elif num_terms <= 9:
        return 9.0, 0.5
    elif num_terms <= 15:
        return 10.0, 0.5
    else:
        return 12.0, 0.5
 

Copydef normalize_bm25(raw_score: float, midpoint: float, steepness: float) -> float:
    """Normalize BM25 score to [0, 1] using logistic sigmoid.
 
    Args:
        raw_score: Raw BM25 score (unbounded, typically 0-20+).
        midpoint: Score at which sigmoid outputs 0.5.
        steepness: Controls how quickly sigmoid transitions.
 
    Returns:
        Normalized score in range [0, 1].
    """
    return 1.0 / (1.0 + math.exp(-steepness * (raw_score - midpoint)))
 

Copydef score_and_rank(
    semantic_results: List[Dict[str, Any]],
    bm25_scores: Dict[str, float],
    entity_boosts: Dict[str, float],
    threshold: float,
    top_k: int,
    explain: bool = False,
) -> List[Dict[str, Any]]:
    """Score candidates additively and return top-k results.
 
    For each candidate:
        semantic_score is taken from the result's score field.
        combined = (semantic + bm25 + entity_boost) / max_possible
 
    Threshold gates the semantic score BEFORE combining -- candidates
    below the threshold are excluded even if BM25/entity would boost them.
 
    The divisor adapts based on which signals are active:
        - Semantic only: max_possible = 1.0
        - Semantic + BM25: max_possible = 2.0
        - Semantic + BM25 + entity: max_possible = 2.5
        - Semantic + entity (no BM25): max_possible = 1.5
 
    Args:
        semantic_results: Candidate memories from vector search.
        bm25_scores: Normalized keyword scores keyed by memory ID.
        entity_boosts: Entity-link boosts keyed by memory ID.
        threshold: Minimum semantic score required before hybrid scoring.
        top_k: Maximum number of results to return.
        explain: Include score_details in each result when true.
 
    Returns:
        List of scored result dicts sorted by combined score descending.
    """
    has_bm25 = bool(bm25_scores)
    has_entity = bool(entity_boosts)
 
    max_possible = 1.0
    if has_bm25:
        max_possible += 1.0
    if has_entity:
        max_possible += ENTITY_BOOST_WEIGHT
 

Copydef similarity_search_with_score(
    self,
    query: str,
    k: int = 5,
    filter: Optional[Dict] = None,
) -> List[Tuple[Document, float]]:
    """Return documents most similar to query with scores."""
    logger.debug("Starting similarity search with score")
    logger.debug("Query: %s", query)
    logger.debug("k: %d", k)
    logger.debug("Filter: %s", filter)
 
    embedding = self.embedder.embed_query(query)
    results = self.search(query=query, vectors=embedding, top_k=k, filters=filter)
 
    docs_and_scores = [
        (Document(page_content=result.payload.get("text", ""), metadata=result.payload), result.score)
        for result in results
    ]
    logger.debug("Found %d results", len(docs_and_scores))
    return docs_and_scores
 

Copydef reset(self):
    """
    Reset the Google Matching Engine index.
    """
    logger.warning("Reset operation is not supported for Google Matching Engine")
    pass

Copydef _fetch_relevant_memories(self, messages, user_id, agent_id, run_id, filters, top_k):
    # Currently, only pass the last 6 messages to the search API to prevent long query
    message_input = [f"{message['role']}: {message['content']}" for message in messages][-6:]
    # TODO: Make it better by summarizing the past conversation
    return self.mem0_client.search(
        query="\n".join(message_input),
        user_id=user_id,
        agent_id=agent_id,
        run_id=run_id,
        filters=filters,
        top_k=top_k,
    )

Copydef lemmatize_for_bm25(text: str) -> str:
    """Lemmatize text for BM25 matching.
 
    Returns space-joined lemmas for full-text search. Falls back to
    the original text if spaCy is unavailable.
    """
    from mem0.utils.spacy_models import get_nlp_lemma
 
    nlp = get_nlp_lemma()
    if nlp is None:
        return text
 
    doc = nlp(text.lower())

Copyif nlp is None:
    return text

Copy# This handles noun/verb ambiguity (meeting/meet, attending/attend).
if token.text.endswith("ing") and token.text != lemma and token.text.isalnum():
    tokens.append(token.text)

Copyreturn " ".join(tokens)

Copydef get_bm25_params(query: str, *, lemmatized: Optional[str] = None) -> tuple:
    """Get BM25 sigmoid parameters based on query length.
 
    Longer queries tend to have higher raw BM25 scores, so we adjust
    the sigmoid midpoint and steepness accordingly.
 
    Returns:
        (midpoint, steepness) for sigmoid normalization.
    """
    if lemmatized is None:
        from mem0.utils.lemmatization import lemmatize_for_bm25
 
        lemmatized = lemmatize_for_bm25(query)
    num_terms = len(lemmatized.split()) if lemmatized else 1
 
    if num_terms <= 3:
        return 5.0, 0.7
    elif num_terms <= 6:
        return 7.0, 0.6
    elif num_terms <= 9:
        return 9.0, 0.5
    elif num_terms <= 15:
        return 10.0, 0.5
    else:
        return 12.0, 0.5
 

Copyif lemmatized is None:
    from mem0.utils.lemmatization import lemmatize_for_bm25
 
    lemmatized = lemmatize_for_bm25(query)

Copydef normalize_bm25(raw_score: float, midpoint: float, steepness: float) -> float:
    """Normalize BM25 score to [0, 1] using logistic sigmoid.
 
    Args:
        raw_score: Raw BM25 score (unbounded, typically 0-20+).
        midpoint: Score at which sigmoid outputs 0.5.
        steepness: Controls how quickly sigmoid transitions.
 
    Returns:
        Normalized score in range [0, 1].
    """
    return 1.0 / (1.0 + math.exp(-steepness * (raw_score - midpoint)))

Copydef normalize_bm25(raw_score: float, midpoint: float, steepness: float) -> float:
    return 1.0 / (1.0 + math.exp(-steepness * (raw_score - midpoint)))

CopyENTITY_BOOST_WEIGHT = 0.5

CopyThreshold gates the semantic score BEFORE combining -- candidates
below the threshold are excluded even if BM25/entity would boost them.

Copyif explain:
    scored_result["score_details"] = {
        "semantic_score": semantic_score,
        "bm25_score": bm25_score,
        "entity_boost": entity_boost,
        "raw_score": raw_combined,
        "max_possible_score": max_possible,
        "final_score": combined,
        "threshold": threshold,
    }

Copyscored_result["score_details"] = {
    "semantic_score": semantic_score,
    "bm25_score": bm25_score,
    "entity_boost": entity_boost,
    "raw_score": raw_combined,
    "max_possible_score": max_possible,
    "final_score": combined,
    "threshold": threshold,
}

Copyraw_combined = semantic_score + bm25_score + entity_boost
combined = min(raw_combined / max_possible, 1.0)

Copymem_id_str = str(mem_id)
bm25_score = bm25_scores.get(mem_id_str, 0.0)
entity_boost = entity_boosts.get(mem_id_str, 0.0)
 
raw_combined = semantic_score + bm25_score + entity_boost

Copymem_id = result.get("id")
if mem_id is None:
    continue
 
semantic_score = result.get("score") or 0.0
if semantic_score < threshold:
    continue

Copyif explain:
    scored_result["score_details"] = {
        "semantic_score": semantic_score,
        "bm25_score": bm25_score,
        "entity_boost": entity_boost,
        "raw_score": raw_combined,
        "max_possible_score": max_possible,
        "final_score": combined,
        "threshold": threshold,
    }

Copyraw_combined = semantic_score + bm25_score + entity_boost
combined = min(raw_combined / max_possible, 1.0)

Copyscored.sort(key=lambda x: x["score"], reverse=True)
return scored[:top_k]

Copymem_id = result.get("id")
if mem_id is None:
    continue
 
semantic_score = result.get("score") or 0.0
if semantic_score < threshold:
    continue