Source code for langchain_community.query_constructors.milvus
"""Logic for converting internal query language to a valid Milvus query."""fromtypingimportTuple,Unionfromlangchain_core.structured_queryimport(Comparator,Comparison,Operation,Operator,StructuredQuery,Visitor,)COMPARATOR_TO_BER={Comparator.EQ:"==",Comparator.GT:">",Comparator.GTE:">=",Comparator.LT:"<",Comparator.LTE:"<=",Comparator.IN:"in",Comparator.LIKE:"like",}UNARY_OPERATORS=[Operator.NOT]
[docs]defprocess_value(value:Union[int,float,str],comparator:Comparator)->str:"""Convert a value to a string and add double quotes if it is a string. It required for comparators involving strings. Args: value: The value to convert. comparator: The comparator. Returns: The converted value as a string. """#ifisinstance(value,str):ifcomparatorisComparator.LIKE:# If the comparator is LIKE, add a percent sign after it for prefix matching# and add double quotesreturnf'"{value}%"'else:# If the value is already a string, add double quotesreturnf'"{value}"'else:# If the value is not a string, convert it to a string without double quotesreturnstr(value)
[docs]classMilvusTranslator(Visitor):"""Translate Milvus internal query language elements to valid filters.""""""Subset of allowed logical operators."""allowed_operators=[Operator.AND,Operator.NOT,Operator.OR]"""Subset of allowed logical comparators."""allowed_comparators=[Comparator.EQ,Comparator.GT,Comparator.GTE,Comparator.LT,Comparator.LTE,Comparator.IN,Comparator.LIKE,]def_format_func(self,func:Union[Operator,Comparator])->str:self._validate_func(func)value=func.valueifisinstance(func,Comparator):value=COMPARATOR_TO_BER[func]returnf"{value}"
[docs]defvisit_operation(self,operation:Operation)->str:ifoperation.operatorinUNARY_OPERATORSandlen(operation.arguments)==1:operator=self._format_func(operation.operator)returnoperator+"("+operation.arguments[0].accept(self)+")"elifoperation.operatorinUNARY_OPERATORS:raiseValueError(f'"{operation.operator.value}" can have only one argument in Milvus')else:args=[arg.accept(self)forarginoperation.arguments]operator=self._format_func(operation.operator)return"("+(" "+operator+" ").join(args)+")"