langchain_community.query_constructors.milvus

"""将内部查询语言转换为有效的Milvus查询的逻辑。"""
from typing import Tuple, Union

from langchain_core.structured_query import (
    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]def process_value(value: Union[int, float, str], comparator: Comparator) -> str:
    """将一个值转换为字符串，并在其为字符串时添加双引号。

在涉及字符串的比较器中需要这样做。

参数:
    value: 要转换的值。
    comparator: 比较器。

返回:
    转换后的值作为字符串。
"""
    #
    if isinstance(value, str):
        if comparator is Comparator.LIKE:
            # If the comparator is LIKE, add a percent sign after it for prefix matching
            # and add double quotes
            return f'"{value}%"'
        else:
            # If the value is already a string, add double quotes
            return f'"{value}"'
    else:
        # If the value is not a string, convert it to a string without double quotes
        return str(value)


[docs]class MilvusTranslator(Visitor):
    """将Milvus内部查询语言元素翻译为有效的过滤器。"""

    """允许的逻辑运算符的子集。"""
    allowed_operators = [Operator.AND, Operator.NOT, Operator.OR]

    """允许的逻辑比较器子集。"""
    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.value
        if isinstance(func, Comparator):
            value = COMPARATOR_TO_BER[func]
        return f"{value}"

[docs]    def visit_operation(self, operation: Operation) -> str:
        if operation.operator in UNARY_OPERATORS and len(operation.arguments) == 1:
            operator = self._format_func(operation.operator)
            return operator + "(" + operation.arguments[0].accept(self) + ")"
        elif operation.operator in UNARY_OPERATORS:
            raise ValueError(
                f'"{operation.operator.value}" can have only one argument in Milvus'
            )
        else:
            args = [arg.accept(self) for arg in operation.arguments]
            operator = self._format_func(operation.operator)
            return "(" + (" " + operator + " ").join(args) + ")"

[docs]    def visit_comparison(self, comparison: Comparison) -> str:
        comparator = self._format_func(comparison.comparator)
        processed_value = process_value(comparison.value, comparison.comparator)
        attribute = comparison.attribute

        return "( " + attribute + " " + comparator + " " + processed_value + " )"

[docs]    def visit_structured_query(
        self, structured_query: StructuredQuery
    ) -> Tuple[str, dict]:
        if structured_query.filter is None:
            kwargs = {}
        else:
            kwargs = {"expr": structured_query.filter.accept(self)}
        return structured_query.query, kwargs
Source code for langchain_community.query_constructors.milvus
