使用langchain异步获取网络信息

文章目录

• 前言
• 发现问题
• 分析问题
• 阻塞？
• 阻塞！
• 附：用什么阻塞streamlit进程

  前言

  从来没想过在接入通义千问的时候还会遇到NotImplementedError。实在难以理解，处理过后才明白问题。现在总结后给出结果。

  
  （图片来源网络，侵删）

  发现问题

  我们来看个例子。就比如这段代码：

  摘自ranying666/langchain_tongyi_practical中的5-langchain-search-document_loaders.py

  loader = AsyncChromiumLoader(["https://dataea.cn/okr-keyresult-checklist/"])
  html = loader.load()
  html2text = Html2TextTransformer()
  docs_transformed = html2text.transform_documents(html)
  

  如果把这段直接接入streamlit，那就直接报错：

  NotImplementedError

  比较离谱的是，只有这么一个错误，没有其他信息。

  分析问题

  很难理解为什么是这个错。

  随着排查的进行，发现问题好像出在AsyncChromiumLoader中。

  AsyncChromiumLoader继承自BaseLoader，而BaseLoader的load方法中，调用的却是lazy_load。先不用看这个方法的具体内容，就看这个方法的名字，你大概也能猜出来什么问题了：

  懒加载带来的未能实例化的问题。

  简单地说，就是：streamlit已在前面飞，loader.load()还在后面追。终于，追不上了，就爆炸了。而刚好的是，lazy_load中抛出的异常就是这个NotImplementedError。

  众所周知，streamlit构建网页的过程是单线程的。

  所以，当我们需要请求内容的时候，使用异步请求的AsyncChromiumLoader就会出现这种问题。

  那么，该怎么办呢？

  阻塞？

  怎么办呢？阻塞，对吧？很容易想到。

  于是会想当然的这么用：

  loader = AsyncChromiumLoader(["https://dataea.cn/okr-keyresult-checklist/"])
  html = loader.load()
  while html is None:
    time.sleep(1)
  html2text = Html2TextTransformer()
  docs_transformed = html2text.transform_documents(html)
  

  看着很直观，检测html是否有返回值。

  如果真这么简单的话我也不会把它写在这里（🤷‍♂️）。

  结果就是，还是报错。这又是为什么呢？逻辑没问题呀？

  逻辑是没问题，问题出在底层。作为一个异步函数，他怎么可能没有返回值呢？

  我们来回顾一下load方法：

  # Sub-classes should not implement this method directly. Instead, they should implement the lazy load method.
  def load(self) -> List[Document]:
    """Load data into Document objects."""
    return list(self.lazy_load())
  

  那么，lazy_load是怎么回事呢？

  def lazy_load(self) -> Iterator[Document]:
    """A lazy loader for Documents."""
    if type(self).load != BaseLoader.load:
      return iter(self.load())
    raise NotImplementedError(
      f"{self.__class__.__name__} does not implement lazy_load()"
    )
  

  这里比较有意思的是，对实例化的AsyncChromiumLoader对象（就是这个self），判断AsyncChromiumLoader.load与BaseLoader.load是否一致。

  其实这里比较的是地址信息，因为子类如果重写了这个load方法，那么地址就会被改变。如果不一致的话，就会返回一个迭代器，这个迭代器就是为了后续过程中无论返回的是否是list，都能够迭代。

  听起来没问题。

  但是，懒加载呢？

  async def aload(self) -> List[Document]:
    """Load data into Document objects."""
    return [document async for document in self.alazy_load()]
  async def alazy_load(self) -> AsyncIterator[Document]:
    """A lazy loader for Documents."""
    iterator = await run_in_executor(None, self.lazy_load)
    done = object()
    while True:
      doc = await run_in_executor(None, next, iterator, done)  # type: ignore[call-arg, arg-type]
      if doc is done:
        break
      yield doc  # type: ignore[misc]
  

  比较神奇的就出现在这里了。alazy_load方法最终给出来的就是一个Document类的迭代器，然后最终通过yield给到调用方，直到doc在迭代过程中达到了done。

  但是呢，doc变量的结果是await run_in_executor(None, next, iterator, done)，即使run_in_executor返回的是一个迭代器对象，最终由await进行处理，所以是有返回值的，但是返回的是未来需要返回的，是asyncio.futures.Future类。这一点完全可以类比Java中的Future对象。

  所以，最终而言，AsyncChromiumLoader.load并不是直到结束才返回值，而是在执行的过程中不断地通过yield给出返回值，只是在await最终处理为AsyncIterator[Document]类型。

  阻塞！

  为了让streamlit等待异步请求，就需要主线程停下来，直到请求结束了才能继续执行。

  那这回该怎么办呢？直接用asyncio与playwright给阻塞掉。

  首先，我们需要利用asyncio创建一个阻塞事件，并设置所有的事件都需要在阻塞事件结束后执行。

  其次，在执行这个阻塞之间的时候，我们依然使用异步请求，只不过是所有的事件都在等我们。

  于是，可以给出代码如下：

  import asyncio
  import platform
  from playwright.async_api import async_playwright
  # 阻塞事件
  async def fetch_page_content(url):
    async with async_playwright() as p:
      browser = await p.chromium.launch()
      page = await browser.new_page()
      await page.goto(url)
      content = await page.content()
      await browser.close()
      return content
  # 阻塞主进程
  @st.cache_data
  def load_documents(url):
    loop = None
    if platform.system() == 'Windows':
      loop = asyncio.ProactorEventLoop()
    elif platform.system() == 'Linux':
      loop = asyncio.new_event_loop()
    elif platform.system() == 'Darwin':
      loop = asyncio.SelectorEventLoop()
    else:
      return None
    html_content = loop.run_until_complete(fetch_page_content(url))
    html2text = Html2TextTransformer()
    document = Document(page_content=html_content)
    docs_transformed = list(html2text.transform_documents([document]))
    return docs_transformed
  

  其实这里面最核心的就是asyncio下的run_until_complete函数：

  def run_until_complete(self, future):
    """Run until the Future is done.
    If the argument is a coroutine, it is wrapped in a Task.
    WARNING: It would be disastrous to call run_until_complete()
    with the same coroutine twice -- it would wrap it in two
    different Tasks and that can't be good.
    Return the Future's result, or raise its exception.
    """
    self._check_closed()
    self._check_running()
    new_task = not futures.isfuture(future)
    future = tasks.ensure_future(future, loop=self)
    if new_task:
      # An exception is raised if the future didn't complete, so there
      # is no need to log the "destroy pending task" message
      future._log_destroy_pending = False
    future.add_done_callback(_run_until_complete_cb)
    try:
      self.run_forever()
    except:
      if new_task and future.done() and not future.cancelled():
        # The coroutine raised a BaseException. Consume the exception
        # to not log a warning, the caller doesn't have access to the
        # local task.
        future.exception()
      raise
    finally:
      future.remove_done_callback(_run_until_complete_cb)
    if not future.done():
      raise RuntimeError('Event loop stopped before Future completed.')
    return future.result()
  

  这个函数最大的特点就是会创建一个任务并执行。直到任务执行完成或者报错中断之前，其他所有任务都得等着这个任务的回调函数。

  于是，这个函数就阻塞了streamlit的进程，直到异步任务完成。

  附：用什么阻塞streamlit进程

  其实这段文字本来应该接在上面这段的。但是这个坑实在太神奇了，单独拉出来说明。

  这里面还有一个很神奇的坑：用什么东西阻塞。

  就像上面这段代码，针对Windows、Linux、Darwin，分别采用了ProactorEventLoop、new_event_loop、SelectorEventLoop阻塞streamlit进程。

  如果在Linux平台中使用ProactorEventLoop，那么streamlit进程依然不会阻塞，因为他们都只能在各自的操作系统中起作用。