DuckDB を用いて日々のストック記事をハイブリッド検索する

Apr 30, 2025

🎧 音声概要 (※ NotebookLM で生成されており、誤りを含む場合があります)

今回は、DuckDB を使って、日々ストックしている Web 記事などのハイブリッド検索（全文検索とベクトル類似度検索の組み合わせ）を実現する試みについてまとめます。

モチベーション

自分は普段、情報収集や後で読みたい記事の保存に Readwise Reader というサービスを利用しています。RSS フィードの購読や記事のアーカイブができ、非常に便利なツールです [1] 。

この Readwise Reader にも検索機能（全文検索）はあるのですが、どこか微妙さを感じていました。特に日本語の記事を探す際に意図した記事が見つからないことが多々あり、不満ポイントでした (海外サービスあるある)。

そんな矢先、時雨堂さんの「DuckDB でハイブリッド検索」の記事を拝見し、DuckDB を用いることで比較的に簡単にハイブリッド検索を実現できることに感銘を受けました。

幸い Readwise Reader では公式で Reader API が用意されており、ストックした記事情報を取得できます。こちらの API から記事情報を取得しつつ、ローカル環境でハイブリッド検索システムを試しに構築してみることにしました。

Readwise から記事を取得する

まずは DuckDB に格納する、検索対象になるデータを集めます。前述の Reader API を用いて、試しにここ数日でストックした記事のデータを取得します。

スクリプト全体はこちらで公開しています。

スクリプトを実行すると、以下のようなフォーマットの JSON を取得できます。 title に記事のタイトル、html_content に保存した記事の HTML が格納されていたり、その他記事に関するあらゆる情報が返ってきます。

[
    {
        "id": "01jskz8xxxxx",
        "url": "https://read.readwise.io/read/xxx",
        "title": "How Heroku Is ‘Re-Platforming’ Its Platform",
        "author": "Heather Joslyn",
        "source": "Reader RSS",
        "category": "rss",
        "location": "new",
        "tags": {},
        "site_name": "The New Stack",
        "word_count": 604,
        "created_at": "2025-04-24T13:19:37.157537+00:00",
        "updated_at": "2025-04-25T05:01:57.499511+00:00",
        "published_date": "2025-04-24",
        "summary": "Building an internal platform, or moving a legacy monolith to microservices and the cloud, can be a huge undertaking. Such\nThe post How Heroku Is ‘Re-Platforming’ Its Platform appeared first on The New Stack.",
        "image_url": "https://cdn.thenewstack.io/media/2025/04/56e4407a-kccnc-eu-25_betty-junod_featured-1024x576.png",
        "content": null,
        "source_url": "https://thenewstack.io/how-heroku-is-re-platforming-its-platform/",
        "notes": "",
        "parent_id": null,
        "reading_progress": 0,
        "first_opened_at": "2025-04-25T05:01:28.715000+00:00",
        "last_opened_at": "2025-04-25T05:01:28.715000+00:00",
        "saved_at": "2025-04-24T13:19:34.098000+00:00",
        "last_moved_at": "2025-04-25T05:01:17.234000+00:00",
        "html_content": "...HTML コンテンツが入る (省略)..."
    }
]

今回は簡単のため、主に title と html_content の情報を利用して、検索用のデータベースを構築していきます。

データの整形をする

取得した html_content は HTML なので、そのまま検索対象とすると不要な情報がノイズになる可能性があります。そこで markdownify パッケージを使って HTML を Markdown 形式に変換します。ソースコードはこちらで公開しています。

Markdown に変換する該当のコードは下記のとおりです:

# ...

# to_markdown 関数で HTML の文字列を markdown に変換す
def to_markdown(html:str)->str:
    return markdownify(html or "", heading_style="ATX")

json_files = glob.glob("articles/*.json")
records = []
for path in json_files:
    data = json.load(open(path, encoding="utf-8"))
    for rec in (data if isinstance(data, list) else [data]):
        md = to_markdown(rec.get("html_content",""))
        title = rec.get("title","")
        rec["fts_text"] = ja_tokens(title + "。 " + md)
        # vss_text は後続のステップでベクトル化する
        rec["vss_text"] = title + "。 " + md
        rec["markdown_content"] = md
        records.append(rec)

# ...

この処理を実行することで、markdown_content に Markdown 形式のコンテンツが格納されます。また、作成した Markdown のデータは fts_text (全文検索用テキスト) や vss_text (ベクトル検索用テキスト) のデータとしても利用します。

DuckDB に格納する

次に、整形したデータを DuckDB に格納します。ここでは、DuckDB にデータを INSERT する直前に embedding し、ベクトルを格納しています。

DuckDB や tokenizer、embedding model の初期化などのコードはこちらで公開しています [2] 。

# ...

con = duckdb.connect("article_search.duckdb")
con.install_extension("vss")
con.load_extension("vss")
con.install_extension("fts")
con.load_extension("fts")


# 1. 先に空テーブルを作成
con.sql("""
CREATE TABLE IF NOT EXISTS articles (
    id VARCHAR PRIMARY KEY,
    title VARCHAR,
    markdown_content VARCHAR,
    fts_text VARCHAR,
    embedding FLOAT[2048]
);
""")

# 既存データ削除して入れ直す
con.sql("DELETE FROM articles;")

# 2. Embedding 計算
for i, rec in df.iterrows():
    # 既に同じ id が DB にあればスキップ
    already = con.sql(
        "SELECT COUNT(*) FROM articles WHERE id = ?", params=[rec["id"]]
    ).fetchone()[0]
    if already:
        print(f"SKIP: {rec['id']} (already in DB)")
        continue

    try:
        with torch.inference_mode():
            emb = v_model.encode_document([rec["vss_text"]], v_tokenizer)[0]
        con.execute(
            "INSERT INTO articles VALUES (?, ?, ?, ?, ?)",
            [
                rec["id"],
                rec["title"],
                rec["markdown_content"],
                rec["fts_text"],
                emb.cpu().squeeze().numpy().tolist(),
            ],
        )
        print(f"INSERT: {rec['id']}")
    except Exception as e:
        print(f"ERROR: {rec['id']} - {e}")

# 3. FTS インデックス
# 既存のインデックスがあれば削除
try:
    con.sql("PRAGMA drop_fts_index('articles');")
except Exception as e:
    print("FTSインデックス削除時の例外:", e)
    
con.sql("""
PRAGMA create_fts_index(
    'articles',
    'id',
    'fts_text',
    stemmer='none', stopwords='none', ignore='', lower=false, strip_accents=false
);
""")

以上で DB の準備は完了です。

参考までに、Apple M2 Pro (メモリ32GB) の環境において、テストで240件のデータをループ処理するのに約7分30秒ほどかかりました。

検索を実行する

DB が整ったら、実際にクエリを投げて検索してみます。以下のようにクエリ用の関数を用意します。

def fts_search(conn, query, k=5):
    q = ja_tokens(query)
    return conn.sql(f"""
        SELECT id, title,
               fts_main_articles.match_bm25(id, '{q}') AS score
        FROM articles
        WHERE score IS NOT NULL
        ORDER BY score DESC
        LIMIT {k}
    """).fetchdf()

def vss_search(conn, query, k=5):
    with torch.inference_mode():
        q_emb = v_model.encode_query(query, v_tokenizer)
    return conn.sql(
        f"""
        SELECT id, title,
               array_cosine_distance(embedding, ?::FLOAT[2048]) AS dist
        FROM articles
        ORDER BY dist ASC
        LIMIT {k}
        """,
        params=[q_emb.cpu().squeeze().numpy().tolist()],
    ).fetchdf()

from sentence_transformers import CrossEncoder
reranker = CrossEncoder(
    "hotchpotch/japanese-bge-reranker-v2-m3-v1", device=device, max_length=512
)

def hybrid_search(conn, query, k_fts=10, k_vss=10, top_n=5):
    fts_df = fts_search(conn, query, k_fts)
    vss_df = vss_search(conn, query, k_vss)
    pool = pd.concat([fts_df, vss_df]).drop_duplicates("id")
    pairs = [(query, txt) for txt in pool["title"]]
    scores = reranker.predict(pairs)
    pool["score"] = scores
    return pool.sort_values("score", ascending=False).head(top_n)[["id","title","score"]]

実際に実行すると以下のような結果を得られます。

display(hybrid_search(con, "AI エージェント", top_n=5))

表示された結果を見ると、クエリに関連する記事が返ってきていそうです！

おわりに

今回、DuckDB と Readwise API を活用することで、比較的簡単にローカル環境にハイブリッド検索システムを構築することができました。全文検索とベクトル検索を組み合わせ、さらにリランキングを行うことで、Readwise Reader 標準の検索機能より優れた検索基盤ができそうな予感です。

さらに、MCP サーバに仕立て上げるなどして Retriever 化すれば LLM と連携させることもできるので、活用の幅が広がりそうです。

2025-05-18 追記

MCP Server を実装して Raycast から使ってみた、という記事を書きました。

DuckDB で構築したハイブリッド検索システムを MCP 経由で Raycast から呼び出す

DuckDB で作成したハイブリッド検索システムを、Raycast から利用できるようにしてみました

www.yukioh.net

まだまだチューニングの余地はあるかと思いますが、第一歩として満足のいく結果が得られました。

ここまでお読みいただきありがとうございました。

[1]

弊ポストも是非参照してください。↑

[2]

時雨堂さんの記事に記載のサンプルコードを参考にさせていただきながら、独自フロー構築のため変更を加えています。ライセンスなど問題ございましたらご指摘いただけますと幸いです。↑