knowledge

knowledge

Covers how to build, maintain, and query knowledge systems: embedding pipelines, vector stores, RAG architecture, and retrieval quality evaluation.

1) Embedding generation

from openai import AsyncOpenAI
import anthropic

client_oai = AsyncOpenAI(api_key=OPENAI_API_KEY)

async def embed_text(text: str, model: str = "text-embedding-3-small") -> list[float]:
    res = await client_oai.embeddings.create(input=text, model=model)
    return res.data[0].embedding

# Batch embedding (efficient)
async def embed_batch(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    res = await client_oai.embeddings.create(input=texts, model=model)
    return [r.embedding for r in res.data]

# Dimensions by model
EMBEDDING_DIMS = {
    "text-embedding-3-small": 1536,   # cheap, good for most use cases
    "text-embedding-3-large": 3072,   # better recall, 6× cost
    "text-embedding-ada-002":  1536,   # legacy
}

2) Semantic chunking

def chunk_document(text: str, chunk_size: int = 512, overlap: int = 64) -> list[str]:
    """Sliding window chunking with overlap to preserve context at boundaries."""
    words = text.split()
    chunks = []
    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i:i + chunk_size])
        if chunk.strip():
            chunks.append(chunk)
    return chunks

def chunk_by_section(text: str) -> list[dict]:
    """Prefer semantic boundaries (headings) over fixed windows."""
    import re
    sections = re.split(r'\n(?=#{1,3} )', text)
    return [
        {"content": s.strip(), "heading": (re.match(r'^#+\s+(.+)', s) or [None, ""])[1]}
        for s in sections if len(s.strip()) > 50
    ]

3) Qdrant vector store patterns

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue

qdrant = QdrantClient(host="localhost", port=6333)

# Create collection
qdrant.create_collection(
    collection_name="articles",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

# Upsert vectors
async def upsert_document(doc_id: str, text: str, metadata: dict):
    embedding = await embed_text(text)
    qdrant.upsert(
        collection_name="articles",
        points=[PointStruct(
            id=hash(doc_id) % (2**63),   # Qdrant needs integer or UUID
            vector=embedding,
            payload={**metadata, "text": text, "doc_id": doc_id},
        )],
    )

# Semantic search with metadata filter
async def search(query: str, domain: str = None, limit: int = 5) -> list[dict]:
    embedding = await embed_text(query)
    filter_ = Filter(must=[FieldCondition(key="domain", match=MatchValue(value=domain))]) if domain else None
    results = qdrant.search(
        collection_name="articles",
        query_vector=embedding,
        query_filter=filter_,
        limit=limit,
        with_payload=True,
    )
    return [{"score": r.score, **r.payload} for r in results]