LangChainとDIY: Oracle AI Databaseによるベクトル検索 (2026/02/12)

https://medium.com/oracledevs/langchain-vs-diy-vector-search-with-oracle-ai-database-fc2fd53c214c

投稿者：Anders Swanson

PythonとOracle AIデータベースを使用した2つの異なるベクトル検索実装の比較

重要なポイント

• Python を使用して Oracle AI Databaseでベクトル検索を実行する 2 つの方法、つまり、同じ DB、埋め込みモデル、データセットを使用して LangChain を使用する方法と SQL を直接記述する方法を比較します。
• LangChain は、スキーマの作成、インデックス作成、クエリを抽象化することでプロトタイピングを高速化します (SQL は不要)。ただし、スキーマの所有権などの特定の操作は困難です。
• 手書きの SQL アプローチでは、より多くのコードと配管の所有権を犠牲にして、スキーマ、インデックス タイプ/パラメータ、トランザクション、および複雑なクエリ/フィルタを完全に制御できます。
• 実用的な方法は、両方を組み合わせることです。つまり、柔軟性とパフォーマンスのために重要な部分をカスタムのままにしながら、フレームワーク コンポーネント (埋め込み、ローダーなど) を使用します。

フレームワークを使うべきか、それとも独自の実装を書くべきか、それは一体どういう場合でしょうか？特にAIに関しては。フレームワークは本番環境への最短ルートとなることもありますが、後々障害となることもあります。設計上の選択が明確でないことが多々あります。

この質問をより深く理解するために、Python と Oracle AI Database を使用した 2 つの異なるベクトル検索実装を比較し、それぞれの利点と欠点を見ていきます。

それぞれの例では、DB、埋め込みモデル、データセットは一定のままですが、API レイヤー、スキーマ制御、インデックス制御、および可観測性は変更されます。

1. まず、LangChainを使用して面倒な処理を実行します。
2. 次に、袖をまくってベクトル検索 SQL クエリを手書きします。

この記事で紹介するサンプルでは、​​Testcontainersと OpenAI を使用しています。

• Oracle AI Database 用の Python Testcontainers 実装については、こちらを参照してください。
• 一緒に学習を進めたい場合は、こちらからOpenAI APIキーを生成することもできます。また、お好みのLLMプロバイダーを使用するようにコードをカスタマイズすることも可能です。
• 記事を飛ばしてコードだけを見たい場合は、ここをクリックしてください。

ベクターデータベースが何なのか分からないですか？大丈夫です ->ここから始めましょう。

LangChain実装

まずはLangChainのベクトル検索実装です。コードはGitHubでこちらから入手できます。

依存関係から始めましょう。LangChainを使用するため、関連するPythonパッケージがいくつか必要になります。

# LangChain
"langchain-oracledb (>=1.2.0,<2.0.0)",
"langchain-openai (>=1.1.7,<2.0.0)",
"langchain-community (>=0.4.1,<0.5.0)"

次に、LangChain ( vector_search_sample.py )を使って、ごく標準的なベクトル検索の例を実装します。手順を順に見ていきましょう。

1. OpenAIの埋め込みモデルtest-embeddings-3-smallを使用するように設定されたLangChain Embeddingsオブジェクトを作成します。これを使用して、テキストをベクトル表現に変換します。
2. Oracle AI DatabaseコンテナDBを起動し、データベース接続を取得します。便宜上、Testcontainersモジュールを使用しています。
3. テーブル名とベクトル距離関数を指定して、 LangChainlangchain_oracledb.OracleVSベクトルストアを作成します。ベクトルストアオブジェクトは、テーブルとインデックスを透過的に作成し、ベクトルストアを操作するためのメソッド（検索など）を提供します。これは、関連するSQLを記述する必要がない（あるいは知る必要さえない）という点で便利です。
4. ベクトルを挿入するには、メソッドを使用しますadd_texts。先ほど作成した埋め込みモデルは、挿入時にテキストを埋め込みに変換するのに役立ちます。
5. 最後に、ベクターストアを使用してセマンティック検索を行います。ここでも、埋め込みモデルがクエリテキストを検索用のクエリベクターに変換します（この場合も、SQL文は1行も記述しません）。

import getpass 
import os 

from langchain_community.vectorstores.utils import DistanceStrategy 
from langchain_openai import OpenAIEmbeddings 
from langchain_oracledb import OracleVS 
from src.python_oracle.testcontainers_sample.oracle_database_container import OracleDatabaseContainer 

def  main (): 
    dimensions = 1536 
    # 環境からOpenAI APIキーをロードします
    if  not os.getenv( "OPENAI_API_KEY" ): 
        os.environ[ "OPENAI_API_KEY" ] = getpass.getpass( "OpenAI APIキーを入力してください: " ) 
        # OpenAI埋め込みモデルを作成します
    embeddings = OpenAIEmbeddings( 
        model= "text-embedding-3-small" , 
        dimensions=dimensions, 
        api_key=os.getenv( "OPENAI_API_KEY" ) 
    ) 
    with OracleDatabaseContainer() as oracledb: 
        conn = oracledb.get_connection() 
        vector_store = OracleVS(conn, 
                 embeddings, 
                 table_name= "sample_vectors" , 
                 distance_strategy=DistanceStrategy.COSINE) 
        print ( "#### Oracle AIデータベースへのデータの埋め込み ####" ) 
        # テキストを埋め込みとしてベクターDBに保存します
        vector_store.add_texts([ 
            "Reset the user's password, clear MFA lockouts, and unlock the account after verifying identity.",
            "Reinstall the application, clear local cache/temp files, and validate the PDF upload workflow end-to-end.",
            "Submit an access request for the finance dashboard, confirm the required role, and route it to the user's manager for approval.",
            "Upgrade/reinstall the VPN client, disable Wi‑Fi power-saving for the adapter, and collect logs to confirm the disconnect cause.",
           "Review the workload needs, then recommend object vs. block storage and share the internal storage standards/decision guide.",        ])         # 検証ベクトルはデータベースに保存されます
        cursor = oracledb.get_connection().cursor() 
        print ( "#### 埋め込みデータを表示 ####:" ) 
        for row in cursor.execute( 'SELECT id, text, metadata,「sample_vectors」からの埋め込み:
            if row is  None : 
                print ( "クエリからの結果がありません!" ) 
            print ( f"id (binary): {row[ 0 ]} , text: {row[ 1 ]} , metadata: {row[ 2 ]} , embedding: vector[ { len (row[ 3 ])} ]" ) 

        print ( "\n#### 類似検索 ####" ) 
        query = "Wi-Fi を使用しているときに VPN が数分ごとに切断され続けますが、イーサネットでは接続されたままです。修正できますか?" 
        print ( f"検索クエリ: ' {query} '" ) 
        documents = vector_store.similarity_search(query, k= 1 ) 
        print ( "\n#### 上位クエリ一致 ####" ) 
        for doc in documents: 
            print (doc.page_content) 
if __name__ == "__main__" : 
    main()

python-oracleディレクトリから、poetry を使ってこのサンプルを実行できます。サンプルを実行するには、docker互換の環境が必要です。

# 依存関係をインストール
poetry install 
# 仮想環境をアクティブにするコマンドを出力
poetry env activate 
# OpenAI API キーが必要になります :) 
# https://platform.openai.com/api-keys 
export OPENAI_API_KEY=<Your OpenAI API Key> 
# それから実行してください!
 python src/python_oracle/langchain/vector_search_sample.py

ベクトル検索が正常に機能したことを示す次のような出力が表示されます。

# ... initializiation logs, creating vectors and such

#### Similarity Search ####
Search query: 'My VPN keeps disconnecting every few minutes when I'm on Wi‑Fi, but it stays connected on Ethernet. Can you fix it?'
#### Top Query Match ####
Upgrade/reinstall the VPN client, disable Wi‑Fi power-saving for the adapter, and collect logs to confirm the disconnect cause.

もちろん、これはOracle AI Databaseベクターで可能なことのほんの一部に過ぎませんが、ライブラリの概要とベクター検索の入門として役立ちます。OracleVS LangChainモジュールは、ドキュメントのロード、ハイブリッド/全文検索などの追加機能を提供します。

LangChainなしでは？

LangChainを使わず、Pythonを使って似たようなものを実装してみましょう。今回は、oracledbとopenaiPython パッケージだけを使用します。

SQLクエリを自分で書かなければなりません😩

でも、そんなにひどいことじゃないって約束するよ。さあ、始めましょう！

SQLクエリ

vector_search_native.pyファイルには完全な実装が含まれています。これを SQL クエリとメインメソッドの 2 つの部分に分けてみましょう。まず、ヘッダーと SQL クエリです。

1. CREATE_TABLE ステートメント：vectorデータ型のsample_vectorsテーブルを作成します。次元数（OpenAI text-embedding-3-small モデルに基づく 1536 ）とベクターデータ形式（FLOAT64）を（オプションで）指定します。テーブルを作成するので、列や外部キーを追加して自由にカスタマイズできます。ただし、この例ではシンプルなままにしておきます。
2. CREATE_INDEX ステートメント：sample_vectorsテーブルの埋め込み列にベクトルインデックスを作成します。COSINE 距離、95% のターゲット精度、10 近傍パーティションの転置ファイルインデックスを使用します。もちろん、これは特定のユースケースに合わせてカスタマイズ可能です。AIベクトル検索ユーザーガイドをご覧ください。このインデックスを選んだのは、シンプルでテキスト埋め込みに最適だからです。
3. INSERT_TEXT_EMBEDDINGS ステートメント:データ (テキスト コンテンツとベクター埋め込み) をsample_vectorsテーブルに追加します。
4. SIMILARITY_SEARCH_QUERY文：関数を使用してsample_vectorsテーブルを照会しますvector_distance。この関数を使用して「類似度スコア」を作成し、結果をランク付けして、最初の（最も類似した）結果を取得します。

import getpass
import os
import array
from typing import Any

import oracledb
from openai import OpenAI
from src.python_oracle.testcontainers_sample.oracle_database_container import OracleDatabaseContainer
DIMENSIONS = 1536
CREATE_TABLE = f"""
create table if not exists sample_vectors (
    id        number generated always as identity primary key,
    content   clob,
    embedding vector({DIMENSIONS}, FLOAT64) 
        annotations(Distance 'COSINE', IndexType 'IVF')
)"""
CREATE_INDEX = """
create vector index if not exists idx_sample_vectors on sample_vectors (embedding)
    organization neighbor partitions
    distance COSINE
    with target accuracy 95
    parameters (
        type IVF, 
        neighbor partitions 10
    )"""
INSERT_TEXT_EMBEDDING = """
insert into sample_vectors 
    (content, embedding) values (:1, :2)"""

SIMILARITY_SEARCH_QUERY = """
select id, content, embedding, (1 - vector_distance(embedding, :1, COSINE)) as score
from sample_vectors
order by score desc
fetch first 1 rows only"""

SQL初心者で、ここまでの内容をすべて理解できなくても大丈夫です。しかし、これらはベクトル検索を動かす低レベルのプリミティブなので、理解を深めるためにも、実際に触れてみる価値はあります。

Python実装

コードに移りましょう。高レベルの構造は同じですが、LangChainに依存せずにこれらのSQL文を呼び出します。mainメソッドを分解すると、以下のようになります。

1. まず、OpenAIクライアントを入手します。これを使って、テキストからベクトル埋め込みを作成します。
2. Oracle AI DatabaseコンテナDBを起動し、データベース接続を取得します。ここでも、便宜上、Testcontainersモジュールを使用しています。
3. CREATE_TABLEおよびCREATE_INDEXステートメントを実行し、 sample_vectorsテーブルを使用してベクトルを保存/クエリするようにデータベースを初期化します。
4. insert_text_embeddingsメソッドを呼び出して、 sample_vectorsテーブルにテキスト埋め込みデータを入力します。このメソッドは、OpenAI クライアントを使用して各文を埋め込みベクトルに変換し、INSERT_TEXT_EMBEDDINGステートメントを使用してデータをバッチとしてデータベースにロードします。
5. 最後に、similarity_search メソッドを呼び出してベクトル埋め込みをクエリします。このメソッドはクエリベクトルを埋め込みに変換し、SIMLARITY_SEARCH_QUERYステートメントを使用してsample_vectorsテーブル内の類似ベクトルを検索します。

def main():
    # Load OpenAI API Key from environment
    if not os.getenv("OPENAI_API_KEY"):
        os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")
    openai = OpenAI()
    with OracleDatabaseContainer() as oracledb:
        conn = oracledb.get_connection()
        cursor = conn.cursor()
        print("Creating table if not exists sample_vectors")
        cursor.execute(CREATE_TABLE)
        print("Creating index if not exists idx_sample_vectors")
        cursor.execute(CREATE_INDEX)
        conn.commit()
        print("Loading data into sample_vectors table")
        insert_text_embeddings(conn, openai, texts=[
            "Reset the user's password, clear MFA lockouts, and unlock the account after verifying identity.",
            "Reinstall the application, clear local cache/temp files, and validate the PDF upload workflow end-to-end.",
            "Submit an access request for the finance dashboard, confirm the required role, and route it to the user's manager for approval.",
            "Upgrade/reinstall the VPN client, disable Wi‑Fi power-saving for the adapter, and collect logs to confirm the disconnect cause.",
            "Review the workload needs, then recommend object vs. block storage and share the internal storage standards/decision guide.",
        ])
        print("\n#### Display Embedded Data ####:")
        for row in cursor.execute('SELECT id, content, embedding FROM sample_vectors'):
            if row is None:
                print("No result from query!")
            print(f"id: {row[0]}, content: {row[1]}, embedding: vector[{len(row[2])}]")
        print("#### Similarity Search ####")
        query = "My VPN keeps disconnecting every few minutes when I'm on Wi‑Fi, but it stays connected on Ethernet. Can you fix it?"
        print(f"Search query: '{query}'")
        result = similarity_search(conn, openai, query)
        if not result:
            print("No result from similarity search query!")
        else:
            print("#### Top Query Match ####")
            print(f"id: {result[0]}, content: {result[1]}, embedding: vector[{len(result[2])}]")

def embed_text(openai: OpenAI, text: str) -> tuple[str, array.array[float]]:
    embedding = openai.embeddings.create(
        input=text,
        model="text-embedding-3-small"
    ).data[0].embedding
    # python array's "d" typecode corresponds to 64-bit floating point,
    # which is what we use in our sample_vectors table.
    vector = array.array("d", embedding)
    return text, vector
def insert_text_embeddings(conn: oracledb.Connection, openai: OpenAI, texts: list[str]):
    # load embeddings: you may also do this asynchronously
    # as it's a series of network calls.
    data = [embed_text(openai, x) for x in texts]
    # save embeddings as batch
    with conn.cursor() as cursor:
        cursor.executemany(
            INSERT_TEXT_EMBEDDING,
            data
        )
        conn.commit()
def similarity_search(conn: oracledb.Connection, openai: OpenAI, query: str) -> Any:
    _, vector = embed_text(openai, query)
    with conn.cursor() as cursor:
        for row in cursor.execute(
                statement=SIMILARITY_SEARCH_QUERY,
                parameters=(vector,) # expects a tuple!
        ):
            return row
    return None
if __name__ == "__main__":
    main()

python-oracleディレクトリから、次のpoetryコマンドでこのサンプルを実行できます:

# If you ran the prior example you can skip these three commands:
poetry install
poetry env activate
export OPENAI_API_KEY=<Your OpenAI API Key> 

# Then run it!
python src/python_oracle/database/vector_search_native.py

出力は前の例とかなり似ているはずです。LangChainを使わずにほぼ同じことをしていることになります。

# ... initializiation logs, creating vectors and such

#### Similarity Search ####
Search query: 'My VPN keeps disconnecting every few minutes when I’m on Wi‑Fi, but it stays connected on Ethernet. Can you fix it?'
#### Top Query Match ####
id: 4, content: Upgrade/reinstall the VPN client, disable Wi‑Fi power-saving for the adapter, and collect logs to confirm the disconnect cause., embedding: vector[1536]

まとめ: フレームワークを使うべきか、独自のソリューションを作るべきか

2つの実装を比較すると、LangChainソリューションの方が明らかにシンプルです。コード行数が少なく、高レベルの抽象化を採用しており、SQLを一切書く必要がありません。

しかし、フレームワークは細部を隠蔽する側面もあります。LangChainはスキーマを作成・管理するため、リレーションシップの管理、カスタムデータ型の使用（ベクターをJSONドキュメントに保存したい場合はどうすればいいでしょうか？）、イベントなどの高度なデータベース機能の使用が難しくなります。LangChainのようなフレームワークでは、以下の基本的な操作が困難になります。

• スキーマの所有権と命名規則
• インデックスの種類/パラメータとその調整方法
• トランザクション境界とコミット動作
• 高度な SQL 構造 (CTE、分析関数、結合) を使用する能力
• 削除/更新の処理とワークフローの再埋め込み

類似検索機能もその一例です。手書きのSQLを使用することで、必要に応じてカスタマイズ、調整、他のデータとの関連付けを行うことができます。

たとえば、類似性検索に述語を追加する:

select id, content, embedding, (1 - vector_distance(embedding, :1, COSINE)) as score
  from sample_vectors
  -- basic predicate filter, endless possibilities
  where department = :2
  order by score desc
  fetch first 1 rows only

判決はどうなりましたか？

• LangChain : プロトタイピング、ストア間の移植性、小規模チーム、高速反復。
• DIY : 厳格なガバナンス、カスタム スキーマ、複雑な結合/フィルター、パフォーマンス チューニング、長期にわたるシステム。

結局のところ、それは抽象化の利便性と明示的な制御の問題です。

この質問に明確な答えがあるとは思いません。私が言いたいのは、フレームワークを選ぶ際には、自分が何に同意するのかを理解することです。このような設計上の選択は、プロジェクトのライフサイクルの深部に入ってしまうと、なかなか変更できないことがよくあります。

逆に、DIY では再試行ロジック、バッチ処理、非同期埋め込み呼び出し、インデックス/クエリのチューニングなど、追加の負荷が発生します。

バランスの取れたアプローチとしては、フレームワークの機能を段階的に導入していくことが挙げられます。LangChainを使用しているからといって、すべてにLangChainを使用する必要はありません。例えば、埋め込みとドキュメントの読み込みにはLangChainを使用し、取得ステップには独自のSQLを記述するといったことも可能です。フレームワークを自分のニーズに合わせて変更するのではなく、フレームワークの機能を最大限活用しましょう。

よくある質問（FAQ）

フレームワークとカスタム SQL はいつ選択すればよいですか?

高速な反復、移植性、小規模チームにはフレームワークを使用し、厳格なガバナンス、カスタム スキーマ、複雑な結合/フィルター、長期システムのパフォーマンス チューニングには DIY を選択します。

両方のアプローチを組み合わせることはできますか?

はい。フレームワーク コンポーネント (埋め込み、ドキュメントの読み込みなど) を選択的に採用し、取得 SQL を手書きして、重要なところで観測可能性、制御、最適化を維持します。

全体的なトレードオフについてはどのように考えればよいでしょうか?

スピードとコントロールの選択です。フレームワークを使用すると迅速かつ一貫した生産性が得られ、カスタム SQL を使用すると精度と長期的な柔軟性が得られます。プロジェクトのタイムライン、ガバナンスのニーズ、予想されるチューニングの量に基づいて選択してください。