web_browsing_agent/web_browsing_agent.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "65997815-9919-45e4-9a43-9d357249f3c4",
   "metadata": {},
   "outputs": [],
   "source": [
    "%pip install -q helium pillow selenium langgraph langchain-core langchain-community langchain_ollama"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ee515b4f-c73d-4cb5-b0a7-b53595c21b34",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os, getpass\n",
    "\n",
    "def _set_env(var: str):\n",
    "    # Check if the variable is set in the OS environment\n",
    "    env_value = os.environ.get(var)\n",
    "    if not env_value:\n",
    "        # If not set, prompt the user for input\n",
    "        env_value = getpass.getpass(f\"{var}: \")\n",
    "    \n",
    "    # Set the environment variable for the current process\n",
    "    os.environ[var] = env_value\n",
    "\n",
    "_set_env(\"LANGCHAIN_API_KEY\")\n",
    "\n",
    "import os\n",
    "os.environ[\"LANGCHAIN_TRACING_V2\"] = \"true\"\n",
    "os.environ[\"LANGCHAIN_PROJECT\"] = \"web browsing agent\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "403eab62-7f74-44dc-8350-ea8e8f022a3e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import warnings\n",
    "warnings.filterwarnings(\"ignore\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "886df61f-6a5b-4d85-8d9e-6f087967b417",
   "metadata": {},
   "source": [
    "# LangGraph"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "05695e5e-d57d-46fc-a573-e04a87b73cfb",
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "import helium\n",
    "\n",
    "from IPython.display import Image, display\n",
    "\n",
    "from typing import Annotated, Dict, List, Literal, Optional\n",
    "\n",
    "from selenium.webdriver.common.by import By\n",
    "from selenium import webdriver\n",
    "from selenium.webdriver.support.ui import WebDriverWait\n",
    "from selenium.webdriver.support import expected_conditions as EC\n",
    "from selenium.common.exceptions import ElementNotInteractableException, TimeoutException\n",
    "\n",
    "from langchain_core.tools import tool\n",
    "from langgraph.prebuilt import InjectedState, ToolNode, tools_condition\n",
    "from langgraph.graph import END, START, MessagesState, StateGraph\n",
    "from langchain_fireworks import ChatFireworks\n",
    "\n",
    "from langchain_core.prompts import ChatPromptTemplate\n",
    "\n",
    "from langchain_core.messages import AIMessage, HumanMessage, SystemMessage\n",
    "from langgraph.checkpoint.memory import MemorySaver\n",
    "\n",
    "from langchain.agents import AgentType, initialize_agent\n",
    "from langgraph.prebuilt import create_react_agent\n",
    "from langchain_ollama import ChatOllama"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1c90e94e-0a1d-4174-99f2-f1e6459316dc",
   "metadata": {},
   "outputs": [],
   "source": [
    "logging.basicConfig(\n",
    "    level=logging.INFO,\n",
    "    format='%(asctime)s - %(levelname)s - %(message)s'\n",
    ")\n",
    "\n",
    "logging = logging.getLogger(__name__)\n",
    "\n",
    "BROWSER_DRIVER = None"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5b91d242-25e4-4391-ac29-d43e56e45451",
   "metadata": {},
   "source": [
    "## Utils"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "97f2ddf6-b3ce-4b7e-9aec-051ac5dec396",
   "metadata": {},
   "outputs": [],
   "source": [
    "def open_browser(\n",
    "    headless: bool = False, \n",
    "    window_width: int = 980, \n",
    "    window_height: int = 1300\n",
    ") -> Optional[webdriver.Chrome]:\n",
    "    \"\"\"\n",
    "    Open a Chrome browser with specified options.\n",
    "    \n",
    "    Args:\n",
    "        headless (bool): Run browser in headless mode.\n",
    "        window_width (int): Browser window width.\n",
    "        window_height (int): Browser window height.\n",
    "    \n",
    "    Returns:\n",
    "        Optional[webdriver.Chrome]: Opened browser driver.\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # Configure Chrome options\n",
    "        chrome_options = webdriver.ChromeOptions()\n",
    "        chrome_options.add_argument(\"--force-device-scale-factor=1\")\n",
    "        chrome_options.add_argument(f\"--window-size={window_width},{window_height}\")\n",
    "        chrome_options.add_argument(\"--disable-pdf-viewer\")\n",
    "        chrome_options.add_argument(\"--no-sandbox\")\n",
    "        chrome_options.add_argument(\"--disable-dev-shm-usage\")\n",
    "        \n",
    "        if headless:\n",
    "            chrome_options.add_argument(\"--headless\")\n",
    "        \n",
    "        # Start Chrome with Helium\n",
    "        driver = helium.start_chrome(headless=headless, options=chrome_options)\n",
    "        \n",
    "        logging.info(\"Browser opened successfully.\")\n",
    "        return driver\n",
    "    \n",
    "    except Exception as e:\n",
    "        logging.error(f\"Failed to open browser: {e}\")\n",
    "        return None\n",
    "\n",
    "def is_browser_open(driver: Optional[webdriver.Chrome]) -> bool:\n",
    "    \"\"\"\n",
    "    Check if the browser driver is open and functional\n",
    "    \n",
    "    Args:\n",
    "        driver (WebDriver): Selenium WebDriver instance\n",
    "    \n",
    "    Returns:\n",
    "        bool: True if browser is open, False otherwise\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # Multiple checks to ensure browser is truly open\n",
    "        return (\n",
    "            driver is not None and \n",
    "            hasattr(driver, 'current_window_handle') and\n",
    "            driver.current_window_handle is not None and\n",
    "            driver.title is not None\n",
    "        )\n",
    "    except Exception:\n",
    "        return False\n",
    "\n",
    "def close_browser(driver: Optional[webdriver.Chrome]):\n",
    "    \"\"\"\n",
    "    Close the current browser session.\n",
    "    \n",
    "    Args:\n",
    "        driver (WebDriver): Selenium WebDriver instance to close\n",
    "    \"\"\"\n",
    "    try:\n",
    "        if is_browser_open(driver):\n",
    "            driver.quit()\n",
    "            logging.info(\"Browser closed successfully.\")\n",
    "    except Exception as e:\n",
    "        logging.error(f\"Error closing browser: {e}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7ecb8b33-6794-45ba-86bf-b7976d9dc1f4",
   "metadata": {},
   "source": [
    "## State"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ebc0ef45-2b60-4cdc-b35d-6864eb36ac45",
   "metadata": {},
   "outputs": [],
   "source": [
    "class BrowserAgentState(MessagesState):\n",
    "    current_url: Optional[str]\n",
    "    error: Optional[str]\n",
    "    next: Optional[Literal[\"continue\", END]]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "656bb6e6-64de-493f-9c28-c25ede3009c6",
   "metadata": {},
   "source": [
    "## Tools"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "78e0b943-1b82-4a2e-b322-537d4b5c3676",
   "metadata": {},
   "outputs": [],
   "source": [
    "@tool\n",
    "def search_item_ctrl_f(text: str, nth_result: int = 1) -> str:\n",
    "    \"\"\"\n",
    "    Searches for the specified text on the current webpage using a Ctrl+F-like functionality.\n",
    "    \"\"\"\n",
    "    global BROWSER_DRIVER\n",
    "    if not BROWSER_DRIVER:\n",
    "        return \"No browser driver available\"\n",
    "    \n",
    "    try:\n",
    "        elements = BROWSER_DRIVER.find_elements(By.XPATH, f\"//*[contains(text(), '{text}')]\")\n",
    "        if nth_result > len(elements):\n",
    "            raise Exception(f\"Match n°{nth_result} not found (only {len(elements)} matches found)\")\n",
    "        \n",
    "        result = f\"Found {len(elements)} matches for '{text}'.\"\n",
    "        elem = elements[nth_result - 1]\n",
    "        BROWSER_DRIVER.execute_script(\"arguments[0].scrollIntoView(true);\", elem)\n",
    "        result += f\"Focused on element {nth_result} of {len(elements)}\"\n",
    "        return result\n",
    "    \n",
    "    except Exception as e:\n",
    "        return f\"Search error: {str(e)}\"\n",
    "\n",
    "@tool\n",
    "def go_to(url: str) -> str:\n",
    "    \"\"\"\n",
    "    Navigates the browser to the specified URL.\n",
    "    \"\"\"\n",
    "    global BROWSER_DRIVER\n",
    "    if not BROWSER_DRIVER:\n",
    "        return \"No browser driver available\"\n",
    "    \n",
    "    try:\n",
    "        BROWSER_DRIVER.get(url)\n",
    "        return f\"Navigated to {url}\"\n",
    "    except Exception as e:\n",
    "        return f\"Navigation error: {str(e)}\"\n",
    "\n",
    "@tool\n",
    "def go_back() -> str:\n",
    "    \"\"\"\n",
    "    Navigates the browser to the previous page in the session history.\n",
    "    \"\"\"\n",
    "    global BROWSER_DRIVER\n",
    "    if not BROWSER_DRIVER:\n",
    "        return \"No browser driver available\"\n",
    "    \n",
    "    try:\n",
    "        BROWSER_DRIVER.back()\n",
    "        return \"Navigated to previous page\"\n",
    "    except Exception as e:\n",
    "        return f\"Navigation error: {str(e)}\"\n",
    "\n",
    "@tool\n",
    "def close_popups() -> str:\n",
    "    \"\"\"\n",
    "    Closes any visible modal dialogs or pop-up windows on the current webpage.\n",
    "    \"\"\"\n",
    "    global BROWSER_DRIVER\n",
    "    if not BROWSER_DRIVER:\n",
    "        return \"No browser driver available\"\n",
    "\n",
    "    modal_selectors = [\n",
    "        \"button[class*='close']\",\n",
    "        \"[class*='modal']\",\n",
    "        \"[class*='modal'] button\",\n",
    "        \"[class*='CloseButton']\",\n",
    "        \"[aria-label*='close']\",\n",
    "        \".modal-close\",\n",
    "        \".close-modal\",\n",
    "        \".modal .close\",\n",
    "        \".modal-backdrop\",\n",
    "        \".modal-overlay\",\n",
    "        \"[class*='overlay']\",\n",
    "    ]\n",
    "\n",
    "    wait = WebDriverWait(BROWSER_DRIVER, timeout=0.5)\n",
    "\n",
    "    for selector in modal_selectors:\n",
    "        try:\n",
    "            elements = wait.until(EC.presence_of_all_elements_located((By.CSS_SELECTOR, selector)))\n",
    "\n",
    "            for element in elements:\n",
    "                if element.is_displayed():\n",
    "                    try:\n",
    "                        BROWSER_DRIVER.execute_script(\"arguments[0].click();\", element)\n",
    "                    except ElementNotInteractableException:\n",
    "                        element.click()\n",
    "\n",
    "        except TimeoutException:\n",
    "            continue\n",
    "        except Exception as e:\n",
    "            print(f\"Error handling selector {selector}: {str(e)}\")\n",
    "            continue\n",
    "    return \"Modals closed\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4219f350-72c7-45fc-875b-fe2930e2dd56",
   "metadata": {},
   "outputs": [],
   "source": [
    "tools = [search_item_ctrl_f, go_to, go_back, close_popups]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "578b369f-85f5-4e2c-b76f-479f46c179da",
   "metadata": {},
   "source": [
    "# Prompt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9411ffb7-aba4-4da8-aa27-3a65bb890a9b",
   "metadata": {},
   "outputs": [],
   "source": [
    "prompt = \"\"\"\n",
    "You can use helium to access websites. Don't bother about the helium driver, it's already managed.\n",
    "First you need to import everything from helium, then you can do other actions!\n",
    "Code:\n",
    "```py\n",
    "from helium import *\n",
    "go_to('github.com/trending')\n",
    "```<end_code>\n",
    "\n",
    "You can directly click clickable elements by inputting the text that appears on them.\n",
    "Code:\n",
    "```py\n",
    "click(\"Top products\")\n",
    "```<end_code>\n",
    "\n",
    "If it's a link:\n",
    "Code:\n",
    "```py\n",
    "click(Link(\"Top products\"))\n",
    "```<end_code>\n",
    "\n",
    "If you try to interact with an element and it's not found, you'll get a LookupError.\n",
    "In general stop your action after each button click to see what happens on your screenshot.\n",
    "Never try to login in a page.\n",
    "\n",
    "To scroll up or down, use scroll_down or scroll_up with as an argument the number of pixels to scroll from.\n",
    "Code:\n",
    "```py\n",
    "scroll_down(num_pixels=1200) # This will scroll one viewport down\n",
    "```<end_code>\n",
    "\n",
    "When you have pop-ups with a cross icon to close, don't try to click the close icon by finding its element or targeting an 'X' element (this most often fails).\n",
    "Just use your built-in tool `close_popups` to close them:\n",
    "Code:\n",
    "```py\n",
    "close_popups()\n",
    "```<end_code>\n",
    "\n",
    "You can use .exists() to check for the existence of an element. For example:\n",
    "Code:\n",
    "```py\n",
    "if Text('Accept cookies?').exists():\n",
    "    click('I accept')\n",
    "```<end_code>\n",
    "\n",
    "Proceed in several steps rather than trying to solve the task in one shot.\n",
    "And at the end, only when you have your answer, return your final answer.\n",
    "Code:\n",
    "```py\n",
    "final_answer(\"YOUR_ANSWER_HERE\")\n",
    "```<end_code>\n",
    "\n",
    "If pages seem stuck on loading, you might have to wait, for instance `import time` and run `time.sleep(5.0)`. But don't overuse this!\n",
    "To list elements on page, DO NOT try code-based element searches like 'contributors = find_all(S(\"ol > li\"))': just look at the latest screenshot you have and read it visually, or use your tool search_item_ctrl_f.\n",
    "Of course, you can act on buttons like a user would do when navigating.\n",
    "After each code blob you write, you will be automatically provided with an updated screenshot of the browser and the current browser url.\n",
    "But beware that the screenshot will only be taken at the end of the whole action, it won't see intermediate states.\n",
    "Don't kill the browser.\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d0052297-41c3-4c9d-b635-33b7d367c66d",
   "metadata": {},
   "source": [
    "## Graph"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "34efe219-b578-4b4d-8cd0-47695c438866",
   "metadata": {},
   "outputs": [],
   "source": [
    "model = ChatOllama(\n",
    "    base_url=\"http://localhost:11434\",\n",
    "    model=\"llama3.2\",\n",
    "    temperature=0,\n",
    "    streaming=True\n",
    ").bind_tools(tools)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "46eb487f-207f-420b-939b-a1f94451c115",
   "metadata": {},
   "outputs": [],
   "source": [
    "def agent_node(state: BrowserAgentState):\n",
    "    global BROWSER_DRIVER\n",
    "    \n",
    "    if not BROWSER_DRIVER or not is_browser_open(BROWSER_DRIVER):\n",
    "        try:\n",
    "            BROWSER_DRIVER = open_browser(headless=False)\n",
    "        except Exception as e:\n",
    "            return BrowserAgentState(\n",
    "                messages=state.messages + [AIMessge(content=f\"Error opening browser: {str(e)}\")],\n",
    "                current_url=None,\n",
    "                error=str(e),\n",
    "                next=None\n",
    "            )\n",
    "            \n",
    "    messages = [\n",
    "        SystemMessage(content=prompt),\n",
    "        *state[\"messages\"]\n",
    "    ]\n",
    "\n",
    "    response = model.invoke(messages)\n",
    "\n",
    "    return {\n",
    "        **state,\n",
    "        \"current_url\": BROWSER_DRIVER.current_url if BROWSER_DRIVER else None,\n",
    "        \"messages\": state[\"messages\"] + [response]\n",
    "    }\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "48ec605f-e083-46c5-999e-0b91b57efaa9",
   "metadata": {},
   "outputs": [],
   "source": [
    "def should_continue(state: BrowserAgentState) -> str:    \n",
    "    messages = state['messages']\n",
    "    last_message = messages[-1]\n",
    "    \n",
    "    return \"continue\" if last_message.tool_calls else END"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dfc87db9-bc05-4b45-a673-a7931cd00460",
   "metadata": {},
   "outputs": [],
   "source": [
    "builder = StateGraph(BrowserAgentState)\n",
    "    \n",
    "builder.add_node(\"agent\", agent_node)\n",
    "builder.add_node(\"tools\", ToolNode(tools))\n",
    "\n",
    "builder.set_entry_point(\"agent\")\n",
    "\n",
    "builder.add_conditional_edges(\n",
    "    \"agent\",\n",
    "    should_continue,\n",
    "    {\n",
    "        \"continue\": \"tools\",\n",
    "        END: END\n",
    "    }\n",
    ")\n",
    "\n",
    "builder.add_edge(\"tools\", \"agent\")\n",
    "\n",
    "\n",
    "graph = builder.compile()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "35eb253d-d913-4f2a-af65-dce3824ff56f",
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.display import Image, display\n",
    "\n",
    "try:\n",
    "    display(Image(graph.get_graph().draw_mermaid_png()))\n",
    "except Exception:\n",
    "    pass"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0b761ab7-ec38-416a-bc7b-a4efd303b795",
   "metadata": {},
   "source": [
    "# Test graph"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "5d16dd5b-e4ac-4567-bfae-a9a65fbedb1b",
   "metadata": {},
   "outputs": [],
   "source": [
    "message = \"please navigate to https://www.langchain.com/about and explain me their mission\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "724560d8-c002-47f3-9375-75ed16afbc63",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025-01-25 22:00:15,584 - INFO - Browser opened successfully.\n",
      "2025-01-25 22:00:25,505 - INFO - HTTP Request: POST http://localhost:11434/api/chat \"HTTP/1.1 200 OK\"\n",
      "2025-01-25 22:00:33,837 - INFO - HTTP Request: POST http://localhost:11434/api/chat \"HTTP/1.1 200 OK\"\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "================================\u001b[1m Human Message \u001b[0m=================================\n",
      "\n",
      "please navigate to https://www.langchain.com/about and explain me their mission\n",
      "==================================\u001b[1m Ai Message \u001b[0m==================================\n",
      "Tool Calls:\n",
      "  go_to (461a8eac-7ddb-4757-a7f7-7bf94cc3f068)\n",
      " Call ID: 461a8eac-7ddb-4757-a7f7-7bf94cc3f068\n",
      "  Args:\n",
      "    url: https://www.langchain.com/about\n",
      "=================================\u001b[1m Tool Message \u001b[0m=================================\n",
      "Name: go_to\n",
      "\n",
      "Navigated to https://www.langchain.com/about\n",
      "==================================\u001b[1m Ai Message \u001b[0m==================================\n",
      "\n",
      "According to the website, LangChain's mission is to \"empower developers and researchers to build scalable, reliable, and maintainable AI systems by providing a set of tools and frameworks that simplify the process of integrating multiple AI models and services into a single workflow.\"\n"
     ]
    }
   ],
   "source": [
    "config = {\"recursion_limit\": 10}\n",
    "messages = [HumanMessage(content=message)]\n",
    "messages = graph.invoke({\"messages\": messages}, config)\n",
    "for m in messages['messages']:\n",
    "    m.pretty_print()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e7fb7f60-1696-45e5-9649-e60a235e7e00",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4ee7d00f-0a6d-4dbd-9697-4bea0f89accc",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "env",
   "language": "python",
   "name": "env"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}