Self-querying with PGVector

[djo10]

Checked

• I searched existing ideas and did not find a similar one
• I added a very descriptive title
• I've clearly described the feature request and motivation for it

Feature request

Self querying feature lacks support for pgvector.

https://python.langchain.com/docs/modules/data_connection/retrievers/self_query
https://python.langchain.com/docs/integrations/retrievers/self_query

Motivation

PGVector support for Self-querying

Proposal (If applicable)

step 1. is accepting improvements for pgvector metadata filtering - PR #17067 (issue #17064 )

step 2. PGVector Visitor to translate StructuredQuery to filter payload that will be later handled by changes from step 1

from typing import Dict, Tuple, Union

from langchain.chains.query_constructor.ir import (
    Comparator,
    Comparison,
    Operation,
    Operator,
    StructuredQuery,
    Visitor,
)


class PGVectorTranslator(Visitor):
    """Translate PGVector internal query language elements to valid filters."""

    allowed_comparators = (
        Comparator.EQ,
        Comparator.NE,
        Comparator.LT,
        Comparator.LTE,
        Comparator.GT,
        Comparator.GTE,
        Comparator.IN,
        Comparator.NIN,
    )
    """Subset of allowed logical comparators."""
    allowed_operators = (Operator.AND, Operator.OR, Operator.NOT)
    """Subset of allowed logical operators."""

    def _format_func(self, func: Union[Operator, Comparator]) -> str:
        self._validate_func(func)
        return func.value

    def visit_operation(self, operation: Operation) -> Dict:
        args = [arg.accept(self) for arg in operation.arguments]
        return {self._format_func(operation.operator): args}

    def visit_comparison(self, comparison: Comparison) -> Dict:
        if comparison.comparator in (Comparator.IN, Comparator.NIN) and not isinstance(
            comparison.value, list
        ):
            comparison.value = [comparison.value]
        return {
            comparison.attribute: {
                self._format_func(comparison.comparator): comparison.value
            }
        }

    def visit_structured_query(
        self, structured_query: StructuredQuery
    ) -> Tuple[str, dict]:
        if structured_query.filter is None:
            kwargs = {}
        else:
            kwargs = {"filter": structured_query.filter.accept(self)}
        return structured_query.query, kwargs