Coverage for src/app/adapters/faiss_adapter.py: 30%

1from ports.similarity_search_port import SimilaritySearchPort 

2from ports.add_chunks_port import AddChunksPort 

3 

4from repositories.faiss_repository import FaissRepository 

5 

6from models.question_model import QuestionModel 

7from models.context_model import ContextModel 

8from models.file_chunk_model import FileChunkModel 

9 

10from entities.query_entity import QueryEntity 

11from entities.file_chunk_entity import FileChunkEntity 

12 

13 

14class FaissAdapter(SimilaritySearchPort, AddChunksPort): 

15 

16 def __init__(self, faiss_repository: FaissRepository): 

17 self.faiss_repository = faiss_repository 

18 

19 def similarity_search(self, question_model: QuestionModel) -> list[ContextModel]: 

20 """ 

21 Perform a similarity search using the provided question model. 

22 

23 Args: 

24 question_model (QuestionModel): The model containing the question to search for. 

25 

26 Returns: 

27 list[ContextModel]: A list of context models that are similar to the question. 

28 """ 

29 if not question_model.get_question().strip(): 

30 raise ValueError("Question cannot be empty") 

31 

32 try: 

33 query = QueryEntity(question_model.get_user_id , question_model.get_question()) 

34 result = self.faiss_repository.similarity_search(query) 

35 

36 context_list = [] 

37 

38 for context in result: 

39 context_list.append( ContextModel(context.get_content()) ) 

40 

41 return context_list 

42 

43 except Exception as e: 

44 return str(e) 

45 

46 def load_chunks(self, chunks: list[FileChunkModel]): 

47 """ 

48 Load chunks into the FAISS repository. 

49 

50 Args: 

51 chunks (list[FileChunkModel]): A list of file chunk models to be loaded. 

52 

53 Raises: 

54 ValueError: If no chunks are provided. 

55 

56 Returns: 

57 str: Result of the loading process or an error message. 

58 """ 

59 if not chunks: 

60 raise ValueError("No chunks to load.") 

61 

62 try: 

63 chunks_entity = [] 

64 for chunk in chunks: 

65 chunks_entity.append( FileChunkEntity(chunk.get_chunk_content(), chunk.get_metadata()) ) 

66 

67 result = self.faiss_repository.load_chunks(chunks_entity) 

68 

69 print(result) 

70 print("Chunks loaded successfully") 

71 

72 except Exception as e: 

73 return str(e) 

74 

75