vectordb

ベクトルデータベーススキル

Golden Armada AI Agent Fleet Platform 向けに、包括的なベクトルデータベース機能を提供します。

このスキルを使用するタイミング

以下の作業を行う際に、このスキルを有効にします。

• セマンティック検索の実装
• RAG（検索拡張生成）
• 埋め込みの保存と取得
• 類似性検索
• ベクトルインデックスの管理

埋め込みの生成

```python import openai from anthropic import Anthropic

OpenAI の埋め込み

def get_openai_embedding(text: str) -> list[float]: response = openai.embeddings.create( model="text-embedding-3-small", input=text ) return response.data[0].embedding

バッチ埋め込み

def get_batch_embeddings(texts: list[str]) -> list[list[float]]: response = openai.embeddings.create( model="text-embedding-3-small", input=texts ) return [item.embedding for item in response.data] ```

Pinecone

```python from pinecone import Pinecone, ServerlessSpec

初期化

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

インデックスを作成

pc.create_index( name="agents", dimension=1536, metric="cosine", spec=ServerlessSpec( cloud="aws", region="us-west-2" ) )

インデックスを取得

index = pc.Index("agents")

ベクトルをアップサート

index.upsert( vectors=[ { "id": "agent-1", "values": embedding, "metadata": { "name": "Claude Agent", "type": "claude", "description": "汎用アシスタント" } } ], namespace="production" )

クエリ

results = index.query( vector=query_embedding, top_k=10, include_metadata=True, namespace="production", filter={"type": {"$eq": "claude"}} )

for match in results.matches: print(f"{match.id}: {match.score} - {match.metadata}")

削除

index.delete(ids=["agent-1"], namespace="production") ```

Chroma

```python import chromadb from chromadb.config import Settings

初期化

client = chromadb.PersistentClient(path="./chroma_db")

コレクションを作成

collection = client.get_or_create_collection( name="agents", metadata={"hnsw:space": "cosine"} )

ドキュメントを追加

collection.add( ids=["agent-1", "agent-2"], embeddings=[embedding1, embedding2], documents=["ドキュメント 1 のテキスト", "ドキュメント 2 のテキスト"], metadatas=[ {"type": "claude", "version": "3"}, {"type": "gpt", "version": "4"} ] )

クエリ

results = collection.query( query_embeddings=[query_embedding], n_results=10, where={"type": "claude"}, include=["documents", "metadatas", "distances"] )

更新

collection.update( ids=["agent-1"], metadatas=[{"type": "claude", "version": "3.5"}] )

削除

collection.delete(ids=["agent-1"]) ```

pgvector（PostgreSQL）

```sql -- 拡張機能を有効化 CREATE EXTENSION vector;

-- テーブルを作成 CREATE TABLE documents ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), content TEXT NOT NULL, embedding vector(1536), metadata JSONB DEFAULT '{}' );

-- インデックスを作成（大規模データセットには IVFFlat） CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

-- または、再現率を高めるには HNSW を使用 CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);

-- 挿入 INSERT INTO documents (content, embedding, metadata) VALUES ('ドキュメントの内容', '[0.1, 0.2, ...]', '{"type": "manual"}');

-- 類似性検索 SELECT id, content, 1 - (embedding <=> $1) AS similarity FROM documents ORDER BY embedding <=> $1 LIMIT 10;

-- フィルター付き SELECT id, content, 1 - (embedding <=> $1) AS similarity FROM documents WHERE metadata->>'type' = 'manual' ORDER BY embedding <=> $1 LIMIT 10; ```

pgvector を使用した Python

from pgvector.sqlalchemy import Vector
from sqlalchemy import Column, String, JSON
from sqlalchemy.dialects.postgresql import UUID

class Document(Base):
    __tablename__ = 'documents'

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    content = Column(String, nullable=False)
    embedding = Column(Vector(1536))
    metadata = Column(JSON, default={})

# 挿入
doc = Document(
    content="Document content",
    embedding=embedding,
    metadata={"type": "manual"}
)
session.add(doc)
session.commit()

# クエリ
from sqlalchemy import select
from pgvector.sqlalchemy import cosine_distance

results = session.execute(
    select(Document)
    .order_by(cosine_distance(Document.embedding, query_embedding))
    .limit(10)
).scalars().all()

RAG の実装

class RAGService:
    def __init__(self, vector_store, llm_client):
        self.vector_store = vector_store
        self.llm = llm_client

    async def query(self, question: str, top_k: int = 5) -> str:
        # 1. クエリの埋め込みを生成
        query_embedding = await self.get_embedding(question)

        # 2. 関連ドキュメントを取得
        docs = await self.vector_store.search(
            embedding=query_embedding,
            top_k=top_k
        )

        # 3. コンテキストを構築
        context = "\n\n".join([
            f"Document {i+1}:\n{doc.content}"
            for i, doc in enumerate(docs)
        ])

        # 4. 回答を生成
        prompt = f"""次のコンテキストに基づいて質問に回答してください。

Context:
{context}

Question: {question}

Answer:"""

        response = await self.llm.generate(prompt)
        return response

    async def add_document(self, content: str, metadata: dict = None):
        # ドキュメントをチャンクに分割
        chunks = self.chunk_text(content)

        # 埋め込みを生成
        embeddings = await self.get_batch_embeddings(chunks)

        # 保存
        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
            await self.vector_store.upsert(
                id=f"{metadata.get('doc_id', 'doc')}-{i}",
                embedding=embedding,
                content=chunk,
                metadata=metadata
            )

    def chunk_text(self, text: str, chunk_size: int = 1000, overlap: int = 200) -> list[str]:
        chunks = []
        start = 0
        while start < len(text):
            end = start + chunk_size
            chunk = text[start:end]
            chunks.append(chunk)
            start += chunk_size - overlap
        return chunks

ベストプラクティス

1. 適切なインデックスの種類を選択する（再現率を重視する場合は HNSW、スケーラビリティを重視する場合は IVFFlat）
2. ドキュメントを適切にチャンク分割する（通常は 500～1000 トークン）
3. チャンク間にオーバーラップを含める（10～20％）
4. フィルタリング用にメタデータを保存する
5. データを整理するために名前空間／コレクションを使用する
6. クエリのレイテンシとインデックスのパフォーマンスを監視する
7. 一括挿入にはバッチ処理を使用する