使用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:breakyield 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 = Noneif platform.system() == 'Windows':loop = asyncio.ProactorEventLoop()elif platform.system() == 'Linux':loop = asyncio.new_event_loop()elif platform.system() == 'Darwin':loop = asyncio.SelectorEventLoop()else:return Nonehtml_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 twodifferent 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" messagefuture._log_destroy_pending = Falsefuture.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()raisefinally:future.remove_done_callback(_run_until_complete_cb)if not future.done():raise RuntimeError('Event loop stopped before Future completed.')return future.result()