return f'{md_filter_key}{md_filter_operator}"{md_value}"'
        def search_data() -> Dict[str, Any]:
            request = QueryRequest(
                table_name=self.config.table_name,
                namespace=self.config.namespace,
                vector=embedding,
                include_vector=True,
                output_fields=self.config.output_fields,
                filter=generate_filter_query(),
                top_k=k,
            )
            query_result = self.ha3_engine_client.query(request)
            return json.loads(query_result.body)
        from alibabacloud_ha3engine_vector.models import QueryRequest
        try:
            json_response = search_data()
            if (
                "errorCode" in json_response
                and "errorMsg" in json_response
                and len(json_response["errorMsg"]) > 0
            ):
                logger.error(
                    f"query {self.config.endpoint} {self.config.instance_id} "
                    f"failed:{json_response['errorMsg']}."
                )
            else:
                return json_response
        except Exception as e:
            logger.error(
                f"query instance endpoint:{self.config.endpoint} "
                f"instance_id:{self.config.instance_id} failed.",
                e,
            )
        return {}
[docs]    def create_results(self, json_result: Dict[str, Any]) -> List[Document]:
        """Assemble documents."""
        items = json_result["result"]
        query_result_list: List[Document] = []
        for item in items:
            if (
                "fields" not in item
                or self.config.field_name_mapping["document"] not in item["fields"]
            ):
                query_result_list.append(Document())
            else:
                fields = item["fields"]
                query_result_list.append(
                    Document(