)
                return ""
            md_filter_key = expr[0].strip()
            md_filter_operator = expr[1].strip()
            if isinstance(md_value, numbers.Number):
                return f"{md_filter_key} {md_filter_operator} {md_value}"
            return f'{md_filter_key}{md_filter_operator}"{md_value}"'
        def search_data(single_query_str: str) -> Dict[str, Any]:
            search_query = models.SearchQuery(query=single_query_str)
            search_request = models.SearchRequestModel(
                self.options_headers, search_query
            )
            return json.loads(self.ha3EngineClient.search(search_request).body)
        from alibabacloud_ha3engine import models
        try:
            query_str = generate_embedding_query()
            json_response = search_data(query_str)
            if len(json_response["errors"]) != 0:
                logger.error(
                    f"query {self.config.endpoint} {self.config.instance_id} "
                    f"errors:{json_response['errors']} failed."
                )
            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]:
        items = json_result["result"]["items"]
        query_result_list: List[Document] = []
        for item in items:
            fields = item["fields"]
            query_result_list.append(
                Document(
                    page_content=fields[self.config.field_name_mapping["document"]],
                    metadata=create_metadata(fields),
                )
            )
        return query_result_list