使用 LangChain 将 S3 数据发送到 Qdrant 向量存储

boto3
langchain-community
langchain
python-dotenv
unstructured
unstructured[pdf]
qdrant_client
fastembed

ACCESS_KEY = ""
SECRET_ACCESS_KEY = ""
QDRANT_KEY = ""

from langchain_community.document_loaders import S3DirectoryLoader

# Initialize the S3 document loader
loader = S3DirectoryLoader(
   "product-dataset",  # S3 bucket name
   "p_1", #S3 Folder name containing the data for the first product
   aws_access_key_id=aws_access_key_id,  # AWS Access Key
   aws_secret_access_key=aws_secret_access_key  # AWS Secret Access Key
)

# Load documents from the specified S3 bucket
docs = loader.load()

from PIL import Image

def process_text(doc):
    source = doc.metadata['source']  # Extract document source (e.g., S3 URL)

    text = doc.page_content  # Extract the content from the text file
    print(f"Processing text from {source}")
    return source, text

def process_image(doc):
    source = doc.metadata['source']  # Extract document source (e.g., S3 URL)
    print(f"Processing image from {source}")

    bucket_name, object_key = parse_s3_url(source)  # Parse the S3 URL
    response = s3.get_object(Bucket=bucket_name, Key=object_key)  # Fetch image from S3
    img_bytes = response['Body'].read()

    img = Image.open(io.BytesIO(img_bytes))
    return source, img

def parse_s3_url(s3_url):
    parts = s3_url.replace("s3://", "").split("/", 1)
    bucket_name = parts[0]
    object_key = parts[1]
    return bucket_name, object_key

def create_collection(collection_name):
    qdrant_client.create_collection(
        collection_name,
        vectors_config={
            "text_embedding": models.VectorParams(
                size=384,  # Dimension of text embeddings
                distance=models.Distance.COSINE,  # Cosine similarity is used for comparison
            ),
            "image_embedding": models.VectorParams(
                size=512,  # Dimension of image embeddings
                distance=models.Distance.COSINE,  # Cosine similarity is used for comparison
            ),
        },
    )

create_collection("products-data")

def ingest_data(points):
    operation_info = qdrant_client.upsert(
        collection_name="products-data",  # Collection where data is being inserted
        points=points
    )
    return operation_info

from qdrant_client import models

if __name__ == "__main__":
    collection_name = "products-data"
    create_collection(collection_name)
    for i in range(1,6): # Five documents
        folder = f"p_{i}"
        loader = S3DirectoryLoader(
            "product-dataset",
            folder,
            aws_access_key_id=aws_access_key_id,
            aws_secret_access_key=aws_secret_access_key
        )
        docs = loader.load()
        points, text_review, product_image = [], "", ""
        for idx, doc in enumerate(docs):
            source = doc.metadata['source']
            if source.endswith(".txt") or source.endswith(".pdf"):
                _text_review_source, text_review = process_text(doc)
            elif source.endswith(".png"):
                product_image_source, product_image = process_image(doc)
        if text_review:
            point = models.PointStruct(
                id=idx,  # Unique identifier for each point
                vector={
                    "text_embedding": models.Document(
                        text=text_review, model="sentence-transformers/all-MiniLM-L6-v2"
                    ),
                    "image_embedding": models.Image(
                        image=product_image, model="Qdrant/clip-ViT-B-32-vision"
                    ),
                },
                payload={"review": text_review, "product_image": product_image_source},
            )
            points.append(point)
    operation_info = ingest_data(points)
    print(operation_info)