如何在LangChain中运行异步方法？chain.aapply调用异常排查

搭建基于欧洲语言共同参考框架（CEFR）的文本分类LLMChain时，测试chain.apply与chain.aapply的耗时差异出现异常：chain.aapply返回协程对象，同时抛出coroutine 'LLMChain.aapply' was never awaited运行时警告。

测试代码 #

import os
from time import time

import openai
from dotenv import load_dotenv, find_dotenv
from langchain.chains import LLMChain
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

_ = load_dotenv(find_dotenv())
openai.api_key = os.getenv('OPENAI_API_KEY')

llm = ChatOpenAI(temperature=0)

prompt = ChatPromptTemplate.from_template(
    'Classify the text based on the Common European Framework of Reference '
    'for Languages (CEFR). Give a single value: {text}',
)
chain = LLMChain(llm=llm, prompt=prompt)

texts = [
    {'text': 'Hallo, ich bin 25 Jahre alt.'},
    {'text': 'Wie geht es dir?'},
    {'text': 'In meiner Freizeit, spiele ich gerne Fussball.'}
]

start = time()
res_a = chain.apply(texts)
print(res_a)
print(f"apply time taken: {time() - start:.2f} seconds")
print()

start = time()
res_aa = chain.aapply(texts)
print(res_aa)
print(f"aapply time taken: {time() - start:.2f} seconds")

运行输出 #

[{'text': 'Based on the given text "Hallo, ich bin 25 Jahre alt," it can be classified as CEFR level A1.'}, {'text': 'A2'}, {'text': 'A2'}]
apply time taken: 2.24 seconds

<coroutine object LLMChain.aapply at 0x0000025EA95BE3B0>
aapply time taken: 0.00 seconds

C:\Users\User\AppData\Local\Temp\ipykernel_13620\1566967258.py:34: RuntimeWarning: coroutine 'LLMChain.aapply' was never awaited
  res_aa = chain.aapply(texts)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

问题原因 #

aapply是LangChain提供的异步批量调用方法，和同步的apply本质不同：

• 同步方法apply调用后会直接执行并返回结果
• 异步方法aapply调用后仅返回协程对象，不会自动执行，必须通过await关键字等待其完成，且异步代码需要在异步运行环境中执行。直接调用而不等待，就会触发未等待协程的警告，同时无法得到实际运行结果。

修复后的代码 #

import os
import asyncio
from time import time

import openai
from dotenv import load_dotenv, find_dotenv
from langchain.chains import LLMChain
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

_ = load_dotenv(find_dotenv())
openai.api_key = os.getenv('OPENAI_API_KEY')

llm = ChatOpenAI(temperature=0)

prompt = ChatPromptTemplate.from_template(
    'Classify the text based on the Common European Framework of Reference '
    'for Languages (CEFR). Give a single value: {text}',
)
chain = LLMChain(llm=llm, prompt=prompt)

texts = [
    {'text': 'Hallo, ich bin 25 Jahre alt.'},
    {'text': 'Wie geht es dir?'},
    {'text': 'In meiner Freizeit, spiele ich gerne Fussball.'}
]

# 同步调用测试
start = time()
res_a = chain.apply(texts)
print(res_a)
print(f"apply time taken: {time() - start:.2f} seconds")
print()

# 异步调用测试
async def run_async_batch():
    start = time()
    res_aa = await chain.aapply(texts)
    print(res_aa)
    print(f"aapply time taken: {time() - start:.2f} seconds")

# 启动异步任务
asyncio.run(run_async_batch())

修改说明 #

1. 导入asyncio库，用于创建和运行异步执行环境
2. 定义异步函数run_async_batch，在函数内部用await等待chain.aapply完成，确保异步任务实际执行
3. 用asyncio.run()启动异步函数，触发协程执行，这样就能得到正确的异步调用结果和耗时统计

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