如何结合asyncio实现LangChain RetrievalQA的并行处理？

问题原因 #

你当前的代码无法运行，核心原因是RetrievalQA默认的调用方法是同步阻塞的，不能直接用await发起异步调用。要实现多查询并行，需要改用LangChain的异步组件，或者将同步调用包装为异步执行。

可行方案1：使用LangChain原生异步组件 #

LangChain提供了异步版本的LLM和链方法，能原生支持异步操作，性能更优（要求LangChain版本≥0.1.0）。

代码示例 #

import langchain
from langchain.chat_models import AsyncChatOpenAI  # 改用异步LLM
from langchain.chains import RetrievalQA
from langchain.vectorstores import FAISS
from langchain.schema.vectorstore import VectorStoreRetriever
import asyncio
import nest_asyncio

# 初始化向量检索器
retriever = VectorStoreRetriever(vectorstore=FAISS(...))

# 使用异步版本的ChatOpenAI
chat = AsyncChatOpenAI(model="gpt-3.5-turbo-16k", temperature=0.7)

# 初始化RetrievalQA链
qa_chain = RetrievalQA.from_llm(
    chat,
    retriever=retriever,
    return_source_documents=True
)

queries = ['query1', 'query2', 'query3']
data_to_append = []

async def process_query(query):
    vectordbkwargs = {"search_distance": 0.9}
    # 使用链的异步调用方法ainvoke
    result = await qa_chain.ainvoke({"query": query, "vectordbkwargs": vectordbkwargs})
    data_to_append.append({
        "Query": query,
        "Source_Documents": result["source_documents"],
        "Generated_Text": result["result"]
    })

async def main():
    tasks = [process_query(q) for q in queries]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    nest_asyncio.apply()
    asyncio.run(main())

可行方案2：将同步调用包装为异步执行 #

如果使用的LangChain版本不支持异步组件，可通过asyncio.to_thread将同步调用包装到线程池中，实现并行处理。

代码示例 #

import langchain
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.vectorstores import FAISS
from langchain.schema.vectorstore import VectorStoreRetriever
import asyncio
import nest_asyncio

retriever = VectorStoreRetriever(vectorstore=FAISS(...))
chat = ChatOpenAI(model="gpt-3.5-turbo-16k", temperature=0.7)

qa_chain = RetrievalQA.from_llm(
    chat,
    retriever=retriever,
    return_source_documents=True
)

queries = ['query1', 'query2', 'query3']
data_to_append = []

async def process_query(query):
    vectordbkwargs = {"search_distance": 0.9}
    # 用to_thread将同步调用包装为异步
    result = await asyncio.to_thread(
        qa_chain,
        {"query": query, "vectordbkwargs": vectordbkwargs}
    )
    data_to_append.append({
        "Query": query,
        "Source_Documents": result["source_documents"],
        "Generated_Text": result["result"]
    })

async def main():
    tasks = [process_query(q) for q in queries]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    nest_asyncio.apply()
    asyncio.run(main())

注意事项 #

• 方案1是原生异步实现，无线程切换开销，推荐优先使用。
• 方案2本质是多线程并行，适合IO密集型场景（如OpenAI API调用、向量检索），能有效提升处理效率。
• 若向量库检索本身是同步阻塞的，两种方案都能并行处理检索操作。

内容的提问来源于stack exchange，提问作者Abhishek R