feat: Upload code from repo

app.py

@@ -1,8 +1,84 @@
+import json
 import gradio as gr
+from typing import Tuple, Dict
+from paddleocr import PaddleOCR
 
-def greet(name):
-    return "Hello " + name + "!!"
+from qdrant_client import QdrantClient
+from fastembed import TextEmbedding
 
-demo = gr.Interface(fn=greet, inputs="text", outputs="text")
-demo.launch()
+from llama_index.llms.openai import OpenAI
+from food_recommender.utils import extract_food_items, synthesize_food_item
+from food_recommender.main import RecommendationEngine
 
+ocr = PaddleOCR(use_angle_cls=True, lang="en", use_gpu=False)
+llm = OpenAI(model="gpt-3.5-turbo")
+rec_engine = RecommendationEngine("food", QdrantClient(), TextEmbedding())
+
+
+def run_ocr(img_path):
+    result = ocr.ocr(img_path, cls=True)[0]
+    return "\n".join([line[1][0] for line in result])
+
+
+def recommend(
+    likes_str, dislikes_str, img_path
+) -> Tuple[str, str, str, Dict[str, float]]:
+    likes = [c.strip() for c in likes_str.split(",")]
+    dislikes = [c.strip() for c in dislikes_str.split(",")]
+    print(likes, dislikes)
+
+    rec_engine.reset()
+    for food_name in likes:
+        rec_engine.like(synthesize_food_item(food_name, llm))
+    for food_name in dislikes:
+        rec_engine.dislike(synthesize_food_item(food_name, llm))
+
+    ocr_text = run_ocr(img_path)
+
+    food_names = extract_food_items(ocr_text, llm)
+    food_items = [synthesize_food_item(name, llm) for name in food_names]
+
+    print("New food items from menu", food_items)
+
+    recommendations = rec_engine.recommend_from_given(food_items)
+    print(recommendations)
+
+    return (
+        ocr_text,
+        json.dumps(food_names, indent=4),
+        json.dumps([item.model_dump() for item in food_items], indent=4),
+        recommendations,
+    )
+
+
+title = "Food recommender"
+description = "Food recommender by <a href='https://kshivendu.dev/bio'>KShivendu</a> using Qdrant Recommendation API + OpenAI Function calling + FastEmbed embeddings"
+article = "<a href='https://github.com/KShivendu/rag-cookbook'>Github Repo</a></p>"
+examples = [
+    [
+        "fanta, waffles, chicken biriyani, most of indian food",
+        "virgin mojito, any pork dishes",
+        "sf-menu3.jpg",
+    ]
+]
+
+step1_ocr = gr.Text(label="OCR Output")
+step2_extraction = gr.Code(language="json", label="Extracted food items")
+step3_enrichment = gr.Code(language="json", label="Enriched food items")
+step4_recommend = gr.Label(label="Recommendations")
+
+app = gr.Interface(
+    fn=recommend,
+    inputs=[
+        gr.Textbox(label="Likes (comma seperated)"),
+        gr.Textbox(label="Dislikes (comma seperated)"),
+        gr.Image(type="filepath", label="Input", width=20),
+    ],
+    outputs=[step1_ocr, step2_extraction, step3_enrichment, step4_recommend],
+    title=title,
+    description=description,
+    article=article,
+    examples=examples,
+)
+app.queue(max_size=10)
+app.launch(debug=True)

main.py

@@ -0,0 +1,118 @@
+import uuid
+from typing import List, Dict
+from qdrant_client import QdrantClient, models as qmodels
+from llama_index.llms.openai import OpenAI
+from fastembed import TextEmbedding
+
+from food_recommender.models import FoodItem
+from food_recommender.utils import synthesize_food_item
+
+likes = ["dosa", "fanta", "croissant", "waffles"]
+dislikes = ["virgin mojito"]
+
+menu = ["croissant", "mango", "jalebi"]
+
+
+class RecommendationEngine:
+    def __init__(
+        self, category: str, qdrant: QdrantClient, fastembed_model: TextEmbedding
+    ) -> None:
+        self.collection = f"{category}_preferences"
+        self.qdrant = qdrant
+        self.embedding_model = fastembed_model
+
+        if self.qdrant.collection_exists(self.collection):
+            self.counter = self.qdrant.count(self.collection, exact=True).count
+        else:
+            self.reset()
+            self.counter = 0
+
+    def reset(self):
+        self.qdrant.recreate_collection(
+            self.collection,
+            vectors_config=qmodels.VectorParams(
+                size=384, distance=qmodels.Distance.COSINE
+            ),
+        )
+
+    def _generate_vector(self, model_json: dict):
+        embedding_txt = ""
+        for k, v in model_json.items():
+            embedding_txt += f"{k}: {v}"
+        return list(self.embedding_model.passage_embed([embedding_txt]))[0]
+
+    def _insert_preference(self, item: FoodItem, *args, **kwargs):
+        model_json: dict = item.model_dump()
+        embedding = self._generate_vector(model_json)
+
+        model_json.update(kwargs)
+
+        self.qdrant.upsert(
+            self.collection,
+            points=[
+                qmodels.PointStruct(
+                    id=self.counter, payload=model_json, vector=embedding
+                )
+            ],
+        )
+        self.counter += 1
+
+    def like(self, item: FoodItem):
+        self._insert_preference(item, liked=True)
+
+    def dislike(self, item: FoodItem):
+        self._insert_preference(item, liked=False)
+
+    def recommend_from_given(
+        self, items: List[FoodItem], limit: int = 3
+    ) -> Dict[str, int]:
+        liked_points, _offset = self.qdrant.scroll(
+            self.collection,
+            scroll_filter={"must": [{"key": "liked", "match": {"value": True}}]},
+        )
+
+        disliked_points, _offset = self.qdrant.scroll(
+            self.collection,
+            scroll_filter={"must": [{"key": "liked", "match": {"value": False}}]},
+        )
+
+        # Insert points in DB so they can be recommended:
+        # A bit ugly but this is the best possible thing at the moment.
+        query_id = str(uuid.uuid1())
+        for item in items:
+            self._insert_preference(item, query_id=query_id)
+
+        scored_points = self.qdrant.recommend(
+            self.collection,
+            positive=[p.id for p in liked_points],
+            negative=[p.id for p in disliked_points],
+            query_filter={"must": [{"key": "query_id", "match": {"value": query_id}}]},
+            with_payload=True,
+            strategy="best_score",
+        )
+        self.qdrant.delete(self.collection, [p.id for p in scored_points])
+
+        return {point.payload["name"]: point.score for point in scored_points}
+
+
+if __name__ == "__main__":
+    llm = OpenAI(model="gpt-3.5-turbo")
+    qdrant = QdrantClient()
+    fastembed_model = TextEmbedding()
+    rec_engine = RecommendationEngine("food", qdrant, fastembed_model)
+
+    if rec_engine.counter != len(likes) + len(dislikes):
+        rec_engine.reset()
+        print("Filling with starter data")
+        for food_name in likes:
+            food_item = synthesize_food_item(food_name, llm)
+            rec_engine.like(food_item)
+
+        for food_name in dislikes:
+            food_item = synthesize_food_item(food_name, llm)
+            rec_engine.dislike(food_item)
+
+    new_items = [synthesize_food_item(food_name, llm) for food_name in menu]
+    recommendations = rec_engine.recommend_from_given(items=new_items)
+
+    print(recommendations)

models.py

@@ -0,0 +1,23 @@
+from typing import List
+from pydantic import BaseModel, Field
+
+
+class FoodItem(BaseModel):
+    """Details of a food item or dish"""
+
+    name: str
+    ingredients_and_approach: str = Field(
+        description="Short description of the ingredients and the overall approach to cook"
+    )
+    taste_and_texture: str = Field(
+        description="Short description of how does it taste and feel in mouth"
+    )
+    is_vegetarian: bool  # TODO: Can be ambiguous. LLM could ask more question while asking for preferences?
+
+
+class ExtractedFoodName(BaseModel):
+    """Food items / dishes extracted from output of an OCR"""
+
+    food_names: List[str] = Field(
+        description="Each item must be an actual food item because OCR data can have lot of noise. If doubtful, discard. I don't want False positives. You can also fix small typos based on your understanding"
+    )

requirements.txt

@@ -0,0 +1,3 @@
+qdrant-client
+llama-index
+fastembed

utils.py

@@ -0,0 +1,31 @@
+from typing import List
+from llama_index.program.openai import OpenAIPydanticProgram
+from llama_index.core.llms.llm import LLM
+
+from food_recommender.models import FoodItem, ExtractedFoodName
+
+
+def synthesize_food_item(food_name: str, llm: LLM) -> FoodItem:
+    prompt = """Tell me what you know about the food item / dish '{food_name}' and return as a JSON object"""
+
+    program = OpenAIPydanticProgram.from_defaults(
+        output_cls=FoodItem,
+        llm=llm,
+        prompt_template_str=prompt,
+        verbose=True,
+    )
+    result: FoodItem = program(food_name=food_name)
+    return result
+
+
+def extract_food_items(text: str, llm: LLM) -> List[str]:
+    prompt = """You're world's best bot for parsing data from noisy OCR output on images. Exact all the food items that you can find in this menu. Please avoid parsing things that are names of the sections. Generally section comes before the food items: '{text}'. Return result as a JSON object"""
+
+    program = OpenAIPydanticProgram.from_defaults(
+        output_cls=ExtractedFoodName,
+        llm=llm,
+        prompt_template_str=prompt,
+        verbose=True,
+    )
+    result: ExtractedFoodName = program(text=text)
+    return [item.lower() for item in result.food_names]