Create app.py

app.py

@@ -0,0 +1,202 @@
+import os
+import re
+from langgraph.checkpoint.sqlite import SqliteSaver
+from langgraph.graph import StateGraph, END
+from langchain_core.messages import SystemMessage, HumanMessage
+from langchain_core.pydantic_v1 import BaseModel
+from tavily import TavilyClient
+from typing import TypedDict, List
+from langchain_google_genai import ChatGoogleGenerativeAI
+from dotenv import load_dotenv
+import sys
+import io
+import gradio as gr
+
+load_dotenv()
+
+memory = SqliteSaver.from_conn_string(":memory:")
+
+class AgentState(TypedDict):
+    task: str
+    plan: str
+    draft: str
+    critique: str
+    content: List[str]
+    revision_number: int
+    max_revisions: int
+
+model = ChatGoogleGenerativeAI(model="gemini-1.5-pro", temperature=0.4)
+
+PLAN_PROMPT = """You are an expert meal outline planner tasked with creating a meal plan outline. 
+Give the outline of the meal plan along with any relevant notes, calories,
+recipes based on user preferences, shopping list based on ingredients, available ingredients or instructions for the recipe."""
+
+WRITER_PROMPT = """You are an excellent meal planner generator tasked with writing excellent meal plans with schedules.
+Write a detailed and concise final meal plan Following this template: 
+Breakfast -
+Lunch -
+Dinner -
+Add optional snacks in between these times.
+Please include the shopping list, calories, protein, and ingredients for the meal plan. 
+Generate the best meal plan possible for the user's request based on the provided template,
+Provide every detail concisely. 
+If the user provides critique, respond with a revised version of your previous attempts.
+Use all the information below as needed: 
+------ 
+{content}"""
+
+REFLECTION_PROMPT = """You are a critic reviewing a meal plan. 
+Generate critique and recommendations for the user's meal plan. 
+Select the best recipes considering nutritional requirements and dietary restrictions. 
+Filter the recipes to ensure they meet the user's nutritional requirements and dietary restrictions considering calories and protein."""
+
+RESEARCH_PLAN_PROMPT = """You are a researcher tasked with providing information to be used in writing a detailed meal plan according to the user meal plan outline. 
+Generate a list of search queries to gather relevant information regarding calories, protein, ingredients, and recipes. Generate a maximum of 3 queries."""
+
+RESEARCH_CRITIQUE_PROMPT = """You are a researcher charged with providing information that can 
+be used when making any requested revisions. Generate a list of search queries to gather relevant information. Generate a maximum of 3 queries."""
+
+class Queries(BaseModel):
+    queries: List[str]
+
+tavily = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
+
+
+def plan_node(state: AgentState):
+    messages = [
+        SystemMessage(content=PLAN_PROMPT),
+        HumanMessage(content=state['task'])
+    ]
+    response = model.invoke(messages)
+    plan_node_result = print("Plan agent Response: ", response.content)
+    plan_node_result
+    return {"plan": response.content}
+
+def parse_queries(response_content):
+    # Extract queries from the response content using regex
+    pattern = r'\*\*"(.*?)"\*\*'
+    queries = re.findall(pattern, response_content)
+    return queries
+
+def research_meal_plan_node(state: AgentState):
+    messages = [
+        SystemMessage(content=RESEARCH_PLAN_PROMPT),
+        HumanMessage(content=state['task'])
+    ]
+    response = model.invoke(messages)
+    queries = parse_queries(response.content)
+    content = state['content'] or []
+    for q in queries:
+        search_response = tavily.search(query=q, max_results=2)
+        for r in search_response['results']:
+            content.append(r['content'])
+    research_meal_plan_node = print("Research Meal Plan Response:", response.content)  # Debug print
+    research_meal_plan_node
+    return {"content": content}
+
+def generation_node(state: AgentState):
+    content = "\n\n".join(state['content'] or [])
+    user_message = HumanMessage(
+        content=f"{state['task']}\n\nHere is my meal plan:\n\n{state['plan']}")
+    messages = [
+        SystemMessage(
+            content=WRITER_PROMPT.format(content=content)
+        ),
+        user_message
+    ]
+    response = model.invoke(messages)
+    generation_node = print("Generation Response: ", response.content)
+    generation_node
+    return {
+        "draft": response.content,
+        "revision_number": state.get("revision_number", 1) + 1
+    }
+
+def reflection_node(state: AgentState):
+    messages = [
+        SystemMessage(content=REFLECTION_PROMPT),
+        HumanMessage(content=state['draft'])
+    ]
+    response = model.invoke(messages)
+    reflection_node = print("Reflection Response:", response.content)
+    reflection_node
+    return {"critique": response.content}
+
+def research_critique_node(state: AgentState):
+    messages = [
+        SystemMessage(content=RESEARCH_CRITIQUE_PROMPT),
+        HumanMessage(content=state['critique'])
+    ]
+    response = model.invoke(messages)
+    research_critique_node = print("Research Critique Response:", response.content)  # Debug print
+    research_critique_node
+    queries = parse_queries(response.content)
+    content = state['content'] or []
+    for q in queries:
+        search_response = tavily.search(query=q, max_results=2)
+        for r in search_response['results']:
+            content.append(r['content'])
+    return {"content": content}
+
+def should_continue(state):
+    if state["revision_number"] > state["max_revisions"]:
+        return END
+    return "reflect_plan"
+
+builder = StateGraph(AgentState)
+
+builder.add_node("meal_planner", plan_node)
+builder.add_node("generate", generation_node)
+builder.add_node("reflect_plan", reflection_node)
+builder.add_node("research_meal_plan", research_meal_plan_node)
+builder.add_node("research_critique", research_critique_node)
+
+builder.set_entry_point("meal_planner")
+
+builder.add_conditional_edges(
+    "generate", 
+    should_continue, 
+    {END: END, "reflect_plan": "reflect_plan"}
+)
+
+builder.add_edge("meal_planner", "research_meal_plan")
+builder.add_edge("research_meal_plan", "generate")
+builder.add_edge("reflect_plan", "research_critique")
+builder.add_edge("research_critique", "generate")
+
+graph = builder.compile(checkpointer=memory)
+
+def start_agents(task, max_revisions):
+    # Save the current stdout
+    old_stdout = sys.stdout
+    sys.stdout = io.StringIO()  # Redirect stdout to a buffer
+    
+    try:
+        thread = {"configurable": {"thread_id": "1"}}
+        responses = list(graph.stream({
+            'task': task,
+            "max_revisions": max_revisions,
+            "revision_number": 1
+        }, thread))
+    finally:
+        # Restore the original stdout
+        sys.stdout = old_stdout
+
+    if responses:
+        draft = responses[-1].get('generate', {}).get('draft', 'No draft found')
+        return draft
+    else:
+        return "No responses received"
+
+interface = gr.Interface(
+    fn = start_agents,
+    inputs= [
+        gr.Textbox(lines=2, placeholder="Enter your meal planning task..."),
+        gr.Slider(1, 3, step=1, label="Max Revisions")
+    ],
+    outputs="text",
+    title="AI Meal Planner",
+    description="Generate meal plans based on your dietary requirements and preferences."
+)
+
+interface.launch()