如何使用LangChain Python并行执行多个查询？

你已经通过LangChain实现了单查询的RAG链，现在要处理多查询列表并实现并行执行，可通过以下两种常用方案实现：

方案一：使用batch()方法（推荐） #

LangChain的Runnable接口内置了batch()方法，支持批量并行处理多个输入，底层会自动优化执行效率，代码实现最简单：

# 直接传入查询列表，批量并行执行
results = rag_chain.batch(queries)

# 遍历输出每个查询对应的结果
for query, answer in zip(queries, results):
    print(f"查询: {query}")
    print(f"回答: {answer}\n")

方案二：使用异步调用ainvoke() + asyncio #

如果需要更灵活的异步流程控制，可以结合Python的asyncio模块，通过异步调用实现并行执行：

import asyncio

async def run_parallel_queries():
    # 为每个查询创建异步任务
    tasks = [rag_chain.ainvoke(query) for query in queries]
    # 并行执行所有异步任务
    results = await asyncio.gather(*tasks)
    return results

# 运行异步函数获取结果
results = asyncio.run(run_parallel_queries())

# 输出结果
for query, answer in zip(queries, results):
    print(f"查询: {query}")
    print(f"回答: {answer}\n")

注意事项 #

• 确保你的LLM和Retriever支持异步操作（大部分主流模型和检索器都支持）
• batch()方法默认会并行执行，无需额外配置，适合大多数场景
• 异步方式适合需要和其他异步逻辑结合的复杂场景

内容的提问来源于stack exchange，提问作者Nicolas REY