Source code for langchain_community.query_constructors.deeplake
"""将内部查询语言转换为有效的Chroma查询的逻辑。"""
from typing import Tuple, Union
from langchain_core.structured_query import (
Comparator,
Comparison,
Operation,
Operator,
StructuredQuery,
Visitor,
)
COMPARATOR_TO_TQL = {
Comparator.EQ: "==",
Comparator.GT: ">",
Comparator.GTE: ">=",
Comparator.LT: "<",
Comparator.LTE: "<=",
}
OPERATOR_TO_TQL = {
Operator.AND: "and",
Operator.OR: "or",
Operator.NOT: "NOT",
}
[docs]def can_cast_to_float(string: str) -> bool:
"""检查一个字符串是否可以转换为浮点数。"""
try:
float(string)
return True
except ValueError:
return False
[docs]class DeepLakeTranslator(Visitor):
"""将`DeepLake`内部查询语言元素翻译为有效的过滤器。"""
allowed_operators = [Operator.AND, Operator.OR, Operator.NOT]
"""允许的逻辑运算符的子集。"""
allowed_comparators = [
Comparator.EQ,
Comparator.GT,
Comparator.GTE,
Comparator.LT,
Comparator.LTE,
]
"""允许的逻辑比较器子集。"""
def _format_func(self, func: Union[Operator, Comparator]) -> str:
self._validate_func(func)
if isinstance(func, Operator):
value = OPERATOR_TO_TQL[func.value] # type: ignore
elif isinstance(func, Comparator):
value = COMPARATOR_TO_TQL[func.value] # type: ignore
return f"{value}"
[docs] def visit_operation(self, operation: Operation) -> str:
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)
values = comparison.value
if isinstance(values, list):
tql = []
for value in values:
comparison.value = value
tql.append(self.visit_comparison(comparison))
return "(" + (" or ").join(tql) + ")"
if not can_cast_to_float(comparison.value):
values = f"'{values}'"
return f"metadata['{comparison.attribute}'] {comparator} {values}"
[docs] def visit_structured_query(
self, structured_query: StructuredQuery
) -> Tuple[str, dict]:
if structured_query.filter is None:
kwargs = {}
else:
tqL = f"SELECT * WHERE {structured_query.filter.accept(self)}"
kwargs = {"tql": tqL}
return structured_query.query, kwargs