Coverage for app \ rag \ qdrant_search.py: 100%

1import requests 

2from typing import Dict, List, Any 

3 

4from app.processing.embedding import embed_texts 

5from app.vector_store.qdrant_store import COLLECTION 

6 

7 

8def qdrant_hybrid_search( 

9 question: str, 

10 user_context: Dict[str, Any], 

11 expanded_entities: Dict[str, List[str]], 

12 top_k: int = 5, 

13) -> List[Dict[str, Any]]: 

14 terms = [] 

15 terms.extend(user_context.get("conditions", [])) 

16 terms.extend(user_context.get("drugs", [])) 

17 

18 for k in expanded_entities: 

19 terms.extend(expanded_entities[k]) 

20 

21 query_text = question + "\nContext: " + ", ".join(set(terms)) 

22 query_vector = embed_texts([query_text])[0] 

23 

24 resp = requests.post( 

25 f"http://localhost:6333/collections/{COLLECTION}/points/search", 

26 json={"vector": query_vector, "limit": top_k, "with_payload": True}, 

27 timeout=10, 

28 ) 

29 resp.raise_for_status() 

30 

31 results = [] 

32 for hit in resp.json().get("result", []): 

33 payload = hit["payload"] 

34 results.append( 

35 { 

36 "score": hit["score"], 

37 "pmid": payload.get("pmid"), 

38 "title": payload.get("title"), 

39 "text_preview": payload.get("text", "")[:500], 

40 "entities": payload.get("entities"), 

41 } 

42 ) 

43 

44 return results