Enterprise Search & Knowledge Base: Architecture Blueprint & Production Stack

from qdrant_client import QdrantClient, models
import cohere
from typing import List, Dict, Any

# Initialize Qdrant and Cohere Clients
qdrant = QdrantClient(url="http://qdrant-cluster.internal:6333", api_key="VPC_QDRANT_KEY")
cohere_client = cohere.ClientV2("COHERE_PRODUCTION_API_KEY")

def execute_permission_aware_search(
    user_query: str, 
    user_groups: List[str], 
    top_k: int = 10
) -> List[Dict[str, Any]]:
    """Performs Qdrant hybrid vector search with ACL payload security filtering."""
    
    # 1. Construct Qdrant security filter enforcing active user group access
    security_filter = models.Filter(
        must=[
            models.FieldCondition(
                key="allowed_groups",
                match=models.MatchAny(any=user_groups)
            )
        ]
    )
    
    # 2. Embed user query using dense vector model
    # (Dummy embedding array representing BGE-M3 / OpenAI 1536-dim vector)
    query_vector = [0.014, -0.092, 0.412] + [0.0] * 1533
    
    # 3. Query Qdrant Collection
    search_results = qdrant.search(
        collection_name="enterprise_kb",
        query_vector=query_vector,
        query_filter=security_filter,
        limit=50,  # Fetch top 50 candidates for reranking
        with_payload=True
    )
    
    # Extract candidate document texts
    documents = [hit.payload["text_chunk"] for hit in search_results]
    
    # 4. Apply Cohere Rerank v3 Cross-Encoder Scoring
    rerank_response = cohere_client.rerank(
        model="rerank-v3.5",
        query=user_query,
        documents=documents,
        top_n=top_k
    )
    
    # Assemble final reranked context payload with citations
    final_results = []
    for item in rerank_response.results:
        original_hit = search_results[item.index]
        final_results.append({
            "chunk_id": original_hit.id,
            "document_title": original_hit.payload["document_title"],
            "page_number": original_hit.payload["page_number"],
            "relevance_score": item.relevance_score,
            "content": original_hit.payload["text_chunk"]
        })
        
    return final_results

# Example Usage
if __name__ == "__main__":
    results = execute_permission_aware_search(
        user_query="What are the Q3 SLA requirements for cloud vector databases?",
        user_groups=["eng-lead", "devops-team"]
    )
    print(f"Retrieved {len(results)} cited context chunks. Top score: {results[0]['relevance_score']:.4f}")