Source code for langchain_experimental.sql.vector_sql
"""Vector SQL Database Chain Retriever"""
from __future__ import annotations
from typing import Any, Dict, List, Optional, Union
from langchain.callbacks.manager import CallbackManagerForChainRun
from langchain.chains.llm import LLMChain
from langchain.chains.sql_database.prompt import PROMPT, SQL_PROMPTS
from langchain.prompts.prompt import PromptTemplate
from langchain.schema import BaseOutputParser, BasePromptTemplate
from langchain.schema.embeddings import Embeddings
from langchain.schema.language_model import BaseLanguageModel
from langchain.tools.sql_database.prompt import QUERY_CHECKER
from langchain.utilities.sql_database import SQLDatabase
from langchain_experimental.sql.base import INTERMEDIATE_STEPS_KEY, SQLDatabaseChain
[docs]class VectorSQLOutputParser(BaseOutputParser[str]):
    """Output Parser for Vector SQL
    1. finds for `NeuralArray()` and replace it with the embedding
    2. finds for `DISTANCE()` and replace it with the distance name in backend SQL
    """
    model: Embeddings
    """Embedding model to extract embedding for entity"""
    distance_func_name: str = "distance"
    """Distance name for Vector SQL"""
    class Config:
        arbitrary_types_allowed = 1
    @property
    def _type(self) -> str:
        return "vector_sql_parser"
[docs]    @classmethod
    def from_embeddings(
        cls, model: Embeddings, distance_func_name: str = "distance", **kwargs: Any
    ) -> BaseOutputParser:
        return cls(model=model, distance_func_name=distance_func_name, **kwargs)
[docs]    def parse(self, text: str) -> str:
        text = text.strip()
        start = text.find("NeuralArray(")