Translating Natural Language to BlendSQL

nl_to_blendsql

Takes a natural language question, and attempts to parse BlendSQL representation for answering against a databse.

Parameters:

Name	Type	Description	Default
`question`	`str`	The natural language question to parse	required
`db`	`Database`	Database to use in translating	required
`model`	`Model`	BlendSQL model to use in translating the question	required
`ingredients`	`Optional[Collection[Type[Ingredient]]]`	Which ingredients to treat as valid in the output parse. Only these ingredient descriptions are included in the system prompt.	required
`few_shot_examples`	`Union[str, FewShot]`	String prompt introducing few shot nl-to-blendsql examples.	`''`
`args`	`Optional[NLtoBlendSQLArgs]`	Optional NLtoBlendSQLArgs object, containing additional parameters.	`None`
`verbose`	`bool`	Boolean defining whether to run in logger mode	`False`

Returns:

Name	Type	Description
`ret_prediction`	`str`	Final BlendSQL query prediction

Examples:

from blendsql import LLMMap, LLMQA
from blendsql.models import TransformersLLM, OllamaLLM
from blendsql.nl_to_blendsql import nl_to_blendsql, NLtoBlendSQLArgs
from blendsql.db import SQLite
from blendsql.utils import fetch_from_hub
from blendsql.prompts import FewShot

db = SQLite(
    fetch_from_hub("1884_New_Zealand_rugby_union_tour_of_New_South_Wales_1.db")
)
parser_model = OllamaLLM("phi3", caching=False)
correction_model = TransformersLLM("Qwen/Qwen1.5-0.5B")

ingredients = {LLMMap, LLMQA}
filtered_few_shot = FewShot.hybridqa.filter(ingredients)

blendsql = nl_to_blendsql(
    "What was the result of the game played 120 miles west of Sydney?",
    db=db,
    model=parser_model,
    correction_model=correction_model,
    ingredients=ingredients,
    few_shot_examples=filtered_few_shot,
    verbose=True,
    args=NLtoBlendSQLArgs(
        max_grammar_corrections=5,
        use_tables=["w"],
        include_db_content_tables=["w"],
        num_serialized_rows=3,
        use_bridge_encoder=True,
    ),
)

Source code in blendsql/nl_to_blendsql/nl_to_blendsql.py

def nl_to_blendsql(
    question: str,
    db: Database,
    model: Model,
    ingredients: Optional[Collection[Type[Ingredient]]],
    correction_model: Optional[Model] = None,
    few_shot_examples: Union[str, FewShot] = "",
    args: Optional[NLtoBlendSQLArgs] = None,
    verbose: bool = False,
) -> str:
    """Takes a natural language question, and attempts to parse BlendSQL representation for answering against a databse.

    Args:
        question: The natural language question to parse
        db: Database to use in translating
        model: BlendSQL model to use in translating the question
        ingredients: Which ingredients to treat as valid in the output parse.
            Only these ingredient descriptions are included in the system prompt.
        few_shot_examples: String prompt introducing few shot nl-to-blendsql examples.
        args: Optional NLtoBlendSQLArgs object, containing additional parameters.
        verbose: Boolean defining whether to run in logger mode

    Returns:
        ret_prediction: Final BlendSQL query prediction

    Examples:
        ```python
        from blendsql import LLMMap, LLMQA
        from blendsql.models import TransformersLLM, OllamaLLM
        from blendsql.nl_to_blendsql import nl_to_blendsql, NLtoBlendSQLArgs
        from blendsql.db import SQLite
        from blendsql.utils import fetch_from_hub
        from blendsql.prompts import FewShot

        db = SQLite(
            fetch_from_hub("1884_New_Zealand_rugby_union_tour_of_New_South_Wales_1.db")
        )
        parser_model = OllamaLLM("phi3", caching=False)
        correction_model = TransformersLLM("Qwen/Qwen1.5-0.5B")

        ingredients = {LLMMap, LLMQA}
        filtered_few_shot = FewShot.hybridqa.filter(ingredients)

        blendsql = nl_to_blendsql(
            "What was the result of the game played 120 miles west of Sydney?",
            db=db,
            model=parser_model,
            correction_model=correction_model,
            ingredients=ingredients,
            few_shot_examples=filtered_few_shot,
            verbose=True,
            args=NLtoBlendSQLArgs(
                max_grammar_corrections=5,
                use_tables=["w"],
                include_db_content_tables=["w"],
                num_serialized_rows=3,
                use_bridge_encoder=True,
            ),
        )
        ```
    """
    if verbose:
        logger.setLevel(logging.DEBUG)
    else:
        logger.setLevel(logging.ERROR)
    if args is None:
        args = NLtoBlendSQLArgs()
    if correction_model is None:
        correction_model = model
    parser: EarleyParser = load_cfg_parser(ingredients)
    system_prompt: str = create_system_prompt(
        ingredients=ingredients, few_shot_examples=few_shot_examples
    )
    serialized_db = db.to_serialized(
        use_tables=args.use_tables,
        num_rows=args.num_serialized_rows,
        include_content=args.include_db_content_tables,
        use_bridge_encoder=args.use_bridge_encoder,
        question=question,
    )
    if args.max_grammar_corrections == 0:
        return model.predict(
            program=ParserProgram,
            system_prompt=system_prompt,
            question=question,
            serialized_db=serialized_db,
            stream=verbose,
        )
    num_correction_left = args.max_grammar_corrections
    partial_program_prediction = ""
    ret_prediction, initial_prediction = None, None
    while num_correction_left > 0 and ret_prediction is None:
        residual_program_prediction = model.predict(
            program=ParserProgram,
            system_prompt=system_prompt,
            question=question,
            serialized_db=serialized_db,
            stream=verbose,
        )

        # if the prediction is empty, return the initial prediction
        if initial_prediction is None:
            initial_prediction = residual_program_prediction
        program_prediction = (
            partial_program_prediction + " " + residual_program_prediction
        )

        if validate_program(program_prediction, parser):
            ret_prediction = program_prediction
            continue

        # find the max score from a list of score
        prefix, candidates, pos_in_stream = obtain_correction_pairs(
            program_prediction, parser
        )
        # candidates = [i for i in candidates if i.strip() != ""]
        if len(candidates) == 0:
            logger.debug(
                Fore.LIGHTMAGENTA_EX + "No correction pairs found" + Fore.RESET
            )
            return prefix
        elif len(candidates) == 1:
            # If we only have 1 candidate, no need to call LLM
            selected_candidate = candidates.pop()
        else:
            # Generate the continuation candidate with the highest probability
            selected_candidate = correction_model.predict(
                program=CorrectionProgram,
                system_prompt=system_prompt,
                question=question,
                serialized_db=serialized_db,
                partial_completion=prefix,
                candidates=candidates,
            )

        # Try to use our selected candidate in a few ways
        # 1) Insert our selection into the index where the error occurred, and add left/right context
        #   Example: SELECT a b FROM table -> SELECT a, b FROM table
        inserted_candidate = (
            prefix + selected_candidate + program_prediction[pos_in_stream:]
        )
        if validate_program(inserted_candidate, parser):
            ret_prediction = inserted_candidate
            continue
        # 2) If rest of our query is also broken, we just keep up to the prefix + candidate
        partial_program_prediction = prefix + selected_candidate
        for p in {inserted_candidate, partial_program_prediction}:
            if validate_program(p, parser):
                ret_prediction = p

        num_correction_left -= 1

    if ret_prediction is None:
        logger.debug(
            Fore.RED
            + f"cannot find a valid prediction after {args.max_grammar_corrections} retries"
            + Fore.RESET
        )
        ret_prediction = initial_prediction
    ret_prediction = post_process_blendsql(
        ret_prediction, db, use_tables=args.use_tables
    )
    logger.debug(Fore.GREEN + ret_prediction + Fore.RESET)
    return ret_prediction

NLtoBlendSQLArgs

Source code in blendsql/nl_to_blendsql/args.py

@dataclass
class NLtoBlendSQLArgs:
    max_grammar_corrections: int = field(
        default=0,
        metadata={
            "help": "Optional int defining maximum CFG-guided correction steps to be taken. This is based on the method in https://arxiv.org/pdf/2305.19234."
        },
    )

    include_db_content_tables: Union[List[str], str] = field(
        default="all",
        metadata={
            "help": "Which database tables to add `num_serialized_rows` worth of content for in serialization."
        },
    )

    num_serialized_rows: int = field(
        default=3,
        metadata={
            "help": "How many example rows to include in serialization of database"
        },
    )

    use_tables: Collection[str] = field(
        default=None,
        metadata={"help": "Collection of tables to use in serialization to string"},
    )

    use_bridge_encoder: bool = field(
        default=True,
        metadata={
            "help": "Whether to use Bridge Content Encoder during input serialization"
        },
    )

Grammar-Based Correction

If you use the grammar correction feature of BlendSQL, please cite the original grammar prompting paper below.

@article{wang2024grammar,
  title={Grammar prompting for domain-specific language generation with large language models},
  author={Wang, Bailin and Wang, Zi and Wang, Xuezhi and Cao, Yuan and A Saurous, Rif and Kim, Yoon},
  journal={Advances in Neural Information Processing Systems},
  volume={36},
  year={2024}
}

FewShot

A collection of few-shot examples, with some utility functions for easy manipulation.

Examples:

from blendsql import LLMMap, LLMQA
from blendsql.prompts import FewShot, Examples
# Fetch the examples for HybridQA
fewshot_prompts: Examples = FewShot.hybridqa
print(f"We have {len(fewshot_prompts)} examples")
# We can select a subset by indexing
first_three_examples = fewshot_prompts[:3]
# Additionally, we can filter to keep only those examples using specified ingredients
filtered_fewshot = fewshot_prompts.filter({LLMQA, LLMMap})

Source code in blendsql/prompts/_prompts.py

@dataclass
class FewShot:
    """A collection of few-shot examples, with some utility functions for easy manipulation.

    Examples:
        ```python
        from blendsql import LLMMap, LLMQA
        from blendsql.prompts import FewShot, Examples
        # Fetch the examples for HybridQA
        fewshot_prompts: Examples = FewShot.hybridqa
        print(f"We have {len(fewshot_prompts)} examples")
        # We can select a subset by indexing
        first_three_examples = fewshot_prompts[:3]
        # Additionally, we can filter to keep only those examples using specified ingredients
        filtered_fewshot = fewshot_prompts.filter({LLMQA, LLMMap})
        ```
    """

    hybridqa = Examples(open(Path(__file__).parent / "./few_shot/hybridqa.txt").read())

Examples

Class for holding few-shot examples.

Examples:

from blendsql.prompts import FewShot, Examples
fewshot_prompts: Examples = FewShot.hybridqa
print(fewshot_prompts[:2])

Examples:

This is the first example

---

This is the second example

Source code in blendsql/prompts/_prompts.py

@attrs
class Examples:
    """Class for holding few-shot examples.

    Examples:
        ```python
        from blendsql.prompts import FewShot, Examples
        fewshot_prompts: Examples = FewShot.hybridqa
        print(fewshot_prompts[:2])
        ```
        ```text
        Examples:

        This is the first example

        ---

        This is the second example
        ```
    """

    data: str = attrib()

    split_data: List[str] = attrib(init=False)

    def __attrs_post_init__(self):
        self.data = self.data.strip()
        self.split_data: list = self.data.split("---")

    def __getitem__(self, subscript):
        newline = (
            "\n\n"
            if (isinstance(subscript, int) and subscript == 0)
            or (isinstance(subscript, slice) and subscript.start in {0, None})
            else ""
        )
        return "Examples:" + newline + "---".join(self.split_data[subscript])

    def __repr__(self):
        return "Examples:\n\n" + self.data

    def __str__(self):
        return "Examples:\n\n" + self.data

    def __len__(self):
        return len(self.split_data)

    def is_valid_query(self, query: str, ingredient_names: Set[str]) -> bool:
        """Checks if a given query is valid given the ingredient_names passed.
        A query is invalid if it includes an ingredient that is not specified in ingredient_names.
        """
        stack = [query]
        while len(stack) > 0:
            for res, _start, _end in peg_grammar.scanString(stack.pop()):
                if res.get("function").upper() not in ingredient_names:
                    return False
                for arg in res.get("args"):
                    stack.append(arg)
        return True

    def filter(self, ingredients: Iterable[Type[Ingredient]]) -> "Examples":
        """Retrieve only those prompts which do not include any ingredient not specified in `ingredients`."""
        ingredient_names: Set[str] = {
            ingredient.__name__.upper() for ingredient in ingredients
        }
        filtered_split_data = []
        for d in self.split_data:
            if self.is_valid_query(d, ingredient_names=ingredient_names):
                filtered_split_data.append(d)
        return Examples("---".join(filtered_split_data))