Multi-Agent LangGraph Course – Web research and asynchronous tools

/ Computer Science, OpenAI / By

Hello and welcome back to the last part of the LangGraph/LangChain/LangSmith course. In this part, we’ll learn how to deal with asynchronous tools by building a graph that will do some web research for us, where one of the tools is going to be visiting several websites at once to feed info back into the graph.

This type of asynchronous action is very helpful when there are multiple steps or actions that can be performed at the same time for optimization as it will save a lot of time and make the user experience much better. It is a bit different to set up and work with though, which is why we’ll be going through it in this part.

I will try to cover the bare basics of async Python programming here, as it can look quite confusing and I want all skill levels to be able to follow along. If you are already very familiar with async programming the level of explanation may be a bit excessive for you and you can probably skip over some of the explanations and just look at the code.

Web research tool

Let’s start by building our tool as usual. This tool is going to visit a bunch of web URLs at the same time (asynchronously) and return the HTML content of each page. We will need to install the BeautifulSoup library to parse the HTML content of the pages. Run the following command in the terminal:

pipenv install beautifulsoup4==4.12.3

Then go ahead and create a new file called web.py in the tools directory:

📂 FINX_LANGGRAPH
    📂 images
    📂 output
    📂 tools
        📄 __init__.py
        📄 image.py
        📄 pdf.py
        📄 weather.py
        📄 web.py    ✨New file
    📄 .env
    📄 langchain_basics.py
    📄 multi_agent.py
    📄 multi_agent_prompts.py
    📄 Pipfile
    📄 Pipfile.lock
    📄 setup_environment.py
    📄 simple_langgraph.py

In the web.py file let’s start with our imports as usual:

import asyncio
import json
import sys

import aiohttp
from bs4 import BeautifulSoup
from langchain.tools import tool
from pydantic import BaseModel, Field

We import asyncio to work with asynchronous code, aiohttp to make HTTP requests asynchronously, and BeautifulSoup to parse the HTML content of the pages. The tool decorator and pydantic imports are the same as for the other tools and json is to return the JSON responses in string format.

Async and event loops

First of all, we’ll use the sys import to set the type of event loop to use for the asynchronous code:

if sys.platform.startswith("win"):
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

Without going into too much detail, there is a known issue with the Python asyncio library on Windows specifically that happens when the Proactor event loop (the default on Windows) is closed while there are still outstanding tasks. It doesn’t affect the correct execution of the code, but something on Windows + aysncio + LangChain/LangGraph triggers it. We’ll use the selector event loop policy to avoid this issue (this is only needed/triggers if you’re on Windows.).

While this tutorial part is way too short to really go in-depth on Python’s asynchronous programming, we’ll try to cover the basics as we go along. Basically, we get an event loop, and we can put tasks in there. Normally a task like fetching a webpage would block the code until it’s done, but with asyncio we can put it in the event loop and continue with other tasks while it’s being fetched. This allows us to run multiple operations at the same time.

This is not to be confused with multi-threading or multi-processing, which are quite different in nature:

  • Multi-processing: is about spreading tasks over a computer’s CPU cores, and is well suited for tasks that require lots of mathematical computations.
  • Multi-threading: is about running multiple threads in the same process, and is well suited for tasks that are I/O bound (like fetching webpages).
  • Asynchronous programming: is a single-process, single-threaded design that uses coroutines to handle multiple tasks concurrently. Async functions are able to sort of pause and resume their execution, allowing other tasks to run in the meantime during this pause.

Async programming in Python is very similar to the JavaScript async/await pattern, and it’s a great way to handle I/O-bound tasks like fetching web pages. If you’re a bit new to this all, just keep going and you’ll get a feel for how it works.

Parsing HTML content

First, we’ll write a very basic function that uses BeautifulSoup to parse some HTML content:

def parse_html(html_content: str) -> str:
    soup = BeautifulSoup(html_content, "html.parser")
    for tag in ["nav", "footer", "aside", "script", "style", "img", "header"]:
        for match in soup.find_all(tag):
            match.decompose()

    text_content = soup.get_text()
    text_content = " ".join(text_content.split())
    return text_content[:8_000]

This function takes the HTML content of a webpage as a string and returns a string with the text content of the page. First we instantiate a new BeautifulSoup object passing in the html.parser string to select the parser. We then make a list of all the HTML tags we want to filter out, namely the navigation, footer, aside, script, style, image and header tags. We’re interested in the main content and don’t want all this pollution.

For each tag in this list of HTML tags, we run soup.find_all(tag) to find all the tags with that name in the HTML content, which returns all the matches for that tag. This allows us to loop over each match in soup.find_all(tag) and call match.decompose() to remove the tag from the HTML content.

We then get the text content of the page with soup.get_text() to remove as much HTML and unneeded stuff as possible from what was left.

Then we call text_content.split() to split the text content into a list of words, which has the side effects of removing long sequences of whitespace, tab, and newline characters. We then join the list of words back into a string with " ".join so that we’re left with only a single space between all words to save space. The LLM does not care about formatting and sending tons of whitespace to it is just a waste of space.

Finally, we return the first 8,000 characters of the text content, to make sure we don’t exceed the context limit if we load like 5 or 6 pages at once. You can set this higher if you use GPT-4-turbo instead of 3.5-turbo

Fetching webpages

Notice that the parse_html function is just a regular synchronous function. Now let’s get started on the asynchronous part. The first thing we’ll do is write a function to fetch the HTML content of a single webpage, and then we can just call this function multiple times to fetch the content of multiple pages at once.

async def get_webpage_content(url: str) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            html_content = await response.text()

    text_content = parse_html(html_content)
    print(f"URL: {url} - fetched successfully.")
    return text_content

First, we declare our async function using async def instead of the normal def. This will allow us to later call await on this function to make the code non-blocking and run other tasks while we wait for the response. We take a URL string and return a string.

Where we would normally fetch a webpage with the requests library, here we need to use aiohttp which is an asynchronous HTTP client/server library for Python that allows us to write this non-blocking code. The ClientSession object represents a single web session, so you could set headers or cookies here that apply to all requests in this session.

The whole thing can be used as a context manager giving us the async with aiohttp.ClientSession() as session syntax and any indented code afterward now takes place inside this context. Then we call get(url) on the session object and use that as a context manager in the same exact manner as the line above it.

The line after calls await on the response.text() and will then save this in the html_content variable. This await keyword is the magic, and whenever you see this keyword it sort of pauses this code, as time is needed to fetch the webpage. While this is happening, other tasks can run in the event loop.

When the html_content has finished fetching, we move outside of the two async context managers and call our parse_html function to get the text content of the page. We then print a message to the console that the URL was fetched successfully and return the text content.

Note that we could easily edit the above function to fetch the whole list of URLs we have inside the same ClientSession context manager, but as the overhead to calling this function a couple of times is minimal, I’ll just keep it as is for now.

Another fair point to make is that the parse_html function is technically blocking non-async code, but as it doesn’t take long to run at all, it’s fine to keep it here. The main time-waster is the fetching of the webpages and we made that asynchronous.

Input arguments and the tool

Before we get to the actual tool itself we need to make sure to define our pydantic object with the input arguments for the tool:

class ResearchInput(BaseModel):
    research_urls: list[str] = Field(description="Must be a list of valid URLs.")

No surprises here, we just want a list of URLs in string format. We’ve used this type of object several times before.

Now let’s write our tool, starting with the first half:

@tool("research", args_schema=ResearchInput)
async def research(research_urls: list[str]) -> str:
    """Get content of provided URLs for research purposes."""
    tasks = [asyncio.create_task(get_webpage_content(url)) for url in research_urls]

We use the @tool decorator to define our tool, passing in the name and the argument schema as always. We declare the function making sure to use async def, and we declare the same research_urls argument as we defined in the ResearchInput class. Again mind the docstring description for the LLM to use.

Then we use a list comprehension, let’s read it from the right to the left. for each url in the list of research_urls, we call asyncio.create_task(get_webpage_content(url)) to create a task for each URL. The asyncio.create_task() function schedules the coroutine to run on the event loop and returns a Task object. However, it doesn’t automatically await the task.

What this means is that it will create our async task and also start it for us, but it won’t await it, or wait for it to finish, which would block the code. We are left with a list full of these task objects of tasks that are currently running but not yet finished.

Let’s finish our tool:

@tool("research", args_schema=ResearchInput)
async def research(research_urls: list[str]) -> str:
    """Get content of provided URLs for research purposes."""
    tasks = [asyncio.create_task(get_webpage_content(url)) for url in research_urls]
    contents = await asyncio.gather(*tasks, return_exceptions=True)
    return json.dumps(contents)

The asyncio.gather() function is used to schedule multiple tasks to run and waits for all of them to complete. It will wait for all our tasks from the previous line to fetch their web pages and then gather the results. This is why we await this function, and then save the results in contents. *tasks is a way to unpack the list of tasks into separate arguments passing them into the function.

The return_exceptions parameter in asyncio.gather() determines how exceptions are handled. If return_exceptions is set to False, gather() will immediately raise the first exception it encounters. When set to True, instead of raising exceptions, it will return them in the result list so that contents will be a list of results or exceptions. We use this as we want to go ahead and fetch the rest of the pages even if one fails.

Finally, dump the response to a JSON string and return it, as naturally, LLMs need string input.

Testing the tool

Now let’s add a quick test to this file to test our tool in isolation and make sure there are no problems:

if __name__ == "__main__":
    import time

    TEST_URLS = [
        "https://en.wikipedia.org/wiki/SpongeBob_SquarePants",
        "https://en.wikipedia.org/wiki/Stephen_Hillenburg",
        "https://en.wikipedia.org/wiki/The_SpongeBob_Movie:_Sponge_Out_of_Water",
    ]

    async def main():
        result = await research.ainvoke({"research_urls": TEST_URLS})

        with open("test.json", "w") as f:
            json.dump(result, f)

    start_time = time.time()
    asyncio.run(main())
    end_time = time.time()
    print(f"Async time: {end_time - start_time} seconds")

We’ve covered the if __name__ == "__main__": block before, so only if we run this file directly will the code inside this block run. We define a list of test URLs to use and then define an async function called main() to run our tool with these test URLs. Instead of invoking the tool as we normally do we now use ainvoke for the async version, and we have to await the result. This is why the main function is async as well.

We then open a file called test.json in write mode and dump the result to it so we can have a quick look to check if the output is as expected. Finally we run the main() function with asyncio.run(main())

asyncio.run is a useful function that creates a new event loop, runs the given coroutine which is main in our case, closes the loop, and then returns the result. This makes it a convenient way to run async code from a synchronous context as it handles the whole event loop thing for us.

I’ve also sneaked a start and end timer in there using time.time() to see how long it takes to run the async code.

Now go ahead and run the web.py file and you’ll see something like this:

URL: https://en.wik