Oracle AI 向量搜索教程：使用 Python 存储与查询嵌入向量

度量

比较的内容

适用场景

注意事项

余弦

向量方向

文本嵌入与语义相似性

结果依赖于使用同一嵌入模型与一致的预处理

欧氏

直线距离

小型示例、几何直觉、部分数值特征向量

幅度会影响距离，因而缩放可能很重要

点积

方向与幅度共同作用

围绕归一化向量或内积打分设计的工作流

在比较结果前需要理解分值方向与归一化规则

索引家族

基本思想

优势

取舍

可能的使用场景

IVF / 邻接分区

对向量空间进行分区，再搜索可能的分区

心智模型简单，适用于许多批量加载的数据集

召回与速度取决于分区与搜索设置

我们希望为不断增长的数据集提供直接的近似搜索路径

HNSW / 邻接图

构建连接近邻向量的图，以便快速遍历

对许多最近邻工作负载有较强的召回与延迟表现

更偏内存，且配置影响更大

我们需要低延迟搜索，并能承担内存与调优投入

mkdir oracle-ai-vector-search-python
 cd oracle-ai-vector-search-python技术支持

docker run --name oracle-free-26ai-vector \
   --detach \
   --publish 1521:1521 \
   --publish 5500:5500 \
   --shm-size=1g \
   --env ORACLE_PWD="replace-with-a-strong-password" \
   container-registry.oracle.com/database/free:23.26.1.0-lite-amd64技术支持

docker logs oracle-free-26ai-vector技术支持

# Create the application user and grant the privileges needed for the demo.
 docker exec -i oracle-free-26ai-vector \
   sqlplus -s 'sys/"replace-with-oracle-pwd"@FREEPDB1 as sysdba' <<'SQL'
 WHENEVER SQLERROR EXIT SQL.SQLCODE

 CREATE USER vector_demo IDENTIFIED BY "replace-with-vector-demo-password";

 GRANT CREATE SESSION, CREATE TABLE, CREATE SEQUENCE, UNLIMITED TABLESPACE TO vector_demo;

 EXIT
 SQL技术支持

python -m venv .venv
 source .venv/bin/activate
 python -m pip install --upgrade pip技术支持

python -m pip install oracledb openai技术支持

export DB_USER="vector_demo"
 export DB_PASSWORD="replace-with-vector-demo-password"
 export DB_DSN="localhost:1521/FREEPDB1"
 export OPENAI_API_KEY="replace-with-your-api-key"
 export EMBEDDING_MODEL="text-embedding-3-small"
 export EMBEDDING_DIM="1536"技术支持

import os

 import oracledb

 ADMIN_USERS = {"SYS", "SYSTEM", "PDBADMIN"}


 def required_env(name):
 	value = os.getenv(name)
 	if not value:
         raise SystemExit(f"Set {name} before running this script.")
 	return value


 def main():
 	# Load required database settings from environment variables.
 	user = required_env("DB_USER")
     password = required_env("DB_PASSWORD")
 	dsn = required_env("DB_DSN")

 	if user.upper() in ADMIN_USERS:
         raise SystemExit(
             "Use the vector_demo application user, not an admin user."
     	)

 	# Open a Thin-mode connection with the application user.
 	with oracledb.connect(user=user, password=password, dsn=dsn) as connection:
         mode = "thin" if oracledb.is_thin_mode() else "thick"
         print("Connected to Oracle Database.")
         print(f"Driver mode: {mode}")
         print(f"Database version: {connection.version}")


 if __name__ == "__main__":
 	main()技术支持

python 01_check_connection.py技术支持

Connected to Oracle Database.
 Driver mode: thin
 Database version: 23.26.1.0.0技术支持

import array
 import os

 import oracledb

 ADMIN_USERS = {"SYS", "SYSTEM", "PDBADMIN"}


 def required_env(name):
 	value = os.getenv(name)
 	if not value:
         raise SystemExit(f"Set {name} before running this script.")
 	return value


 def main():
 	user = required_env("DB_USER")
 	if user.upper() in ADMIN_USERS:
         raise SystemExit(
             "Use the vector_demo application user, not an admin user."
     	)

 	# Define three small vectors so the distance calculation is easy to inspect.
 	rows = [
         ("apple", array.array("f", [1.0, 0.0, 0.0])),
         ("banana", array.array("f", [0.9, 0.1, 0.0])),
         ("car", array.array("f", [0.0, 1.0, 0.0])),
 	]

 	with oracledb.connect(
         user=user,
         password=required_env("DB_PASSWORD"),
         dsn=required_env("DB_DSN"),
 	) as connection:
         with connection.cursor() as cursor:
             # Recreate the table so the script can be run repeatedly.
             cursor.execute(
                 """
                 BEGIN
                 	EXECUTE IMMEDIATE 'DROP TABLE manual_vectors PURGE';
                 EXCEPTION
                 	WHEN OTHERS THEN
                     	IF SQLCODE <> -942 THEN
                         	RAISE;
                     	END IF;
                 END;
                 """
             )
             cursor.execute(
                 """
                 CREATE TABLE manual_vectors (
                 	id NUMBER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
                 	label VARCHAR2(100),
                 	embedding VECTOR(3, FLOAT32)
                 )
                 """
             )
             # Bind each vector as a 32-bit floating-point array.
             cursor.executemany(
                 """
                 INSERT INTO manual_vectors (label, embedding)
                 VALUES (:1, :2)
                 """,
                 rows,
             )
             connection.commit()

             # Rank rows by Euclidean distance from the query vector.
             cursor.execute(
                 """
                 SELECT label,
                    	VECTOR_DISTANCE(
                        	embedding,
                        	:query_vector,
                        	EUCLIDEAN
                    	) AS distance
                 FROM manual_vectors
                 ORDER BY distance
                 FETCH EXACT FIRST 2 ROWS ONLY
                 """,
                 query_vector=array.array("f", [1.0, 0.0, 0.0]),
             )

             print("Created manual_vectors.")
             print("Inserted 3 manual vectors.")
             print("\nNearest manual vectors:")
             for rank, (label, distance) in enumerate(cursor, start=1):
                 print(f"{rank}. {label:<6} distance={float(distance):.4f}")


 if __name__ == "__main__":
 	main()技术支持

python 02_manual_vectors.py技术支持

Created manual_vectors.
 Inserted 3 manual vectors.

 Nearest manual vectors:
 1. apple  distance=0.0000
 2. banana distance=0.1414技术支持

import os

 import oracledb

 ADMIN_USERS = {"SYS", "SYSTEM", "PDBADMIN"}


 def required_env(name):
 	value = os.getenv(name)
 	if not value:
         raise SystemExit(f"Set {name} before running this script.")
 	return value


 def configured_dimension():
 	try:
         dimension = int(required_env("EMBEDDING_DIM"))
 	except ValueError as exc:
         raise SystemExit("EMBEDDING_DIM must be an integer.") from exc

 	if dimension <= 0:
         raise SystemExit("EMBEDDING_DIM must be a positive integer.")

 	return dimension


 def main():
 	user = required_env("DB_USER")
 	if user.upper() in ADMIN_USERS:
         raise SystemExit(
             "Use the vector_demo application user, not an admin user."
     	)

 	# Validate the configured model name and vector dimension.
     model_name = required_env("EMBEDDING_MODEL")
     dimension = configured_dimension()

 	with oracledb.connect(
         user=user,
         password=required_env("DB_PASSWORD"),
         dsn=required_env("DB_DSN"),
 	) as connection:
         with connection.cursor() as cursor:
             # Recreate the documents table so the script is repeatable.
             cursor.execute(
                 """
                 BEGIN
                 	EXECUTE IMMEDIATE 'DROP TABLE documents PURGE';
                 EXCEPTION
                 	WHEN OTHERS THEN
                     	IF SQLCODE <> -942 THEN
                         	RAISE;
                     	END IF;
                 END;
                 """
             )
             cursor.execute(
                 f"""
                 CREATE TABLE documents (
                 	id NUMBER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
                 	title VARCHAR2(200) NOT NULL,
                 	content VARCHAR2(1000) NOT NULL,
                 	embedding VECTOR({dimension}, FLOAT32)
                 )
                 """
             )

     print(f"Embedding model: {model_name}")
     print(f"Embedding dimension: {dimension}")
     print("Recreated documents table.")


 if __name__ == "__main__":
 	main()技术支持

python 03_create_documents_table.py技术支持

Embedding model: text-embedding-3-small
 Embedding dimension: 1536
 Recreated documents table.技术支持

import os

 from openai import OpenAI


 def required_env(name):
 	value = os.getenv(name)
 	if not value:
         raise SystemExit(f"Set {name} before running this script.")
 	return value


 def configured_dimension():
 	try:
         return int(required_env("EMBEDDING_DIM"))
 	except ValueError as exc:
         raise SystemExit("EMBEDDING_DIM must be an integer.") from exc


 def main():
     required_env("OPENAI_API_KEY")

     model_name = required_env("EMBEDDING_MODEL")
     expected_dimension = configured_dimension()

 	# Request one embedding to confirm the model output dimension.
 	client = OpenAI()
     response = client.embeddings.create(
         model=model_name,
         input="Oracle AI Vector Search dimension check",
 	)

     returned_dimension = len(response.data[0].embedding)

     print(f"Embedding model: {model_name}")
     print(f"Returned embedding length: {returned_dimension}")
     print(f"Configured EMBEDDING_DIM: {expected_dimension}")

 	# Stop before loading data if the table dimension is incompatible.
 	if returned_dimension != expected_dimension:
         raise SystemExit(
             "Dimension check failed. Update EMBEDDING_DIM, rerun "
             "03_create_documents_table.py, and try again."
     	)

     print("Dimension check passed.")


 if __name__ == "__main__":
 	main()技术支持

python 04_check_embedding_dimension.py技术支持

Embedding model: text-embedding-3-small
 Returned embedding length: 1536
 Configured EMBEDDING_DIM: 1536
 Dimension check passed.技术支持

import array
 import os

 import oracledb
 from openai import OpenAI

 DOCUMENTS = [
 	(
         "Oracle AI Vector Search",
         "Oracle Database can store vector embeddings in native VECTOR "
         "columns and query them with SQL similarity search.",
 	),
 	(
         "Semantic Search for Applications",
         "Semantic search ranks results by meaning, helping applications "
         "find related content even when exact words differ.",
 	),
 	(
         "Vector Index Workflow",
         "Vector indexes help approximate nearest-neighbor search scale "
         "when datasets grow beyond small examples.",
 	),
 	(
         "Local Docker Development",
         "Docker containers make it practical to run a local database for "
         "development and repeatable tutorials.",
 	),
 	(
         "Python Database Code",
         "Python applications can connect to Oracle Database with the "
         "oracledb driver and use environment variables for credentials.",
 	),
 	(
         "Acoustic Guitar Practice",
         "Daily guitar practice improves timing, chord transitions, and "
         "confidence when learning new songs.",
 	),
 	(
         "Weeknight Vegetable Soup",
         "A simple soup can combine onions, carrots, beans, herbs, and "
         "broth for an easy weeknight meal.",
 	),
 ]

 ADMIN_USERS = {"SYS", "SYSTEM", "PDBADMIN"}


 def required_env(name):
 	value = os.getenv(name)
 	if not value:
         raise SystemExit(f"Set {name} before running this script.")
 	return value


 def configured_dimension():
 	try:
         return int(required_env("EMBEDDING_DIM"))
 	except ValueError as exc:
         raise SystemExit("EMBEDDING_DIM must be an integer.") from exc


 def main():
 	user = required_env("DB_USER")
 	if user.upper() in ADMIN_USERS:
         raise SystemExit(
             "Use the vector_demo application user, not an admin user."
     	)

     required_env("OPENAI_API_KEY")
     model_name = required_env("EMBEDDING_MODEL")
     expected_dimension = configured_dimension()

 	# Generate embeddings for all sample documents in one API request.
 	client = OpenAI()
     response = client.embeddings.create(
         model=model_name,
         input=[content for _, content in DOCUMENTS],
 	)

 	# Restore the response order using each item's index.
     embeddings = [
         item.embedding
     	for item in sorted(
             response.data,
             key=lambda item: item.index,
     	)
 	]

 	if any(len(embedding) != expected_dimension for embedding in embeddings):
         raise SystemExit(
             "An embedding length did not match EMBEDDING_DIM. "
             "Rerun 04_check_embedding_dimension.py."
     	)

 	# Bind each embedding as a 32-bit floating-point array.
 	rows = [
         (title, content, array.array("f", embedding))
     	for (title, content), embedding in zip(DOCUMENTS, embeddings)
 	]

 	# Replace the sample rows and commit the transaction.
 	with oracledb.connect(
         user=user,
         password=required_env("DB_PASSWORD"),
         dsn=required_env("DB_DSN"),
 	) as connection:
         with connection.cursor() as cursor:
             cursor.execute("DELETE FROM documents")
             cursor.executemany(
                 """
                 INSERT INTO documents (title, content, embedding)
                 VALUES (:1, :2, :3)
                 """,
                 rows,
             )
             cursor.execute("SELECT COUNT(*) FROM documents")
             row_count = cursor.fetchone()[0]
             connection.commit()

     print(f"Embedding model: {model_name}")
     print(f"Generated embeddings for {len(DOCUMENTS)} documents.")
     print(f"Inserted {len(DOCUMENTS)} documents.")
     print(f"documents row count: {row_count}")


 if __name__ == "__main__":
 	main()技术支持

python 05_insert_embeddings.py技术支持

Embedding model: text-embedding-3-small
 Generated embeddings for 7 documents.
 Inserted 7 documents.
 documents row count: 7技术支持

import array
 import os

 import oracledb
 from openai import OpenAI

 QUERY_TEXT = "Where should an AI app keep meaning-based search data?"
 ADMIN_USERS = {"SYS", "SYSTEM", "PDBADMIN"}


 def required_env(name):
 	value = os.getenv(name)
 	if not value:
         raise SystemExit(f"Set {name} before running this script.")
 	return value


 def main():
 	user = required_env("DB_USER")
 	if user.upper() in ADMIN_USERS:
         raise SystemExit(
             "Use the vector_demo application user, not an admin user."
     	)

     required_env("OPENAI_API_KEY")

 	# Generate an embedding for the natural-language query.
 	client = OpenAI()
     response = client.embeddings.create(
         model=required_env("EMBEDDING_MODEL"),
         input=QUERY_TEXT,
 	)
     query_vector = array.array("f", response.data[0].embedding)

 	with oracledb.connect(
         user=user,
         password=required_env("DB_PASSWORD"),
         dsn=required_env("DB_DSN"),
 	) as connection:
         with connection.cursor() as cursor:
             # Rank stored documents by cosine distance from the query vector.
             cursor.execute(
                 """
                 SELECT title,
                    	content,
                    	VECTOR_DISTANCE(
                        	embedding,
                        	:query_vector,
                        	COSINE
                    	) AS distance
                 FROM documents
                 ORDER BY distance
                 FETCH EXACT FIRST 5 ROWS ONLY
                 """,
                 query_vector=query_vector,
             )
             rows = cursor.fetchall()

 	if not rows:
         raise SystemExit("No documents found. Run 05_insert_embeddings.py first.")

     print(f"Query: {QUERY_TEXT}")
     print("\nTop 5 results:")

 	for rank, (title, content, distance) in enumerate(rows, start=1):
         print(f"{rank}. {title}")
         print(f"   distance: {float(distance):.6f}")
         print(f"   {content[:90]}...")


 if __name__ == "__main__":
 	main()技术支持

python 06_semantic_search.py技术支持

Query: Where should an AI app keep meaning-based search data?

 Top 5 results:
 1. Oracle AI Vector Search
    distance: <cosine distance from your run>
    Oracle Database can store vector embeddings in native VECTOR columns and query them with SQL...

 2. Semantic Search for Applications
    distance: <cosine distance from your run>
    Semantic search ranks results by meaning, helping applications find related content even when...

 3. Python Database Code
    distance: <cosine distance from your run>
    Python applications can connect to Oracle Database with the oracledb driver and use environment...

 4. Vector Index Workflow
    distance: <cosine distance from your run>
    Vector indexes help approximate nearest-neighbor search scale when datasets grow beyond small...

 5. Local Docker Development
    distance: <cosine distance from your run>
    Docker containers make it practical to run a local database for development and repeatable...技术支持

import array
 import os

 import oracledb
 from openai import OpenAI

 SEMANTIC_QUERY = "Where should an AI app keep meaning-based search data?"
 KEYWORD_PHRASE = "meaning-based search data"
 ADMIN_USERS = {"SYS", "SYSTEM", "PDBADMIN"}


 def required_env(name):
 	value = os.getenv(name)
 	if not value:
         raise SystemExit(f"Set {name} before running this script.")
 	return value


 def main():
 	user = required_env("DB_USER")
 	if user.upper() in ADMIN_USERS:
         raise SystemExit(
             "Use the vector_demo application user, not an admin user."
     	)

     required_env("OPENAI_API_KEY")

 	# Generate an embedding for the semantic query.
 	client = OpenAI()
     response = client.embeddings.create(
         model=required_env("EMBEDDING_MODEL"),
         input=SEMANTIC_QUERY,
 	)
     query_vector = array.array("f", response.data[0].embedding)

 	with oracledb.connect(
         user=user,
         password=required_env("DB_PASSWORD"),
         dsn=required_env("DB_DSN"),
 	) as connection:
         with connection.cursor() as cursor:
             # Rank documents by semantic similarity.
             cursor.execute(
                 """
                 SELECT title,
                        VECTOR_DISTANCE(embedding, :query_vector, COSINE)
                 FROM documents
                 ORDER BY VECTOR_DISTANCE(embedding, :query_vector, COSINE)
                 FETCH EXACT FIRST 3 ROWS ONLY
                 """,
                 query_vector=query_vector,
             )
             semantic_rows = cursor.fetchall()

             # Compare semantic results with an exact phrase predicate.
             cursor.execute(
                 """
                 SELECT title
                 FROM documents
                 WHERE LOWER(title || ' ' || content)
                   	LIKE '%' || :phrase || '%'
                 ORDER BY title
                 FETCH FIRST 3 ROWS ONLY
                 """,
                 phrase=KEYWORD_PHRASE.lower(),
             )
             keyword_rows = cursor.fetchall()

     print(f"Semantic query: {SEMANTIC_QUERY}")
     print(f"Keyword phrase: {KEYWORD_PHRASE}")

     print("\nSemantic results:")
 	for rank, (title, distance) in enumerate(semantic_rows, start=1):
         print(f"{rank}. {title}")
         print(f"   distance: {float(distance):.6f}")

     print("\nKeyword results:")
 	if keyword_rows:
     	for rank, (title,) in enumerate(keyword_rows, start=1):
             print(f"{rank}. {title}")
 	else:
         print("0 exact phrase matches")


 if __name__ == "__main__":
 	main()技术支持

python 07_semantic_vs_keyword.py技术支持

Semantic query: Where should an AI app keep meaning-based search data?
 Keyword phrase: meaning-based search data

 Semantic results:
 1. Oracle AI Vector Search
    distance: <cosine distance from your run>
 2. Semantic Search for Applications
    distance: <cosine distance from your run>
 3. Python Database Code
    distance: <cosine distance from your run>

 Keyword results:
 0 exact phrase matches技术支持

import array
 import os

 import oracledb
 from openai import OpenAI

 INDEX_NAME = "DOCUMENT_EMBEDDING_IDX"
 QUERY_TEXT = "Where should an AI app keep meaning-based search data?"
 ADMIN_USERS = {"SYS", "SYSTEM", "PDBADMIN"}


 def required_env(name):
 	value = os.getenv(name)
 	if not value:
         raise SystemExit(f"Set {name} before running this script.")
 	return value


 def main():
 	user = required_env("DB_USER")
 	if user.upper() in ADMIN_USERS:
         raise SystemExit(
             "Use the vector_demo application user, not an admin user."
     	)

     required_env("OPENAI_API_KEY")

 	# Generate the query vector used for the validation search.
 	client = OpenAI()
     response = client.embeddings.create(
         model=required_env("EMBEDDING_MODEL"),
         input=QUERY_TEXT,
 	)
     query_vector = array.array("f", response.data[0].embedding)

 	with oracledb.connect(
         user=user,
         password=required_env("DB_PASSWORD"),
         dsn=required_env("DB_DSN"),
 	) as connection:
         with connection.cursor() as cursor:
             # Remove a prior index so the script can be rerun safely.
             try:
                 cursor.execute(f"DROP INDEX {INDEX_NAME}")
             except oracledb.DatabaseError as exc:
                 error = exc.args[0]
                 if error.code != 1418:
                 	raise

             # Create an approximate neighbor-partitions vector index.
             cursor.execute(
                 """
                 CREATE VECTOR INDEX document_embedding_idx
                 ON documents (embedding)
                 ORGANIZATION NEIGHBOR PARTITIONS
                 DISTANCE COSINE
                 WITH TARGET ACCURACY 95
                 """
             )

             # Confirm that Oracle reports the vector index as valid.
             cursor.execute(
                 """
                 SELECT index_name, index_type, status
                 FROM user_indexes
                 WHERE index_name = :index_name
                 """,
                 index_name=INDEX_NAME,
             )
             index_row = cursor.fetchone()

             # Confirm that semantic search still returns rows after index creation.
             cursor.execute(
                 """
                 SELECT title
                 FROM documents
                 ORDER BY VECTOR_DISTANCE(embedding, :query_vector, COSINE)
                 FETCH EXACT FIRST 3 ROWS ONLY
                 """,
                 query_vector=query_vector,
             )
             rows = cursor.fetchall()

 	if not index_row:
         raise SystemExit("Vector index was not found in USER_INDEXES.")

     print("Created vector index DOCUMENT_EMBEDDING_IDX.")
     print("\nIndex validation:")
     print(f"Index name: {index_row[0]}")
     print(f"Index type: {index_row[1]}")
     print(f"Status: {index_row[2]}")
     result_count = len(rows)

 	print(
         "\nSemantic search still returns "
         f"{result_count} results after index creation."
 	)


 if __name__ == "__main__":
 	main()技术支持

python 08_create_vector_index.py技术支持

Created vector index DOCUMENT_EMBEDDING_IDX.

 Index validation:
 Index name: DOCUMENT_EMBEDDING_IDX
 Index type: VECTOR
 Status: VALID

 Semantic search still returns 3 results after index creation.技术支持

docker stop oracle-free-26ai-vector
docker rm oracle-free-26ai-vector技术支持