| self._model = model | self._model = model | ||||
| self._load_balancing_configs = load_balancing_configs | self._load_balancing_configs = load_balancing_configs | ||||
| for load_balancing_config in self._load_balancing_configs: | |||||
| for load_balancing_config in self._load_balancing_configs[:]: # Iterate over a shallow copy of the list | |||||
| if load_balancing_config.name == "__inherit__": | if load_balancing_config.name == "__inherit__": | ||||
| if not managed_credentials: | if not managed_credentials: | ||||
| # FIXME: Mutation to loop iterable `self._load_balancing_configs` during iteration | |||||
| # remove __inherit__ if managed credentials is not provided | # remove __inherit__ if managed credentials is not provided | ||||
| self._load_balancing_configs.remove(load_balancing_config) | self._load_balancing_configs.remove(load_balancing_config) | ||||
| else: | else: |
| raise NotImplementedError | raise NotImplementedError | ||||
| def _filter_duplicate_texts(self, texts: list[Document]) -> list[Document]: | def _filter_duplicate_texts(self, texts: list[Document]) -> list[Document]: | ||||
| for text in texts: | |||||
| for text in texts[:]: | |||||
| doc_id = text.metadata['doc_id'] | doc_id = text.metadata['doc_id'] | ||||
| exists_duplicate_node = self.text_exists(doc_id) | exists_duplicate_node = self.text_exists(doc_id) | ||||
| if exists_duplicate_node: | if exists_duplicate_node: | ||||
| # FIXME: Mutation to loop iterable `texts` during iteration | |||||
| texts.remove(text) | texts.remove(text) | ||||
| return texts | return texts |
| raise NotImplementedError | raise NotImplementedError | ||||
| def _filter_duplicate_texts(self, texts: list[Document]) -> list[Document]: | def _filter_duplicate_texts(self, texts: list[Document]) -> list[Document]: | ||||
| for text in texts: | |||||
| for text in texts[:]: | |||||
| doc_id = text.metadata['doc_id'] | doc_id = text.metadata['doc_id'] | ||||
| exists_duplicate_node = self.text_exists(doc_id) | exists_duplicate_node = self.text_exists(doc_id) | ||||
| if exists_duplicate_node: | if exists_duplicate_node: | ||||
| # FIXME: Mutation to loop iterable `texts` during iteration | |||||
| texts.remove(text) | texts.remove(text) | ||||
| return texts | return texts |
| return CacheEmbedding(embedding_model) | return CacheEmbedding(embedding_model) | ||||
| def _filter_duplicate_texts(self, texts: list[Document]) -> list[Document]: | def _filter_duplicate_texts(self, texts: list[Document]) -> list[Document]: | ||||
| for text in texts: | |||||
| for text in texts[:]: | |||||
| doc_id = text.metadata['doc_id'] | doc_id = text.metadata['doc_id'] | ||||
| exists_duplicate_node = self.text_exists(doc_id) | exists_duplicate_node = self.text_exists(doc_id) | ||||
| if exists_duplicate_node: | if exists_duplicate_node: | ||||
| # FIXME: Mutation to loop iterable `texts` during iteration | |||||
| texts.remove(text) | texts.remove(text) | ||||
| return texts | return texts |
| load_balancing_configs.insert(0, inherit_config) | load_balancing_configs.insert(0, inherit_config) | ||||
| else: | else: | ||||
| # move the inherit configuration to the first | # move the inherit configuration to the first | ||||
| for i, load_balancing_config in enumerate(load_balancing_configs): | |||||
| for i, load_balancing_config in enumerate(load_balancing_configs[:]): | |||||
| if load_balancing_config.name == '__inherit__': | if load_balancing_config.name == '__inherit__': | ||||
| # FIXME: Mutation to loop iterable `load_balancing_configs` during iteration | |||||
| inherit_config = load_balancing_configs.pop(i) | inherit_config = load_balancing_configs.pop(i) | ||||
| load_balancing_configs.insert(0, inherit_config) | load_balancing_configs.insert(0, inherit_config) | ||||