import streamlit as st import pymysql import myloginpath import ollama import bcrypt import requests import string import random import smtplib from email.message import EmailMessage import pandas as pd from unit_converter import UnitConverter def local_web_search(query: str) -> str: try: req = requests.get(f'http://127.0.0.1:8080/search', params={'q': query, 'format': 'json'}) if req.status_code == 200: data = req.json() results = data.get('results', []) if not results: return f"No results found on the web for '{query}'." snippets = [f"Source: {r.get('url')}\nContent: {r.get('content')}" for r in results[:3]] return "\n\n".join(snippets) return "Search engine returned an error." except Exception as e: return f"Local search engine unreachable: {e}" search_tool_schema = { 'type': 'function', 'function': { 'name': 'local_web_search', 'description': 'Search the internet for info not in DB.', 'parameters': {'type': 'object', 'properties': {'query': {'type': 'string'}}, 'required': ['query']}, }, } def search_nutrition_db(query: str) -> str: conn = get_db_connection('app_reader') if not conn: return "Database connection failed." try: with conn.cursor() as cursor: # Query products view via natural language match on core table sql = """ SELECT c.product_name, m.proteins_100g, m.fat_100g, m.carbohydrates_100g, m.sugars_100g FROM food_db.products_core c LEFT JOIN food_db.products_macros m ON c.code = m.code WHERE MATCH(c.product_name, c.ingredients_text) AGAINST(%s IN NATURAL LANGUAGE MODE) LIMIT 5 """ cursor.execute(sql, (query,)) results = cursor.fetchall() if not results: return f"No database records found for '{query}'." snippets = [] for r in results: snippets.append(f"- {r['product_name']}: Protein {r['proteins_100g']}g, Fat {r['fat_100g']}g, Carbs {r['carbohydrates_100g']}g, Sugars {r['sugars_100g']}g (per 100g)") return "\n".join(snippets) except Exception as e: return f"Database query failed: {e}" finally: conn.close() db_search_tool_schema = { 'type': 'function', 'function': { 'name': 'search_nutrition_db', 'description': 'Search the local medical nutrition database for product macros and ingredients. ALWAYS prioritize this over web search.', 'parameters': {'type': 'object', 'properties': {'query': {'type': 'string', 'description': 'The product or food name to search for (e.g. apple, chicken, bread)'}}, 'required': ['query']}, }, } def get_db_connection(login_path): try: conf = myloginpath.parse(login_path) if not conf or not conf.get('user'): st.error(f"⚠️ MySQL configuration missing for `{login_path}`. If you are testing locally on Windows, this app must be run on the Ubuntu server where `mysql_config_editor` is properly configured.") return None return pymysql.connect( host=conf.get('host', '127.0.0.1'), user=conf.get('user'), password=conf.get('password'), database='food_db', cursorclass=pymysql.cursors.DictCursor ) except Exception as e: st.error(f"Connection Failed: {e}") return None def verify_login(username, password): conn = get_db_connection('app_auth') if not conn: return False with conn.cursor() as cursor: cursor.execute("SELECT password_hash FROM users WHERE username = %s LIMIT 1", (username,)) result = cursor.fetchone() conn.close() if result: return bcrypt.checkpw(password.encode('utf-8'), result['password_hash'].encode('utf-8')) return False def get_user_id(username): conn = get_db_connection('app_auth') if not conn: return None with conn.cursor() as cursor: cursor.execute("SELECT id FROM users WHERE username = %s LIMIT 1", (username,)) result = cursor.fetchone() conn.close() return result['id'] if result else None def get_eav_profile(username): uid = get_user_id(username) if not uid: return [] conn = get_db_connection('app_auth') with conn.cursor() as cursor: cursor.execute("SELECT id, illness_health_condition_diet_dislikes_name as name, illness_health_condition_diet_dislikes_value as value FROM user_health_profiles WHERE user_id = %s", (uid,)) res = cursor.fetchall() conn.close() return res def get_user_limit(username): conn = get_db_connection('app_auth') if not conn: return "50" with conn.cursor() as cursor: cursor.execute("SELECT search_limit FROM users WHERE username = %s LIMIT 1", (username,)) result = cursor.fetchone() conn.close() return result['search_limit'] if (result and result['search_limit']) else "50" def register_user(username, password, email): conn = get_db_connection('app_auth') if not conn: return False hashed = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') try: with conn.cursor() as cursor: cursor.execute("INSERT INTO users (username, password_hash, email) VALUES (%s, %s, %s)", (username, hashed, email)) conn.commit() conn.close() send_email(email, "Welcome to Local Food AI", f"Hello {username}, your account was securely created!", to_name=username.title()) return True except pymysql.err.IntegrityError: return False def send_email(to_email, subject, body, to_name="User"): msg = EmailMessage() msg.set_content(body) msg['Subject'] = subject msg['From'] = '"Clinical Food AI System" ' msg['To'] = f'"{to_name}" <{to_email}>' import time for attempt in range(5): try: s = smtplib.SMTP('localhost', 25) s.send_message(msg) s.quit() return True except Exception as e: if attempt == 4: return f"SMTP Delivery Failed: {str(e)}" time.sleep(2) return "Unknown Error Occurred" def reset_password(username, email): conn = get_db_connection('app_auth') if not conn: return False with conn.cursor() as cursor: cursor.execute("SELECT id, email FROM users WHERE username = %s", (username,)) user = cursor.fetchone() if user and user['email'] == email: new_pass = ''.join(random.choices(string.ascii_letters + string.digits, k=10)) hashed = bcrypt.hashpw(new_pass.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') cursor.execute("UPDATE users SET password_hash = %s WHERE id = %s", (hashed, user['id'])) conn.commit() conn.close() status = send_email(email, "Password Reset", f"Your new temporary password is: {new_pass}", to_name=username.title()) if status is True: return True return status return False # UI Theming st.set_page_config(page_title="Food AI Explorer", page_icon="πŸ”", layout="wide") st.markdown(""" """, unsafe_allow_html=True) if "authenticated_user" not in st.session_state: st.session_state["authenticated_user"] = None with st.sidebar: st.title("User Portal πŸ”") if st.session_state["authenticated_user"]: st.success(f"Logged in as: {st.session_state['authenticated_user']}") if st.button("Logout"): st.session_state["authenticated_user"] = None st.rerun() st.markdown("---") st.subheader("πŸ₯ Dynamic Health Profile") eav_data = get_eav_profile(st.session_state["authenticated_user"]) uid = get_user_id(st.session_state["authenticated_user"]) user_lim = get_user_limit(st.session_state["authenticated_user"]) with st.expander("βš™οΈ Account Preferences"): opts = ["10", "20", "50", "100", "All"] idx = opts.index(user_lim) if user_lim in opts else 2 new_lim = st.selectbox("Default Search Limit", opts, index=idx) if new_lim != user_lim: conn = get_db_connection('app_auth') with conn.cursor() as c: c.execute("UPDATE users SET search_limit = %s WHERE id = %s", (new_lim, uid)) conn.commit() st.rerun() with st.expander("βž• Add Condition / Diet"): new_cat = st.selectbox("Category", ["Condition", "Illness", "Diet", "Dislike", "Allergy"]) new_val = st.text_input("Value (e.g. 'vegan', 'diabetes', 'broccoli')").strip().lower() if st.button("Add to Profile") and new_val and uid: conn = get_db_connection('app_auth') with conn.cursor() as c: c.execute("INSERT INTO user_health_profiles (user_id, illness_health_condition_diet_dislikes_name, illness_health_condition_diet_dislikes_value) VALUES (%s, %s, %s)", (uid, new_cat, new_val)) conn.commit() st.rerun() if eav_data: st.markdown("#### Active Flags") for e in eav_data: col1, col2 = st.columns([4, 1]) col1.info(f"**{e['name']}:** {e['value'].title()}") if col2.button("X", key=f"del_eav_{e['id']}"): conn = get_db_connection('app_auth') with conn.cursor() as c: c.execute("DELETE FROM user_health_profiles WHERE id = %s", (e['id'],)) conn.commit() st.rerun() else: tab1, tab2, tab3 = st.tabs(["Login", "Register", "Reset"]) with tab1: l_user = st.text_input("Username", key="l_user") l_pass = st.text_input("Password", type="password", key="l_pass") if st.button("Login"): if verify_login(l_user, l_pass): st.session_state["authenticated_user"] = l_user st.rerun() else: st.error("Invalid login.") with tab2: r_user = st.text_input("Username", key="r_user") r_email = st.text_input("Email Address", key="r_email") r_pass = st.text_input("Password", type="password", key="r_pass") if st.button("Register"): if len(r_pass) < 6: st.error("Password too short.") elif register_user(r_user, r_pass, r_email): st.success("Registered safely!") else: st.error("Username exists.") with tab3: f_user = st.text_input("Username", key="f_user") f_email = st.text_input("Registered Email", key="f_email") if st.button("Send Reset Link"): status = reset_password(f_user, f_email) if status is True: st.success("Password reset emailed.") else: st.error(f"Failed: {status}") if not st.session_state["authenticated_user"]: st.title("πŸ” Food AI Medical Explorer") st.info("Please login to interrogate the Clinical Data.") st.stop() st.title("πŸ” Food AI Clinical Explorer") conn_reader = get_db_connection('app_reader') tab_chat, tab_explore, tab_plate, tab_planner = st.tabs(["πŸ’¬ AI Chat", "πŸ”¬ Clinical Search", "🍽️ My Plate Builder", "πŸ€– AI Meal Planner"]) with tab_chat: st.subheader("Chat with the Context") if "messages" not in st.session_state: st.session_state["messages"] = [{"role": "assistant", "content": "How can I help you analyze the food data today?"}] for msg in st.session_state.messages: st.chat_message(msg["role"]).write(msg["content"]) if prompt := st.chat_input("Ask about the food items..."): st.session_state.messages.append({"role": "user", "content": prompt}) st.chat_message("user").write(prompt) user_eav = get_eav_profile(st.session_state["authenticated_user"]) profile_text = ", ".join([f"{p['name']}: {p['value']}" for p in user_eav]) if user_eav else "None" sys_prompt = f"""You are a helpful medical data analyst AI. The user has the following health profile / conditions: {profile_text}. You MUST act as a specialized clinical dietitian. Autonomously deduce what foods are recommended, forbidden, or accepted for these specific conditions and apply these rules to all your answers. ALWAYS query the local database using the search_nutrition_db tool to answer questions about food, macros, and nutrients before answering or searching the web! If it's not in the DB, you can use local_web_search.""" with st.spinner("Analyzing..."): try: temp_messages = [{"role": "system", "content": sys_prompt}] + [m for m in st.session_state.messages if m["role"] != "tool"] response = ollama.chat(model='mistral', messages=temp_messages, tools=[search_tool_schema, db_search_tool_schema]) if response.get('message', {}).get('tool_calls'): for tool in response['message']['tool_calls']: if tool['function']['name'] == 'local_web_search': query_arg = tool['function']['arguments'].get('query') st.info(f"πŸ” Web Search triggered for: '{query_arg}'") search_data = local_web_search(query_arg) st.session_state.messages.append(response['message']) st.session_state.messages.append({'role': 'tool', 'content': search_data, 'name': 'local_web_search'}) elif tool['function']['name'] == 'search_nutrition_db': query_arg = tool['function']['arguments'].get('query') st.info(f"πŸ—„οΈ Database Search triggered for: '{query_arg}'") db_data = search_nutrition_db(query_arg) st.session_state.messages.append(response['message']) st.session_state.messages.append({'role': 'tool', 'content': db_data, 'name': 'search_nutrition_db'}) temp_messages = [{"role": "system", "content": sys_prompt}] + st.session_state.messages response = ollama.chat(model='mistral', messages=temp_messages) ai_reply = response['message']['content'] except Exception as e: ai_reply = f"Hold on! Engine execution fault: {e}" st.session_state.messages.append({"role": "assistant", "content": ai_reply}) st.chat_message("assistant").write(ai_reply) def highlight_medical_warnings(row): try: val = str(row.get('Medical Warning', '')) if '⚠️' in val: return ['background-color: rgba(255, 0, 0, 0.4); color: white;'] * len(row) if 'πŸ’š' in val: return ['background-color: rgba(0, 255, 0, 0.3); color: white;'] * len(row) except: pass return [''] * len(row) with tab_explore: st.subheader("Clinical Data Search") sq = st.text_input("Search Product Name or Ingredient") cols = st.columns(5) min_pro = cols[0].number_input("Min Protein (g)", 0, 1000, 0) min_fat = cols[1].number_input("Min Fat (g)", 0, 1000, 0) min_carb = cols[2].number_input("Min Carbs (g)", 0, 1000, 0) max_sug = cols[3].number_input("Max Sugar (g)", 0, 1000, 1000) # Load dynamically fetched limit as index opts = [10, 20, 50, 100, "All"] user_lim_str = get_user_limit(st.session_state["authenticated_user"]) user_lim_val = "All" if user_lim_str == "All" else int(user_lim_str) idx = opts.index(user_lim_val) if user_lim_val in opts else 2 limit_rc = cols[4].selectbox("Limit Results", opts, index=idx) if st.button("Search Database") and sq and conn_reader: with st.spinner("Processing massive clinical query..."): try: with conn_reader.cursor() as cursor: l_str = "" if limit_rc == "All" else f"LIMIT {limit_rc}" query = f""" SELECT c.code, c.product_name, c.generic_name, c.brands, c.ingredients_text, a.allergens, m.`energy-kcal_100g`, m.proteins_100g, m.fat_100g, m.carbohydrates_100g, m.sugars_100g, m.fiber_100g, m.sodium_100g, m.salt_100g, m.cholesterol_100g, v.`vitamin-a_100g`, v.`vitamin-b1_100g`, v.`vitamin-b2_100g`, v.`vitamin-pp_100g`, v.`vitamin-b6_100g`, v.`vitamin-b9_100g`, v.`vitamin-b12_100g`, v.`vitamin-c_100g`, v.`vitamin-d_100g`, v.`vitamin-e_100g`, v.`vitamin-k_100g`, min.calcium_100g, min.iron_100g, min.magnesium_100g, min.potassium_100g, min.zinc_100g FROM food_db.products_core c LEFT JOIN food_db.products_allergens a ON c.code = a.code LEFT JOIN food_db.products_macros m ON c.code = m.code LEFT JOIN food_db.products_vitamins v ON c.code = v.code LEFT JOIN food_db.products_minerals min ON c.code = min.code WHERE MATCH(c.product_name, c.ingredients_text) AGAINST(%s IN NATURAL LANGUAGE MODE) AND (m.proteins_100g >= %s OR m.proteins_100g IS NULL) AND (m.fat_100g >= %s OR m.fat_100g IS NULL) AND (m.carbohydrates_100g >= %s OR m.carbohydrates_100g IS NULL) AND (m.sugars_100g <= %s OR m.sugars_100g IS NULL) {l_str} """ cursor.execute(query, (sq, min_pro, min_fat, min_carb, max_sug)) results = cursor.fetchall() if results: # Fetch EAV Medical Profile eav_profile = get_eav_profile(st.session_state["authenticated_user"]) df = pd.DataFrame(results) st.markdown("### πŸ› οΈ Dynamic Column Display") default_columns = [ 'code', 'product_name', 'generic_name', 'brands', 'allergens', 'ingredients_text', 'proteins_100g', 'fat_100g', 'carbohydrates_100g', 'sugars_100g', 'sodium_100g', 'energy-kcal_100g', 'vitamin-c_100g', 'iron_100g', 'calcium_100g' ] all_fetched_cols = list(df.columns) valid_defaults = [c for c in default_columns if c in all_fetched_cols] if "selected_columns" not in st.session_state or st.button("Reset Default Columns"): st.session_state["selected_columns"] = valid_defaults st.rerun() chosen_cols = st.multiselect("Customize Dataset View", all_fetched_cols, default=st.session_state["selected_columns"], key="multi_cols") st.session_state["selected_columns"] = chosen_cols # Filter dataframe gracefully, but we retain a copy for background analytics df_display = df[chosen_cols].copy() warnings_col = [] for idx, row in df.iterrows(): warns = [] ing_text = str(row['ingredients_text']).lower() all_text = str(row['allergens']).lower() for param in eav_profile: cat = param['name'].lower() val = param['value'] # Disease Analytics if cat == 'illness': if val == 'diabetes' and pd.notnull(row.get('sugars_100g')) and float(row['sugars_100g']) > 10.0: warns.append("⚠️ High Sugar (Diabetes)") if (val == 'hypertension' or val == 'high bp') and pd.notnull(row.get('sodium_100g')) and float(row['sodium_100g']) > 1.5: warns.append("⚠️ High Salt (Hypertension)") if val == 'scurvy' and pd.notnull(row.get('vitamin-c_100g')) and float(row['vitamin-c_100g']) > 0.005: warns.append("πŸ’š High Vitamin C (Scurvy Recommended)") if val == 'anemia' and pd.notnull(row.get('iron_100g')) and float(row['iron_100g']) > 0.002: warns.append("πŸ’š High Iron (Anemia Recommended)") # Condition Analytics if cat == 'condition': if val == 'pregnant': if ('cru' in ing_text or 'raw' in ing_text or 'viande crue' in ing_text): warns.append("⚠️ Raw Foods (Pregnancy Toxoplasmosis)") if pd.notnull(row.get('iron_100g')) and float(row['iron_100g']) > 0.002: warns.append("πŸ’š Med-High Iron (Pregnancy Health)") if val == 'low fat' and pd.notnull(row.get('fat_100g')) and float(row['fat_100g']) > 20.0: warns.append("⚠️ High Fat") if val == 'osteoporosis' and pd.notnull(row.get('calcium_100g')) and float(row['calcium_100g']) > 0.1: warns.append("πŸ’š High Calcium (Bone Health)") if eav_data: ing_text = str(row.get('ingredients_text', '')).lower() all_text = str(row.get('allergens', '')).lower() product_name_text = str(row.get('product_name', '')).lower() for e in eav_data: cat = str(e['name']).lower() val = str(e['value']).lower() # Clinical Trace Checks... if cat == 'condition' and (val == 'pregnant' or val == 'pregnancy' or val == 'breastfeeding'): # Forbidden / High Risk (Toxoplasmosis & Listeria) if any(x in ing_text or x in product_name_text for x in ['cru', 'raw', 'viande crue', 'sushi', 'sashimi', 'poisson cru']): warns.append("⚠️ Forbidden: Raw Meat/Fish (Toxoplasmosis/Parasite Risk)") if any(x in ing_text or x in product_name_text for x in ['lait cru', 'unpasteurized', 'non pasteurisΓ©']): warns.append("⚠️ Forbidden: Unpasteurized Dairy (Listeria Risk)") if any(x in ing_text or x in product_name_text for x in ['alcool', 'wine', 'alcohol', 'beer']): warns.append("⚠️ Forbidden: Contains Alcohol") # Recommended (Iron & Calcium) if float(row.get('iron_100g', 0) or 0) > 0.003: warns.append("πŸ’š Recommended: High Iron (Pregnancy Health)") if float(row.get('calcium_100g', 0) or 0) > 0.120: warns.append("πŸ’š Recommended: High Calcium (Bone Health / Breastfeeding)") if cat == 'illness' and val == 'osteoporosis': if float(row.get('calcium_100g', 0) or 0) < 0.120: warns.append("⚠️ Low Calcium (Osteoporosis Risk)") else: warns.append("πŸ’š Recommended (High Calcium)") if cat == 'illness' and val == 'scurvy': if float(row.get('vitamin-c_100g', 0) or 0) < 0.010: warns.append("⚠️ Low Vitamin C (Scurvy Risk)") else: warns.append("πŸ’š Recommended (High Vitamin C)") if cat == 'diet' and val in ['vegan', 'vegetarian']: if any(x in ing_text for x in ['meat', 'beef', 'chicken', 'fish', 'gelatin', 'whey', 'pork', 'porc', 'poulet']): warns.append("⚠️ Contains Animal Products") if cat == 'diet' and val == 'halal': if any(x in ing_text for x in ['pork', 'pig', 'porc', 'wine', 'alcohol', 'beer', 'vin']): warns.append("⚠️ Probable Haram Ingredients (e.g. Pork/Wine)") if cat in ['dislike', 'allergy']: if val in ing_text or val in all_text or val in product_name_text: warns.append(f"⚠️ Contains: {val.upper()}") warnings_col.append(" | ".join(list(set(warns))) if warns else "βœ… Safe for Profile") df_display.insert(0, 'Medical Warning', warnings_col) styled_df = df_display.style.apply(highlight_medical_warnings, axis=1) st.success(f"Analysed {len(results)} records utilizing dynamic Partitions!") st.dataframe(styled_df, use_container_width=True) if st.button("πŸ€– Ask AI to Evaluate This Table"): with st.spinner("AI is dynamically evaluating these records against your profile..."): user_eav = get_eav_profile(st.session_state["authenticated_user"]) profile_text = ", ".join([f"{p['name']}: {p['value']}" for p in user_eav]) if user_eav else "None" eval_prompt = f"The user has this profile: {profile_text}. Evaluate these foods and state which are highly recommended or strictly forbidden: {df_display.to_dict('records')}" try: response = ollama.chat(model='mistral', messages=[{'role': 'user', 'content': eval_prompt}]) st.info(response['message']['content']) except Exception as e: st.error(f"AI Evaluation Failed: {e}") else: st.warning("No products found matching those strict terms.") except Exception as e: st.error(f"SQL/Pandas Error: {e}") with tab_plate: st.subheader("🍽️ My Plate Builder") uid = get_user_id(st.session_state["authenticated_user"]) conn = get_db_connection('app_auth') if conn and uid: with conn.cursor() as cursor: cursor.execute("SELECT id, plate_name FROM plates WHERE user_id = %s", (uid,)) plates = cursor.fetchall() with st.expander("βž• Create a New Plate"): new_plate_name = st.text_input("Plate Name") if st.button("Create Plate"): cursor.execute("INSERT INTO plates (user_id, plate_name) VALUES (%s, %s)", (uid, new_plate_name)) conn.commit() st.rerun() if plates: selected_plate = st.selectbox("Select Active Plate", [p['plate_name'] for p in plates]) active_p_id = next(p['id'] for p in plates if p['plate_name'] == selected_plate) cursor.execute(""" SELECT i.id, i.product_code, i.quantity_grams, p.product_name, p.proteins_100g, p.fat_100g, p.carbohydrates_100g FROM plate_items i LEFT JOIN products p ON i.product_code = p.code WHERE i.plate_id = %s """, (active_p_id,)) items = cursor.fetchall() if items: st.dataframe(items, use_container_width=True) total_pro = sum((float(i['proteins_100g'] or 0) * (float(i['quantity_grams'])/100.0)) for i in items) total_fat = sum((float(i['fat_100g'] or 0) * (float(i['quantity_grams'])/100.0)) for i in items) total_carb = sum((float(i['carbohydrates_100g'] or 0) * (float(i['quantity_grams'])/100.0)) for i in items) st.info(f"**Total Protein:** {total_pro:.1f}g | **Total Fat:** {total_fat:.1f}g | **Total Carbs:** {total_carb:.1f}g") st.markdown("---") st.markdown("#### βž• Add Food to Plate") add_search = st.text_input("Search Product Name") if add_search: cursor.execute("SELECT code, product_name FROM food_db.products_core WHERE MATCH(product_name, ingredients_text) AGAINST(%s IN NATURAL LANGUAGE MODE) LIMIT 10", (add_search,)) search_res = cursor.fetchall() if search_res: options = {f"{r['product_name']} ({r['code']})": r for r in search_res} selected_str = st.selectbox("Select Product", list(options.keys())) selected_product = options[selected_str] add_amount_str = st.text_input("Portion Quantity (e.g., '100g', '2 tbsp', '1.5 cups', '1 pinch')", value="100g") if st.button("Add Item to Plate"): # Use UnitConverter to parse grams = UnitConverter.parse_and_convert(add_amount_str, product_name=selected_product['product_name']) if grams is not None: cursor.execute("INSERT INTO plate_items (plate_id, product_code, quantity_grams) VALUES (%s, %s, %s)", (active_p_id, selected_product['code'], grams)) conn.commit() st.success(f"Added {grams}g of {selected_product['product_name']}!") st.rerun() else: st.error("Could not parse unit. Please use format like '100g' or '1 cup'.") else: st.warning("No products found.") with tab_planner: st.subheader("πŸ€– AI Meal Planner") p_col1, p_col2, p_col3 = st.columns(3) target_cal = p_col1.number_input("Target Daily Calories (kcal)", 1000, 5000, 2000, 50) diet_pref = p_col2.selectbox("Dietary Preference", ["Omnivore", "Vegetarian", "Vegan", "Keto", "Paleo"]) meal_count = p_col3.slider("Number of Meals", 2, 6, 3) extra_notes = st.text_input("Any additional allergies or goals?") if st.button("Generate Professional Menu"): with st.spinner("AI is formulating and interrogating the local database..."): user_eav = get_eav_profile(st.session_state["authenticated_user"]) profile_text = ", ".join([f"{p['name']}: {p['value']}" for p in user_eav]) if user_eav else "None" sys_prompt = f"""You are a professional clinical Dietitian planner. Target: {target_cal}kcal over {meal_count} meals. Dietary constraint: {diet_pref}. Additional notes: {extra_notes}. The user has the following health profile / conditions: {profile_text}. You MUST autonomously deduce what foods are recommended, forbidden, or accepted for these specific conditions and ensure the menu perfectly respects their medical requirements! CRITICAL INSTRUCTIONS: - YOU MUST USE the `search_nutrition_db` tool to find real products and their exact macros before constructing the menu! - If you cannot find appropriate products in the local DB, use `local_web_search`. - ALWAYS output exactly as a strict Markdown table including Columns: | Meal | Food | Calories | Salt | Fat | Iron | - DO NOT output | separated text outside of standard strict markdown block ` ```markdown ` or standard rendering. - Convert ALL cooking measurements to Grams (g). Use these equivalents STRICTLY: 1 tbsp = 15g, 1 tsp = 5g, 1 cup = 200g, 1 mustard glass = 100g. 1 cl of liquid = 10g. """ temp_messages = [{'role': 'system', 'content': sys_prompt}, {'role': 'user', 'content': 'Generate my meal plan. Find real foods from the DB.'}] response = ollama.chat(model='mistral', messages=temp_messages, tools=[search_tool_schema, db_search_tool_schema]) # Simple loop to handle multiple tool calls (up to 3 times to prevent infinite loops) for _ in range(3): if response.get('message', {}).get('tool_calls'): temp_messages.append(response['message']) for tool in response['message']['tool_calls']: if tool['function']['name'] == 'local_web_search': query_arg = tool['function']['arguments'].get('query') st.info(f"πŸ” Planner Web Search triggered for: '{query_arg}'") search_data = local_web_search(query_arg) temp_messages.append({'role': 'tool', 'content': search_data, 'name': 'local_web_search'}) elif tool['function']['name'] == 'search_nutrition_db': query_arg = tool['function']['arguments'].get('query') st.info(f"πŸ—„οΈ Planner DB Search triggered for: '{query_arg}'") db_data = search_nutrition_db(query_arg) temp_messages.append({'role': 'tool', 'content': db_data, 'name': 'search_nutrition_db'}) response = ollama.chat(model='mistral', messages=temp_messages, tools=[search_tool_schema, db_search_tool_schema]) else: break st.markdown(response['message']['content']) if conn_reader: conn_reader.close()