文本分块策略

def fixed_size_chunk(text, chunk_size=200, overlap=50):
    words = text.split()
    chunks = []
    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i:i + chunk_size])
        chunks.append(chunk)
    return chunks

# ! pip install nltk
from nltk.tokenize import sent_tokenize


def sentence_chunk(text, max_words=150):
    sentences = sent_tokenize(text)
    chunks, buffer, length = [], [], 0

    for sent in sentences:
        count = len(sent.split())
        if length + count > max_words:
            chunks.append(" ".join(buffer))
            buffer, length = [], 0
        buffer.append(sent)
        length += count

    if buffer:
        chunks.append(" ".join(buffer))
    return chunks

def paragraph_chunk(text):
    return [p.strip() for p in text.split("\n\n") if p.strip()]

def sliding_window(text, window=200, stride=100):
    words = text.split()
    chunks = []
    for i in range(0, len(words) - window + 1, stride):
        chunk = " ".join(words[i:i + window])
        chunks.append(chunk)
    return chunks

# ! pip install langchain
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512, chunk_overlap=100, separators=["\n\n", "\n", ". ", " ", ""]
)

text = """
Hello, world! More text here. another line.
Hello, world! More text here. another line......
"""
chunks = splitter.split_text(text)

from sentence_transformers import SentenceTransformer
import numpy as np

def semantic_chunking(text, similarity_threshold=0.5):
    model = SentenceTransformer('all-MiniLM-L6-v2')
    sentences = text.split('.')
    embeddings = model.encode(sentences)
    
    chunks = []
    current_chunk = [sentences[0]]
    
    for i in range(1, len(sentences)):
        # Calculate cosine similarity between consecutive sentences
        similarity = np.dot(embeddings[i-1], embeddings[i]) / (
            np.linalg.norm(embeddings[i-1]) * np.linalg.norm(embeddings[i])
        )
        
        if similarity < similarity_threshold:
            chunks.append('. '.join(current_chunk))
            current_chunk = [sentences[i]]
        else:
            current_chunk.append(sentences[i])
    
    chunks.append('. '.join(current_chunk))
    return chunks

{
  "document_id": "collection-config-guide",
  "document_title": "What is a Vector Database",
  "section_title": "What Is a Vector",
  "chunk_index": 7,
  "chunk_count": 15,
  "url": "https://qdrant.org.cn/documentation/concepts/collections/",
  "tags": ["qdrant", "vector search", "point", "vector", "payload"],
  "source_type": "documentation", 
  "created_at": "2025-01-15T10:00:00Z",
  "content": "There are three key elements that define a vector in vector search: the ID, the dimensions, and the payload. These components work together to represent a vector effectively within the system...",
  "word_count": 45,
  "char_count": 287
}

from qdrant_client import models

# Only show results from a specific article
filter = models.Filter(
    must=[
        models.FieldCondition(
            key="document_id", match=models.MatchValue(value="collection-config-guide")
        )
    ]
)

# Find vectors that also contain the keyword "HNSW" in their content
filter = models.Filter(
    must=[
        models.FieldCondition(
            key="content", # The field with the full-text index
            match=models.MatchText(text="HNSW algorithm")
        )
    ]
)

# Top result per document - get the most relevant chunk from each source
group_by = "document_id"

# Filter by user permissions
filter = models.Filter(
    must=[
        models.FieldCondition(
            key="access_level", match=models.MatchValue(value="public")
        )
    ]
)

def search_with_filters(query, document_type=None, date_range=None):
    """Search with metadata filtering"""

    # Build filter conditions
    filter_conditions = []

    if document_type:
        filter_conditions.append(
            models.FieldCondition(
                key="source_type", match=models.MatchValue(value=document_type)
            )
        )

    if date_range:
        filter_conditions.append(
            models.FieldCondition(
                key="created_at",
                range=models.Range(gte=date_range["start"], lte=date_range["end"]),
            )
        )

    # Execute search
    query_filter = models.Filter(must=filter_conditions) if filter_conditions else None

    results = client.query_points(
        collection_name="documents",
        query=generate_embedding(query),
        query_filter=query_filter,
        limit=5,
    )

    return results