Qdrantベクターデータベースの構築とホスト¶

!pip install "langchain-community==0.4.1" \
             "langchain_text_splitters==1.0.0" \
             "qdrant-client==1.16.1" \
             "sentence-transformers==5.1.2" \
             "datarobot" \
             "datarobot-predict"

import datarobot as dr
from datarobot.models.genai.vector_database import CustomModelVectorDatabaseValidation
from datarobot.models.genai.vector_database import VectorDatabase
import os
from pathlib import Path
from qdrant_client import QdrantClient, models
from sentence_transformers import SentenceTransformer
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
import json
import requests
import zipfile
import io
import re

QDRANT_DATA_PATH = "qdrant"
EMBEDDING_MODEL_NAME = "all-MiniLM-L6-v2"
COLLECTION_NAME = "my_documents"
SOURCE_DOCUMENTS_ZIP_URL = "https://s3.amazonaws.com/datarobot_public_datasets/ai_accelerators/datarobot_english_documentation_5th_December.zip"
UNZIPPED_DOCS_DIR = "datarobot_english_documentation"

doc_path = Path(UNZIPPED_DOCS_DIR)
if doc_path.exists():
    txt_files = list(doc_path.rglob("*.txt"))
    print(f"✓ Found {len(txt_files)} .txt files in {UNZIPPED_DOCS_DIR}/")
    if txt_files:
        print(f"  Sample files:")
        for f in txt_files[:5]:
            print(f"    - {f}")
else:
    # Download some example docs
    r = requests.get(SOURCE_DOCUMENTS_ZIP_URL)
    z = zipfile.ZipFile(io.BytesIO(r.content))
    z.extractall()

# Create vectorDB
def create_database():
    """Create and populate the Qdrant database"""
    
    # Initialize encoder
    encoder = SentenceTransformer(EMBEDDING_MODEL_NAME)
    
    # Initialize Qdrant client
    client = QdrantClient(path=QDRANT_DATA_PATH)
    
    try:
        # Create collection
        print("Creating collection...")
        client.create_collection(
            collection_name=COLLECTION_NAME,
            vectors_config=models.VectorParams(
                size=encoder.get_sentence_embedding_dimension(),
                distance=models.Distance.COSINE,
            ),
        )

        # Load text files
        print(f"Loading documents from {UNZIPPED_DOCS_DIR}/...")
        docs = []
        doc_path = Path(UNZIPPED_DOCS_DIR)
        
        for file_path in doc_path.rglob("*.txt"):
            loader = TextLoader(str(file_path))
            loaded_docs = loader.load()
            docs.extend(loaded_docs)

        print(f"Loaded {len(docs)} documents")

        # Split documents into chunks
        splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)
        split_docs = splitter.split_documents(docs)
        print(f"Split into {len(split_docs)} chunks")
        
        # Process metadata
        for doc in split_docs:
            # Convert file path to documentation URL
            doc.metadata['source'] = re.sub(
                r'datarobot_english_documentation/datarobot_docs/en/(.+)\.(txt|md)',
                r'https://docs.datarobot.com/en/docs/\1.html',
                doc.metadata.get('source', '')
            )
            # Extract category from source
            doc.metadata["category"] = doc.metadata["source"].split("/")[-1].split(".")[0]

        # Batch encode all documents
        print("Encoding all documents (this may take a few minutes)...")
        all_contents = [doc.page_content for doc in split_docs]
        all_vectors = encoder.encode(
            all_contents, 
            show_progress_bar=False, 
            batch_size=32,
            convert_to_numpy=True
        )
        
        # Create points
        print("Creating point structures...")
        points_to_upload = [
            models.PointStruct(
                id=idx,
                vector=all_vectors[idx].tolist(),
                payload={
                    "content": doc.page_content,
                    "source": doc.metadata.get("source", ""),
                    "category": doc.metadata.get("category", ""),
                    **doc.metadata
                }
            )
            for idx, doc in enumerate(split_docs)
        ]

        # Upload to Qdrant
        print(f"Uploading {len(points_to_upload)} points to Qdrant...")
        client.upload_points(
            collection_name=COLLECTION_NAME,
            points=points_to_upload
        )

        # Verify
        collection_info = client.get_collection(collection_name=COLLECTION_NAME)
        print(f"✓ Collection '{COLLECTION_NAME}' created with {collection_info.points_count} points")
        print(f"✓ Data saved to: {QDRANT_DATA_PATH}/")

    finally:
        # ALWAYS close the client
        client.close()
        print("✓ Database client closed")

# Run the creation (only once)
create_database()

# Test a query to the vectorDB
def test_database():
    """Test the database with a sample query"""
    
    encoder = SentenceTransformer(EMBEDDING_MODEL_NAME)
    client = QdrantClient(path=QDRANT_DATA_PATH)
    
    try:
        question = "What is MLOps?"
        query_vector = encoder.encode(question).tolist()

        results = client.query_points(
            collection_name=COLLECTION_NAME,
            query=query_vector,
            limit=5,
        ).points

        print(f"Found {len(results)} results for: '{question}'\n")
        for hit in results:
            print(f"Score: {hit.score:.4f}")
            print(f"Content: {hit.payload.get('content', '')[:200]}...")
            print(f"Source: {hit.payload.get('source', '')}")
            print(f"Category: {hit.payload.get('category', '')}\n")
            
    finally:
        # ALWAYS close the client
        client.close()
        print("✓ Test client closed")

# Run the test
test_database()

def load_model(input_dir):
    """Custom model hook for loading our Qdrant knowledge base."""
    
    import os
    from qdrant_client import QdrantClient
    from sentence_transformers import SentenceTransformer
    print("Loading model")
    
    EMBEDDING_MODEL_NAME = 'all-MiniLM-L6-v2'
    COLLECTION_NAME = "my_documents"
    
    # When deploying model="qdrant/", the files are at input_dir/qdrant/
    # not directly at input_dir
    if input_dir:
        QDRANT_DATA_PATH = os.path.join(input_dir, "qdrant")
    else:
        QDRANT_DATA_PATH = "qdrant"
    
    print(f'QDRANT_DATA_PATH = {QDRANT_DATA_PATH}')
    print(f'EMBEDDING_MODEL_NAME = {EMBEDDING_MODEL_NAME}')
    print(f'COLLECTION_NAME = {COLLECTION_NAME}')
    
    # Initialize the embedding model (downloads if needed)
    encoder = SentenceTransformer(EMBEDDING_MODEL_NAME)
    
    # Initialize Qdrant client
    client = QdrantClient(path=QDRANT_DATA_PATH)
    
    # Get collection info to verify it loaded
    collection_info = client.get_collection(collection_name=COLLECTION_NAME)
    print(f'Loaded Qdrant collection "{COLLECTION_NAME}" with {collection_info.points_count} points')
    
    return {
        "client": client,
        "encoder": encoder,
        "collection_name": COLLECTION_NAME
    }


def score_unstructured(model, data, **kwargs) -> str:
    """Custom model hook for retrieving relevant docs with our Qdrant knowledge base.

    When requesting predictions from the deployment, pass a dictionary
    with the following keys:
    - 'question' the question to be passed to the vector store retriever
    - 'filter' any metadata filter
    - 'k' the number of results to return (default: 10)

    datarobot-user-models (DRUM) handles loading the model and calling
    this function with the appropriate parameters.

    Returns:
    --------
    rv : str
        Json dictionary with keys:
            - 'question' user's original question
            - 'relevant' the retrieved document contents
            - 'metadata' - metadata for each document including similarity scores
            - 'error' - error message if exception in handling request
    """
    import json
    from qdrant_client import models
    try:

        # Loading data
        data_dict = json.loads(data)
        question = data_dict['question']
        metadata_filter = data_dict.get("filter", None)
        top_k = data_dict.get("k", 10)

        # Defining info
        client = model["client"]
        encoder = model["encoder"]
        collection_name = model["collection_name"]
        
        # Encode the question
        query_vector = encoder.encode(question).tolist()
        
        # Build query parameters
        query_params = {
            "collection_name": collection_name,
            "query": query_vector,
            "limit": top_k,
        }
        
        # Only add filter if it exists
        if metadata_filter is not None:
            query_params["query_filter"] = models.Filter(**metadata_filter)
        
        # Perform the search
        results = client.query_points(**query_params).points
        
        print(f'Returned {len(results)} results')
        
        relevant, metadata = [], []
        for hit in results:
            # Extract the content from payload
            content = hit.payload.get('content', '')
            relevant.append(content)
            
            # Add similarity score to metadata
            hit_metadata = dict(hit.payload)
            hit_metadata["similarity_score"] = hit.score
            metadata.append(hit_metadata)
        
        rv = {
            "question": question,
            "relevant": relevant,
            "metadata": metadata,
        }
    except Exception as e:
        rv = {'error': f"{e.__class__.__name__}: {str(e)}"}
    return json.dumps(rv), {"mimetype": "application/json", "charset": "utf8"}

# Testing to ensure they work locally before deploying
def test_hooks():
    """Test the DataRobot hooks locally"""
    
    # Load the model - pass None so it uses "qdrant" as the path
    model = load_model(None)
    
    try:
        # Test scoring
        print("=" * 80)
        print("TEST: Basic search")
        print("=" * 80)
        result = score_unstructured(
            model,
            json.dumps({
                "question": "What is MLOps?",
                "filter": models.Filter(
                    must=[
                        models.FieldCondition(
                            key="category",
                            match=models.MatchValue(value="datarobot_docs|en|more-info|eli5")
                        )
                    ]
                ).model_dump(), # converting to be json serializable (will be convert back before scoring)
                "k": 3,
            })
        )
        response = json.loads(result[0])
        print(f"Question: {response['question']}")
        print(f"Found {len(response['relevant'])} results:\n")
        
        for idx, (content, meta) in enumerate(zip(response['relevant'], response['metadata'])):
            print(f"{idx+1}. Score: {meta['similarity_score']:.4f}")
            print(f"   Source: {meta.get('source', 'N/A')}")
            print(f"   Category: {meta.get('category', 'N/A')}")
            print(f"   Content: {content[:150]}...")
            print()
            
    finally:
        # ALWAYS close the client after testing
        model["client"].close()
        print("✓ Test client closed")

# Run the test
test_hooks()

import inspect
from pathlib import Path

CUSTOM_MODEL_NAME = "Qdrant Vector Database"
DEPLOY_DIR = Path("deploy_model")
DEPLOY_DIR.mkdir(parents=True, exist_ok=True)

(DEPLOY_DIR / "custom.py").write_text(
    inspect.getsource(load_model) + "\n\n" + inspect.getsource(score_unstructured)
)
(DEPLOY_DIR / "requirements.txt").write_text(
    "qdrant-client\n"
    "sentence-transformers\n"
)

model_files = [
    (str(DEPLOY_DIR / "custom.py"), "custom.py"),
    (str(DEPLOY_DIR / "requirements.txt"), "requirements.txt"),
]
source_root = Path("qdrant")
for path in source_root.rglob("*"):
    if path.is_file():
        model_files.append(
            (str(path), f"qdrant/{path.relative_to(source_root).as_posix()}")
        )

genai_environment = dr.ExecutionEnvironment.list(
    search_for="[GenAI] Python 3.12 with Moderations"
)[0]

existing_models = [m for m in dr.CustomInferenceModel.list() if m.name == CUSTOM_MODEL_NAME]
if existing_models:
    custom_model = existing_models[0]
else:
    custom_model = dr.CustomInferenceModel.create(
        name=CUSTOM_MODEL_NAME,
        target_type=dr.TARGET_TYPE.UNSTRUCTURED,
        language="python",
    )

print("Uploading custom model version")
model_version = dr.CustomModelVersion.create_clean(
    custom_model_id=custom_model.id,
    base_environment_id=genai_environment.id,
    files=model_files,
    maximum_memory=8 * 1024 * 1024 * 1024,
    max_wait=60 * 30,
)

print("Building custom model dependency image")
try:
    build_info = dr.CustomModelVersionDependencyBuild.start_build(
        custom_model_id=custom_model.id,
        custom_model_version_id=model_version.id,
        max_wait=60 * 60,
    )
except dr.errors.ClientError as e:
    if "already has a dependency image" in str(e):
        print("Dependency build already exists, skipping build step")
        build_info = dr.CustomModelVersionDependencyBuild.get_build_info(
            custom_model.id,
            model_version.id,
        )
    else:
        raise

if build_info is not None and build_info.build_status != "success":
    raise RuntimeError(
        f"Dependency build failed with status '{build_info.build_status}'.\n"
        f"{build_info.get_log()}"
    )

print("Creating deployment")
pred_server = dr.PredictionServer.list()[0]
deployment = dr.Deployment.create_from_custom_model_version(
    model_version.id,
    label=CUSTOM_MODEL_NAME,
    default_prediction_server_id=pred_server.id,
    max_wait=60 * 30,
)

print(f"Deployment ID: {deployment.id}")

from datarobot_predict.deployment import predict_unstructured

# Now with metadata filtering
data = {
    "question": "How do I replace a custom model on an existing custom environment?",
    "filter": models.Filter(
        must=[
            models.FieldCondition(
                key="category",
                match=models.MatchValue(value="datarobot_docs|en|modeling|special-workflows|cml|cml-custom-env")
            )
        ]
    ).model_dump(),
    "k": 5,
}

# Prediction request
content, response_headers = predict_unstructured(
    deployment=deployment,
    data=data,
)

# Check output
content

use_case_id = os.environ['DATAROBOT_DEFAULT_USE_CASE']
use_case = dr.UseCase.get(use_case_id)

# UNCOMMENT if you want to create a new Use Case
# use_case = dr.UseCase.create()

external_vdb_validation = CustomModelVectorDatabaseValidation.create(
    prompt_column_name="question", 
    target_column_name="relevant",
    deployment_id=deployment.id,
    use_case=use_case,
    wait_for_completion=True
)
external_vdb_validation

assert external_vdb_validation.validation_status == "PASSED"

vdb = VectorDatabase.create_from_custom_model(
    name="Qdrant Vector Database",
    use_case=use_case,
    validation_id=external_vdb_validation.id
)
vdb

assert vdb.execution_status == "COMPLETED"

print(f"Vector Database ID: {vdb.id}")