RAG Meal Planning Assistant

Introduction

30 Second Summary

Deciding what to eat every day is surprisingly exhausting. Factor in dietary restrictions, allergies, and nutritional goals, and meal planning becomes a chore that most people give up on within a week.

In this project, you will build a conversational AI meal planning assistant that recommends personalized recipes based on your dietary preferences. Behind the scenes, it uses a multi-stage retrieval pipeline that searches a local recipe database, judges the quality of results, and falls back to web search when needed.

What You'll Build

A Streamlit chat application powered by a LangGraph state machine that orchestrates query optimization, vector retrieval, LLM-based quality judging, and adaptive web search fallback.

By the end of this project, you'll have:

• A ChromaDB vector store loaded with ~500 recipes, embedded using Gemini's embedding model.
• A LangGraph StateGraph with four nodes (Optimize, Retrieve, Judge, Respond) and conditional routing that retries or falls back to web search.
• A full Streamlit chat interface with sidebar profile management and SQLite persistence for user preferences and meal history.
• Secret Mission: a weekly meal plan generator that balances macros across 7 days.

Are there any prerequisites?

You should be comfortable with basic Python (functions, classes, and dictionaries). You also need API keys for Google AI Studio and Tavily before starting. Both offer free tiers that cover this entire project at no cost.

Before We Start

You're about to build a conversational meal planning assistant powered by a multi-stage RAG pipeline.

Before we dive into building the pipeline itself, let's get your project folder, packages, and API connections ready to go.

In this step, get ready to:

• Create your project folder and a Python virtual environment.
• Install all required Python packages.
• Configure API keys and verify connectivity to Gemini and Tavily.

Create the project directory and virtual environment

A virtual environment keeps this project's packages isolated from other Python projects on your machine.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir meal-planner
cd meal-planner

• Create a Python virtual environment by running:

python3 -m venv venv

• Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

What is a virtual environment?

A virtual environment is an isolated Python workspace. Packages you install here won't affect other projects on your machine.

You'll know it's active when you see (venv) at the start of your terminal prompt.

• Confirm your folder was created by running:

ls

You should see the venv folder listed in the output.

Don't see (venv) in your prompt?

Make sure you ran the activation command for your operating system. The macOS command starts with source while the Windows command uses venv\Scripts\activate.

help me activate my Python virtual environment

Install required packages

This project uses seven packages that handle everything from graph orchestration to vector storage to the chat UI.

• Install all the packages by running:

pip install langgraph chromadb google-genai tavily-python streamlit pandas python-dotenv

What are these packages?

• langgraph orchestrates your RAG pipeline as a stateful graph.
• chromadb is a vector database for storing and searching recipe embeddings.
• google-genai is the official Google GenAI SDK for calling Gemini models.
• tavily-python provides web search as a fallback when local retrieval fails.
• streamlit builds the chat interface with minimal code.
• pandas loads and processes the recipe dataset.
• python-dotenv loads environment variables from your .env file.

The install may take a minute or two. You should see a success message for each package once it finishes.

Seeing installation errors?

Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt before running pip install.

If you see a permissions error, do not use sudo. Instead, deactivate and recreate your virtual environment.

help me fix my pip install errors

Create your .env file and verify connectivity

Your API keys need to live in a .env file so the application can load them securely without hardcoding them in your scripts.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Cursor and press Enter to open it.
• In Cursor, click File in the top menu bar.
• Click Open Folder.
• Select the meal-planner folder on your Desktop, then click Open.

• Create a new file called .env in the root of your meal-planner folder by clicking the new file icon in the Explorer sidebar.
• Paste the following into your .env file, replacing the placeholder values with your actual API keys:

GEMINI_API_KEY=your-gemini-api-key-here
TAVILY_API_KEY=your-tavily-api-key-here

Where do I find my API keys?

Your GEMINI_API_KEY comes from Google AI Studio. Your TAVILY_API_KEY comes from your Tavily account dashboard.

Replace your-gemini-api-key-here and your-tavily-api-key-here with the actual key strings. Do not add quotes around the values.

• Save .env.

Now let's verify that both APIs are reachable. This catches any key typos before you build the full pipeline.

• In your terminal (make sure you're still in the meal-planner folder with (venv) active), run this verification script:

python3 -c "
from dotenv import load_dotenv
load_dotenv()
import os
from google import genai
from tavily import TavilyClient

# Test Gemini
client = genai.Client()
response = client.models.generate_content(model='gemini-2.5-flash', contents='Say hello in one word.')
print('Gemini:', response.text.strip())

# Test Tavily
tavily = TavilyClient(api_key=os.environ['TAVILY_API_KEY'])
result = tavily.search(query='test', max_results=1)
print('Tavily:', len(result.get('results', [])), 'result(s) returned')
print('All good! Both APIs are connected.')
"

You should see output confirming both Gemini and Tavily responded successfully.

Seeing an API error?

Double-check that your .env file is saved in the meal-planner folder (not inside venv or a subfolder).

Verify there are no extra spaces around the = signs in your .env file, and no quotes wrapping the key values.

help me debug my API connectivity issue

✔️ Awesome, I've got everything!

Great. Make sure you've saved your .env file.

ⓧ I'd like to double check the full code

Here's exactly what your .env file should look like (with your real keys instead of the placeholders):

GEMINI_API_KEY=your-gemini-api-key-here
TAVILY_API_KEY=your-tavily-api-key-here

Your environment is ready. Next up, you'll load a recipe dataset and embed it into a vector store that your RAG pipeline can search.

Load and Prepare the Recipe Dataset

Your development environment is set up and both APIs are confirmed working. Now you need data for your assistant to recommend from.

In this step, you will download a real recipe dataset from Kaggle, process it into a searchable format, and store the results in a ChromaDB vector database. This is the foundation of the RAG pipeline. Your assistant will search these stored recipes later when a user asks for meal recommendations.

In this step, get ready to:

• Download the Food Ingredients and Recipes CSV from Kaggle.
• Build an ingestion script that loads recipes and creates text representations.
• Generate vector embeddings with Gemini and store them in ChromaDB.

Download the recipe dataset from Kaggle

Your meal planner needs a knowledge base of recipes to recommend from. You will use a public dataset from Kaggle containing hundreds of recipes with their titles, ingredients, and cooking instructions.

• Open the dataset page in your browser:
• https://www.kaggle.com/datasets/pes12017000148/food-ingredients-and-recipe-dataset-with-images
• Click the Download button near the top of the page.
• Sign in with a Kaggle account if prompted (creating one is free).
• Extract the downloaded ZIP file.
• Find the CSV file inside (its name is long: Food Ingredients and Recipe Dataset with Image Name Mapping.csv).
• Rename it to recipes.csv and move it into your meal-planner project folder.

• Confirm the file is in place by running:

ls recipes.csv

You should see recipes.csv listed in the output.

Don't see the file?

Make sure you renamed the long filename to recipes.csv and placed it directly inside your meal-planner folder (not in a subfolder from the ZIP extraction).

help me find where my downloaded Kaggle dataset ended up

Build the ingestion script

Raw CSV data is not searchable by meaning. To enable semantic search, you need to convert each recipe into a text chunk and then into a numerical vector (an embedding). This process is called ingestion.

You will create a script called ingest.py that handles the full pipeline: reading the CSV, formatting recipes as text, generating embeddings with Gemini, and storing everything in ChromaDB.

• Create a new file called ingest.py in your meal-planner folder.
• Add the imports and client setup by pasting this code into the file:

import os
import pandas as pd
import chromadb
from google import genai
from google.genai import types
from dotenv import load_dotenv

load_dotenv()

client = genai.Client()
chroma_client = chromadb.PersistentClient(path="./chroma_db")

collection = chroma_client.get_or_create_collection(
    name="recipes",
    embedding_function=None
)

What does this code do?

• load_dotenv() reads your .env file so the Gemini API key is available automatically.
• genai.Client() creates a Gemini client that picks up your GEMINI_API_KEY from the environment.
• PersistentClient(path="./chroma_db") creates a ChromaDB database stored on disk so your embeddings survive between script runs.
• embedding_function=None tells ChromaDB you will provide pre-computed embeddings rather than having ChromaDB generate them.

• Below the collection setup, add the function that converts a CSV row into a searchable text chunk:

def create_recipe_text(row):
    title = str(row.get("Title", ""))
    ingredients = str(row.get("Ingredients", ""))
    instructions = str(row.get("Instructions", ""))
    return f"Recipe: {title}\nIngredients: {ingredients}\nInstructions: {instructions}"

What does this function do?

Each recipe in the CSV has separate columns for title, ingredients, and instructions. This function combines them into a single string so the embedding model can capture the full context of each recipe in one vector.

Generate embeddings and store in ChromaDB

Vector embeddings are numerical representations of text that capture meaning. Two recipes with similar ingredients and techniques will have vectors that are close together, enabling semantic search.

What are vector embeddings?

Think of an embedding as a coordinate in a high-dimensional space. The gemini-embedding-2 model converts each recipe text into a list of 768 numbers. Recipes about pasta will cluster near other pasta recipes, making it possible to find relevant results by comparing distances between vectors.

• Add the embedding function below create_recipe_text:

def embed_texts(texts, batch_size=20):
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        result = client.models.embed_content(
            model="gemini-embedding-2",
            contents=batch,
            config=types.EmbedContentConfig(output_dimensionality=768),
        )
        for embedding in result.embeddings:
            all_embeddings.append(embedding.values)
    return all_embeddings

Why batch in groups of 20?

The Gemini embedding API accepts multiple texts in a single call. Sending 20 at a time balances speed (fewer API calls) with reliability (avoiding timeouts on large payloads). For 500 recipes, this means roughly 25 API calls instead of 500.

• Add the main function that orchestrates the full ingestion pipeline:

def main():
    df = pd.read_csv("recipes.csv")
    df = df.dropna(subset=["Title", "Ingredients", "Instructions"]).head(500)

    documents = []
    ids = []
    metadatas = []

    for idx, row in df.iterrows():
        doc_text = create_recipe_text(row)
        documents.append(doc_text)
        ids.append(f"recipe_{idx}")
        metadatas.append({
            "title": str(row.get("Title", "")),
            "ingredients": str(row.get("Ingredients", ""))[:500],
        })

    print(f"Embedding {len(documents)} recipes...")
    embeddings = embed_texts(documents)

    print("Adding to ChromaDB...")
    batch_size = 100
    for i in range(0, len(documents), batch_size):
        end = min(i + batch_size, len(documents))
        collection.add(
            ids=ids[i:end],
            documents=documents[i:end],
            embeddings=embeddings[i:end],
            metadatas=metadatas[i:end],
        )

    print(f"Done! {collection.count()} recipes indexed in ChromaDB.")


if __name__ == "__main__":
    main()

What does the main function do?

• Reads recipes.csv, drops rows missing critical columns, and keeps the first 500 recipes.
• Builds a list of text documents, unique IDs, and metadata for each recipe.
• Calls embed_texts to generate embeddings via the Gemini API.
• Stores everything in ChromaDB in batches of 100 (ChromaDB handles large inserts better in chunks).

• Save ingest.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before running it.

ⓧ I'd like to double check the full code

import os
import pandas as pd
import chromadb
from google import genai
from google.genai import types
from dotenv import load_dotenv

load_dotenv()

client = genai.Client()
chroma_client = chromadb.PersistentClient(path="./chroma_db")

collection = chroma_client.get_or_create_collection(
    name="recipes",
    embedding_function=None
)


def create_recipe_text(row):
    title = str(row.get("Title", ""))
    ingredients = str(row.get("Ingredients", ""))
    instructions = str(row.get("Instructions", ""))
    return f"Recipe: {title}\nIngredients: {ingredients}\nInstructions: {instructions}"


def embed_texts(texts, batch_size=20):
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        result = client.models.embed_content(
            model="gemini-embedding-2",
            contents=batch,
            config=types.EmbedContentConfig(output_dimensionality=768),
        )
        for embedding in result.embeddings:
            all_embeddings.append(embedding.values)
    return all_embeddings


def main():
    df = pd.read_csv("recipes.csv")
    df = df.dropna(subset=["Title", "Ingredients", "Instructions"]).head(500)

    documents = []
    ids = []
    metadatas = []

    for idx, row in df.iterrows():
        doc_text = create_recipe_text(row)
        documents.append(doc_text)
        ids.append(f"recipe_{idx}")
        metadatas.append({
            "title": str(row.get("Title", "")),
            "ingredients": str(row.get("Ingredients", ""))[:500],
        })

    print(f"Embedding {len(documents)} recipes...")
    embeddings = embed_texts(documents)

    print("Adding to ChromaDB...")
    batch_size = 100
    for i in range(0, len(documents), batch_size):
        end = min(i + batch_size, len(documents))
        collection.add(
            ids=ids[i:end],
            documents=documents[i:end],
            embeddings=embeddings[i:end],
            metadatas=metadatas[i:end],
        )

    print(f"Done! {collection.count()} recipes indexed in ChromaDB.")


if __name__ == "__main__":
    main()

• Run the ingestion script by executing this command:

python ingest.py

This will take 1-2 minutes as Gemini processes 500 recipes in batches. You should see output like this:

Embedding 500 recipes...
Adding to ChromaDB...
Done! 500 recipes indexed in ChromaDB.

Seeing an API error?

• If you see a permissions or authentication error, double check that your .env file contains a valid GEMINI_API_KEY and that the file is in the same folder as ingest.py.
• If you see a rate limit error, wait 60 seconds and run the script again. The free tier has generous limits but can throttle rapid bursts.
• If you see a file not found error for recipes.csv, make sure the CSV is in your meal-planner folder and named exactly recipes.csv.

help me debug my ingest.py script that is failing when trying to embed recipes

Your recipe knowledge base is now searchable by meaning. Next up, you will build the LangGraph pipeline that queries this vector store and orchestrates the full recommendation flow.

Build the LangGraph RAG Pipeline

Your recipe database is loaded and searchable. Now you need an intelligent system that can take a user's question, find the best recipes, and decide whether those results are good enough.

In this step, you will build a multi-stage RAG pipeline using LangGraph's StateGraph. Each stage of the pipeline is a node (a Python function) that transforms the shared state. Conditional routing between nodes lets the pipeline make decisions like "are these results good enough, or should I try again?"

In this step, get ready to:

• Define the pipeline state schema with a TypedDict.
• Implement four node functions for query optimization, retrieval, judgment, and response generation.
• Wire the nodes into a StateGraph with conditional edges and retry logic.

Define the graph state and initialize clients

Every LangGraph pipeline starts with a state definition. The state is a Python TypedDict that every node in the graph can read from and write to.

• Create a new file called graph.py in your meal-planner folder.
• Add the following imports, client setup, and state definition:

import os
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from google import genai
from google.genai import types
import chromadb
from tavily import TavilyClient
from dotenv import load_dotenv

load_dotenv()

gemini_client = genai.Client()
chroma_client = chromadb.PersistentClient(path="./chroma_db")
collection = chroma_client.get_or_create_collection(name="recipes", embedding_function=None)
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])


class MealPlannerState(TypedDict):
    user_query: str
    user_profile: dict
    optimized_query: str
    retrieved_recipes: list
    web_results: list
    relevance_score: str
    retry_count: int
    response: str

What does this code do?

The top section loads environment variables and creates clients for Gemini (LLM calls and embeddings), ChromaDB (vector search), and Tavily (web search fallback).

The MealPlannerState class defines eight fields that flow through the pipeline. Each node reads what it needs from the state and returns updates to specific fields.

• Save graph.py.
• Verify the file loads correctly by running this in your terminal:

python -c "from graph import MealPlannerState; print(MealPlannerState.__annotations__)"

You should see a dictionary of all eight state fields printed to the terminal.

Seeing a ModuleNotFoundError?

• Make sure your virtual environment is activated. You should see (venv) or similar at the start of your terminal prompt.
• Confirm your .env file is in the same meal-planner folder as graph.py.
• Check that TAVILY_API_KEY is set in your .env file. The TavilyClient reads it at import time.

help me debug this ModuleNotFoundError in graph.py

Implement the node functions

The pipeline has four nodes. Each one is a Python function that receives the current state and returns a dictionary of field updates.

• In graph.py, add the optimize_query and retrieve_recipes functions below the MealPlannerState class:

def optimize_query(state: MealPlannerState) -> dict:
    profile = state.get("user_profile", {})
    dietary_info = ""
    if profile:
        restrictions = profile.get("restrictions", [])
        preferences = profile.get("preferences", [])
        if restrictions:
            dietary_info += f" Dietary restrictions: {', '.join(restrictions)}."
        if preferences:
            dietary_info += f" Preferences: {', '.join(preferences)}."

    prompt = f"""Rewrite the following user query into an optimized search query for a recipe database.
Include relevant keywords for ingredients, cuisine type, and cooking method.
{dietary_info}

User query: {state['user_query']}

Return ONLY the optimized search query, nothing else."""

    response = gemini_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    return {"optimized_query": response.text.strip(), "retry_count": state.get("retry_count", 0) + 1}


def retrieve_recipes(state: MealPlannerState) -> dict:
    query_text = state["optimized_query"]
    query_embedding = gemini_client.models.embed_content(
        model="gemini-embedding-2",
        contents=query_text,
        config=types.EmbedContentConfig(output_dimensionality=768),
    )
    results = collection.query(
        query_embeddings=[query_embedding.embeddings[0].values],
        n_results=5,
        include=["documents", "metadatas", "distances"],
    )
    recipes = []
    if results["documents"] and results["documents"][0]:
        for i, doc in enumerate(results["documents"][0]):
            recipes.append({
                "text": doc,
                "metadata": results["metadatas"][0][i] if results["metadatas"] else {},
                "distance": results["distances"][0][i] if results["distances"] else 0,
            })
    return {"retrieved_recipes": recipes}

What do these functions do?

• optimize_query calls gemini-2.5-flash to rewrite the user's casual question into a keyword-rich search query. It also injects dietary preferences from the user profile. The retry_count increments each time this node runs, which prevents infinite retry loops later.
• retrieve_recipes embeds the optimized query using gemini-embedding-2 (the same model used during ingestion) and queries ChromaDB for the 5 closest recipe matches.

• Save graph.py.
• You won't see a change yet. The next chunk adds the judgment and response functions that complete the pipeline.

Seeing a syntax error on the f-string?

• Triple-quoted f-strings (f"""...""") must not have unescaped curly braces. Make sure state['user_query'] uses single quotes inside the braces so they don't collide with the outer double quotes.
• Check that you have matching opening and closing triple quotes on the prompt string.

help me fix this f-string syntax error in optimize_query

• Now add the judge_relevance and generate_response functions below retrieve_recipes:

def judge_relevance(state: MealPlannerState) -> dict:
    recipes = state.get("retrieved_recipes", [])
    if not recipes:
        return {"relevance_score": "insufficient"}

    recipe_summaries = "\n".join(
        [f"- {r['metadata'].get('title', 'Unknown')}" for r in recipes[:5]]
    )
    prompt = f"""You are judging whether retrieved recipes are relevant to the user's request.

User query: {state['user_query']}
Optimized query: {state['optimized_query']}

Retrieved recipes:
{recipe_summaries}

Are these recipes relevant and helpful for the user's request?
Answer with exactly one word: "sufficient" or "insufficient"."""

    response = gemini_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    score = response.text.strip().lower()
    if "sufficient" in score and "insufficient" not in score:
        return {"relevance_score": "sufficient"}
    return {"relevance_score": "insufficient"}


def generate_response(state: MealPlannerState) -> dict:
    profile = state.get("user_profile", {})
    dietary_info = ""
    if profile:
        restrictions = profile.get("restrictions", [])
        if restrictions:
            dietary_info = f"\nUser dietary restrictions: {', '.join(restrictions)}"

    context_parts = []
    recipes = state.get("retrieved_recipes", [])
    if recipes:
        context_parts.append("Local recipes found:")
        for r in recipes[:5]:
            context_parts.append(r["text"])

    web_results = state.get("web_results", [])
    if web_results:
        context_parts.append("\nWeb search results:")
        for w in web_results:
            context_parts.append(f"- {w['title']}: {w['content'][:300]}")

    context = "\n\n".join(context_parts)

    prompt = f"""You are a helpful meal planning assistant. Based on the following recipe information,
provide a personalized recommendation to the user.{dietary_info}

User request: {state['user_query']}

Available information:
{context}

Provide a friendly, helpful response with specific recipe recommendations.
Include key ingredients and brief cooking instructions when possible."""

    response = gemini_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    return {"response": response.text}

What do these functions do?

• judge_relevance implements the LLM-as-judge pattern. It sends the retrieved recipe titles to gemini-2.5-flash and asks for a single-word verdict: "sufficient" or "insufficient". This determines whether the pipeline moves forward or retries.
• generate_response combines local recipes and any web results into a context block, then asks Gemini to produce a friendly recommendation. It respects dietary restrictions from the user profile.

• Save graph.py.
• You won't see a change yet. The next section wires these functions into a graph with conditional routing.

IndentationError or unexpected indent?

• Make sure each function starts at the top level (no indentation before def).
• Check that you have exactly one blank line between the end of retrieve_recipes and the start of judge_relevance.

help me fix this IndentationError in graph.py

Wire the StateGraph with conditional edges

With all four nodes written, you can now assemble them into a StateGraph. The key feature is conditional routing after the Judge node. If results are good, the pipeline moves to Respond. If not, it loops back to Optimize for a refined query.

• Add the routing function and graph builder below generate_response in graph.py:

def route_after_judge(state: MealPlannerState) -> Literal["web_search", "generate_response", "optimize_query"]:
    if state["relevance_score"] == "sufficient":
        return "generate_response"
    if state.get("retry_count", 0) >= 2:
        return "web_search"
    return "optimize_query"


def build_graph():
    graph = StateGraph(MealPlannerState)

    graph.add_node("optimize_query", optimize_query)
    graph.add_node("retrieve_recipes", retrieve_recipes)
    graph.add_node("judge_relevance", judge_relevance)
    graph.add_node("generate_response", generate_response)

    graph.add_edge(START, "optimize_query")
    graph.add_edge("optimize_query", "retrieve_recipes")
    graph.add_edge("retrieve_recipes", "judge_relevance")
    graph.add_conditional_edges("judge_relevance", route_after_judge)
    graph.add_edge("generate_response", END)

    return graph.compile()

What does this code do?

• route_after_judge is the conditional routing function. It checks relevance_score and retry_count to decide the next node. If retrieval quality is good, go to generate_response. If retries are exhausted (2+), fall back to web_search. Otherwise, loop back to optimize_query for another attempt.
• build_graph registers all nodes, connects them with fixed edges, and uses add_conditional_edges after the judge to branch based on the routing function's return value. The compiled graph is ready to be invoked with a state dictionary.

The web_search node referenced in the routing function does not exist yet. You will add it in the next step.

• Save graph.py.
• Verify the module loads by running:

python -c "from graph import MealPlannerState, optimize_query, retrieve_recipes, judge_relevance, generate_response, route_after_judge; print('All nodes defined successfully')"

You should see All nodes defined successfully printed in the terminal.

Seeing a NameError or ImportError?

• A NameError means one of your functions references a variable that is not defined above it. Check that gemini_client, collection, and tavily_client are all defined at the top of the file.
• If you see KeyError: 'TAVILY_API_KEY', confirm your .env file has TAVILY_API_KEY= followed by your actual key.
• Make sure your terminal is in the meal-planner directory when you run the command.

help me debug why graph.py won't import

✔️ Awesome, I've got everything!

Great. Double check you have saved graph.py with all four node functions, the routing function, and the build_graph function.

ⓧ I'd like to double check the full code

import os
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from google import genai
from google.genai import types
import chromadb
from tavily import TavilyClient
from dotenv import load_dotenv

load_dotenv()

gemini_client = genai.Client()
chroma_client = chromadb.PersistentClient(path="./chroma_db")
collection = chroma_client.get_or_create_collection(name="recipes", embedding_function=None)
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])


class MealPlannerState(TypedDict):
    user_query: str
    user_profile: dict
    optimized_query: str
    retrieved_recipes: list
    web_results: list
    relevance_score: str
    retry_count: int
    response: str


def optimize_query(state: MealPlannerState) -> dict:
    profile = state.get("user_profile", {})
    dietary_info = ""
    if profile:
        restrictions = profile.get("restrictions", [])
        preferences = profile.get("preferences", [])
        if restrictions:
            dietary_info += f" Dietary restrictions: {', '.join(restrictions)}."
        if preferences:
            dietary_info += f" Preferences: {', '.join(preferences)}."

    prompt = f"""Rewrite the following user query into an optimized search query for a recipe database.
Include relevant keywords for ingredients, cuisine type, and cooking method.
{dietary_info}

User query: {state['user_query']}

Return ONLY the optimized search query, nothing else."""

    response = gemini_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    return {"optimized_query": response.text.strip(), "retry_count": state.get("retry_count", 0) + 1}


def retrieve_recipes(state: MealPlannerState) -> dict:
    query_text = state["optimized_query"]
    query_embedding = gemini_client.models.embed_content(
        model="gemini-embedding-2",
        contents=query_text,
        config=types.EmbedContentConfig(output_dimensionality=768),
    )
    results = collection.query(
        query_embeddings=[query_embedding.embeddings[0].values],
        n_results=5,
        include=["documents", "metadatas", "distances"],
    )
    recipes = []
    if results["documents"] and results["documents"][0]:
        for i, doc in enumerate(results["documents"][0]):
            recipes.append({
                "text": doc,
                "metadata": results["metadatas"][0][i] if results["metadatas"] else {},
                "distance": results["distances"][0][i] if results["distances"] else 0,
            })
    return {"retrieved_recipes": recipes}


def judge_relevance(state: MealPlannerState) -> dict:
    recipes = state.get("retrieved_recipes", [])
    if not recipes:
        return {"relevance_score": "insufficient"}

    recipe_summaries = "\n".join(
        [f"- {r['metadata'].get('title', 'Unknown')}" for r in recipes[:5]]
    )
    prompt = f"""You are judging whether retrieved recipes are relevant to the user's request.

User query: {state['user_query']}
Optimized query: {state['optimized_query']}

Retrieved recipes:
{recipe_summaries}

Are these recipes relevant and helpful for the user's request?
Answer with exactly one word: "sufficient" or "insufficient"."""

    response = gemini_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    score = response.text.strip().lower()
    if "sufficient" in score and "insufficient" not in score:
        return {"relevance_score": "sufficient"}
    return {"relevance_score": "insufficient"}


def generate_response(state: MealPlannerState) -> dict:
    profile = state.get("user_profile", {})
    dietary_info = ""
    if profile:
        restrictions = profile.get("restrictions", [])
        if restrictions:
            dietary_info = f"\nUser dietary restrictions: {', '.join(restrictions)}"

    context_parts = []
    recipes = state.get("retrieved_recipes", [])
    if recipes:
        context_parts.append("Local recipes found:")
        for r in recipes[:5]:
            context_parts.append(r["text"])

    web_results = state.get("web_results", [])
    if web_results:
        context_parts.append("\nWeb search results:")
        for w in web_results:
            context_parts.append(f"- {w['title']}: {w['content'][:300]}")

    context = "\n\n".join(context_parts)

    prompt = f"""You are a helpful meal planning assistant. Based on the following recipe information,
provide a personalized recommendation to the user.{dietary_info}

User request: {state['user_query']}

Available information:
{context}

Provide a friendly, helpful response with specific recipe recommendations.
Include key ingredients and brief cooking instructions when possible."""

    response = gemini_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    return {"response": response.text}


def route_after_judge(state: MealPlannerState) -> Literal["web_search", "generate_response", "optimize_query"]:
    if state["relevance_score"] == "sufficient":
        return "generate_response"
    if state.get("retry_count", 0) >= 2:
        return "web_search"
    return "optimize_query"


def build_graph():
    graph = StateGraph(MealPlannerState)

    graph.add_node("optimize_query", optimize_query)
    graph.add_node("retrieve_recipes", retrieve_recipes)
    graph.add_node("judge_relevance", judge_relevance)
    graph.add_node("generate_response", generate_response)

    graph.add_edge(START, "optimize_query")
    graph.add_edge("optimize_query", "retrieve_recipes")
    graph.add_edge("retrieve_recipes", "judge_relevance")
    graph.add_conditional_edges("judge_relevance", route_after_judge)
    graph.add_edge("generate_response", END)

    return graph.compile()

Your RAG pipeline structure is complete. Next up, you will add the Tavily web search fallback so the web_search node actually works when local retrieval falls short.

Add Tavily Web Search Fallback

Your RAG pipeline can now retrieve recipes from ChromaDB and judge whether they match the user's request. But what happens when your local database of 500 recipes simply does not have what the user needs?

In this step, you will add a fallback path that searches the web using Tavily when local retrieval fails. The Judge node already routes to a web_search node after retries are exhausted. You just need to build that node and wire it into the graph.

In this step, get ready to:

• Add a web_search node function that queries Tavily for recipes online.
• Wire the web search node into the LangGraph StateGraph with proper edges.
• Verify that web results merge into the response generation.

Add the web_search node function

When the Judge determines that local retrieval has failed (after 2 retry attempts), the conditional routing sends the pipeline to a web_search node. This node uses the Tavily API to search the web for recipes that match the optimized query.

• In graph.py, find the judge_relevance function.
• Add the following web_search function below it (before generate_response):

def web_search(state: MealPlannerState) -> dict:
    # Build a search query by prepending "recipe" to the optimized query
    search_query = f"recipe {state['optimized_query']}"
    response = tavily_client.search(
        query=search_query,
        max_results=5,
        topic="general",
    )
    # Extract title, content, and URL from each result
    web_results = []
    for result in response.get("results", []):
        web_results.append({
            "title": result.get("title", ""),
            "content": result.get("content", ""),
            "url": result.get("url", ""),
        })
    return {"web_results": web_results}

What does this code do?

This function calls Tavily's search API to find recipes on the web when the local ChromaDB collection does not have good enough results.

• search_query prepends "recipe" to the optimized query so Tavily focuses on food-related pages.
• tavily_client.search() sends the query with max_results=5 and topic="general" to get the top 5 web pages matching the query.
• The function returns a list of dictionaries (title, content, URL) that gets stored in the web_results state field for the Respond node to use.

Why Tavily instead of a generic search API?

Tavily is purpose-built for AI applications. It returns clean, structured content snippets rather than raw HTML, which means you can feed results directly into an LLM prompt without parsing.

The free tier gives you 1,000 API credits per month, and each basic search costs 1 credit. That is plenty of budget for development and testing.

Wire the web search node into the graph

The web_search function exists, but the graph does not know about it yet. You need to register it as a node and add an edge so it connects to generate_response.

• In graph.py, scroll down to the build_graph() function.
• Find this line:

    graph.add_node("generate_response", generate_response)

• Add this line directly above it:

    graph.add_node("web_search", web_search)

• Now find this line:

    graph.add_edge("generate_response", END)

• Add this line directly above it:

    graph.add_edge("web_search", "generate_response")

What do these edges mean?

The route_after_judge function you built in the previous step already returns "web_search" when retry_count >= 2. Now that the node is registered, LangGraph can actually route to it.

The edge from web_search to generate_response ensures that after Tavily returns results, the pipeline always moves forward to generate the final answer.

• Save graph.py.
• Verify the graph compiles without errors by running this in your terminal:

python -c "from graph import build_graph; g = build_graph(); print('Graph compiled with nodes:', list(g.nodes.keys()))"

• You should see a list of node names including web_search in the output.

Seeing a NameError or missing node?

• Make sure your web_search function is defined before build_graph() in the file. Python reads top to bottom, so the function must exist before it is referenced.
• Check that graph.add_node("web_search", web_search) uses the exact string "web_search" to match what route_after_judge returns.
• Confirm your virtual environment is activated. If tavily cannot be imported, re-run pip install tavily-python.

Help me debug my web_search node in LangGraph

Understand how web results blend into the response

Your generate_response function (which you wrote in the previous step) already handles web results. No additional code changes are needed here.

• Scroll to the generate_response function in graph.py and locate this block:

    web_results = state.get("web_results", [])
    if web_results:
        context_parts.append("\nWeb search results:")
        for w in web_results:
            context_parts.append(f"- {w['title']}: {w['content'][:300]}")

How does the data flow end to end?

Here is the full fallback path through your graph:

• The Judge scores retrieval as "insufficient" and the retry count has reached 2.
• route_after_judge returns "web_search", sending the pipeline to the Tavily node.
• The web_search node queries Tavily and stores results in web_results.
• The edge sends the pipeline to generate_response, which combines local recipes and web results into a single prompt for Gemini.

✔️ Awesome, I've got everything!

Great. Double check you have saved graph.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete graph.py at this point in the project:

import os
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from google import genai
from google.genai import types
import chromadb
from tavily import TavilyClient
from dotenv import load_dotenv

load_dotenv()

gemini_client = genai.Client()
chroma_client = chromadb.PersistentClient(path="./chroma_db")
collection = chroma_client.get_or_create_collection(name="recipes", embedding_function=None)
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])


class MealPlannerState(TypedDict):
    user_query: str
    user_profile: dict
    optimized_query: str
    retrieved_recipes: list
    web_results: list
    relevance_score: str
    retry_count: int
    response: str


def optimize_query(state: MealPlannerState) -> dict:
    profile = state.get("user_profile", {})
    dietary_info = ""
    if profile:
        restrictions = profile.get("restrictions", [])
        preferences = profile.get("preferences", [])
        if restrictions:
            dietary_info += f" Dietary restrictions: {', '.join(restrictions)}."
        if preferences:
            dietary_info += f" Preferences: {', '.join(preferences)}."

    prompt = f"""Rewrite the following user query into an optimized search query for a recipe database.
Include relevant keywords for ingredients, cuisine type, and cooking method.
{dietary_info}

User query: {state['user_query']}

Return ONLY the optimized search query, nothing else."""

    response = gemini_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    return {"optimized_query": response.text.strip(), "retry_count": state.get("retry_count", 0) + 1}


def retrieve_recipes(state: MealPlannerState) -> dict:
    query_text = state["optimized_query"]
    query_embedding = gemini_client.models.embed_content(
        model="gemini-embedding-2",
        contents=query_text,
        config=types.EmbedContentConfig(output_dimensionality=768),
    )
    results = collection.query(
        query_embeddings=[query_embedding.embeddings[0].values],
        n_results=5,
        include=["documents", "metadatas", "distances"],
    )
    recipes = []
    if results["documents"] and results["documents"][0]:
        for i, doc in enumerate(results["documents"][0]):
            recipes.append({
                "text": doc,
                "metadata": results["metadatas"][0][i] if results["metadatas"] else {},
                "distance": results["distances"][0][i] if results["distances"] else 0,
            })
    return {"retrieved_recipes": recipes}


def judge_relevance(state: MealPlannerState) -> dict:
    recipes = state.get("retrieved_recipes", [])
    if not recipes:
        return {"relevance_score": "insufficient"}

    recipe_summaries = "\n".join(
        [f"- {r['metadata'].get('title', 'Unknown')}" for r in recipes[:5]]
    )
    prompt = f"""You are judging whether retrieved recipes are relevant to the user's request.

User query: {state['user_query']}
Optimized query: {state['optimized_query']}

Retrieved recipes:
{recipe_summaries}

Are these recipes relevant and helpful for the user's request?
Answer with exactly one word: "sufficient" or "insufficient"."""

    response = gemini_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    score = response.text.strip().lower()
    if "sufficient" in score and "insufficient" not in score:
        return {"relevance_score": "sufficient"}
    return {"relevance_score": "insufficient"}


def web_search(state: MealPlannerState) -> dict:
    search_query = f"recipe {state['optimized_query']}"
    response = tavily_client.search(
        query=search_query,
        max_results=5,
        topic="general",
    )
    web_results = []
    for result in response.get("results", []):
        web_results.append({
            "title": result.get("title", ""),
            "content": result.get("content", ""),
            "url": result.get("url", ""),
        })
    return {"web_results": web_results}


def generate_response(state: MealPlannerState) -> dict:
    profile = state.get("user_profile", {})
    dietary_info = ""
    if profile:
        restrictions = profile.get("restrictions", [])
        if restrictions:
            dietary_info = f"\nUser dietary restrictions: {', '.join(restrictions)}"

    context_parts = []
    recipes = state.get("retrieved_recipes", [])
    if recipes:
        context_parts.append("Local recipes found:")
        for r in recipes[:5]:
            context_parts.append(r["text"])

    web_results = state.get("web_results", [])
    if web_results:
        context_parts.append("\nWeb search results:")
        for w in web_results:
            context_parts.append(f"- {w['title']}: {w['content'][:300]}")

    context = "\n\n".join(context_parts)

    prompt = f"""You are a helpful meal planning assistant. Based on the following recipe information,
provide a personalized recommendation to the user.{dietary_info}

User request: {state['user_query']}

Available information:
{context}

Provide a friendly, helpful response with specific recipe recommendations.
Include key ingredients and brief cooking instructions when possible."""

    response = gemini_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    return {"response": response.text}


def route_after_judge(state: MealPlannerState) -> Literal["web_search", "generate_response", "optimize_query"]:
    if state["relevance_score"] == "sufficient":
        return "generate_response"
    if state.get("retry_count", 0) >= 2:
        return "web_search"
    return "optimize_query"


def build_graph():
    graph = StateGraph(MealPlannerState)

    graph.add_node("optimize_query", optimize_query)
    graph.add_node("retrieve_recipes", retrieve_recipes)
    graph.add_node("judge_relevance", judge_relevance)
    graph.add_node("web_search", web_search)
    graph.add_node("generate_response", generate_response)

    graph.add_edge(START, "optimize_query")
    graph.add_edge("optimize_query", "retrieve_recipes")
    graph.add_edge("retrieve_recipes", "judge_relevance")
    graph.add_conditional_edges("judge_relevance", route_after_judge)
    graph.add_edge("web_search", "generate_response")
    graph.add_edge("generate_response", END)

    return graph.compile()

Your pipeline now has a complete fallback strategy. When local recipes are not good enough, it automatically reaches out to the web. Next up, you will build the Streamlit chat interface so users can interact with this pipeline through a conversational UI.

Create the Streamlit Chat Interface

Your RAG pipeline is fully operational. It optimizes queries, retrieves recipes from ChromaDB, judges relevance, and falls back to web search when needed.

Now you need a way for users to actually talk to it. Streamlit is a Python framework that turns scripts into web apps with minimal frontend code. In this step, you'll build a chat interface where users set dietary preferences in a sidebar and ask for recipe recommendations.

In this step, get ready to:

• Set up the Streamlit app configuration and session state.
• Add a sidebar for dietary preferences and restrictions.
• Build the chat interface and connect it to the LangGraph pipeline.

Set up the app configuration and session state

Streamlit apps re-run the entire script on every user interaction. Session state lets you persist data (like chat history and the compiled graph) across those reruns.

• Create a new file called app.py in your meal-planner folder.
• Add the following code to configure the page and initialize session state:

import streamlit as st
from graph import build_graph

# Configure the page layout and metadata
st.set_page_config(page_title="Meal Planner AI", page_icon="🍽️", layout="wide")

# Initialize session state for chat history and the compiled graph
if "messages" not in st.session_state:
    st.session_state.messages = []
if "graph" not in st.session_state:
    st.session_state.graph = build_graph()

What does this code do?

• st.set_page_config() sets the browser tab title, favicon, and uses a wide layout so the chat has room to breathe.
• The two if blocks check whether messages and graph already exist in session state. If not, they create them. This prevents your chat history from being wiped on every rerun.
• build_graph() is the function you wrote in graph.py. Storing the compiled graph in session state means it only compiles once per session.

• Save app.py.

Add a sidebar for dietary preferences

The sidebar collects the user's dietary profile. These preferences get passed into the LangGraph state so the Optimize and Respond nodes can personalize recommendations.

• In app.py, add the following sidebar code below the session state initialization:

with st.sidebar:
    st.header("🥗 Your Profile")

    # Dietary preference options
    preferences = st.multiselect(
        "Dietary Preferences",
        ["Vegetarian", "Vegan", "Keto", "Paleo", "Mediterranean", "Low-carb", "High-protein"],
        default=[],
    )
    # Allergy and restriction options
    restrictions = st.multiselect(
        "Restrictions / Allergies",
        ["Gluten-free", "Dairy-free", "Nut-free", "Soy-free", "Egg-free", "Shellfish-free"],
        default=[],
    )
    # Daily calorie target input
    calorie_target = st.number_input(
        "Daily Calorie Target",
        min_value=1000,
        max_value=5000,
        value=2000,
        step=100,
    )

    # Placeholder save button (wired to SQLite in the next step)
    if st.button("Save Profile"):
        st.success("Profile saved!")

What does this code do?

• with st.sidebar: places all nested widgets in a collapsible left panel, keeping the main area free for the chat.
• st.multiselect() creates dropdown menus where users can select multiple dietary preferences or restrictions at once.
• st.number_input() renders a numeric field with increment/decrement buttons for the calorie target.
• The Save Profile button shows a success message for now. You'll wire it to SQLite in the next step.

• Save app.py.

Build the chat interface and connect the pipeline

This is where everything comes together. The chat interface displays message history and sends each new user query through your full LangGraph pipeline.

• In app.py, add the following code below the sidebar block:

# App title and description
st.title("🍽️ Meal Planner AI")
st.caption("Ask me for recipe recommendations based on your dietary preferences!")

# Display existing chat history
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

What does this code do?

• st.title() and st.caption() render the main heading and a subtle description at the top of the page.
• The for loop iterates over saved messages and re-renders each one using st.chat_message(). This keeps the full conversation visible after Streamlit reruns.

• Now add the chat input handler that invokes your LangGraph pipeline. Paste this below the history loop:

# Handle new user input
if prompt := st.chat_input("What would you like to eat?"):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    with st.chat_message("assistant"):
        with st.spinner("Finding recipes..."):
            # Build the user profile from sidebar selections
            user_profile = {
                "preferences": preferences,
                "restrictions": restrictions,
                "calorie_target": calorie_target,
            }
            # Invoke the full LangGraph pipeline
            result = st.session_state.graph.invoke({
                "user_query": prompt,
                "user_profile": user_profile,
                "optimized_query": "",
                "retrieved_recipes": [],
                "web_results": [],
                "relevance_score": "",
                "retry_count": 0,
                "response": "",
            })
            response_text = result.get("response", "I could not find a suitable recipe. Please try a different query.")
            st.markdown(response_text)

    st.session_state.messages.append({"role": "assistant", "content": response_text})

What does this code do?

• st.chat_input() renders a text box at the bottom of the page. It returns the user's message (or None if empty).
• The walrus operator (:=) assigns and checks in one step. If the user typed something, the block runs.
• st.session_state.graph.invoke() sends a full MealPlannerState dictionary through your entire LangGraph pipeline (Optimize, Retrieve, Judge, and either Respond or Web Search).
• The sidebar variables (preferences, restrictions, calorie_target) are bundled into user_profile so the pipeline can personalize its output.

• Save app.py.
• Launch the app by running this command in your terminal:

streamlit run app.py

Your browser should open automatically to http://localhost:8501. You should see the title "Meal Planner AI", a sidebar with dietary preference dropdowns, and a chat input box at the bottom.

• Type a message like quick vegan dinner into the chat box and press Enter.
• You should see a spinner that says "Finding recipes..." followed by a personalized recipe recommendation from the assistant.

App not loading or showing an error?

• If you see a ModuleNotFoundError, make sure your virtual environment is activated. The terminal prompt should show your venv name.
• If the page loads but the chat returns an error about missing API keys, confirm your .env file is in the same folder as app.py and contains valid keys.
• If you see an error about the ChromaDB collection being empty, make sure you ran ingest.py successfully in a previous step. The ./chroma_db folder should exist.

Help me debug my Streamlit app

✔️ Awesome, I've got everything!

Great. Double check you've saved app.py and your chat is responding to queries.

ⓧ I'd like to double check the full code

import streamlit as st
from graph import build_graph

st.set_page_config(page_title="Meal Planner AI", page_icon="🍽️", layout="wide")

if "messages" not in st.session_state:
    st.session_state.messages = []
if "graph" not in st.session_state:
    st.session_state.graph = build_graph()

with st.sidebar:
    st.header("🥗 Your Profile")

    preferences = st.multiselect(
        "Dietary Preferences",
        ["Vegetarian", "Vegan", "Keto", "Paleo", "Mediterranean", "Low-carb", "High-protein"],
        default=[],
    )
    restrictions = st.multiselect(
        "Restrictions / Allergies",
        ["Gluten-free", "Dairy-free", "Nut-free", "Soy-free", "Egg-free", "Shellfish-free"],
        default=[],
    )
    calorie_target = st.number_input(
        "Daily Calorie Target",
        min_value=1000,
        max_value=5000,
        value=2000,
        step=100,
    )

    if st.button("Save Profile"):
        st.success("Profile saved!")

st.title("🍽️ Meal Planner AI")
st.caption("Ask me for recipe recommendations based on your dietary preferences!")

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

if prompt := st.chat_input("What would you like to eat?"):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    with st.chat_message("assistant"):
        with st.spinner("Finding recipes..."):
            user_profile = {
                "preferences": preferences,
                "restrictions": restrictions,
                "calorie_target": calorie_target,
            }
            result = st.session_state.graph.invoke({
                "user_query": prompt,
                "user_profile": user_profile,
                "optimized_query": "",
                "retrieved_recipes": [],
                "web_results": [],
                "relevance_score": "",
                "retry_count": 0,
                "response": "",
            })
            response_text = result.get("response", "I could not find a suitable recipe. Please try a different query.")
            st.markdown(response_text)

    st.session_state.messages.append({"role": "assistant", "content": response_text})

Your meal planner is live and responding to queries. Next up, you'll add SQLite persistence so dietary preferences and meal history survive between sessions.

Add SQLite Persistence for User Profiles

Your meal planner works, but it has a major usability gap. Every time the Streamlit app refreshes, the learner's dietary preferences reset to empty defaults.

In this step, you will add SQLite persistence so user profiles and meal history survive across sessions. You will create a dedicated database module, then wire it into your Streamlit app.

In this step, get ready to:

• Create a SQLite database with tables for user profiles and meal history.
• Wire the sidebar save button to persist profile data.
• Load saved profiles on startup and record every recommendation.

Create the database module

SQLite stores data in a single file on disk with no separate server process. This makes it ideal for local applications like your meal planner.

• Create a new file called database.py in your meal-planner folder.
• Add the connection helper and table-creation function by pasting this code:

import sqlite3
import json
from datetime import datetime


def get_connection():
    conn = sqlite3.connect("meal_planner.db", check_same_thread=False)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS user_profiles (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT DEFAULT 'default',
            preferences TEXT DEFAULT '[]',
            restrictions TEXT DEFAULT '[]',
            calorie_target INTEGER DEFAULT 2000,
            updated_at TEXT
        )
    """)
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS meal_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER DEFAULT 1,
            date TEXT,
            recipe_title TEXT,
            query TEXT,
            response TEXT,
            created_at TEXT
        )
    """)
    conn.commit()
    conn.close()

What does this code do?

get_connection() opens (or creates) a file called meal_planner.db in your project folder. The check_same_thread=False flag lets Streamlit's multi-threaded runner access the database safely.

init_db() creates two tables if they do not already exist. user_profiles stores dietary preferences as JSON strings. meal_history records every recipe recommendation the assistant makes.

• Verify the database file is created by running this command:

python -c "from database import init_db; init_db()"

• Confirm the file exists by running:

ls meal_planner.db

• You should see meal_planner.db in the output.

Seeing ModuleNotFoundError?

Make sure you are running the command from inside your meal-planner folder where database.py lives. Also confirm your virtual environment is activated.

help me debug a ModuleNotFoundError when importing database.py

• Now add the three CRUD functions below the init_db() function in database.py:

def save_profile(preferences, restrictions, calorie_target=2000):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT id FROM user_profiles WHERE name = 'default'")
    existing = cursor.fetchone()
    now = datetime.now().isoformat()
    if existing:
        cursor.execute("""
            UPDATE user_profiles
            SET preferences = ?, restrictions = ?, calorie_target = ?, updated_at = ?
            WHERE name = 'default'
        """, (json.dumps(preferences), json.dumps(restrictions), calorie_target, now))
    else:
        cursor.execute("""
            INSERT INTO user_profiles (name, preferences, restrictions, calorie_target, updated_at)
            VALUES (?, ?, ?, ?, ?)
        """, ("default", json.dumps(preferences), json.dumps(restrictions), calorie_target, now))
    conn.commit()
    conn.close()


def load_profile():
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM user_profiles WHERE name = 'default'")
    row = cursor.fetchone()
    conn.close()
    if row:
        return {
            "preferences": json.loads(row["preferences"]),
            "restrictions": json.loads(row["restrictions"]),
            "calorie_target": row["calorie_target"],
        }
    return {"preferences": [], "restrictions": [], "calorie_target": 2000}


def save_meal(recipe_title, query, response):
    conn = get_connection()
    cursor = conn.cursor()
    now = datetime.now().isoformat()
    cursor.execute("""
        INSERT INTO meal_history (user_id, date, recipe_title, query, response, created_at)
        VALUES (?, ?, ?, ?, ?, ?)
    """, (1, now[:10], recipe_title, query, response, now))
    conn.commit()
    conn.close()

What do these functions do?

save_profile() uses an upsert pattern. It checks if a profile row already exists and updates it, or inserts a new one. Preferences and restrictions are stored as JSON strings.

load_profile() reads the saved profile and parses the JSON back into Python lists. If no profile exists yet, it returns safe defaults.

save_meal() logs every recommendation to the meal_history table. This lets you track what the assistant recommended over time.

• Save database.py.
• You won't see a change in the app yet. The next substep wires these functions into app.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before moving on.

ⓧ I'd like to double check the full code

import sqlite3
import json
from datetime import datetime


def get_connection():
    conn = sqlite3.connect("meal_planner.db", check_same_thread=False)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS user_profiles (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT DEFAULT 'default',
            preferences TEXT DEFAULT '[]',
            restrictions TEXT DEFAULT '[]',
            calorie_target INTEGER DEFAULT 2000,
            updated_at TEXT
        )
    """)
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS meal_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER DEFAULT 1,
            date TEXT,
            recipe_title TEXT,
            query TEXT,
            response TEXT,
            created_at TEXT
        )
    """)
    conn.commit()
    conn.close()


def save_profile(preferences, restrictions, calorie_target=2000):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT id FROM user_profiles WHERE name = 'default'")
    existing = cursor.fetchone()
    now = datetime.now().isoformat()
    if existing:
        cursor.execute("""
            UPDATE user_profiles
            SET preferences = ?, restrictions = ?, calorie_target = ?, updated_at = ?
            WHERE name = 'default'
        """, (json.dumps(preferences), json.dumps(restrictions), calorie_target, now))
    else:
        cursor.execute("""
            INSERT INTO user_profiles (name, preferences, restrictions, calorie_target, updated_at)
            VALUES (?, ?, ?, ?, ?)
        """, ("default", json.dumps(preferences), json.dumps(restrictions), calorie_target, now))
    conn.commit()
    conn.close()


def load_profile():
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM user_profiles WHERE name = 'default'")
    row = cursor.fetchone()
    conn.close()
    if row:
        return {
            "preferences": json.loads(row["preferences"]),
            "restrictions": json.loads(row["restrictions"]),
            "calorie_target": row["calorie_target"],
        }
    return {"preferences": [], "restrictions": [], "calorie_target": 2000}


def save_meal(recipe_title, query, response):
    conn = get_connection()
    cursor = conn.cursor()
    now = datetime.now().isoformat()
    cursor.execute("""
        INSERT INTO meal_history (user_id, date, recipe_title, query, response, created_at)
        VALUES (?, ?, ?, ?, ?, ?)
    """, (1, now[:10], recipe_title, query, response, now))
    conn.commit()
    conn.close()

Wire the sidebar save button to upsert the user profile

Right now, the Save Profile button shows a success toast but does not actually store anything. Let's connect it to the database.

• In app.py, find the import line at the top of the file:

from graph import build_graph

• Add a new import line directly below it, so the top of your file looks like this:

import streamlit as st
from graph import build_graph
from database import init_db, save_profile, load_profile, save_meal

• Next, find the st.set_page_config(...) line. Add init_db() on the very next line:

st.set_page_config(page_title="Meal Planner AI", page_icon="🍽️", layout="wide")

init_db()

Why call init_db() at the top?

Streamlit re-runs the entire script on every interaction. Calling init_db() at the top ensures the tables exist before any read or write attempts. The CREATE TABLE IF NOT EXISTS clause makes this safe to call repeatedly.

• Now find the Save Profile button block. It currently looks like this:

    if st.button("Save Profile"):
        st.success("Profile saved!")

• Replace it with the following to actually persist the profile:

    if st.button("Save Profile"):
        save_profile(preferences, restrictions, calorie_target)
        st.success("Profile saved!")

• Save app.py.
• You won't test this yet. The next substep adds profile loading so you can verify the full round trip.

Load the profile on app startup and record meal history

The save button now writes to SQLite, but the sidebar still defaults to empty lists on every refresh. You need to load the stored profile and use it as the default values.

• In app.py, find the sidebar block. It currently starts like this:

with st.sidebar:
    st.header("🥗 Your Profile")

    preferences = st.multiselect(

• Add a profile = load_profile() call right after the header, and update the three widget defaults to use the loaded profile. The updated sidebar block should look like this:

with st.sidebar:
    st.header("🥗 Your Profile")
    profile = load_profile()

    preferences = st.multiselect(
        "Dietary Preferences",
        ["Vegetarian", "Vegan", "Keto", "Paleo", "Mediterranean", "Low-carb", "High-protein"],
        default=profile.get("preferences", []),
    )
    restrictions = st.multiselect(
        "Restrictions / Allergies",
        ["Gluten-free", "Dairy-free", "Nut-free", "Soy-free", "Egg-free", "Shellfish-free"],
        default=profile.get("restrictions", []),
    )
    calorie_target = st.number_input(
        "Daily Calorie Target",
        min_value=1000,
        max_value=5000,
        value=profile.get("calorie_target", 2000),
        step=100,
    )

    if st.button("Save Profile"):
        save_profile(preferences, restrictions, calorie_target)
        st.success("Profile saved!")

What changed here?

load_profile() reads the saved profile from SQLite and returns a dictionary. The default=profile.get(...) calls populate each widget with the stored values instead of empty lists.

If no profile has been saved yet, load_profile() returns safe defaults (empty lists and 2000 calories). So the app works identically on first run.

• Finally, find the line where the assistant displays its response. It currently looks like this:

            response_text = result.get("response", "I could not find a suitable recipe. Please try a different query.")
            st.markdown(response_text)

• Add a save_meal() call directly after st.markdown(response_text), so the section looks like this:

            response_text = result.get("response", "I could not find a suitable recipe. Please try a different query.")
            st.markdown(response_text)
            save_meal("recommendation", prompt, response_text)

Why record meal history?

Every recommendation gets logged with the date, the user's original query, and the full response. This creates a history you can review later or use as training data for improved recommendations.

• Save app.py.
• Run your Streamlit app by running this command:

streamlit run app.py

• In the sidebar, select a few dietary preferences (e.g. Vegan and Gluten-free).
• Click Save Profile.
• Refresh the browser page. Your saved preferences should still appear in the sidebar widgets.

Preferences reset after refresh?

Check that profile = load_profile() appears before the st.multiselect calls in your sidebar block. If it comes after, the widgets use empty defaults before the profile loads.

Also verify that init_db() is called before load_profile(). Without the tables created first, the query inside load_profile will fail silently and return defaults.

help me debug why my Streamlit sidebar preferences reset after page refresh

✔️ Awesome, I've got everything!

Great. Double check you have saved both files before moving on.

ⓧ I'd like to double check the full code

import streamlit as st
from graph import build_graph
from database import init_db, save_profile, load_profile, save_meal

st.set_page_config(page_title="Meal Planner AI", page_icon="🍽️", layout="wide")

init_db()

if "messages" not in st.session_state:
    st.session_state.messages = []
if "graph" not in st.session_state:
    st.session_state.graph = build_graph()

with st.sidebar:
    st.header("🥗 Your Profile")
    profile = load_profile()

    preferences = st.multiselect(
        "Dietary Preferences",
        ["Vegetarian", "Vegan", "Keto", "Paleo", "Mediterranean", "Low-carb", "High-protein"],
        default=profile.get("preferences", []),
    )
    restrictions = st.multiselect(
        "Restrictions / Allergies",
        ["Gluten-free", "Dairy-free", "Nut-free", "Soy-free", "Egg-free", "Shellfish-free"],
        default=profile.get("restrictions", []),
    )
    calorie_target = st.number_input(
        "Daily Calorie Target",
        min_value=1000,
        max_value=5000,
        value=profile.get("calorie_target", 2000),
        step=100,
    )

    if st.button("Save Profile"):
        save_profile(preferences, restrictions, calorie_target)
        st.success("Profile saved!")

st.title("🍽️ Meal Planner AI")
st.caption("Ask me for recipe recommendations based on your dietary preferences!")

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

if prompt := st.chat_input("What would you like to eat?"):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    with st.chat_message("assistant"):
        with st.spinner("Finding recipes..."):
            user_profile = {
                "preferences": preferences,
                "restrictions": restrictions,
                "calorie_target": calorie_target,
            }
            result = st.session_state.graph.invoke({
                "user_query": prompt,
                "user_profile": user_profile,
                "optimized_query": "",
                "retrieved_recipes": [],
                "web_results": [],
                "relevance_score": "",
                "retry_count": 0,
                "response": "",
            })
            response_text = result.get("response", "I could not find a suitable recipe. Please try a different query.")
            st.markdown(response_text)
            save_meal("recommendation", prompt, response_text)

    st.session_state.messages.append({"role": "assistant", "content": response_text})

Your meal planner now remembers who it is talking to. Preferences survive page refreshes and every recommendation gets logged. Next up, you will tackle the Secret Mission and build a weekly meal plan generator that balances macros across 7 days.

Secret mission

Weekly Meal Plan Generator

Your RAG pipeline can recommend individual recipes. But what if you could generate an entire week of balanced meals with a single click? In this secret mission, you'll build a weekly meal plan generator that invokes the graph 7 times, ensures variety across days, tracks macronutrients, and persists the plan to SQLite.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. Both the Gemini and Tavily APIs use free tiers with no credit card required.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• Project directory on your Desktop (includes Python files, virtual environment, .env, chroma_db/, meal_planner.db, and recipes.csv).
• Google AI Studio API key (managed at aistudio.google.com/api-keys).
• Tavily API key (managed in the API Keys section of the Tavily Dashboard).

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the meal planner.

• Your project files remain on your Desktop.
• Your API keys stay active on the free tier. Neither Gemini nor Tavily charges you while idle.
• The ChromaDB database and SQLite database persist on disk and load automatically when you run the app again.

Pause - I'll come back to this later

Shut down any running processes to free up system memory. Your files and data stay on disk for when you return.

• Stop the Streamlit server by pressing Ctrl+C in the terminal where it is running.
• Deactivate your virtual environment by running:

deactivate

• Your project files, ChromaDB data, and SQLite database remain on your Desktop for next time.

Delete - I don't want to use this again

Remove all project resources and revoke your API keys to start completely fresh.

• Stop the Streamlit server by pressing Ctrl+C in the terminal if it is still running.
• Deactivate your virtual environment by running:

deactivate

• Delete the project directory by running the appropriate command for your operating system.

macOS

rm -rf ~/Desktop/meal-planner

Windows

rmdir /s /q %USERPROFILE%\Desktop\meal-planner

• Delete your Gemini API key. Go to aistudio.google.com/api-keys, find the key you created for this project, and delete it.
• Delete your Tavily API key. Log in to the Tavily Dashboard, navigate to the API Keys section, and delete the key you created for this project.

Nice Work!

Nice Work!

You just built a production-grade RAG application with intelligent orchestration powered by LangGraph. Your meal planning assistant can retrieve recipes, judge their relevance, retry with refined queries, and fall back to web search when needed.

You've learned how to:

• Built a multi-stage RAG pipeline with a LangGraph StateGraph containing four nodes (Optimize, Retrieve, Judge, Respond) connected by conditional edges.
• Implemented the LLM-as-judge pattern with conditional retry logic and Tavily web search as an adaptive fallback when local retrieval fails.
• Created a full Streamlit chat interface with sidebar profile management and SQLite persistence for user preferences and meal history.
• Generated balanced weekly meal plans with macro tracking across 7 days.

Ready to quiz yourself?