langgraph/examples/web-navigation/web_voyager.ipynb

Web Voyager

WebVoyager by He, et. al., is a vision-enabled web-browsing agent capable of controlling the mouse and keyboard.

It works by viewing annotated browser screenshots for each turn, then choosing the next step to take. The agent architecture is a basic reasoning and action (ReAct) loop. The unique aspects of this agent are:

• It's usage of Set-of-Marks-like image annotations to serve as UI affordances for the agent
• It's application in the browser by using tools to control both the mouse and keyboard

The overall design looks like the following:

Configure environment

We will first set up LangSmith tracing. Though optional, this lets us inspect and debug agent's trajectory for a given input.

You can sign up at smith.langchain.com to get an API key.

In [ ]:

%pip install -U --quiet langgraph langsmith langchain_openai

In [2]:

# Optional: add tracing to visualize the agent trajectories
import os
from getpass import getpass


def _getpass(env_var: str):
    if not os.environ.get(env_var):
        os.environ[env_var] = getpass(f"{env_var}=")


os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "Web-Voyager"
_getpass("LANGCHAIN_API_KEY")
_getpass("OPENAI_API_KEY")

Install Agent requirements

The only additional requirement we have is the playwright browser. Uncomment and install below:

In [ ]:

# %pip install --upgrade --quiet  playwright > /dev/null
# !playwright install

In [1]:

import nest_asyncio

# This is just required for running async playwright in a Jupyter notebook
nest_asyncio.apply()

Define Graph State

The state provides the inputs to each node in the graph.

In our case, the agent will track the webpage object (within the browser), annotated images + bounding boxes, the user's initial request, and the messages containing the agent scratchpad, system prompt, and other information.

In [4]:

from typing import List, Optional, TypedDict

from langchain_core.messages import BaseMessage, SystemMessage
from playwright.async_api import Page


class BBox(TypedDict):
    x: float
    y: float


class Prediction(TypedDict):
    action: str
    args: Optional[List[str]]


# This represents the state of the agent
# as it proceeds through execution
class AgentState(TypedDict):
    page: Page  # The Playwright web page lets us interact with the web environment
    input: str  # User request
    img: str  # b64 encoded screenshot
    bboxes: List[BBox]  # The bounding boxes from the browser annotation function
    prediction: Prediction  # The Agent's output
    # A system message (or messages) containing the intermediate steps
    scratchpad: List[BaseMessage]
    observation: str  # The most recent response from a tool

Define tools

The agent has 6 simple tools:

1. Click (at labeled box)
2. Type
3. Scroll
4. Wait
5. Go back
6. Go to search engine (Google)

We define them below here as functions:

In [43]:

import asyncio
import platform


async def click(state: AgentState):
    # - Click [Numerical_Label]
    page = state["page"]
    click_args = state["prediction"]["args"]
    if click_args is None or len(click_args) != 1:
        return f"Failed to click bounding box labeled as number {click_args}"
    bbox_id = click_args[0]
    bbox_id = int(bbox_id)
    try:
        bbox = state["bboxes"][bbox_id]
    except:
        return f"Error: no bbox for : {bbox_id}"
    x, y = bbox["x"], bbox["y"]
    res = await page.mouse.click(x, y)
    # TODO: In the paper, they automatically parse any downloaded PDFs
    # We could add something similar here as well and generally
    # improve response format.
    return f"Clicked {bbox_id}"


async def type_text(state: AgentState):
    page = state["page"]
    type_args = state["prediction"]["args"]
    if type_args is None or len(type_args) != 2:
        return (
            f"Failed to type in element from bounding box labeled as number {type_args}"
        )
    bbox_id = type_args[0]
    bbox_id = int(bbox_id)
    bbox = state["bboxes"][bbox_id]
    x, y = bbox["x"], bbox["y"]
    text_content = type_args[1]
    await page.mouse.click(x, y)
    # Check if MacOS
    select_all = "Meta+A" if platform.system() == "Darwin" else "Control+A"
    await page.keyboard.press(select_all)
    await page.keyboard.press("Backspace")
    await page.keyboard.type(text_content)
    await page.keyboard.press("Enter")
    return f"Typed {text_content} and submitted"


async def scroll(state: AgentState):
    page = state["page"]
    scroll_args = state["prediction"]["args"]
    if scroll_args is None or len(scroll_args) != 2:
        return "Failed to scroll due to incorrect arguments."

    target, direction = scroll_args

    if target.upper() == "WINDOW":
        # Not sure the best value for this:
        scroll_amount = 500
        scroll_direction = (
            -scroll_amount if direction.lower() == "up" else scroll_amount
        )
        await page.evaluate(f"window.scrollBy(0, {scroll_direction})")
    else:
        # Scrolling within a specific element
        scroll_amount = 200
        target_id = int(target)
        bbox = state["bboxes"][target_id]
        x, y = bbox["x"], bbox["y"]
        scroll_direction = (
            -scroll_amount if direction.lower() == "up" else scroll_amount
        )
        await page.mouse.move(x, y)
        await page.mouse.wheel(0, scroll_direction)

    return f"Scrolled {direction} in {'window' if target.upper() == 'WINDOW' else 'element'}"


async def wait(state: AgentState):
    sleep_time = 5
    await asyncio.sleep(sleep_time)
    return f"Waited for {sleep_time}s."


async def go_back(state: AgentState):
    page = state["page"]
    await page.go_back()
    return f"Navigated back a page to {page.url}."


async def to_google(state: AgentState):
    page = state["page"]
    await page.goto("https://www.google.com/")
    return "Navigated to google.com."

Define Agent

The agent is driven by a multi-modal model and decides the action to take for each step. It is composed of a few runnable objects:

1. A mark_page function to annotate the current page with bounding boxes
2. A prompt to hold the user question, annotated image, and agent scratchpad
3. GPT-4V to decide the next steps
4. Parsing logic to extract the action

Let's first define the annotation step:

Browser Annotations

This function annotates all buttons, inputs, text areas, etc. with numbered bounding boxes. GPT-4V then just has to refer to a bounding box when taking actions, reducing the complexity of the overall task.

In [44]:

import asyncio
import base64

from langchain_core.runnables import chain as chain_decorator

# Some javascript we will run on each step
# to take a screenshot of the page, select the
# elements to annotate, and add bounding boxes
with open("mark_page.js") as f:
    mark_page_script = f.read()


@chain_decorator
async def mark_page(page):
    await page.evaluate(mark_page_script)
    for _ in range(10):
        try:
            bboxes = await page.evaluate("markPage()")
            break
        except:
            # May be loading...
            asyncio.sleep(3)
    screenshot = await page.screenshot()
    # Ensure the bboxes don't follow us around
    await page.evaluate("unmarkPage()")
    return {
        "img": base64.b64encode(screenshot).decode(),
        "bboxes": bboxes,
    }

Agent definition

Now we'll compose this function with the prompt, llm and output parser to complete our agent.

In [45]:

from langchain import hub
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI


async def annotate(state):
    marked_page = await mark_page.with_retry().ainvoke(state["page"])
    return {**state, **marked_page}


def parse(text: str) -> dict:
    action_prefix = "Action: "
    if not text.strip().split("\n")[-1].startswith(action_prefix):
        return {"action": "retry", "args": f"Could not parse LLM Output: {text}"}
    action_block = text.strip().split("\n")[-1]

    action_str = action_block[len(action_prefix) :]
    split_output = action_str.split(" ", 1)
    if len(split_output) == 1:
        action, action_input = split_output[0], None
    else:
        action, action_input = split_output
    action = action.strip()
    if action_input is not None:
        action_input = [
            inp.strip().strip("[]") for inp in action_input.strip().split(";")
        ]
    return {"action": action, "args": action_input}


# Will need a later version of langchain to pull
# this image prompt template
prompt = hub.pull("wfh/web-voyager")

In [46]:

llm = ChatOpenAI(model="gpt-4-vision-preview", max_tokens=4096)
agent = annotate | RunnablePassthrough.assign(
    prediction=prompt | llm | StrOutputParser() | parse
)

Define graph

We've created most of the important logic. We have one more function to define that will help us update the graph state after a tool is called.

In [48]:

import re


def update_scratchpad(state: AgentState):
    """After a tool is invoked, we want to update
    the scratchpad so the agent is aware of its previous steps"""
    old = state.get("scratchpad")
    if old:
        txt = old[0].content
        last_line = txt.rsplit("\n", 1)[-1]
        step = int(re.match(r"\d+", last_line).group()) + 1
    else:
        txt = "Previous action observations:\n"
        step = 1
    txt += f"\n{step}. {state['observation']}"

    return {**state, "scratchpad": [SystemMessage(content=txt)]}

Now we can compose everything into a graph:

In [49]:

from langchain_core.runnables import RunnableLambda
from langgraph.graph import END, StateGraph

graph_builder = StateGraph(AgentState)


graph_builder.add_node("agent", agent)
graph_builder.set_entry_point("agent")

graph_builder.add_node("update_scratchpad", update_scratchpad)
graph_builder.add_edge("update_scratchpad", "agent")

tools = {
    "Click": click,
    "Type": type_text,
    "Scroll": scroll,
    "Wait": wait,
    "GoBack": go_back,
    "Google": to_google,
}


for node_name, tool in tools.items():
    graph_builder.add_node(
        node_name,
        # The lambda ensures the function's string output is mapped to the "observation"
        # key in the AgentState
        RunnableLambda(tool) | (lambda observation: {"observation": observation}),
    )
    # Always return to the agent (by means of the update-scratchpad node)
    graph_builder.add_edge(node_name, "update_scratchpad")


def select_tool(state: AgentState):
    # Any time the agent completes, this function
    # is called to route the output to a tool or
    # to the end user.
    action = state["prediction"]["action"]
    if action == "ANSWER":
        return END
    if action == "retry":
        return "agent"
    return action


graph_builder.add_conditional_edges("agent", select_tool)

graph = graph_builder.compile()

Run agent

Now that we've created the whole agent executor, we can run it on a few questions! We'll start our browser at "google.com" and then let it control the rest.

Below is a helper function to help print out the steps to the notebook (and display the intermediate screenshots).

In [50]:

import playwright
from IPython import display
from playwright.async_api import async_playwright

browser = await async_playwright().start()
# We will set headless=False so we can watch the agent navigate the web.
browser = await browser.chromium.launch(headless=False, args=None)
page = await browser.new_page()
_ = await page.goto("https://www.google.com")


async def call_agent(question: str, page, max_steps: int = 150):
    event_stream = graph.astream(
        {
            "page": page,
            "input": question,
            "scratchpad": [],
        },
        {
            "recursion_limit": max_steps,
        },
    )
    final_answer = None
    steps = []
    async for event in event_stream:
        # We'll display an event stream here
        if "agent" not in event:
            continue
        pred = event["agent"].get("prediction") or {}
        action = pred.get("action")
        action_input = pred.get("args")
        display.clear_output(wait=False)
        steps.append(f"{len(steps) + 1}. {action}: {action_input}")
        print("\n".join(steps))
        display.display(display.Image(base64.b64decode(event["agent"]["img"])))
        if "ANSWER" in action:
            final_answer = action_input[0]
            break
    return final_answer

In [35]:

res = await call_agent("Could you explain the WebVoyager paper (on arxiv)?", page)
print(f"Final response: {res}")

1. Type: ['7', 'WebVoyager paper arxiv']
2. Click: ['27']
3. ANSWER;: ['The WebVoyager paper presents a study on building an end-to-end web agent with large multimodal models. The research addresses the advancement of large language models, such as GPT-3, and their impact on various tasks, specifically highlighting their ability to generate human-like text. However, these models are not specifically trained to interact with and navigate web environments. The paper introduces WebVoyager, an agent trained using multimodal WebGPT models, to interact with the web. The model combines visual and textual information to perform tasks on the web. It improves upon the capabilities of text-only language models by enabling the agent to understand and interact with web pages in a more human-like manner. The paper likely details the architecture, training, and performance of WebVoyager, as well as the implications and potential applications of such an agent in various web tasks.']

Final response: The WebVoyager paper presents a study on building an end-to-end web agent with large multimodal models. The research addresses the advancement of large language models, such as GPT-3, and their impact on various tasks, specifically highlighting their ability to generate human-like text. However, these models are not specifically trained to interact with and navigate web environments. The paper introduces WebVoyager, an agent trained using multimodal WebGPT models, to interact with the web. The model combines visual and textual information to perform tasks on the web. It improves upon the capabilities of text-only language models by enabling the agent to understand and interact with web pages in a more human-like manner. The paper likely details the architecture, training, and performance of WebVoyager, as well as the implications and potential applications of such an agent in various web tasks.

In [36]:

res = await call_agent(
    "Please explain the today's XKCD comic for me. Why is it funny?", page
)
print(f"Final response: {res}")

1. Google: None
2. Type: ['6', "today's XKCD comic"]
3. Click: ['24']
4. ANSWER;: ['The XKCD comic shown in the image seems to be making a humorous observation about the greenhouse effect. It illustrates a timeline with two significant historical points: the development of a steam engine by James Watt in 1776, which helped kick off the Industrial Revolution, and the observations by scientists Svante Arrhenius and Arvid Högbom in 1896 that industrial activity was adding CO2 to the atmosphere and calculating how much the Earth would heat up if the CO2 concentration doubles. The punchline lies in the statement below that says, "We figured out the greenhouse effect closer to the start of the Industrial Revolution than to today." This is humorous because it suggests that the understanding of the greenhouse effect was realized not long after the beginning of the Industrial Revolution, yet it implies that not enough action has been taken to address it since then, even though we are now closer to the future point in time they were considering ("today"). The humor is thus derived from a juxtaposition of the early recognition of a serious issue with the apparent lack of progress in addressing it over a long period.']

Final response: The XKCD comic shown in the image seems to be making a humorous observation about the greenhouse effect. It illustrates a timeline with two significant historical points: the development of a steam engine by James Watt in 1776, which helped kick off the Industrial Revolution, and the observations by scientists Svante Arrhenius and Arvid Högbom in 1896 that industrial activity was adding CO2 to the atmosphere and calculating how much the Earth would heat up if the CO2 concentration doubles. The punchline lies in the statement below that says, "We figured out the greenhouse effect closer to the start of the Industrial Revolution than to today." This is humorous because it suggests that the understanding of the greenhouse effect was realized not long after the beginning of the Industrial Revolution, yet it implies that not enough action has been taken to address it since then, even though we are now closer to the future point in time they were considering ("today"). The humor is thus derived from a juxtaposition of the early recognition of a serious issue with the apparent lack of progress in addressing it over a long period.

In [37]:

res = await call_agent("What are the latest blog posts from langchain?", page)
print(f"Final response: {res}")

1. Google: None
2. Type: ['6', 'latest blog posts from langchain']
3. Click: ['23']
4. ANSWER;: ['The latest blog posts from LangChain are "OpenGPTs," "LangGraph: Multi-Agent Workflows," and "LangGraph."']

Final response: The latest blog posts from LangChain are "OpenGPTs," "LangGraph: Multi-Agent Workflows," and "LangGraph."

In [51]:

res = await call_agent(
    "Could you check google maps to see when i should leave to get to SFO by 7 o'clock? starting from SF downtown.",
    page,
)
print(f"Final response: {res}")

1. Type: ['6', 'SF downtown to SFO by 7 AM']
2. Click: ['10']
3. Click: ['11']
4. Click: ['0']
5. Click: ['0']
6. Click: ['0']
7. Click: ['16']
8. Click: ['11']
9. Click: ['0']
10. Click: ['2']
11. Type: ['0', 'SF downtown to SFO by 7 AM']
12. Click: ['0']
13. Click: ['0']
14. Type: ['0', 'SF downtown to SFO by 7 AM']
15. Click: ['8']
16. Click: ['2']
17. Click: ['0']
18. Click: ['11']
19. Google: None
20. Type: ['6', 'SF downtown to SFO by 7 AM']
21. Click: ['10']
22. Click: ['18']
23. Type: ['1', 'SF downtown to SFO by 7 AM']
24. Click: ['18']
25. Click: ['2']

In [ ]: