Feature(ai-service): refine ask pipeline v2 (#86)

* allow visualization using one file

* remove default argparse option value

* update setup instructions

* add anthropic model pricing

* add container:ubuntu

* undo

* update

* remove unused files

* add logging for ask api results

* fix postprocessor for ask details

* fix bugs for tests and eval and restructure eval-ask

* remove unused code

* update setup instructions

* resolve conflict

* WIP: add eval for sample data

* remove unused env var

* remove unused code

* allow evalaute sample data

* refine eval sample data output

* update ask eval for sampledata output

* fix bug

* remove duplicates of ask generation results and return original sql instead of no sql in sql details pipeline
This commit is contained in:
Chih-Yu Yeh
2024-04-09 19:04:38 +08:00
committed by GitHub
parent f8f82caaf7
commit d24cdc1090
12 changed files with 451 additions and 38 deletions
+44 -1
View File
@@ -245,7 +245,6 @@ if __name__ == "__main__":
help="Whether to use hard questions for evaluation. Default is False.",
)
parser.add_argument
args = parser.parse_args()
PREDICTION_RESULTS_FILE = args.input_file
@@ -318,6 +317,50 @@ if __name__ == "__main__":
if not Path("./outputs/ask").exists():
Path("./outputs/ask").mkdir(parents=True)
if ENABLE_SEMANTIC_DESCRIPTION:
if os.path.exists(f"./src/eval/data/{DATASET_NAME}_with_semantic_mdl.json"):
print(f"Use the existed {DATASET_NAME}_with_semantic_mdl.json...\n")
else:
print(
f"Generating semantic description for the {DATASET_NAME} dataset...\n"
)
semantics_service = SemanticsService(
pipelines={
"generate_description": description.Generation(),
}
)
with open(f"./src/eval/data/{DATASET_NAME}_mdl.json", "r") as f:
mdl_data = json.load(f)
for model in tqdm(mdl_data["models"]):
semantic_desc = semantics_service.generate_description(
GenerateDescriptionRequest(
mdl=model,
model=model["name"],
identifier="model",
)
)
model["properties"]["description"] = semantic_desc.description
model["properties"]["display_name"] = semantic_desc.display_name
for column in model["columns"]:
semantic_desc = semantics_service.generate_description(
GenerateDescriptionRequest(
mdl=model,
model=model["name"],
identifier="column@" + column["name"],
)
)
column["properties"]["description"] = semantic_desc.description
column["properties"]["display_name"] = semantic_desc.display_name
with open(
f"./src/eval/data/{DATASET_NAME}_with_semantic_mdl.json", "w"
) as f:
json.dump(mdl_data, f)
if not Path("./outputs").exists():
Path("./outputs").mkdir()
print(f"Running ask pipeline evaluation for the {DATASET_NAME} dataset...\n")
if (
PREDICTION_RESULTS_FILE
@@ -0,0 +1,273 @@
import argparse
import json
import os
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from pathlib import Path
import requests
from tqdm import tqdm
from src.pipelines.ask.components.document_store import init_document_store
from src.pipelines.ask.components.embedder import init_embedder
from src.pipelines.ask.components.generator import init_generator
from src.pipelines.ask.components.retriever import init_retriever
from src.pipelines.ask.generation_pipeline import Generation
from src.pipelines.ask.indexing_pipeline import Indexing
from src.pipelines.ask.retrieval_pipeline import Retrieval
from src.pipelines.ask.sql_correction_pipeline import SQLCorrection
from src.utils import load_env_vars
load_env_vars()
def get_mdl_from_wren_engine():
response = requests.get(
f'{os.getenv("WREN_ENGINE_ENDPOINT")}/v1/mdl',
)
assert response.status_code == 200
return response.json()
def process_item(query: str):
retrieval_start = time.perf_counter()
retrieval_result = retrieval_pipeline.run(query)
documents = retrieval_result["post_processor"]["documents"]
retrieval_end = time.perf_counter()
valid_generation_results = []
invalid_generation_results = []
text_to_sql_generation_start = time.perf_counter()
text_to_sql_generation_results = generation_pipeline.run(
query,
contexts=documents,
)
text_to_sql_generation_end = time.perf_counter()
text_to_sql_generation_time_cost = (
text_to_sql_generation_end - text_to_sql_generation_start
)
if text_to_sql_generation_results["post_processor"]["valid_generation_results"]:
valid_generation_results += text_to_sql_generation_results["post_processor"][
"valid_generation_results"
]
sql_correction_results = None
sql_correction_generation_time_cost = 0
if text_to_sql_generation_results["post_processor"]["invalid_generation_results"]:
sql_correction_generation_start = time.perf_counter()
sql_correction_results = sql_correction_pipeline.run(
contexts=documents,
invalid_generation_results=text_to_sql_generation_results["post_processor"][
"invalid_generation_results"
],
)
sql_correction_generation_end = time.perf_counter()
sql_correction_generation_time_cost = (
sql_correction_generation_end - sql_correction_generation_start
)
valid_generation_results += sql_correction_results["post_processor"][
"valid_generation_results"
]
invalid_generation_results += sql_correction_results["post_processor"][
"invalid_generation_results"
]
return {
"question": query,
"valid_generation_results": valid_generation_results,
"invalid_generation_results": invalid_generation_results,
"metadata": {
"generation": {
"text_to_sql": text_to_sql_generation_results["text_to_sql_generator"][
"meta"
][0],
"sql_correction": (
sql_correction_results["sql_correction_generator"]["meta"][0]
if sql_correction_results
else []
),
},
"latency": {
"retrieval": retrieval_end - retrieval_start,
"generation": {
"text_to_sql": text_to_sql_generation_time_cost,
"sql_correction": sql_correction_generation_time_cost,
},
},
},
}
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Evaluate the ask pipeline using the sample dataset: music, ecommerce, nba"
)
parser.add_argument(
"--dataset",
type=str,
default="music",
choices=["music", "ecommerce", "nba"],
)
args = parser.parse_args()
SAMPLE_DATASET_NAME = args.dataset
SAMPLE_DATASET_QUESTIONS = {
"nba": [
"How many assists were made by each player in each game?",
"How many blocks were made by each player in each game?",
"How many field goal attempts were made by each player in each game?",
"How many field goals were made by each player in each game?",
"How many games went into overtime, and which teams were involved?",
"How many games were played by each team in the season?",
"How many personal fouls were committed by each player in each game?",
"How many points were scored by each player in each game, and what are their full names?",
"How many points were scored by each team in each quarter of the game?",
"How many rebounds were grabbed by each player in each game?",
"How many steals were made by each player in each game?",
"How many three-point attempts were made by each player in each game?",
"How many three-pointers were made by each player in each game?",
"How many turnovers were committed by each player in each game?",
"What is the differences in turnover rates between teams with high and low average scores?",
"Which players participated in each game along with their teams and ages?",
"Which players recorded the highest number of triple-doubles throughout the season?",
"Which teams exhibited the most consistent performance in terms of points scored per quarter?",
"Which teams had the highest average points scored per game throughout the season?",
],
"music": [
"How many invoices were issued for tracks in each genre?",
"How many tracks are there in each genre?",
"How many tracks were purchased by each customer, and what is the total cost of those tracks?",
"How many tracks were purchased in each genre, and what is the average unit price of those tracks?",
"How many tracks were purchased in each invoice, along with their unit prices?",
"How many tracks were purchased in each invoice, and what is the average unit price of those tracks?",
"How many tracks were purchased in each invoice, and what is the total cost of those tracks?",
"What are the details of customers who made purchases in each invoice?",
"What are the details of invoices along with the album titles purchased by each customer?",
"What are the details of invoices along with the customer names who made those purchases?",
"What are the details of invoices along with the total quantity of tracks purchased by each customer?",
"What are the details of invoices along with the total quantity of tracks purchased in each invoice?",
"What are the names of composers for tracks in a specific album?",
"What are the titles of albums along with their artists' names and genres?",
"What are the titles of albums along with their corresponding artist names?",
"What are the total sales generated from each customer?",
"What are the top 5 selling albums in the US?",
"What is the total revenue generated from each album?",
"What is the total revenue generated from each genre?",
"What is the total revenue generated from tracks composed by a specific composer?",
"What is the total revenue generated from tracks in each album?",
"Which albums have been purchased by customers in each invoice, along with their track details?",
"Which albums have been purchased by customers in each invoice, and what is the total cost of those albums?",
"Which albums have been purchased by each customer, and what is the total cost of those albums?",
"Which albums have been purchased in each invoice, and what is the total cost of each album?",
"Which customers have made purchases of tracks composed by a specific composer?",
"Which customers have made purchases of tracks from albums by a specific artist?",
"Which customers have made purchases of tracks from albums in each genre?",
"Which tracks are associated with each album and their respective genres?",
"Which tracks have been purchased by each customer in each invoice, along with their composers?",
"Which tracks have been purchased in each invoice along with their composers?",
],
"ecommerce": [
"How many items have been purchased in orders placed by customers in each state?",
"How many items were purchased in each order, and what was the total price?",
"How many orders have been delivered to customers in each city, and when were they delivered?",
"What are the sequential numbers of payments made for each order?",
"What are the top 3 value for orders placed by customers in each city?",
"What is the average score of reviews submitted for orders placed by customers in each city?",
"What is the creation timestamp of each review along with its associated order's status?",
"What is the delivery carrier date of each order along with its customer's city and state?",
"What is the estimated delivery date of each order along with its customer's city?",
"What is the status of each order along with its customer's city and state?",
"What is the total freight value for each order?",
"What is the total freight value for orders placed by customers in each state?",
"What is the total price of products included in each order along with their shipping limit dates?",
"What is the total value of payments made by customers from each state?",
],
}
if not Path("./outputs/ask/sampledata").exists():
Path("./outputs/ask/sampledata").mkdir(parents=True)
# init ask pipeline
document_store = init_document_store(
dataset_name=SAMPLE_DATASET_NAME,
recreate_index=True,
)
embedder = init_embedder()
retriever = init_retriever(
document_store=document_store,
top_k=10,
)
text_to_sql_generator = init_generator()
sql_correction_generator = init_generator()
retrieval_pipeline = Retrieval(
embedder=embedder,
retriever=retriever,
)
generation_pipeline = Generation(
text_to_sql_generator=text_to_sql_generator,
)
sql_correction_pipeline = SQLCorrection(
sql_correction_generator=sql_correction_generator,
)
# indexing
print("Indexing documents...")
mdl = get_mdl_from_wren_engine()
indexing_pipeline = Indexing(document_store=document_store)
indexing_pipeline.run(json.dumps(mdl))
print(
f"Finished indexing documents, document count: {document_store.count_documents()}"
)
print(
f"Running predictions for {len(SAMPLE_DATASET_QUESTIONS[SAMPLE_DATASET_NAME])} questions..."
)
start = time.time()
with ThreadPoolExecutor() as executor:
args_list = [
(question,) for question in SAMPLE_DATASET_QUESTIONS[SAMPLE_DATASET_NAME]
]
outputs = list(
tqdm(
executor.map(lambda p: process_item(*p), args_list),
total=len(args_list),
)
)
end = time.time()
print(f"Time taken: {end - start:.2f}s")
no_valid_generations = list(
filter(
lambda x: (not x["valid_generation_results"])
and x["invalid_generation_results"],
outputs,
)
)
total_invalid_generations = list(
filter(lambda x: x["invalid_generation_results"], outputs)
)
results = {
"mdl": mdl,
"no_valid_generation": {
"count": len(no_valid_generations),
"details": no_valid_generations,
},
"total_invalid_generation": {
"count": len(total_invalid_generations),
"details": total_invalid_generations,
},
"outputs": outputs,
}
with open(
f"./outputs/ask/sampledata/{SAMPLE_DATASET_NAME}_{datetime.now().strftime("%Y%m%d%H%M%S")}.json",
"w",
) as f:
json.dump(results, f, indent=2)
@@ -30,7 +30,7 @@ summary = Summary()
collectors = [Collector(element=element) for element in eval_context]
pipeline = Generation(
generator=init_generator(),
sql_details_generator=init_generator(),
)
for collector in collectors:
@@ -120,8 +120,8 @@ class Collector:
def _destruct(self, response: Dict[str, Any]):
return (
response["post_processor"]["generator"]["replies"][0],
response["post_processor"]["generator"]["meta"][0],
response["post_processor"]["replies"][0],
response["generator"]["meta"][0],
response["post_processor"]["results"],
)
@@ -112,6 +112,62 @@ Current User Query: {{ query }}
Generated SQL Queries amd Summaries:
"""
text_to_sql_with_followup_user_prompt_template = """
### TASK ###
Given a set of user queries that are ambiguous in nature, your task is to interpret these queries in various plausible ways and
generate multiple SQL statements that could potentially answer each interpreted version of the queries and within-10-words summary.
For each ambiguous user query, provide at least three different interpretations and corresponding SQL queries that reflect these interpretations.
Ensure that your SQL queries are diverse, covering a range of possible meanings behind the ambiguous query.
Consider the structure of a generic database which includes common tables like users, orders, products, and transactions.
Here are the ambiguous user queries:
### EXAMPLES ###
Previous SQL Summary: "Users signed up this year."
Previous Generated SQL Query: "SELECT * FROM users WHERE sign_up_date >= '2023-01-01';"
Current User Query: "Who has made a purchase?"
Generated SQL Queries amd Summaries:
{
"results": [
{
"sql": "SELECT users.* FROM users JOIN purchases ON users.id = purchases.user_id WHERE users.sign_up_date >= '2023-01-01';",
"summary": "Users joined in 2023 with purchases."
},
{
"sql": "SELECT DISTINCT users.* FROM users INNER JOIN purchases ON users.id = purchases.user_id WHERE users.sign_up_date >= '2023-01-01';",
"summary": "Unique users with purchases since 2023."
}
]
}
### DATABASE SCHEMA ###
{% for document in documents %}
{{ document.content }}
{% endfor %}
### FINAL ANSWER FORMAT ###
The final answer must be the JSON format like following:
{
"results": [
{"sql": <SQL_QUERY_STRING_1>, "summary": <SUMMARY_STRING_1>},
{"sql": <SQL_QUERY_STRING2>, "summary": <SUMMARY_STRING_2>}
]
}
### NOTICE ###
- Only use the tables and columns mentioned in the database schema.
- If you think you can't generate a valid SQL query for a specific interpretation, you can skip that interpretation and provide the other ones.
- Make sure to map operators and operands correctly based on their data types.
### QUESTION ###
Previous SQL Summary: {{ history.summary }}
Previous Generated SQL Query: {{ history.sql }}
Current User Query: {{ query }}
Generated SQL Queries amd Summaries:
"""
sql_correction_user_prompt_template = """
You are a Trino SQL expert with exceptional logical thinking skills and debugging skills.
@@ -18,34 +18,35 @@ class GenerationPostProcessor:
@component.output_types(
results=Optional[Dict[str, Any]],
)
def run(self, replies: List[str], meta: List[Dict[str, Any]]) -> Dict[str, Any]:
generator = {
"replies": replies,
"meta": meta,
}
def run(self, replies: List[str]) -> Dict[str, Any]:
cleaned_generation_result = json.loads(clean_generation_result(replies[0]))
steps = cleaned_generation_result.get("steps", [])
if not steps:
return {
"generator": generator,
"results": None,
"replies": replies,
"results": {
"description": cleaned_generation_result["description"],
"steps": [],
},
}
sql = _build_cte_query(steps)
if not check_if_sql_executable(os.getenv("WREN_ENGINE_ENDPOINT"), sql):
return {
"generator": generator,
"results": None,
"replies": replies,
"results": {
"description": cleaned_generation_result["description"],
"steps": [],
},
}
# make sure the last step has an empty cte_name
cleaned_generation_result["steps"][-1]["cte_name"] = ""
return {
"generator": generator,
"replies": replies,
"results": cleaned_generation_result,
}
@@ -23,7 +23,6 @@ class Generation(BasicPipeline):
self._pipeline.add_component("generator", generator)
self._pipeline.add_component("post_processor", init_generation_post_processor())
self._pipeline.connect("generator.replies", "post_processor.replies")
self._pipeline.connect("generator.meta", "post_processor.meta")
super().__init__(self._pipeline)
@@ -39,7 +38,7 @@ class Generation(BasicPipeline):
if __name__ == "__main__":
generation_pipeline = Generation(
generator=init_generator(),
sql_details_generator=init_generator(),
)
print("generating generation_pipeline.jpg to outputs/pipelines/ask_details...")
+24
View File
@@ -162,3 +162,27 @@ def generate_ddls_from_semantics(
ddl_commands.append(create_table_ddl)
return ddl_commands
def remove_duplicates(dicts):
"""
Removes duplicates from a list of dictionaries based on 'sql' and 'summary' fields.
Args:
dicts (list of dict): The list of dictionaries to be deduplicated.
Returns:
list of dict: A list of dictionaries after removing duplicates.
"""
# Convert each dictionary to a tuple of (sql, summary) to make them hashable
seen = set()
unique_dicts = []
for d in dicts:
identifier = (
d["sql"],
d["summary"],
) # This assumes 'sql' and 'summary' always exist
if identifier not in seen:
seen.add(identifier)
unique_dicts.append(d)
return unique_dicts
@@ -4,6 +4,8 @@ from typing import List, Literal, Optional
from haystack import Pipeline
from pydantic import BaseModel
from src.utils import remove_duplicates
logging.basicConfig(format="%(asctime)s %(message)s", level=logging.INFO)
@@ -216,6 +218,9 @@ class AskService:
f'sql_correction_results: {sql_correction_results["post_processor"]}'
)
# remove duplicates of valid_generation_results, which consists of a sql and a summary
valid_generation_results = remove_duplicates(valid_generation_results)
logging.debug("After sql correction:")
logging.debug(f"valid_generation_results: {valid_generation_results}")
@@ -66,6 +66,7 @@ class AskDetailsService:
self.ask_details_results[query_id] = AskDetailsResultResponse(
status="understanding"
)
self.ask_details_results[query_id] = AskDetailsResultResponse(
status="searching"
)
@@ -80,12 +81,19 @@ class AskDetailsService:
ask_details_result = generation_result["post_processor"]["results"]
if ask_details_result is None:
if not ask_details_result["steps"]:
ask_details_result["steps"] = [
{
"sql": ask_details_request.sql,
"summary": ask_details_request.summary,
"cte_name": "",
}
]
self.ask_details_results[query_id] = AskDetailsResultResponse(
status="failed",
error=AskDetailsResultResponse.AskDetailsError(
code="NO_RELEVANT_SQL",
message="No relevant SQL",
status="finished",
response=AskDetailsResultResponse.AskDetailsResponseDetails(
**ask_details_result
),
)
else:
@@ -70,10 +70,11 @@ def test_ask_details_with_failed_sql(ask_details_service: AskDetailsService):
# asking details
query_id = str(uuid.uuid4())
sql = "SELECT * FROM xxx"
summary = "This is a summary"
ask_details_request = AskDetailsRequest(
query="How many books are there?'",
sql=sql,
summary="This is a summary",
summary=summary,
)
ask_details_request.query_id = query_id
ask_details_service.ask_details(ask_details_request)
@@ -95,5 +96,8 @@ def test_ask_details_with_failed_sql(ask_details_service: AskDetailsService):
)
)
assert ask_details_result_response.status == "failed"
assert ask_details_result_response.error.code == "NO_RELEVANT_SQL"
assert ask_details_result_response.status == "finished"
assert ask_details_result_response.response.description != ""
assert len(ask_details_result_response.response.steps) == 1
assert ask_details_result_response.response.steps[0].sql == sql
assert ask_details_result_response.response.steps[0].summary != ""
+13 -13
View File
@@ -15,7 +15,7 @@ def test_semantics_description():
# using TestClient as a context manager would trigger startup/shutdown events as well as lifespans.
with TestClient(app) as client:
response = client.post(
url="/v1/semantics-descriptions/",
url="/v1/semantics-descriptions",
json={
"mdl": {
"name": "all_star",
@@ -59,7 +59,7 @@ def test_semantics_preparations():
mdl_str = json.dumps(json.load(f))
response = client.post(
url="/v1/semantics-preparations/",
url="/v1/semantics-preparations",
json={
"mdl": mdl_str,
"id": semantics_preperation_id,
@@ -73,7 +73,7 @@ def test_semantics_preparations():
while status == "indexing":
response = client.get(
url=f"/v1/semantics-preparations/{semantics_preperation_id}/status/"
url=f"/v1/semantics-preparations/{semantics_preperation_id}/status"
)
assert response.status_code == 200
@@ -101,12 +101,12 @@ def test_asks_with_successful_query():
query_id = response.json()["query_id"]
GLOBAL_DATA["query_id"] = query_id
response = client.get(url=f"/v1/asks/{query_id}/result/")
response = client.get(url=f"/v1/asks/{query_id}/result")
while (
response.json()["status"] != "finished"
and response.json()["status"] != "failed"
):
response = client.get(url=f"/v1/asks/{query_id}/result/")
response = client.get(url=f"/v1/asks/{query_id}/result")
assert response.status_code == 200
assert response.json()["status"] == "finished"
@@ -133,12 +133,12 @@ def test_asks_with_failed_query():
query_id = response.json()["query_id"]
GLOBAL_DATA["query_id"] = query_id
response = client.get(url=f"/v1/asks/{query_id}/result/")
response = client.get(url=f"/v1/asks/{query_id}/result")
while (
response.json()["status"] != "finished"
and response.json()["status"] != "failed"
):
response = client.get(url=f"/v1/asks/{query_id}/result/")
response = client.get(url=f"/v1/asks/{query_id}/result")
assert response.status_code == 200
assert response.json()["status"] == "failed"
@@ -162,9 +162,9 @@ def test_stop_asks():
assert response.status_code == 200
assert response.json()["query_id"] == query_id
response = client.get(url=f"/v1/asks/{query_id}/result/")
response = client.get(url=f"/v1/asks/{query_id}/result")
while response.json()["status"] != "stopped":
response = client.get(url=f"/v1/asks/{query_id}/result/")
response = client.get(url=f"/v1/asks/{query_id}/result")
assert response.status_code == 200
assert response.json()["status"] == "stopped"
@@ -173,7 +173,7 @@ def test_stop_asks():
def test_ask_details():
with TestClient(app) as client:
response = client.post(
url="/v1/ask-details/",
url="/v1/ask-details",
json={
"query": "How many books are there?",
"sql": "SELECT COUNT(*) FROM book",
@@ -185,9 +185,9 @@ def test_ask_details():
assert response.json()["query_id"] != ""
query_id = response.json()["query_id"]
response = client.get(url=f"/v1/ask-details/{query_id}/result/")
response = client.get(url=f"/v1/ask-details/{query_id}/result")
while response.json()["status"] != "finished":
response = client.get(url=f"/v1/ask-details/{query_id}/result/")
response = client.get(url=f"/v1/ask-details/{query_id}/result")
assert response.status_code == 200
assert response.json()["status"] == "finished"
@@ -202,7 +202,7 @@ def test_ask_details():
def test_web_error_handler():
with TestClient(app) as client:
response = client.post(
url="/v1/semantics-descriptions/",
url="/v1/semantics-descriptions",
json={},
)