🤖 Nomic Atlasのベクトルサーチの「Range Filter」について

１．はじめに

• Nomic Atlasは、特許データなどの大量の非構造化データ（テキストなど）を視覚化・探索できるプラットフォームで、APIを使えば、Pythonなどのプログラムから直接、埋め込みベクトルを用いた高度な検索を実行可能
• この検索時、単に類似度が高い結果を出すだけでなく、「設立年がこの範囲」や「資金調達額がこの金額以上」といった書誌事項の条件で結果を絞り込めるのが非常に強力。
• が、以下のとおりハマったので後のためにメモ。

２．課題・困ったこと

• ベクトルサーチを行う際、ベクトルの類似以外に、検索条件に何らかのメタデータの条件を付けたいが、どう書けばいいかわからない。例えば、特定の期間に設立された企業や、一定額以上の資金調達を受けている企業のみを対象にしたい、といったケース
• Nomic AtlasのAPIでフィルタリングを行うには、リクエストボディ内の selection フィールドを利用する。
• 公式ドキュメントには、search メソッドを使った文字列によるフィルタリングの記載がある。
• しかし、「設立日がこの日付からこの日付まで」「資金調達額が***以上」といった range フィルターの具体的な書き方が、ドキュメントからは明確に読み取れず。特に、 日付（timestamp） のような数値として扱われるべきフィールドに対して、range フィルターがどのように機能するのか、その引数（リストの形式や、日付のデータ形式など）がわからず、期待通りに動作させるまでに試行錯誤を繰り返すことに。

３．こう書けばよかった

• 試行錯誤の結果、最終的にうまく動作した関数は以下の通り。

url = "https://api-atlas.nomic.ai/v1/query/topk"
import requests
import json
import datetime

# 1. 範囲の開始と終了を定義
start_date = datetime.datetime(2000, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc)
end_date = datetime.datetime(2025, 12, 31, 23, 59, 59, tzinfo=datetime.timezone.utc)

# 2. Unixエポックからのミリ秒に変換（重要！）
start_millis = int(start_date.timestamp() * 1000)
end_millis = int(end_date.timestamp() * 1000)


def get_app_numbers(query,topK=100):
    payload = json.dumps({
        "k": topK,
        "query": query,
        "fields": [
        ~~略~~

        ],
        "selection": {
            "method": "composition",
            "conjunctor": "ALL",
            "filters": [
                { 
                    "method": "range",
                    "field": "total_funding_usd",
                    "range": [0.00000000001, 1000000000000000000] # 例えば資金調達額の範囲
                },
                { 
                    "method": "range",
                    "field": "founded_on",
                    "range": [start_millis, end_millis] # 例えば設立日の範囲（ミリ秒）
                },            
                {
                    "method": "search",
                    "query": "private",
                    "field": "ipo_status"
                }
            ]
        },
        "task_type": "search_document",
        "projection_id": "PROJECTION_ID",
        "includeSimilarity": True
    })

    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Authorization': 'Bearer {TOKEN}'
    }

    response = requests.request("POST", url, headers=headers, data=payload)

    if response.status_code != 200:
        print(f"API Request failed with status code: {response.status_code}")
        print(f"Response text: {response.text}")
        response.raise_for_status()

    return response.json()

4.その他

• 根拠は探してもわからなかったけど、たぶんelasticsearchと同じ書き方でいけると思う。