"""

def __init__(self, collection_name: str = DEFAULT_COLLECTION, default_vector_size: int = 1536):

self.collection_name = collection_name

self.default_vector_size = default_vector_size

self.client = self._create_client()

def _create_client(self) -> QdrantClient:

"""Create and configure the Qdrant client"""

try:

if QDRANT_API_KEY:

client = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)

else:

client = QdrantClient(url=QDRANT_URL)

# Test connection

client.get_collections()

return client

except Exception as e:

raise RuntimeError(f'Failed to connect to Qdrant at {QDRANT_URL}: {e}')

def health_check(self) -> bool:

"""Check if Qdrant is reachable and responsive"""

try:

collections = self.client.get_collections()

return True

except Exception as e:

print(f"Health check failed: {e}")

return False

def get_cluster_info(self) -> Dict[str, Any]:

"""Get comprehensive cluster information"""

try:

info = {

'url': QDRANT_URL,

'collections': [],

'total_points': 0,

'total_collections': 0,

'cluster_info': None

}

# Get all collections

collections_response = self.client.get_collections()

collections = collections_response.collections

info['total_collections'] = len(collections)

# Get detailed info for each collection

for collection in collections:

try:

collection_info = self.client.get_collection(collection.name)

points_count = collection_info.points_count or 0

info['total_points'] += points_count

# Extract vector configuration

vectors_config = collection_info.config.params.vectors

if isinstance(vectors_config, dict):

# Named vectors

vector_info = {name: {'size': vec.size, 'distance': vec.distance.value}

for name, vec in vectors_config.items()}

else:

# Single vector

vector_info = {'default': {'size': vectors_config.size, 'distance': vectors_config.distance.value}}

info['collections'].append({

'name': collection.name,

'points_count': points_count,

'vectors': vector_info,

'status': collection_info.status.value if collection_info.status else 'unknown'

})

except Exception as e:

info['collections'].append({

'name': collection.name,

'error': str(e)

})

# Try to get cluster info (if available)

try:

cluster_info = self.client.get_cluster_info()

info['cluster_info'] = {

'peer_id': cluster_info.peer_id,

'peers': len(cluster_info.peers) if cluster_info.peers else 1

}

except:

pass # Cluster info not available in single-node setups

return info

except Exception as e:

return {'error': str(e)}

def list_collections(self) -> None:

"""List all collections with detailed information"""

info = self.get_cluster_info()

if 'error' in info:

print(f"❌ Error getting cluster info: {info['error']}")

return

print(f"📊 Qdrant Cluster Status")

print(f"URL: {info['url']}")

print(f"Collections: {info['total_collections']}")

print(f"Total Points: {info['total_points']:,}")

if info['cluster_info']:

print(f"Cluster: {info['cluster_info']['peers']} peer(s)")

print("-" * 60)

if info['total_collections'] == 0:

print("No collections found")

return

for col in info['collections']:

if 'error' in col:

print(f" 📁 {col['name']} (❌ Error: {col['error']})")

continue

print(f" 📁 {col['name']}")

print(f" Status: {col['status']}")

print(f" Points: {col['points_count']:,}")

# Display vector configurations

for vec_name, vec_config in col['vectors'].items():

name_suffix = f" ({vec_name})" if vec_name != 'default' else ""

print(f" Vector{name_suffix}: {vec_config['size']}D, {vec_config['distance']}")

# Show sample points if collection has data

if col['points_count'] > 0:

self._show_sample_points(col['name'], limit=2)

print()

def _show_sample_points(self, collection_name: str, limit: int = 3) -> None:

"""Show sample points from a collection"""

try:

# Use scroll to get points without needing a query vector

points, _ = self.client.scroll(collection_name=collection_name, limit=limit)

if points:

print(f" 📝 Sample points:")

for i, point in enumerate(points, 1):

# Extract meaningful text from payload

text = ""

if point.payload:

# Common text fields

text = (point.payload.get('text') or

point.payload.get('content') or

point.payload.get('message') or

str(point.payload))

# Truncate text for display

if isinstance(text, str) and len(text) > 100:

text = text[:97] + "..."

print(f" {i}. ID: {point.id}")

if text:

print(f" Text: {text}")

if point.payload and len(point.payload) > 1:

keys = [k for k in point.payload.keys() if k not in ['text', 'content', 'message']]

if keys:

print(f" Fields: {', '.join(keys[:5])}")

except Exception as e:

print(f" ⚠️ Could not fetch sample points: {e}")

def collection_info(self) -> None:

"""Show detailed information about the specific collection"""

try:

collection_info = self.client.get_collection(self.collection_name)

print(f"📊 Collection: {self.collection_name}")

print("-" * 50)

print(f"Status: {collection_info.status.value}")

print(f"Points: {collection_info.points_count:,}")

# Vector configuration

vectors_config = collection_info.config.params.vectors

if isinstance(vectors_config, dict):

print("Vector configurations:")

for name, config in vectors_config.items():

print(f" {name}: {config.size}D, {config.distance.value}")

else:

print(f"Vector size: {vectors_config.size}D")

print(f"Distance metric: {vectors_config.distance.value}")

# Indexing configuration

if collection_info.config.hnsw_config:

hnsw = collection_info.config.hnsw_config

print(f"HNSW M: {hnsw.m}, EF Construct: {hnsw.ef_construct}")

# Show sample search if there are points

if collection_info.points_count > 0:

print("\n🔍 Sample search (random vector):")

self._perform_sample_search()

except UnexpectedResponse as e:

if e.status_code == 404:

print(f"❌ Collection '{self.collection_name}' does not exist")

else:

print(f"❌ Error getting collection info: {e}")

except Exception as e:

print(f"❌ Error getting collection info: {e}")

def _perform_sample_search(self) -> None:

"""Perform a sample search with a random vector"""

try:

import random

# Create a random vector for search

query_vector = [random.random() for _ in range(self.default_vector_size)]

results = self.client.search(

collection_name=self.collection_name,

query_vector=query_vector,

limit=3

)

for i, result in enumerate(results, 1):

text = ""

if result.payload:

text = (result.payload.get('text') or

result.payload.get('content') or

result.payload.get('message') or

str(result.payload))

if isinstance(text, str) and len(text) > 80:

text = text[:77] + "..."

print(f" {i}. Score: {result.score:.3f}, ID: {result.id}")

if text:

print(f" Text: {text}")

except Exception as e:

print(f" ⚠️ Sample search failed: {e}")

def delete_collection(self, force: bool = False) -> bool:

"""Delete the collection"""

try:

# Check if collection exists

try:

collection_info = self.client.get_collection(self.collection_name)

points_count = collection_info.points_count

except UnexpectedResponse as e:

if e.status_code == 404:

print(f"ℹ️ Collection '{self.collection_name}' does not exist")

return True

raise

if not force:

print(f"⚠️ About to delete collection '{self.collection_name}'")

print(f" This will permanently remove {points_count:,} points")

confirm = input(" Are you sure? (y/N): ").strip().lower()

if confirm != 'y':

print("❌ Deletion cancelled")

return False

self.client.delete_collection(collection_name=self.collection_name)

print(f"✅ Deleted collection '{self.collection_name}' ({points_count:,} points)")

return True

except Exception as e:

print(f"❌ Error deleting collection: {e}")

return False

def create_collection(self, vector_size: Optional[int] = None,

distance: str = 'Cosine',

hnsw_m: int = 16,

hnsw_ef_construct: int = 100) -> bool:

"""Create a new collection with specified parameters"""

try:

vsize = vector_size or self.default_vector_size

# Convert distance string to enum

distance_enum = getattr(Distance, distance.upper(), Distance.COSINE)

vectors_config = VectorParams(

size=vsize,

distance=distance_enum

)

# HNSW configuration for better performance

hnsw_config = models.HnswConfigDiff(

m=hnsw_m,

ef_construct=hnsw_ef_construct

)

self.client.create_collection(

collection_name=self.collection_name,

vectors_config=vectors_config,

hnsw_config=hnsw_config

)

print(f"✅ Created collection '{self.collection_name}'")

print(f" Vector size: {vsize}D")

print(f" Distance: {distance}")

print(f" HNSW M: {hnsw_m}, EF Construct: {hnsw_ef_construct}")

return True

except Exception as e:

print(f"❌ Error creating collection: {e}")

return False

def recreate_collection(self, vector_size: Optional[int] = None, **kwargs) -> bool:

"""Delete and recreate the collection"""

print(f"🔄 Recreating collection '{self.collection_name}'")

if self.delete_collection(force=True):

return self.create_collection(vector_size=vector_size, **kwargs)

return False

def add_point(self, point_id: str, vector: List[float],

payload: Optional[Dict[str, Any]] = None) -> bool:

"""Add a single point to the collection"""

try:

self.client.upsert(

collection_name=self.collection_name,

points=[models.PointStruct(

id=point_id,

vector=vector,

payload=payload or {}

)]

)

print(f"✅ Added point {point_id}")

return True

except Exception as e:

print(f"❌ Error adding point: {e}")

return False

def delete_point(self, point_id: str) -> bool:

"""Delete a single point from the collection"""

try:

self.client.delete(

collection_name=self.collection_name,

points_selector=models.PointIdsList(points=[point_id])

)

print(f"✅ Deleted point {point_id}")

return True

except Exception as e:

print(f"❌ Error deleting point: {e}")

return False

def search_points(self, query_vector: List[float], limit: int = 10,

score_threshold: Optional[float] = None) -> List[models.ScoredPoint]:

"""Search for similar points"""

try:

return self.client.search(

collection_name=self.collection_name,

query_vector=query_vector,

limit=limit,

score_threshold=score_threshold

)

except Exception as e:

print(f"❌ Error searching points: {e}")

return []

def get_storage_info(self) -> None:

"""Display storage and configuration information"""

print("💾 Qdrant Storage Information")

print("-" * 50)

print(f"Connection URL: {QDRANT_URL}")

print(f"API Key: {'Set' if QDRANT_API_KEY else 'Not set'}")

print(f"Target Collection: {self.collection_name}")

print()

# Cluster info

cluster_info = self.get_cluster_info()

if 'error' not in cluster_info:

print(f"Total Collections: {cluster_info['total_collections']}")

print(f"Total Points: {cluster_info['total_points']:,}")

if cluster_info['cluster_info']:

print(f"Cluster Peers: {cluster_info['cluster_info']['peers']}")

print("\n🗑️ Management Options:")

print(" 1. List all collections: --list")

print(" 2. Collection details: --info")

print(" 3. Delete collection: --delete")

print(" 4. Recreate collection: --recreate")

parser = argparse.ArgumentParser(

description='Qdrant Collection Management - Official Python SDK',

formatter_class=argparse.RawDescriptionHelpFormatter,

epilog="""

""")

parser.add_argument('--collection', type=str, default=DEFAULT_COLLECTION,

help=f'Collection name (default: {DEFAULT_COLLECTION})')

parser.add_argument('--list', action='store_true', help='List all collections')

parser.add_argument('--info', action='store_true', help='Show collection information')

parser.add_argument('--delete', action='store_true', help='Delete the collection')

parser.add_argument('--recreate', action='store_true', help='Delete and recreate collection')

parser.add_argument('--storage', action='store_true', help='Show storage information')

parser.add_argument('--health', action='store_true', help='Run health check')

parser.add_argument('--force', action='store_true', help='Force operations without confirmation')

# Collection creation options

parser.add_argument('--vector-size', type=int, help='Vector size for new collections')

parser.add_argument('--distance', choices=['Cosine', 'Euclidean', 'Dot'], default='Cosine',

help='Distance metric for vectors')

parser.add_argument('--hnsw-m', type=int, default=16, help='HNSW M parameter')

parser.add_argument('--hnsw-ef-construct', type=int, default=100, help='HNSW EF construct parameter')

# Point operations

parser.add_argument('--add-point', action='store_true', help='Add a point to the collection')

parser.add_argument('--delete-point', action='store_true', help='Delete a point from the collection')

parser.add_argument('--id', type=str, help='Point ID for add/delete operations')

parser.add_argument('--vector', type=str, help='Comma-separated vector values')

parser.add_argument('--payload', type=str, help='JSON payload for the point')

args = parser.parse_args()

# Default to showing help if no operation specified

operations = [args.list, args.info, args.delete, args.recreate, args.storage,

args.health, args.add_point, args.delete_point]

if not any(operations):

parser.print_help()

return

try:

manager = QdrantManager(collection_name=args.collection)

except RuntimeError as e:

print(f"❌ {e}")

return

# Execute requested operations

if args.health:

ok = manager.health_check()

print('✅ Qdrant is reachable and responsive' if ok else '❌ Qdrant is not reachable')

if args.list:

manager.list_collections()

if args.info:

manager.collection_info()

if args.storage:

manager.get_storage_info()

if args.delete:

manager.delete_collection(force=args.force)

if args.recreate:

kwargs = {}

if args.distance:

kwargs['distance'] = args.distance

if args.hnsw_m:

kwargs['hnsw_m'] = args.hnsw_m

if args.hnsw_ef_construct:

kwargs['hnsw_ef_construct'] = args.hnsw_ef_construct

manager.recreate_collection(vector_size=args.vector_size, **kwargs)

if args.add_point:

if not args.id or not args.vector:

print('❌ Error: --id and --vector are required for --add-point')

return

try:

vector = [float(x.strip()) for x in args.vector.split(',') if x.strip()]

payload = None

if args.payload:

payload = json.loads(args.payload)

manager.add_point(args.id, vector, payload)

except (ValueError, json.JSONDecodeError) as e:

print(f'❌ Error parsing vector or payload: {e}')

if args.delete_point:

if not args.id:

print('❌ Error: --id is required for --delete-point')

return