shopping pipeline

.gitignore

@@ -11,3 +11,4 @@ eval-results/
 eval-queue-bk/
 eval-results-bk/
 logs/
+output-data/

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.9
+FROM python:3.10
 
 WORKDIR /code
 
@@ -6,7 +6,17 @@ COPY ./requirements.txt /code/requirements.txt
 
 RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
 
-COPY . .
+COPY scripts /code/scripts
+COPY src /code/src
+
+COPY app.py /code/app.py
+COPY main_backend.py /code/main_backend.py
+
+# to generate adhoc result files --> to be deleted once generation pipeline is stable
+COPY adhoc.py /code/adhoc.py
+
+# answer and score generation pipeline
+COPY generation.py /code/generation.py
 
 
 EXPOSE 7860

generation.py

@@ -0,0 +1,158 @@
+from lamini.generation.generation_node import GenerationNode
+from lamini.generation.generation_pipeline import GenerationPipeline
+from lamini.generation.base_prompt_object import PromptObject
+from typing import Union, Iterator, AsyncIterator
+from src.backend.lamini_eval.datasets.ecommerce.shopping_dataset import load_shopping_dataset
+from src.envs import DATASET_PATH, LAMINI_API_KEY
+import logging
+import asyncio
+import lamini
+import jsonlines
+from tqdm import tqdm
+
+
+lamini.api_key = LAMINI_API_KEY
+
+logger = logging.getLogger(__name__)
+
+logging.basicConfig(
+    level=logging.DEBUG,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+
+
+class AnswerScorePipeline(GenerationPipeline):
+    def __init__(self):
+        super(AnswerScorePipeline, self).__init__()
+
+        self.answer_generator = AnswerGenerator()
+        self.score_generator = ScoreGenerator()
+
+    def forward(self, x):
+        ans = self.answer_generator(x)
+        score = self.score_generator(ans)
+        return score
+
+
+class AnswerGenerator(GenerationNode):
+    def __init__(self):
+        super(AnswerGenerator, self).__init__(
+            model_name="mistralai/Mistral-7B-Instruct-v0.1", max_tokens=200
+        )
+
+    def generate(
+            self,
+            prompt: Union[Iterator[PromptObject], AsyncIterator[PromptObject]],
+            *args,
+            **kwargs,
+    ):
+        prompts = self.transform_prompt(prompt)
+        op_type = {
+                "product_name": "str",
+                "product_description": "str",
+                "product_id": "str"
+            }
+        results = super(AnswerGenerator, self).generate(prompts, output_type=op_type, *args, **kwargs)
+        processed_results = self.process_results(results)
+        return processed_results
+
+    async def process_results(self, results):
+        async for result in results:
+            logger.info(f"Generated answer for {result}")
+            if result is None:
+                continue
+            yield result
+
+    async def transform_prompt(self, prompts):
+        async for prompt in prompts:
+            prompt.prompt = self.make_prompt(prompt)
+            yield prompt
+
+    def make_prompt(self, chunk):
+        prompt = (
+            "You are an expert shopper at Instacart."
+        )
+        prompt += "You are helping a customer find a product. "
+        prompt += "Include the product name, id, and detailed description in your answer. "
+        prompt += "A product id is a number between 0 and 100,000. "
+        prompt += "The customer asks\n"
+        prompt += chunk.data["question"]
+        print("prompt:", prompt)
+        return prompt
+
+
+class ScoreGenerator(GenerationNode):
+    def __init__(self):
+        super(ScoreGenerator, self).__init__(
+            model_name="mistralai/Mistral-7B-Instruct-v0.1", max_tokens=200
+        )
+
+    def generate(
+            self,
+            prompt: Union[Iterator[PromptObject], AsyncIterator[PromptObject]],
+            *args,
+            **kwargs,
+    ):
+        prompts = self.transform_prompt(prompt)
+        results = super(ScoreGenerator, self).generate(prompts, *args, **kwargs)
+        processed_results = self.process_results(results)
+        return processed_results
+
+    async def process_results(self, results):
+        async for result in results:
+            logger.info(f"Generated score for {result}")
+            if result is None:
+                continue
+            yield result
+
+    async def transform_prompt(self, prompts):
+        async for prompt in prompts:
+            prompt.data["question"] = prompt.prompt
+            prompt.prompt = self.make_prompt(prompt)
+            yield prompt
+
+    def make_prompt(self, chunk):
+        prompt = (
+            "A model is going to answer a question. Your job is to score the answer, comparing it to a golden reference. You are an expert scorer.\n\n"
+        )
+        prompt += f"Rate the answer using a score from 1 (lowest match) to 5 (highest match).\n"
+        prompt += chunk.data["rubric"]
+        prompt += "Use the full range. Read the gold answer carefully. "
+        prompt += "Explain your score in 2-3 short sentences not exceeding 100 words each, then assign a score. "
+        prompt += "Output your score as a JSON object in the format {\"explanation\" : str, \"score\" : int}\n"
+        prompt += "Use single quotes within your explanation. End your explanation with a double quote.\n"
+        prompt += "Prefer answers that are most similar to the gold answer, even if the gold answer refused to answer the question.\n\n"
+        prompt += f"========== question =========\n{chunk.data['question']}\n\n"
+        prompt += f"========== gold answer =========\n{chunk.data['expected_response']}\n\n"
+        prompt += f"========== model answer =========\n{chunk.data['response']}\n\n"
+        prompt += "=" * 40 + "\n\n"
+        prompt += f"How would you score the model's answer compared to the gold answer (using the 1-5 scale defined above)?\n\n"
+        print("prompt:", prompt)
+        return prompt
+
+
+async def main():
+    dataset = load_shopping_dataset(DATASET_PATH)
+
+    answers = AnswerScorePipeline().call(dataset)
+    await print_answers(answers)
+
+
+async def print_answers(answers):
+    path = "/code/data/model-answers.jsonl"
+
+    with jsonlines.open(path, "w") as writer:
+        pbar = tqdm(desc="Saving answers", unit=" answers")
+        async for answer in answers:
+            answer = {
+                "prompt": answer.prompt,
+                "question": answer.data["question"],
+                "answer": answer.response,
+            }
+            print(f"\n\n=======\n{answer}\n\n")
+            writer.write(answer)
+            pbar.update()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

run-adhoc.sh

@@ -1,2 +1,13 @@
+#!/bin/bash
+
+# Safely execute this bash script
+# e exit on first failure
+# x all executed commands are printed to the terminal
+# u unset variables are errors
+# a export all variables to the environment
+# E any trap on ERR is inherited by shell functions
+# -o pipefail | produces a failure code if any stage fails
+set -Eeuoxa pipefail
+
 docker buildx build --platform=linux/amd64 -t adhoc .
-docker run -it --rm --platform=linux/amd64 -e TOKEN=$TOKEN adhoc python adhoc.py
+docker run -it --rm --platform=linux/amd64 adhoc python adhoc.py

run-pipeline.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+
+# Safely execute this bash script
+# e exit on first failure
+# x all executed commands are printed to the terminal
+# u unset variables are errors
+# a export all variables to the environment
+# E any trap on ERR is inherited by shell functions
+# -o pipefail | produces a failure code if any stage fails
+set -Eeuoxa pipefail
+
+LOCAL_DIRECTORY="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+
+docker buildx build --platform=linux/amd64 -t pipeline .
+docker run -it --rm --platform=linux/amd64 -v $LOCAL_DIRECTORY/output-data:/code/data pipeline python3 generation.py

start.sh

@@ -1,5 +1,14 @@
 #!/bin/bash
 
+# Safely execute this bash script
+# e exit on first failure
+# x all executed commands are printed to the terminal
+# u unset variables are errors
+# a export all variables to the environment
+# E any trap on ERR is inherited by shell functions
+# -o pipefail | produces a failure code if any stage fails
+set -Eeuoxa pipefail
+
 for ARGUMENT in "$@"
 do
    KEY=$(echo $ARGUMENT | cut -f1 -d=)
@@ -15,4 +24,4 @@ echo "Run mode is: $RUN_MODE"
 echo "Model passed is: $LOCAL_MODEL_NAME"
 
 docker buildx build --platform=linux/amd64 -t ldr .
-docker run -it --rm -p 7860:7860 --platform=linux/amd64 -e TOKEN=$TOKEN -e RUN_MODE=$RUN_MODE -e LOCAL_MODEL_NAME=$LOCAL_MODEL_NAME ldr python app.py
+docker run -it --rm -p 7860:7860 --platform=linux/amd64 -e RUN_MODE=$RUN_MODE -e LOCAL_MODEL_NAME=$LOCAL_MODEL_NAME ldr python app.py