前言

學習了如何使用 Amazon Bedrock 生成 Embeddings，但光有向量還不夠，我們需要一個高效的方式來儲存和檢索這些向量資料
今天我們將深入探討兩個主流的向量資料庫解決方案：Amazon OpenSearch Service 和 Pinecone，並實作如何將它們整合到我們的 AI 應用中

為什麼我們要向量資料庫?

傳統的關聯是資料庫雖然擅長結構化處理，但處理高維度的向量資料時會面臨到幾個問題

• 效能問題：傳統資料庫在進行向量相似度搜尋時效率低下
• 儲存限制：向量資料通常是高維度的浮點數陣列，佔用大量儲存空間
• 缺乏專門的索引：沒有針對向量檢索優化的索引結構

因此有向量資料庫，姐個這方面的問題

向量資料庫的場景設計

• 高效的近似最近鄰搜尋 (ANN)
• 專門的向量索引演算法 (如 HNSW IVF)
• 水平擴展能力
• 與 AI/ML 工作流程的無縫整合

Amazon OpenSearch Service 簡介

是一個完全託管的搜尋和分析引擎服務，支援向量搜尋功能，它特別適合已經在使用 AWS 生態系統的團隊

OpenSearch 的優勢

• 完全託管：自動處理硬體配置軟體安裝補丁和備份
• 高可用性：支援多可用區部署
• 混合搜尋：可以結合關鍵字搜尋和向量搜尋
• 豐富的視覺化工具：內建 OpenSearch Dashboards

以 python 來說

安裝 openSearch 相關

uv add opensearch-py boto3

or

pip install opensearch-py boto3

接下來用 OpenSearch 建立::

from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
import boto3
import json

class OpenSearchVectorStore:
    def __init__(self, host, region='us-east-1'):
        """
        初始化 OpenSearch 客戶端
        """
        credentials = boto3.Session().get_credentials()
        awsauth = AWS4Auth(
            credentials.access_key,
            credentials.secret_key,
            region,
            'es',
            session_token=credentials.token
        )
        
        self.client = OpenSearch(
            hosts=[{'host': host, 'port': 443}],
            http_auth=awsauth,
            use_ssl=True,
            verify_certs=True,
            connection_class=RequestsHttpConnection
        )
        
    def create_index(self, index_name, dimension=1536):
        """
        創建向量索引
        dimension: 向量維度，預設為 Titan Embeddings 的 1536
        """
        index_body = {
            "settings": {
                "index": {
                    "knn": True,  # 啟用 k-NN 搜尋
                    "knn.algo_param.ef_search": 512
                }
            },
            "mappings": {
                "properties": {
                    "vector": {
                        "type": "knn_vector",
                        "dimension": dimension,
                        "method": {
                            "name": "hnsw",  # 使用 HNSW 演算法
                            "space_type": "cosinesimil",  # 餘弦相似度
                            "engine": "nmslib",
                            "parameters": {
                                "ef_construction": 512,
                                "m": 16
                            }
                        }
                    },
                    "text": {
                        "type": "text"
                    },
                    "metadata": {
                        "type": "object"
                    }
                }
            }
        }
        
        try:
            response = self.client.indices.create(index_name, body=index_body)
            print(f"索引 {index_name} 創建成功")
            return response
        except Exception as e:
            print(f"創建索引時發生錯誤: {e}")
            
    def add_documents(self, index_name, documents):
        """
        批次新增文件
        documents: [{"text": "...", "vector": [...], "metadata": {...}}]
        """
        for i, doc in enumerate(documents):
            try:
                response = self.client.index(
                    index=index_name,
                    body={
                        "text": doc["text"],
                        "vector": doc["vector"],
                        "metadata": doc.get("metadata", {})
                    },
                    id=doc.get("id", None),
                    refresh=True
                )
                print(f"文件 {i+1} 新增成功")
            except Exception as e:
                print(f"新增文件 {i+1} 時發生錯誤: {e}")
                
    def search(self, index_name, query_vector, k=5, filter_query=None):
        """
        向量相似度搜尋
        query_vector: 查詢向量
        k: 返回前 k 個最相似的結果
        filter_query: 可選的過濾條件
        """
        search_body = {
            "size": k,
            "query": {
                "knn": {
                    "vector": {
                        "vector": query_vector,
                        "k": k
                    }
                }
            }
        }
        
        # 如果有過濾條件，使用 bool query
        if filter_query:
            search_body["query"] = {
                "bool": {
                    "must": [
                        {"knn": {"vector": {"vector": query_vector, "k": k}}}
                    ],
                    "filter": filter_query
                }
            }
        
        try:
            response = self.client.search(
                index=index_name,
                body=search_body
            )
            
            results = []
            for hit in response['hits']['hits']:
                results.append({
                    "score": hit['_score'],
                    "text": hit['_source']['text'],
                    "metadata": hit['_source'].get('metadata', {})
                })
            
            return results
        except Exception as e:
            print(f"搜尋時發生錯誤: {e}")
            return []

整合 bedrock 與 openSearch

import boto3
import json

# 初始化 Bedrock 客戶端
bedrock_runtime = boto3.client('bedrock-runtime', region_name='us-east-1')

def generate_embedding(text):
    """使用 Bedrock 生成向量"""
    body = json.dumps({
        "inputText": text
    })
    
    response = bedrock_runtime.invoke_model(
        modelId='amazon.titan-embed-text-v1',
        body=body,
        contentType='application/json',
        accept='application/json'
    )
    
    response_body = json.loads(response['body'].read())
    return response_body['embedding']

# 使用範例
if __name__ == "__main__":
    # 初始化 OpenSearch
    vector_store = OpenSearchVectorStore(
        host='your-opensearch-domain.us-east-1.es.amazonaws.com'
    )
    
    # 創建索引
    vector_store.create_index('knowledge_base')
    
    # 準備文件資料
    documents = [
        {
            "text": "Amazon Bedrock 是一個完全託管的服務，提供多種基礎模型。",
            "metadata": {"category": "AWS", "topic": "Bedrock"}
        },
        {
            "text": "SageMaker 是 AWS 的機器學習平台，支援模型訓練和部署。",
            "metadata": {"category": "AWS", "topic": "SageMaker"}
        },
        {
            "text": "向量資料庫用於儲存和檢索高維度向量資料。",
            "metadata": {"category": "Database", "topic": "Vector"}
        }
    ]
    
    # 生成向量並新增到資料庫
    for doc in documents:
        doc["vector"] = generate_embedding(doc["text"])
    
    vector_store.add_documents('knowledge_base', documents)
    
    # 執行搜尋
    query = "什麼是 AWS 的機器學習服務？"
    query_vector = generate_embedding(query)
    
    results = vector_store.search('knowledge_base', query_vector, k=2)
    
    print("\n搜尋結果：")
    for i, result in enumerate(results, 1):
        print(f"\n{i}. 相似度分數: {result['score']:.4f}")
        print(f"   內容: {result['text']}")
        print(f"   元數據: {result['metadata']}")

Pinecone 簡介

是一個專門的向量資料庫服務，以其簡單易用和優秀的效能著稱

優勢

• 專為向量設計：完全聚焦於向量搜尋的效能優化
• 使用簡單：API 設計直觀，上手快速
• 自動擴展：根據需求自動調整資源
• 即時更新：支援向量的即時插入和更新

這裡安裝 pinecone

uv add pinecone-client

or

pip install pinecone-client

實作 pinecone 存儲

from pinecone import Pinecone, ServerlessSpec
import time

class PineconeVectorStore:
    def __init__(self, api_key, environment='us-east-1'):
        """初始化 Pinecone 客戶端"""
        self.pc = Pinecone(api_key=api_key)
        self.environment = environment
        
    def create_index(self, index_name, dimension=1536, metric='cosine'):
        """
        創建 Pinecone 索引
        metric: 相似度計算方式 ('cosine', 'euclidean', 'dotproduct')
        """
        # 檢查索引是否已存在
        if index_name not in self.pc.list_indexes().names():
            self.pc.create_index(
                name=index_name,
                dimension=dimension,
                metric=metric,
                spec=ServerlessSpec(
                    cloud='aws',
                    region=self.environment
                )
            )
            
            # 等待索引準備完成
            while not self.pc.describe_index(index_name).status['ready']:
                time.sleep(1)
                
            print(f"索引 {index_name} 創建成功")
        else:
            print(f"索引 {index_name} 已存在")
            
        return self.pc.Index(index_name)
    
    def add_documents(self, index_name, documents, namespace=''):
        """
        批次新增文件到 Pinecone
        documents: [{"id": "...", "values": [...], "metadata": {...}}]
        namespace: 命名空間，用於組織資料
        """
        index = self.pc.Index(index_name)
        
        # Pinecone 建議每次最多 upsert 100 個向量
        batch_size = 100
        for i in range(0, len(documents), batch_size):
            batch = documents[i:i + batch_size]
            
            vectors = [
                (
                    doc['id'],
                    doc['values'],
                    doc.get('metadata', {})
                )
                for doc in batch
            ]
            
            index.upsert(vectors=vectors, namespace=namespace)
            print(f"已新增 {min(i + batch_size, len(documents))}/{len(documents)} 個向量")
            
    def search(self, index_name, query_vector, top_k=5, filter_dict=None, namespace=''):
        """
        向量相似度搜尋
        filter_dict: 元數據過濾條件，例如 {"category": "AWS"}
        """
        index = self.pc.Index(index_name)
        
        results = index.query(
            vector=query_vector,
            top_k=top_k,
            filter=filter_dict,
            namespace=namespace,
            include_metadata=True
        )
        
        return [
            {
                "id": match['id'],
                "score": match['score'],
                "metadata": match.get('metadata', {})
            }
            for match in results['matches']
        ]
    
    def delete_index(self, index_name):
        """刪除索引"""
        self.pc.delete_index(index_name)
        print(f"索引 {index_name} 已刪除")

bedrock 與 pinecone 整合範例

import boto3
import json
from datetime import datetime

# 初始化服務
bedrock_runtime = boto3.client('bedrock-runtime', region_name='us-east-1')
pinecone_store = PineconeVectorStore(
    api_key='your-pinecone-api-key',
    environment='us-east-1'
)

def generate_embedding(text):
    """使用 Bedrock 生成向量"""
    body = json.dumps({"inputText": text})
    
    response = bedrock_runtime.invoke_model(
        modelId='amazon.titan-embed-text-v1',
        body=body,
        contentType='application/json',
        accept='application/json'
    )
    
    response_body = json.loads(response['body'].read())
    return response_body['embedding']

# 使用範例
if __name__ == "__main__":
    # 創建索引
    index = pinecone_store.create_index('aws-knowledge-base', dimension=1536)
    
    # 準備文件資料
    raw_documents = [
        {
            "text": "Amazon Bedrock 提供 Claude、Llama 等多種基礎模型。",
            "metadata": {"source": "AWS Docs", "category": "Bedrock", "date": "2024-01-15"}
        },
        {
            "text": "Lambda 是 AWS 的 serverless 運算服務，支援事件驅動架構。",
            "metadata": {"source": "AWS Docs", "category": "Lambda", "date": "2024-01-20"}
        },
        {
            "text": "S3 是物件儲存服務，提供 99.999999999% 的耐久性。",
            "metadata": {"source": "AWS Docs", "category": "S3", "date": "2024-02-01"}
        }
    ]
    
    # 生成向量並準備 Pinecone 格式
    pinecone_documents = []
    for i, doc in enumerate(raw_documents):
        embedding = generate_embedding(doc["text"])
        pinecone_documents.append({
            "id": f"doc-{i}",
            "values": embedding,
            "metadata": {
                **doc["metadata"],
                "text": doc["text"]
            }
        })
    
    # 新增到 Pinecone
    pinecone_store.add_documents('aws-knowledge-base', pinecone_documents)
    
    # 執行搜尋
    query = "AWS 的 AI 服務有哪些？"
    query_vector = generate_embedding(query)
    
    # 可以加上過濾條件
    results = pinecone_store.search(
        'aws-knowledge-base',
        query_vector,
        top_k=3,
        filter_dict={"category": "Bedrock"}  # 只搜尋 Bedrock 相關內容
    )
    
    print("\n搜尋結果：")
    for i, result in enumerate(results, 1):
        print(f"\n{i}. ID: {result['id']}")
        print(f"   相似度分數: {result['score']:.4f}")
        print(f"   內容: {result['metadata']['text']}")
        print(f"   分類: {result['metadata']['category']}")

OpenSearch vs Pinecone 比較

參考資料

AWS opensearch doc
pinecone doc
amazon bedrock