Underground-Digital/Workflow-Engine
0
1import math2import threading3from collections import Counter4from typing import Optional, cast5 6from flask import Flask, current_app7 8from core.app.app_config.entities import DatasetEntity, DatasetRetrieveConfigEntity9from core.app.entities.app_invoke_entities import InvokeFrom, ModelConfigWithCredentialsEntity10from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler11from core.entities.agent_entities import PlanningStrategy12from core.memory.token_buffer_memory import TokenBufferMemory13from core.model_manager import ModelInstance, ModelManager14from core.model_runtime.entities.message_entities import PromptMessageTool15from core.model_runtime.entities.model_entities import ModelFeature, ModelType16from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel17from core.ops.entities.trace_entity import TraceTaskName18from core.ops.ops_trace_manager import TraceQueueManager, TraceTask19from core.ops.utils import measure_time20from core.rag.data_post_processor.data_post_processor import DataPostProcessor21from core.rag.datasource.keyword.jieba.jieba_keyword_table_handler import JiebaKeywordTableHandler22from core.rag.datasource.retrieval_service import RetrievalService23from core.rag.entities.context_entities import DocumentContext24from core.rag.models.document import Document25from core.rag.rerank.rerank_type import RerankMode26from core.rag.retrieval.retrieval_methods import RetrievalMethod27from core.rag.retrieval.router.multi_dataset_function_call_router import FunctionCallMultiDatasetRouter28from core.rag.retrieval.router.multi_dataset_react_route import ReactMultiDatasetRouter29from core.tools.tool.dataset_retriever.dataset_multi_retriever_tool import DatasetMultiRetrieverTool30from core.tools.tool.dataset_retriever.dataset_retriever_base_tool import DatasetRetrieverBaseTool31from core.tools.tool.dataset_retriever.dataset_retriever_tool import DatasetRetrieverTool32from extensions.ext_database import db33from models.dataset import Dataset, DatasetQuery, DocumentSegment34from models.dataset import Document as DatasetDocument35from services.external_knowledge_service import ExternalDatasetService36 37default_retrieval_model = {38 "search_method": RetrievalMethod.SEMANTIC_SEARCH.value,39 "reranking_enable": False,40 "reranking_model": {"reranking_provider_name": "", "reranking_model_name": ""},41 "top_k": 2,42 "score_threshold_enabled": False,43}44 45 46class DatasetRetrieval:47 def __init__(self, application_generate_entity=None):48 self.application_generate_entity = application_generate_entity49 50 def retrieve(51 self,52 app_id: str,53 user_id: str,54 tenant_id: str,55 model_config: ModelConfigWithCredentialsEntity,56 config: DatasetEntity,57 query: str,58 invoke_from: InvokeFrom,59 show_retrieve_source: bool,60 hit_callback: DatasetIndexToolCallbackHandler,61 message_id: str,62 memory: Optional[TokenBufferMemory] = None,63 ) -> Optional[str]:64 """65 Retrieve dataset.66 :param app_id: app_id67 :param user_id: user_id68 :param tenant_id: tenant id69 :param model_config: model config70 :param config: dataset config71 :param query: query72 :param invoke_from: invoke from73 :param show_retrieve_source: show retrieve source74 :param hit_callback: hit callback75 :param message_id: message id76 :param memory: memory77 :return:78 """79 dataset_ids = config.dataset_ids80 if len(dataset_ids) == 0:81 return None82 retrieve_config = config.retrieve_config83 84 # check model is support tool calling85 model_type_instance = model_config.provider_model_bundle.model_type_instance86 model_type_instance = cast(LargeLanguageModel, model_type_instance)87 88 model_manager = ModelManager()89 model_instance = model_manager.get_model_instance(90 tenant_id=tenant_id, model_type=ModelType.LLM, provider=model_config.provider, model=model_config.model91 )92 93 # get model schema94 model_schema = model_type_instance.get_model_schema(95 model=model_config.model, credentials=model_config.credentials96 )97 98 if not model_schema:99 return None100 101 planning_strategy = PlanningStrategy.REACT_ROUTER102 features = model_schema.features103 if features:104 if ModelFeature.TOOL_CALL in features or ModelFeature.MULTI_TOOL_CALL in features:105 planning_strategy = PlanningStrategy.ROUTER106 available_datasets = []107 for dataset_id in dataset_ids:108 # get dataset from dataset id109 dataset = db.session.query(Dataset).filter(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()110 111 # pass if dataset is not available112 if not dataset:113 continue114 115 # pass if dataset is not available116 if dataset and dataset.available_document_count == 0 and dataset.provider != "external":117 continue118 119 available_datasets.append(dataset)120 all_documents = []121 user_from = "account" if invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER} else "end_user"122 if retrieve_config.retrieve_strategy == DatasetRetrieveConfigEntity.RetrieveStrategy.SINGLE:123 all_documents = self.single_retrieve(124 app_id,125 tenant_id,126 user_id,127 user_from,128 available_datasets,129 query,130 model_instance,131 model_config,132 planning_strategy,133 message_id,134 )135 elif retrieve_config.retrieve_strategy == DatasetRetrieveConfigEntity.RetrieveStrategy.MULTIPLE:136 all_documents = self.multiple_retrieve(137 app_id,138 tenant_id,139 user_id,140 user_from,141 available_datasets,142 query,143 retrieve_config.top_k,144 retrieve_config.score_threshold,145 retrieve_config.rerank_mode,146 retrieve_config.reranking_model,147 retrieve_config.weights,148 retrieve_config.reranking_enabled,149 message_id,150 )151 152 dify_documents = [item for item in all_documents if item.provider == "dify"]153 external_documents = [item for item in all_documents if item.provider == "external"]154 document_context_list = []155 retrieval_resource_list = []156 # deal with external documents157 for item in external_documents:158 document_context_list.append(DocumentContext(content=item.page_content, score=item.metadata.get("score")))159 source = {160 "dataset_id": item.metadata.get("dataset_id"),161 "dataset_name": item.metadata.get("dataset_name"),162 "document_name": item.metadata.get("title"),163 "data_source_type": "external",164 "retriever_from": invoke_from.to_source(),165 "score": item.metadata.get("score"),166 "content": item.page_content,167 }168 retrieval_resource_list.append(source)169 document_score_list = {}170 # deal with dify documents171 if dify_documents:172 for item in dify_documents:173 if item.metadata.get("score"):174 document_score_list[item.metadata["doc_id"]] = item.metadata["score"]175 176 index_node_ids = [document.metadata["doc_id"] for document in dify_documents]177 segments = DocumentSegment.query.filter(178 DocumentSegment.dataset_id.in_(dataset_ids),179 DocumentSegment.status == "completed",180 DocumentSegment.enabled == True,181 DocumentSegment.index_node_id.in_(index_node_ids),182 ).all()183 184 if segments:185 index_node_id_to_position = {id: position for position, id in enumerate(index_node_ids)}186 sorted_segments = sorted(187 segments, key=lambda segment: index_node_id_to_position.get(segment.index_node_id, float("inf"))188 )189 for segment in sorted_segments:190 if segment.answer:191 document_context_list.append(192 DocumentContext(193 content=f"question:{segment.get_sign_content()} answer:{segment.answer}",194 score=document_score_list.get(segment.index_node_id, None),195 )196 )197 else:198 document_context_list.append(199 DocumentContext(200 content=segment.get_sign_content(),201 score=document_score_list.get(segment.index_node_id, None),202 )203 )204 if show_retrieve_source:205 for segment in sorted_segments:206 dataset = Dataset.query.filter_by(id=segment.dataset_id).first()207 document = DatasetDocument.query.filter(208 DatasetDocument.id == segment.document_id,209 DatasetDocument.enabled == True,210 DatasetDocument.archived == False,211 ).first()212 if dataset and document:213 source = {214 "dataset_id": dataset.id,215 "dataset_name": dataset.name,216 "document_id": document.id,217 "document_name": document.name,218 "data_source_type": document.data_source_type,219 "segment_id": segment.id,220 "retriever_from": invoke_from.to_source(),221 "score": document_score_list.get(segment.index_node_id, 0.0),222 }223 224 if invoke_from.to_source() == "dev":225 source["hit_count"] = segment.hit_count226 source["word_count"] = segment.word_count227 source["segment_position"] = segment.position228 source["index_node_hash"] = segment.index_node_hash229 if segment.answer:230 source["content"] = f"question:{segment.content} \nanswer:{segment.answer}"231 else:232 source["content"] = segment.content233 retrieval_resource_list.append(source)234 if hit_callback and retrieval_resource_list:235 retrieval_resource_list = sorted(retrieval_resource_list, key=lambda x: x.get("score") or 0.0, reverse=True)236 for position, item in enumerate(retrieval_resource_list, start=1):237 item["position"] = position238 hit_callback.return_retriever_resource_info(retrieval_resource_list)239 if document_context_list:240 document_context_list = sorted(document_context_list, key=lambda x: x.score or 0.0, reverse=True)241 return str("\n".join([document_context.content for document_context in document_context_list]))242 return ""243 244 def single_retrieve(245 self,246 app_id: str,247 tenant_id: str,248 user_id: str,249 user_from: str,250 available_datasets: list,251 query: str,252 model_instance: ModelInstance,253 model_config: ModelConfigWithCredentialsEntity,254 planning_strategy: PlanningStrategy,255 message_id: Optional[str] = None,256 ):257 tools = []258 for dataset in available_datasets:259 description = dataset.description260 if not description:261 description = "useful for when you want to answer queries about the " + dataset.name262 263 description = description.replace("\n", "").replace("\r", "")264 message_tool = PromptMessageTool(265 name=dataset.id,266 description=description,267 parameters={268 "type": "object",269 "properties": {},270 "required": [],271 },272 )273 tools.append(message_tool)274 dataset_id = None275 if planning_strategy == PlanningStrategy.REACT_ROUTER:276 react_multi_dataset_router = ReactMultiDatasetRouter()277 dataset_id = react_multi_dataset_router.invoke(278 query, tools, model_config, model_instance, user_id, tenant_id279 )280 281 elif planning_strategy == PlanningStrategy.ROUTER:282 function_call_router = FunctionCallMultiDatasetRouter()283 dataset_id = function_call_router.invoke(query, tools, model_config, model_instance)284 285 if dataset_id:286 # get retrieval model config287 dataset = db.session.query(Dataset).filter(Dataset.id == dataset_id).first()288 if dataset:289 results = []290 if dataset.provider == "external":291 external_documents = ExternalDatasetService.fetch_external_knowledge_retrieval(292 tenant_id=dataset.tenant_id,293 dataset_id=dataset_id,294 query=query,295 external_retrieval_parameters=dataset.retrieval_model,296 )297 for external_document in external_documents:298 document = Document(299 page_content=external_document.get("content"),300 metadata=external_document.get("metadata"),301 provider="external",302 )303 document.metadata["score"] = external_document.get("score")304 document.metadata["title"] = external_document.get("title")305 document.metadata["dataset_id"] = dataset_id306 document.metadata["dataset_name"] = dataset.name307 results.append(document)308 else:309 retrieval_model_config = dataset.retrieval_model or default_retrieval_model310 311 # get top k312 top_k = retrieval_model_config["top_k"]313 # get retrieval method314 if dataset.indexing_technique == "economy":315 retrieval_method = "keyword_search"316 else:317 retrieval_method = retrieval_model_config["search_method"]318 # get reranking model319 reranking_model = (320 retrieval_model_config["reranking_model"]321 if retrieval_model_config["reranking_enable"]322 else None323 )324 # get score threshold325 score_threshold = 0.0326 score_threshold_enabled = retrieval_model_config.get("score_threshold_enabled")327 if score_threshold_enabled:328 score_threshold = retrieval_model_config.get("score_threshold")329 330 with measure_time() as timer:331 results = RetrievalService.retrieve(332 retrieval_method=retrieval_method,333 dataset_id=dataset.id,334 query=query,335 top_k=top_k,336 score_threshold=score_threshold,337 reranking_model=reranking_model,338 reranking_mode=retrieval_model_config.get("reranking_mode", "reranking_model"),339 weights=retrieval_model_config.get("weights", None),340 )341 self._on_query(query, [dataset_id], app_id, user_from, user_id)342 343 if results:344 self._on_retrieval_end(results, message_id, timer)345 346 return results347 return []348 349 def multiple_retrieve(350 self,351 app_id: str,352 tenant_id: str,353 user_id: str,354 user_from: str,355 available_datasets: list,356 query: str,357 top_k: int,358 score_threshold: float,359 reranking_mode: str,360 reranking_model: Optional[dict] = None,361 weights: Optional[dict] = None,362 reranking_enable: bool = True,363 message_id: Optional[str] = None,364 ):365 if not available_datasets:366 return []367 threads = []368 all_documents = []369 dataset_ids = [dataset.id for dataset in available_datasets]370 index_type_check = all(371 item.indexing_technique == available_datasets[0].indexing_technique for item in available_datasets372 )373 if not index_type_check and (not reranking_enable or reranking_mode != RerankMode.RERANKING_MODEL):374 raise ValueError(375 "The configured knowledge base list have different indexing technique, please set reranking model."376 )377 index_type = available_datasets[0].indexing_technique378 if index_type == "high_quality":379 embedding_model_check = all(380 item.embedding_model == available_datasets[0].embedding_model for item in available_datasets381 )382 embedding_model_provider_check = all(383 item.embedding_model_provider == available_datasets[0].embedding_model_provider384 for item in available_datasets385 )386 if (387 reranking_enable388 and reranking_mode == "weighted_score"389 and (not embedding_model_check or not embedding_model_provider_check)390 ):391 raise ValueError(392 "The configured knowledge base list have different embedding model, please set reranking model."393 )394 if reranking_enable and reranking_mode == RerankMode.WEIGHTED_SCORE:395 weights["vector_setting"]["embedding_provider_name"] = available_datasets[0].embedding_model_provider396 weights["vector_setting"]["embedding_model_name"] = available_datasets[0].embedding_model397 398 for dataset in available_datasets:399 index_type = dataset.indexing_technique400 retrieval_thread = threading.Thread(401 target=self._retriever,402 kwargs={403 "flask_app": current_app._get_current_object(),404 "dataset_id": dataset.id,405 "query": query,406 "top_k": top_k,407 "all_documents": all_documents,408 },409 )410 threads.append(retrieval_thread)411 retrieval_thread.start()412 for thread in threads:413 thread.join()414 415 with measure_time() as timer:416 if reranking_enable:417 # do rerank for searched documents418 data_post_processor = DataPostProcessor(tenant_id, reranking_mode, reranking_model, weights, False)419 420 all_documents = data_post_processor.invoke(421 query=query, documents=all_documents, score_threshold=score_threshold, top_n=top_k422 )423 else:424 if index_type == "economy":425 all_documents = self.calculate_keyword_score(query, all_documents, top_k)426 elif index_type == "high_quality":427 all_documents = self.calculate_vector_score(all_documents, top_k, score_threshold)428 429 self._on_query(query, dataset_ids, app_id, user_from, user_id)430 431 if all_documents:432 self._on_retrieval_end(all_documents, message_id, timer)433 434 return all_documents435 436 def _on_retrieval_end(437 self, documents: list[Document], message_id: Optional[str] = None, timer: Optional[dict] = None438 ) -> None:439 """Handle retrieval end."""440 dify_documents = [document for document in documents if document.provider == "dify"]441 for document in dify_documents:442 query = db.session.query(DocumentSegment).filter(443 DocumentSegment.index_node_id == document.metadata["doc_id"]444 )445 446 # if 'dataset_id' in document.metadata:447 if "dataset_id" in document.metadata:448 query = query.filter(DocumentSegment.dataset_id == document.metadata["dataset_id"])449 450 # add hit count to document segment451 query.update({DocumentSegment.hit_count: DocumentSegment.hit_count + 1}, synchronize_session=False)452 453 db.session.commit()454 455 # get tracing instance456 trace_manager: TraceQueueManager = (457 self.application_generate_entity.trace_manager if self.application_generate_entity else None458 )459 if trace_manager:460 trace_manager.add_trace_task(461 TraceTask(462 TraceTaskName.DATASET_RETRIEVAL_TRACE, message_id=message_id, documents=documents, timer=timer463 )464 )465 466 def _on_query(self, query: str, dataset_ids: list[str], app_id: str, user_from: str, user_id: str) -> None:467 """468 Handle query.469 """470 if not query:471 return472 dataset_queries = []473 for dataset_id in dataset_ids:474 dataset_query = DatasetQuery(475 dataset_id=dataset_id,476 content=query,477 source="app",478 source_app_id=app_id,479 created_by_role=user_from,480 created_by=user_id,481 )482 dataset_queries.append(dataset_query)483 if dataset_queries:484 db.session.add_all(dataset_queries)485 db.session.commit()486 487 def _retriever(self, flask_app: Flask, dataset_id: str, query: str, top_k: int, all_documents: list):488 with flask_app.app_context():489 dataset = db.session.query(Dataset).filter(Dataset.id == dataset_id).first()490 491 if not dataset:492 return []493 494 if dataset.provider == "external":495 external_documents = ExternalDatasetService.fetch_external_knowledge_retrieval(496 tenant_id=dataset.tenant_id,497 dataset_id=dataset_id,498 query=query,499 external_retrieval_parameters=dataset.retrieval_model,500 )501 for external_document in external_documents:502 document = Document(503 page_content=external_document.get("content"),504 metadata=external_document.get("metadata"),505 provider="external",506 )507 document.metadata["score"] = external_document.get("score")508 document.metadata["title"] = external_document.get("title")509 document.metadata["dataset_id"] = dataset_id510 document.metadata["dataset_name"] = dataset.name511 all_documents.append(document)512 else:513 # get retrieval model , if the model is not setting , using default514 retrieval_model = dataset.retrieval_model or default_retrieval_model515 516 if dataset.indexing_technique == "economy":517 # use keyword table query518 documents = RetrievalService.retrieve(519 retrieval_method="keyword_search", dataset_id=dataset.id, query=query, top_k=top_k520 )521 if documents:522 all_documents.extend(documents)523 else:524 if top_k > 0:525 # retrieval source526 documents = RetrievalService.retrieve(527 retrieval_method=retrieval_model["search_method"],528 dataset_id=dataset.id,529 query=query,530 top_k=retrieval_model.get("top_k") or 2,531 score_threshold=retrieval_model.get("score_threshold", 0.0)532 if retrieval_model["score_threshold_enabled"]533 else 0.0,534 reranking_model=retrieval_model.get("reranking_model", None)535 if retrieval_model["reranking_enable"]536 else None,537 reranking_mode=retrieval_model.get("reranking_mode") or "reranking_model",538 weights=retrieval_model.get("weights", None),539 )540 541 all_documents.extend(documents)542 543 def to_dataset_retriever_tool(544 self,545 tenant_id: str,546 dataset_ids: list[str],547 retrieve_config: DatasetRetrieveConfigEntity,548 return_resource: bool,549 invoke_from: InvokeFrom,550 hit_callback: DatasetIndexToolCallbackHandler,551 ) -> Optional[list[DatasetRetrieverBaseTool]]:552 """553 A dataset tool is a tool that can be used to retrieve information from a dataset554 :param tenant_id: tenant id555 :param dataset_ids: dataset ids556 :param retrieve_config: retrieve config557 :param return_resource: return resource558 :param invoke_from: invoke from559 :param hit_callback: hit callback560 """561 tools = []562 available_datasets = []563 for dataset_id in dataset_ids:564 # get dataset from dataset id565 dataset = db.session.query(Dataset).filter(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()566 567 # pass if dataset is not available568 if not dataset:569 continue570 571 # pass if dataset is not available572 if dataset and dataset.provider != "external" and dataset.available_document_count == 0:573 continue574 575 available_datasets.append(dataset)576 577 if retrieve_config.retrieve_strategy == DatasetRetrieveConfigEntity.RetrieveStrategy.SINGLE:578 # get retrieval model config579 default_retrieval_model = {580 "search_method": RetrievalMethod.SEMANTIC_SEARCH.value,581 "reranking_enable": False,582 "reranking_model": {"reranking_provider_name": "", "reranking_model_name": ""},583 "top_k": 2,584 "score_threshold_enabled": False,585 }586 587 for dataset in available_datasets:588 retrieval_model_config = dataset.retrieval_model or default_retrieval_model589 590 # get top k591 top_k = retrieval_model_config["top_k"]592 593 # get score threshold594 score_threshold = None595 score_threshold_enabled = retrieval_model_config.get("score_threshold_enabled")596 if score_threshold_enabled:597 score_threshold = retrieval_model_config.get("score_threshold")598 599 tool = DatasetRetrieverTool.from_dataset(600 dataset=dataset,601 top_k=top_k,602 score_threshold=score_threshold,603 hit_callbacks=[hit_callback],604 return_resource=return_resource,605 retriever_from=invoke_from.to_source(),606 )607 608 tools.append(tool)609 elif retrieve_config.retrieve_strategy == DatasetRetrieveConfigEntity.RetrieveStrategy.MULTIPLE:610 tool = DatasetMultiRetrieverTool.from_dataset(611 dataset_ids=[dataset.id for dataset in available_datasets],612 tenant_id=tenant_id,613 top_k=retrieve_config.top_k or 2,614 score_threshold=retrieve_config.score_threshold,615 hit_callbacks=[hit_callback],616 return_resource=return_resource,617 retriever_from=invoke_from.to_source(),618 reranking_provider_name=retrieve_config.reranking_model.get("reranking_provider_name"),619 reranking_model_name=retrieve_config.reranking_model.get("reranking_model_name"),620 )621 622 tools.append(tool)623 624 return tools625 626 def calculate_keyword_score(self, query: str, documents: list[Document], top_k: int) -> list[Document]:627 """628 Calculate keywords scores629 :param query: search query630 :param documents: documents for reranking631 632 :return:633 """634 keyword_table_handler = JiebaKeywordTableHandler()635 query_keywords = keyword_table_handler.extract_keywords(query, None)636 documents_keywords = []637 for document in documents:638 # get the document keywords639 document_keywords = keyword_table_handler.extract_keywords(document.page_content, None)640 document.metadata["keywords"] = document_keywords641 documents_keywords.append(document_keywords)642 643 # Counter query keywords(TF)644 query_keyword_counts = Counter(query_keywords)645 646 # total documents647 total_documents = len(documents)648 649 # calculate all documents' keywords IDF650 all_keywords = set()651 for document_keywords in documents_keywords:652 all_keywords.update(document_keywords)653 654 keyword_idf = {}655 for keyword in all_keywords:656 # calculate include query keywords' documents657 doc_count_containing_keyword = sum(1 for doc_keywords in documents_keywords if keyword in doc_keywords)658 # IDF659 keyword_idf[keyword] = math.log((1 + total_documents) / (1 + doc_count_containing_keyword)) + 1660 661 query_tfidf = {}662 663 for keyword, count in query_keyword_counts.items():664 tf = count665 idf = keyword_idf.get(keyword, 0)666 query_tfidf[keyword] = tf * idf667 668 # calculate all documents' TF-IDF669 documents_tfidf = []670 for document_keywords in documents_keywords:671 document_keyword_counts = Counter(document_keywords)672 document_tfidf = {}673 for keyword, count in document_keyword_counts.items():674 tf = count675 idf = keyword_idf.get(keyword, 0)676 document_tfidf[keyword] = tf * idf677 documents_tfidf.append(document_tfidf)678 679 def cosine_similarity(vec1, vec2):680 intersection = set(vec1.keys()) & set(vec2.keys())681 numerator = sum(vec1[x] * vec2[x] for x in intersection)682 683 sum1 = sum(vec1[x] ** 2 for x in vec1)684 sum2 = sum(vec2[x] ** 2 for x in vec2)685 denominator = math.sqrt(sum1) * math.sqrt(sum2)686 687 if not denominator:688 return 0.0689 else:690 return float(numerator) / denominator691 692 similarities = []693 for document_tfidf in documents_tfidf:694 similarity = cosine_similarity(query_tfidf, document_tfidf)695 similarities.append(similarity)696 697 for document, score in zip(documents, similarities):698 # format document699 document.metadata["score"] = score700 documents = sorted(documents, key=lambda x: x.metadata["score"], reverse=True)701 return documents[:top_k] if top_k else documents702 703 def calculate_vector_score(704 self, all_documents: list[Document], top_k: int, score_threshold: float705 ) -> list[Document]:706 filter_documents = []707 for document in all_documents:708 if score_threshold is None or document.metadata["score"] >= score_threshold:709 filter_documents.append(document)710 711 if not filter_documents:712 return []713 filter_documents = sorted(filter_documents, key=lambda x: x.metadata["score"], reverse=True)714 return filter_documents[:top_k] if top_k else filter_documents715