Hi guys - I'm trying to create a GraphRAG template, but it seems it dosen't get the variables right.
the goal is to get the LLM to create a recipe based on the nutritions and the correct amount of them, of the user in the database. The inputs are latest with the interaction of the user.
let me know if you need any more specifications.
Thanks a lot in adcanve!
PS: I haven't done the academy in LLM yet. It is on my to do.
I tried the documentation with the .format() but this did not work.
Thanks guys - regards, Gabriel
here my error:
Verbindung zur Neo4j-Datenbank erfolgreich.
Index 'text_embeddings' wurde erfolgreich erstellt.
VectorRetriever erfolgreich initialisiert.
Available users:
- user1
- user2
Select a user by number: 1
Available plans for user1: - muscle
- define
Select a plan by number: 1
Available recipe types (menge_typ) for plan 'muscle' of user1: - Frühstück: Continental
- Frühstück: Smoothie/Müesli
- Abendessen
- Krafttraining - Preworkout
- Jokermahlzeit
- Mittagessen
- Krafttraining - Workout
Select a recipe type (menge_typ) by number: 3
Enter preferred cuisine type (e.g., Italian, Asian, etc.): Asian
Enter diet preference (e.g., Vegetarian, Vegan, With Meat): Vegan
Traceback (most recent call last):
File "/app.py", line 237, in
recipe_suggestion = generate_recipe(selected_user, selected_menu_type, selected_plan, cuisine_type, diet_preference)
File "/app.py", line 198, in generate_recipe
prompt = prompt_template.format(
TypeError: RagTemplate.format() got an unexpected keyword argument 'cuisine_type'
Failed to write data to connection ResolvedIPv4Address()
Failed to write data to connection IPv4Address()
here my code, I guess it's no rocket science:
from dotenv import load_dotenv
import os
import neo4j
from neo4j_graphrag.llm import OpenAILLM
from neo4j_graphrag.embeddings.openai import OpenAIEmbeddings
from neo4j_graphrag.generation import RagTemplate
from neo4j_graphrag.generation.graphrag import GraphRAG
import json
from neo4j_graphrag.indexes import create_vector_index
from neo4j_graphrag.retrievers import VectorRetriever
# Load environment variables
load_dotenv('.env', override=True)
NEO4J_URI = os.getenv('NEO4J_URI')
NEO4J_USERNAME = os.getenv('NEO4J_USERNAME')
NEO4J_PASSWORD = os.getenv('NEO4J_PASSWORD')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
# Initialize Neo4j driver
driver = neo4j.GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USERNAME, NEO4J_PASSWORD))
# Create text embedder
embedder = OpenAIEmbeddings()
# Set up LLM-based RAG templates for question answering
llm = OpenAILLM(model_name="gpt-4o", model_params={"temperature": 0.0})
# Verify Neo4j connection
try:
with driver.session() as session:
session.run("RETURN 1")
print("Verbindung zur Neo4j-Datenbank erfolgreich.")
except Exception as e:
print(f"Fehler bei der Verbindung zur Datenbank: {e}")
# Create text_embeddings index if it does not exist
try:
create_vector_index(driver, name="text_embeddings", label="Chunk",
embedding_property="embedding", dimensions=1536, similarity_fn="cosine")
print("Index 'text_embeddings' wurde erfolgreich erstellt.")
except Exception as e:
print(f"Fehler beim Erstellen des Indexes: {e}")
# Initialize VectorRetriever after creating the index
try:
vector_retriever = VectorRetriever(
driver,
index_name="text_embeddings",
embedder=embedder,
return_properties=["text"],
)
print("VectorRetriever erfolgreich initialisiert.")
except Exception as e:
print(f"Fehler bei der Initialisierung von VectorRetriever: {e}")
# Initialize LLM with OpenAI API
ex_llm = OpenAILLM(
model_name="gpt-4o-mini",
model_params={
"response_format": {"type": "json_object"},
"temperature": 0
}
)
# Define node labels and relationship types
basic_node_labels = ["Nutzer", "Zutat", "Ernährungsplan", "Supplement", "Rezept", "RezeptTyp"]
node_labels = basic_node_labels
rel_types = ["ENTHÄLT", "HAT_PLAN", "VERBUNDEN_MIT", "HAT_TYP"]
"""
GraphRAG
"""
# Define node labels and relationship types
basic_node_labels = ["Nutzer", "Zutat", "Ernährungsplan", "Supplement", "Rezept", "RezeptTyp"]
node_labels = basic_node_labels
rel_types = ["ENTHÄLT", "HAT_PLAN", "VERBUNDEN_MIT", "HAT_TYP"]
# Step 1: List all available users for selection
def list_users():
with driver.session() as session:
result = session.run("MATCH (n:Nutzer) RETURN n.user_email AS user_email")
users = [record["user_email"] for record in result]
return users
def list_plans_for_user(user_email):
with driver.session() as session:
query = f"""
MATCH (n:Nutzer {{user_email: '{user_email}'}})-[:VERBUNDEN_MIT*0..1]-(connected:Nutzer)
MATCH (connected)-[:HAT_PLAN]->(p:Ernährungsplan)
RETURN DISTINCT p.name AS ernährungsplantyp
"""
result = session.run(query)
plans = [record["ernährungsplantyp"] for record in result]
return plans
# Step 3: List available recipe types (menge_typ) based on the selected user and plan
def list_menu_types(user_email, ernährungsplan):
with driver.session() as session:
query = f"""
MATCH (n:Nutzer {{user_email: '{user_email}'}})-[:VERBUNDEN_MIT*0..1]-(connected:Nutzer)
MATCH (connected)-[:HAT_PLAN]->(p:Ernährungsplan {{name: '{ernährungsplan}'}})
MATCH (p)-[z:ENTHÄLT]->(:Zutat)
RETURN DISTINCT z.menge_typ AS menge_typ
"""
result = session.run(query)
menu_types = [record["menge_typ"] for record in result]
return menu_types
# Step 4: Retrieve ingredients based on user, menu type, and selected plan
def get_ingredients_by_category(user_email, menu_type, ernährungsplan):
with driver.session() as session:
ingredient_query = f"""
MATCH (n:Nutzer {{user_email: '{user_email}'}})-[:VERBUNDEN_MIT*0..1]-(connected:Nutzer)
MATCH (connected)-[:HAT_PLAN]->(p:Ernährungsplan {{name: '{ernährungsplan}'}})
MATCH (p)-[r:ENTHÄLT]->(z:Zutat)
WHERE r.menge_typ = '{menu_type}'
RETURN z.kategorie AS category, z.name AS ingredient, z.kcal_pro_g AS kcal_per_g, r.menge AS amount, r.unit AS unit
"""
ingredients_by_category = {}
for record in session.run(ingredient_query):
category = record["category"]
kcal_per_g = record["kcal_per_g"]
amount = record["amount"]
total_kcal = kcal_per_g * amount
# Add ingredient to category and limit by calories per category
if category not in ingredients_by_category:
ingredients_by_category[category] = []
ingredients_by_category[category].append({
"ingredient": record["ingredient"],
"amount": amount,
"unit": record["unit"],
"kcal": total_kcal
})
return ingredients_by_category
from neo4j_graphrag.generation import RagTemplate, GraphRAG
# Definiere das Prompt-Template mit den erforderlichen Platzhaltern
prompt_template = RagTemplate(
template="""
You are a nutrition assistant tasked with creating a recipe based on the user's dietary plan.
Create a recipe for {query_text}, considering the following preferences:
Preferences:
- Cuisine Type: {cuisine_type}
- Diet Preference: {diet_preference}
- Possible ingredients with kcal information: {context}
Recipe Format:
{{
"menu_type": "{query_text}",
"recipe_name": "Generated Recipe Name",
"ingredients": [
{{
"name": "ingredient_name",
"quantity": "specified_quantity",
"unit": "measurement_unit"
}}
],
"preparation_steps": [
"Step 1: Description of the first step.",
"Step 2: Description of the second step.",
"and all other steps of the recipe"
]
}}
""",
expected_inputs=["query_text", "cuisine_type", "diet_preference", "context"]
)
# Initialize GraphRAG with the VectorRetriever and LLM
rag = GraphRAG(retriever=vector_retriever, llm=llm, prompt_template=prompt_template)
# Function to generate a recipe based on user inputs
def generate_recipe(user_email, menu_type, ernährungsplan, cuisine_type, diet_preference):
# Retrieve ingredients by category with kcal limits
ingredients_by_category = get_ingredients_by_category(user_email, menu_type, ernährungsplan)
# Format ingredients list as context text
context = "\n".join([
f"Category: {category}\n" +
"\n".join([f"{ing['ingredient']} ({ing['amount']} {ing['unit']}, {ing['kcal']} kcal)" for ing in ingredients])
for category, ingredients in ingredients_by_category.items()
])
# Fill the template with the required values
prompt = prompt_template.format(
query_text=menu_type,
cuisine_type=cuisine_type,
diet_preference=diet_preference,
context=context
)
# Perform the GraphRAG search with formatted prompt
recipe_result = rag.search(
query_text=prompt
)
return recipe_result.answer
# Interactive user, plan, recipe type, cuisine, and dietary preferences selection
users = list_users()
print("Available users:")
for idx, user in enumerate(users, 1):
print(f"{idx}. {user}")
selected_user = users[int(input("Select a user by number: ")) - 1]
# List available plans for the selected user
plans = list_plans_for_user(selected_user)
print(f"Available plans for {selected_user}:")
for idx, plan in enumerate(plans, 1):
print(f"{idx}. {plan}")
selected_plan = plans[int(input("Select a plan by number: ")) - 1]
# List recipe types based on the selected user and plan
menu_types = list_menu_types(selected_user, selected_plan)
print(f"Available recipe types (menge_typ) for plan '{selected_plan}' of {selected_user}:")
for idx, menu_type in enumerate(menu_types, 1):
print(f"{idx}. {menu_type}")
selected_menu_type = menu_types[int(input("Select a recipe type (menge_typ) by number: ")) - 1]
cuisine_type = input("Enter preferred cuisine type (e.g., Italian, Asian, etc.): ")
diet_preference = input("Enter diet preference (e.g., Vegetarian, Vegan, With Meat): ")
# Generate the recipe with the selected parameters
recipe_suggestion = generate_recipe(selected_user, selected_menu_type, selected_plan, cuisine_type, diet_preference)
print(f"Generated Recipe Suggestion for {selected_menu_type}:\n{recipe_suggestion}")