Upload folder using huggingface_hub

README.md

@@ -1,12 +1,6 @@
 ---
-title: AIAssistant LangGraph
-emoji: 🌖
-colorFrom: yellow
-colorTo: gray
-sdk: gradio
-sdk_version: 5.42.0
+title: AIAssistant_LangGraph
 app_file: app.py
-pinned: false
+sdk: gradio
+sdk_version: 5.34.2
 ---
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,61 @@
+import gradio as gr
+from sidekick import Sidekick
+
+
+async def setup():
+    sidekick = Sidekick()
+    await sidekick.setup()
+    return sidekick
+
+
+async def process_message(sidekick, message, success_criteria, history):
+    results = await sidekick.run_superstep(message, success_criteria, history)
+    return results, sidekick
+
+
+async def reset():
+    new_sidekick = Sidekick()
+    await new_sidekick.setup()
+    return "", "", None, new_sidekick
+
+
+def free_resources(sidekick):
+    print("Cleaning up")
+    try:
+        if sidekick:
+            sidekick.cleanup()
+    except Exception as e:
+        print(f"Exception during cleanup: {e}")
+
+
+with gr.Blocks(title="Personal Assistant", theme=gr.themes.Default(primary_hue="emerald")) as ui:
+    gr.Markdown("## Personal Assistant")
+    sidekick = gr.State(delete_callback=free_resources)
+
+    with gr.Row():
+        chatbot = gr.Chatbot(label="AI Assistant", height=300, type="messages")
+    with gr.Group():
+        with gr.Row():
+            message = gr.Textbox(show_label=False, placeholder="Your request to the AI Assistant")
+        with gr.Row():
+            success_criteria = gr.Textbox(
+                show_label=False, placeholder="What are your success critiera?"
+            )
+    with gr.Row():
+        reset_button = gr.Button("Reset", variant="stop")
+        go_button = gr.Button("Go!", variant="primary")
+
+    ui.load(setup, [], [sidekick])
+    message.submit(
+        process_message, [sidekick, message, success_criteria, chatbot], [chatbot, sidekick]
+    )
+    success_criteria.submit(
+        process_message, [sidekick, message, success_criteria, chatbot], [chatbot, sidekick]
+    )
+    go_button.click(
+        process_message, [sidekick, message, success_criteria, chatbot], [chatbot, sidekick]
+    )
+    reset_button.click(reset, [], [message, success_criteria, chatbot, sidekick])
+
+
+ui.launch(inbrowser=True)

sandbox/dinner.md

@@ -0,0 +1,24 @@
+# Dinner Report: Le Bernardin
+
+## Restaurant Overview
+- **Name**: Le Bernardin
+- **Address**: 155 W 51st St, New York, NY 10019
+- **Phone**: (212) 554-1515
+- **Website**: [le-bernardin.com](http://le-bernardin.com)
+
+## Cuisine
+Le Bernardin specializes in refined seafood dishes, crafted with the utmost respect for the ingredients. The menu includes a variety of seafood preparations, with an emphasis on freshness and simplicity.
+
+## Menu Highlights
+- **Tuna Tartare**: Diced raw tuna, served with toasted sesame and avocado.
+- **Wild Salmon**: Lightly cooked, served with a warm ginger-soy emulsion and bok choy.
+- **Poached Lobster**: Accompanied by truffle butter and a delicate sauce.
+- **Chocolate Soufflé**: A classic dessert, rich and airy.
+
+## Ambiance
+The atmosphere at Le Bernardin is elegant and serene, characterized by minimalist decor and a focus on the dining experience.
+
+## Summary of Reviews
+Le Bernardin has consistently received rave reviews for its exceptional cuisine and impeccable service. It is a three-Michelin star restaurant, recognized as one of the best seafood restaurants not only in New York but also in the world. Diners emphasize the harmonious flavors and the artistry of the dishes, making it a top choice for special occasions.
+
+**In conclusion**, Le Bernardin represents the pinnacle of fine dining, offering diners an unforgettable experience centered around seafood excellence.

sidekick.py

@@ -0,0 +1,222 @@
+from typing import Annotated
+from typing_extensions import TypedDict
+from langgraph.graph import StateGraph, START, END
+from langgraph.graph.message import add_messages
+from dotenv import load_dotenv
+from langgraph.prebuilt import ToolNode
+from langchain_openai import ChatOpenAI
+from langgraph.checkpoint.memory import MemorySaver
+from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
+from typing import List, Any, Optional, Dict
+from pydantic import BaseModel, Field
+from sidekick_tools import playwright_tools, other_tools
+import uuid
+import asyncio
+from datetime import datetime
+
+load_dotenv(override=True)
+
+
+class State(TypedDict):
+    messages: Annotated[List[Any], add_messages]
+    success_criteria: str
+    feedback_on_work: Optional[str]
+    success_criteria_met: bool
+    user_input_needed: bool
+
+
+class EvaluatorOutput(BaseModel):
+    feedback: str = Field(description="Feedback on the assistant's response")
+    success_criteria_met: bool = Field(description="Whether the success criteria have been met")
+    user_input_needed: bool = Field(
+        description="True if more input is needed from the user, or clarifications, or the assistant is stuck"
+    )
+
+
+class Sidekick:
+    def __init__(self):
+        self.worker_llm_with_tools = None
+        self.evaluator_llm_with_output = None
+        self.tools = None
+        self.llm_with_tools = None
+        self.graph = None
+        self.sidekick_id = str(uuid.uuid4())
+        self.memory = MemorySaver()
+        self.browser = None
+        self.playwright = None
+
+    async def setup(self):
+        self.tools, self.browser, self.playwright = await playwright_tools()
+        self.tools += await other_tools()
+        worker_llm = ChatOpenAI(model="gpt-4o-mini")
+        self.worker_llm_with_tools = worker_llm.bind_tools(self.tools)
+        evaluator_llm = ChatOpenAI(model="gpt-4o-mini")
+        self.evaluator_llm_with_output = evaluator_llm.with_structured_output(EvaluatorOutput)
+        await self.build_graph()
+
+    def worker(self, state: State) -> Dict[str, Any]:
+        system_message = f"""You are a helpful assistant that can use tools to complete tasks.
+    You keep working on a task until either you have a question or clarification for the user, or the success criteria is met.
+    You have many tools to help you, including tools to browse the internet, navigating and retrieving web pages.
+    You have a tool to run python code, but note that you would need to include a print() statement if you wanted to receive output.
+    The current date and time is {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
+
+    This is the success criteria:
+    {state["success_criteria"]}
+    You should reply either with a question for the user about this assignment, or with your final response.
+    If you have a question for the user, you need to reply by clearly stating your question. An example might be:
+
+    Question: please clarify whether you want a summary or a detailed answer
+
+    If you've finished, reply with the final answer, and don't ask a question; simply reply with the answer.
+    """
+
+        if state.get("feedback_on_work"):
+            system_message += f"""
+    Previously you thought you completed the assignment, but your reply was rejected because the success criteria was not met.
+    Here is the feedback on why this was rejected:
+    {state["feedback_on_work"]}
+    With this feedback, please continue the assignment, ensuring that you meet the success criteria or have a question for the user."""
+
+        # Add in the system message
+
+        found_system_message = False
+        messages = state["messages"]
+        for message in messages:
+            if isinstance(message, SystemMessage):
+                message.content = system_message
+                found_system_message = True
+
+        if not found_system_message:
+            messages = [SystemMessage(content=system_message)] + messages
+
+        # Invoke the LLM with tools
+        response = self.worker_llm_with_tools.invoke(messages)
+
+        # Return updated state
+        return {
+            "messages": [response],
+        }
+
+    def worker_router(self, state: State) -> str:
+        last_message = state["messages"][-1]
+
+        if hasattr(last_message, "tool_calls") and last_message.tool_calls:
+            return "tools"
+        else:
+            return "evaluator"
+
+    def format_conversation(self, messages: List[Any]) -> str:
+        conversation = "Conversation history:\n\n"
+        for message in messages:
+            if isinstance(message, HumanMessage):
+                conversation += f"User: {message.content}\n"
+            elif isinstance(message, AIMessage):
+                text = message.content or "[Tools use]"
+                conversation += f"Assistant: {text}\n"
+        return conversation
+
+    def evaluator(self, state: State) -> State:
+        last_response = state["messages"][-1].content
+
+        system_message = """You are an evaluator that determines if a task has been completed successfully by an Assistant.
+    Assess the Assistant's last response based on the given criteria. Respond with your feedback, and with your decision on whether the success criteria has been met,
+    and whether more input is needed from the user."""
+
+        user_message = f"""You are evaluating a conversation between the User and Assistant. You decide what action to take based on the last response from the Assistant.
+
+    The entire conversation with the assistant, with the user's original request and all replies, is:
+    {self.format_conversation(state["messages"])}
+
+    The success criteria for this assignment is:
+    {state["success_criteria"]}
+
+    And the final response from the Assistant that you are evaluating is:
+    {last_response}
+
+    Respond with your feedback, and decide if the success criteria is met by this response.
+    Also, decide if more user input is required, either because the assistant has a question, needs clarification, or seems to be stuck and unable to answer without help.
+
+    The Assistant has access to a tool to write files. If the Assistant says they have written a file, then you can assume they have done so.
+    Overall you should give the Assistant the benefit of the doubt if they say they've done something. But you should reject if you feel that more work should go into this.
+
+    """
+        if state["feedback_on_work"]:
+            user_message += f"Also, note that in a prior attempt from the Assistant, you provided this feedback: {state['feedback_on_work']}\n"
+            user_message += "If you're seeing the Assistant repeating the same mistakes, then consider responding that user input is required."
+
+        evaluator_messages = [
+            SystemMessage(content=system_message),
+            HumanMessage(content=user_message),
+        ]
+
+        eval_result = self.evaluator_llm_with_output.invoke(evaluator_messages)
+        new_state = {
+            "messages": [
+                {
+                    "role": "assistant",
+                    "content": f"Evaluator Feedback on this answer: {eval_result.feedback}",
+                }
+            ],
+            "feedback_on_work": eval_result.feedback,
+            "success_criteria_met": eval_result.success_criteria_met,
+            "user_input_needed": eval_result.user_input_needed,
+        }
+        return new_state
+
+    def route_based_on_evaluation(self, state: State) -> str:
+        if state["success_criteria_met"] or state["user_input_needed"]:
+            return "END"
+        else:
+            return "worker"
+
+    async def build_graph(self):
+        # Set up Graph Builder with State
+        graph_builder = StateGraph(State)
+
+        # Add nodes
+        graph_builder.add_node("worker", self.worker)
+        graph_builder.add_node("tools", ToolNode(tools=self.tools))
+        graph_builder.add_node("evaluator", self.evaluator)
+
+        # Add edges
+        graph_builder.add_conditional_edges(
+            "worker", self.worker_router, {"tools": "tools", "evaluator": "evaluator"}
+        )
+        graph_builder.add_edge("tools", "worker")
+        graph_builder.add_conditional_edges(
+            "evaluator", self.route_based_on_evaluation, {"worker": "worker", "END": END}
+        )
+        graph_builder.add_edge(START, "worker")
+
+        # Compile the graph
+        self.graph = graph_builder.compile(checkpointer=self.memory)
+
+    async def run_superstep(self, message, success_criteria, history):
+        config = {"configurable": {"thread_id": self.sidekick_id}}
+
+        state = {
+            "messages": message,
+            "success_criteria": success_criteria or "The answer should be clear and accurate",
+            "feedback_on_work": None,
+            "success_criteria_met": False,
+            "user_input_needed": False,
+        }
+        result = await self.graph.ainvoke(state, config=config)
+        user = {"role": "user", "content": message}
+        reply = {"role": "assistant", "content": result["messages"][-2].content}
+        feedback = {"role": "assistant", "content": result["messages"][-1].content}
+        return history + [user, reply, feedback]
+
+    def cleanup(self):
+        if self.browser:
+            try:
+                loop = asyncio.get_running_loop()
+                loop.create_task(self.browser.close())
+                if self.playwright:
+                    loop.create_task(self.playwright.stop())
+            except RuntimeError:
+                # If no loop is running, do a direct run
+                asyncio.run(self.browser.close())
+                if self.playwright:
+                    asyncio.run(self.playwright.stop())

sidekick_tools.py

@@ -0,0 +1,55 @@
+from playwright.async_api import async_playwright
+from langchain_community.agent_toolkits import PlayWrightBrowserToolkit
+from dotenv import load_dotenv
+import os
+import requests
+from langchain.agents import Tool
+from langchain_community.agent_toolkits import FileManagementToolkit
+from langchain_community.tools.wikipedia.tool import WikipediaQueryRun
+from langchain_experimental.tools import PythonREPLTool
+from langchain_community.utilities import GoogleSerperAPIWrapper
+from langchain_community.utilities.wikipedia import WikipediaAPIWrapper
+
+
+
+load_dotenv(override=True)
+pushover_token = os.getenv("PUSHOVER_TOKEN")
+pushover_user = os.getenv("PUSHOVER_USER")
+pushover_url = "https://api.pushover.net/1/messages.json"
+serper = GoogleSerperAPIWrapper()
+
+async def playwright_tools():
+    playwright = await async_playwright().start()
+    browser = await playwright.chromium.launch(headless=False)
+    toolkit = PlayWrightBrowserToolkit.from_browser(async_browser=browser)
+    return toolkit.get_tools(), browser, playwright
+
+
+def push(text: str):
+    """Send a push notification to the user"""
+    requests.post(pushover_url, data = {"token": pushover_token, "user": pushover_user, "message": text})
+    return "success"
+
+
+def get_file_tools():
+    toolkit = FileManagementToolkit(root_dir="sandbox")
+    return toolkit.get_tools()
+
+
+async def other_tools():
+    push_tool = Tool(name="send_push_notification", func=push, description="Use this tool when you want to send a push notification")
+    file_tools = get_file_tools()
+
+    tool_search =Tool(
+        name="search",
+        func=serper.run,
+        description="Use this tool when you want to get the results of an online web search"
+    )
+
+    wikipedia = WikipediaAPIWrapper()
+    wiki_tool = WikipediaQueryRun(api_wrapper=wikipedia)
+
+    python_repl = PythonREPLTool()
+    
+    return file_tools + [push_tool, tool_search, python_repl,  wiki_tool]
+