Hybrid Search Example in Python

# search.py : python3 search.py solar panels for a small roof
# Needs Python 3.8 or newer and the requests package (pip install requests).
import re
import sys

import requests

CONFIG = {
    "email": "YOUR_EMAIL",
    "api_key": "YOUR_API_KEY",
    "index_name": "YOUR_INDEX",
    "index_url": "https://YOUR_INDEX_HOST/solr/YOUR_INDEX/select",
    "index_user": "opensolr",
    "index_password": "YOUR_API_KEY",
    "rows": 10,
}
QF = "title^0.2 description^0.05 uri^0.01 text^0.01 text_t^0.01"
MM = "2<65% 4<50% 8<40%"

def embed_query(text):
    """Step 1: the meaning vector of the search, or None when there is none."""
    if len(text) < 2:
        return None
    try:
        res = requests.get(
            "https://api.opensolr.com/solr_manager/api/embed",
            params={
                "email": CONFIG["email"],
                "api_key": CONFIG["api_key"],
                "index_name": CONFIG["index_name"],
                "is_query": 1,
                "payload": text,
            },
            timeout=5,
        )
        vector = res.json() if res.status_code == 200 else None
    except (requests.RequestException, ValueError):
        return None
    if isinstance(vector, list) and len(vector) == 1024:
        return [float(x) for x in vector]
    return None

def search(text, page=1, lang=""):
    """Step 2: one Solr request, hybrid when there is a vector, words only when there is not."""
    text = text.strip()[:200]
    page = max(1, min(100, int(page)))
    lexical = '{!edismax qf="%s" mm="%s" v=$uq}' % (QF, MM)
    params = [
        ("uq", text),
        ("fl", "id,uri,title,description,og_image,creation_date,score"),
        ("rows", CONFIG["rows"]),
        ("start", (page - 1) * CONFIG["rows"]),
        ("fq", "content_type:text*"),
        ("hl", "true"),
        ("hl.method", "unified"),
        ("hl.encoder", "html"),
        ("hl.q", '{!edismax qf="title description uri text" mm=0 v=$uq}'),
        ("hl.fl", "title,description,text"),
        ("hl.fragsize", 300),
        ("hl.snippets", 1),
        ("hl.defaultSummary", "true"),
        ("f.title.hl.fragsize", 0),
        ("facet", "true"),
        ("facet.field", "{!ex=lang}meta_detected_language"),
        ("facet.mincount", 1),
        ("spellcheck", "true"),
        ("spellcheck.q", text),
        ("spellcheck.count", 5),
        ("spellcheck.collate", "true"),
        ("spellcheck.maxCollations", 3),
        ("spellcheck.maxCollationTries", 15),
        ("echoParams", "none"),
        ("wt", "json"),
    ]
    if re.fullmatch(r"[a-zA-Z]{2,3}(-[a-zA-Z0-9]{2,8})?", lang or ""):
        params.append(("fq", "{!tag=lang}{!term f=meta_detected_language}" + lang))
    vector = embed_query(text)
    if vector:
        params.append(("q", "{!hybrid lexical=$lexicalQuery vector=$vectorQuery mode=union alpha=0.85 topN=500}"))
        params.append(("lexicalQuery", lexical))
        params.append(("vectorQuery", "{!knn f=embeddings topK=500}[" + ",".join(str(x) for x in vector) + "]"))
    else:
        params.append(("q", lexical))
    # POST, because the vector makes the request too long for a URL
    res = requests.post(
        CONFIG["index_url"],
        data=params,
        auth=(CONFIG["index_user"], CONFIG["index_password"]),
        timeout=10,
    )
    res.raise_for_status()
    return res.json()

if __name__ == "__main__":
    data = search(" ".join(sys.argv[1:]) or "solar panels")
    print(data["response"]["numFound"], "results")
    for doc in data["response"]["docs"]:
        print("%.4f  %s  %s" % (doc.get("score", 0), doc.get("title", ""), doc.get("uri", "")))