Skip to main content

Explainable Results

A result without context forces users to trust it blindly. An explainable result shows why it was returned: which fields matched, which related records reinforce it, and what summary metric supports it.

This tutorial shows three patterns for building explainable results on top of RushDB:

  1. Field-level match explanation — which fields matched the query
  2. Evidence assembly — related records that corroborate the result
  3. Score + signal summary — pairing semantic __score with structured signals

Pattern 1: Field-level match explanation​

After structured retrieval, compute which fields matched the query and return them alongside the record.

from rushdb import RushDB
import os

db = RushDB(os.environ["RUSHDB_API_KEY"], base_url="https://api.rushdb.com/api/v1")

def explain_match(record_data: dict, query: dict) -> list[dict]:
    matches = []
    for field, query_value in query.items():
        if field.startswith("$") or isinstance(query_value, dict):
            continue
        if field in record_data:
            matches.append({"field": field, "value": record_data[field], "queryValue": query_value})
    return matches

def search_with_explanation(where: dict) -> list[dict]:
    result = db.records.find({"labels": ["ARTICLE"], "where": where})
    return [
        {
            "id": a.id,
            "title": a.data.get("title"),
            "matchedFields": explain_match(a.data, where)
        }
        for a in result.data
    ]

Pattern 2: Evidence assembly via graph traversal​

After retrieving a primary result, traverse related records to assemble corroborating evidence.

from concurrent.futures import ThreadPoolExecutor

def assemble_evidence(article_id: str) -> dict:
    def get_author():
        r = db.records.find({
            "labels": ["AUTHOR"],
            "where": {"ARTICLE": {"$relation": {"type": "AUTHORED_BY", "direction": "in"}, "__id": article_id}}
        })
        return {"name": r.data[0].data.get("name")} if r.data else None

    def get_cited_by():
        r = db.records.find({
            "labels": ["ARTICLE"],
            "select": {"count": {"$count": "*"}},
            "where": {"ARTICLE": {"$relation": {"type": "CITES", "direction": "out"}, "__id": article_id}}
        })
        return r.data[0].data.get("count", 0) if r.data else 0

    def get_topics():
        r = db.records.find({
            "labels": ["TOPIC"],
            "where": {"ARTICLE": {"$relation": {"type": "COVERS", "direction": "in"}, "__id": article_id}}
        })
        return [t.data.get("name") for t in r.data]

    with ThreadPoolExecutor(max_workers=3) as pool:
        author_f  = pool.submit(get_author)
        cited_f   = pool.submit(get_cited_by)
        topics_f  = pool.submit(get_topics)

    return {
        "author":        author_f.result(),
        "citedBy":       cited_f.result(),
        "relatedTopics": topics_f.result()
    }

def explained_semantic_search(user_query: str) -> list[dict]:
    results = db.ai.search({
        "query": user_query,
        "propertyName": "content",
        "labels": ["ARTICLE"],
        "where": {"status": "published"},
        "limit": 5
    })

    return [
        {
            "id": a.id,
            "title": a.get("title"),
            "score": a.score,
            "evidence": assemble_evidence(a.id)
        }
        for a in results.data
    ]

Pattern 3: Score + structured signal summary​

For agent contexts, produce a brief natural-language explanation rather than raw data.

def build_explanation_text(result: dict) -> str:
    parts = []
    score_pct = int((result.get("score") or 0) * 100)
    parts.append(f'"{result["title"]}" is a strong match (relevance: {score_pct}%).')

    evidence = result.get("evidence", {})
    if evidence.get("author"):
        parts.append(f'Written by {evidence["author"]["name"]}.')

    topics = evidence.get("relatedTopics", [])
    if topics:
        parts.append(f'Topics: {", ".join(topics[:3])}.')

    cited_by = evidence.get("citedBy", 0)
    if cited_by > 0:
        label = "article" if cited_by == 1 else "articles"
        parts.append(f'Cited by {cited_by} other {label}.')

    return " ".join(parts)

Pattern 4: REST-only evidence pipeline​

If you are building an agent or backend service without an SDK, assemble evidence with sequential REST calls.

# 1. Semantic search for the primary result
results = db.ai.search({
    "query": "reducing latency",
    "propertyName": "content",
    "labels": ["ARTICLE"],
    "limit": 1
})

article = results.data[0]

# 2. Fetch author evidence
author_result = db.records.find({
    "labels": ["AUTHOR"],
    "where": {
        "ARTICLE": {
            "__id": article.data["__id"],
            "$relation": {"type": "AUTHORED_BY", "direction": "in"}
        }
    }
})

author_name = author_result.data[0].data.get("name", "unknown") if author_result.data else "unknown"
print(f"Result: {article.data['title']} (score: {article.data['__score']}) — Author: {author_name}")

When to use explainability​

ContextRecommended patterns
User-facing search resultsField match highlights + related topic tags
AI agent tool responsesScore + structured summary text
Compliance / audit surfacesFull evidence assembly with provenance links
Debug / developmentField-level match + raw score logging

Production caveat​

Evidence assembly makes one additional query per result per evidence type. For 10 results and 3 evidence types, that is 30 extra queries. Cache evidence for repeated result IDs within the same session, or pre-join commonly needed evidence fields during ingestion.

Next steps​