在本文中，我将介绍如何使用LangChain在长工作流中对LLM的异步调用。我们将通过一个包含完整代码的示例来比较顺序执行和异步调用。

以下是内容概述：

1. 基础知识：什么是LangChain

2. 如何使用LangChain运行同步链

3. 如何使用LangChain运行单个异步链

4. 使用异步链进行长工作流技巧

基础知识：什么是LangChain

LangChain是一个用于开发由语言模型驱动的应用程序的框架。这就是LangChain的官方定义。这个框架是最近创建的，并且已经被用作构建由LLM驱动的工具的行业标准。

它是开源的，维护得很好，新功能在很短的时间内发布。

这个库的一个缺点是，由于这些特性是新的，我们不能使用Chat GPT来有效地帮助构建新代码。所以这意味着我们必须以“古老”的方式阅读文档、论坛和教程。

LangChain的文档。确实很好，但是没有很多具体的例子。

 

如何使用LangChain运行同步链

问题：我有一个数据帧有很多行，对于每一行我需要运行多个提示(链)到LLM，并将结果返回到我的数据帧。

当你有多行时，假设是10K，为每个响应运行3个提示(如果服务器没有过载)，大约需要3-5秒，你最终需要等待数天才能完成工作流。

下面我将展示构建同步链的主要步骤和代码，并对数据子集进行计时。

对于这个例子，我将使用数据集Wine Reviews, license。

为此，我创建了两个链，一个用于摘要和情感，另一个将摘要作为输入提取特征。

下面是运行它的代码：

from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain
from langchain.chains import SequentialChain
from langchain.callbacks import get_openai_callback
import time
import pandas as pd

from langchain.output_parsers import ResponseSchema
from langchain.output_parsers import StructuredOutputParser
class SummaryChain:
    """
    Class for creating a summary chain for extracting main sentiment and summary from wine reviews.
    
    Attributes
    ----------
    df : pandas.DataFrame
        The dataframe that contains the wine reviews.
    llm : langchain.chat_models.ChatOpenAI
        The language model for extracting the summary.
    
    Methods
    ----------
    build_chain(): 
        Builds a SequentialChain for sentiment extraction. 
    generate_sequentially(): 
        Generates sentiment and summary sequentially for each review in the dataframe.
    """ 
    def __init__(self, df, llm): 
        self.df = df 
        self.llm = llm 

    def build_chain(self): 
    """ 
    Builds a SequentialChain for sentiment extraction. 

    Returns 
    -------
    tuple 
        A tuple containing the built SequentialChain, the output parser, and the response format. 
    """ 

    llm = self.llm 

    sentiment_schema = ResponseSchema( 
        name="sentiment", 
        description="The main sentiment of the review, limited to 3 words.", 
    ) 
    summary_schema = ResponseSchema( 
        name="summary", 
        description="Brief Summary of the review, limited to one paragraph.", 
    ) 
    sentiment_response_schemas = [sentiment_schema, summary_schema] 

    output_parser = StructuredOutputParser.from_response_schemas( 
        sentiment_response_schemas 
    ) 
    response_format = output_parser.get_format_instructions() 

    sentiment_prompt = ChatPromptTemplate.from_template( 
        """Act like an expert somellier. Your goal is to extract the main sentiment from wine reviews, delimited by triple dashes. Limit the sentiment to 3 words. \ 

        Review: {review} 

        {response_format} 
        """ 
    )  

    sentiment_chain = LLMChain( 
        llm=llm, prompt=sentiment_prompt, output_key="sentiment" 
    ) 

    chain = SequentialChain( 
        chains=[sentiment_chain], 
        input_variables=["review"] + ["response_format"], 
        output_variables=["sentiment"], 
        verbose=False, 
    ) 

    return chain, output_parser, response_format 
def generate_sequentially(self): 
    """ 
    Generates sentiment and summary sequentially for each review in the dataframe. 
    The extracted sentiments, summaries, and costs are added to the dataframe. 
    """ 
    df = self.df 
    chain, output_parser, response_format = self.build_chain() 
    for _, row in df.iterrows(): 
        review = row["description"] 
        unique_id = row["unique_id"] 
        inputs = { 
            "review": review, 
            "response_format": response_format, 
        } 
        with get_openai_callback() as cb: 
            resp = chain.run(inputs) 
            cost = cb.total_cost 
        summary = output_parser.parse(resp)["summary"] 
        sentiment = output_parser.parse(resp)["sentiment"] 
        df.loc[ 
            df["unique_id"] == unique_id, ["summary", "sentiment", "sentiment_cost"] 
        ] = [summary, sentiment, cost] 

class CharacteristicsChain: 
    """ 
    Class for creating a chain for extracting top five main characteristics of the wine. 
    
    Attributes 
    df : pandas.DataFrame 
        The dataframe that contains the wine reviews. 
    llm : langchain.chat_models.ChatOpenAI 
        The language model for extracting the characteristics. 

    Methods 
    build_chain(): 
        Builds a SequentialChain for characteristic extraction. 
    generate_sequentially(): 
        Generates characteristics sequentially for each wine in the dataframe. 
    """ 

    def __init__(self, df, llm): 
        self.df = df 
        self.llm = llm 
    def build_chain(self): 
        """ 
        Builds a SequentialChain for characteristic extraction. 
        Returns 
        tuple 
            A tuple containing the built SequentialChain, the output parser, and the response format. 
        """ 
        llm = self.llm 

        characteristics_schema = [] 
        for i in range(1, 6): 
            characteristics_schema.append( 
                ResponseSchema( 
                    name=f"characteristic_{i}", 
                    description=f"The number {i} characteristic. One or two words long.", 
                 ) 
             ) 
         output_parser = StructuredOutputParser.from_response_schemas( 
             characteristics_schema 
         ) 
         response_format = output_parser.get_format_instructions() 

         characteristics_prompt = ChatPromptTemplate.from_template( 
             """ 
         Act like an expert somellier. You will receive the name, the summary of the review and the county of origin of a given wine, delimited by triple dashes. 
         Your goal is to extract the top five main characteristics of the wine. 
             Wine Name: {wine_name} 
             Country: {country} 
             Summary Review: {summary} 
             --- 
             {response_format} 
             """ 
         ) 
         characteristics_chain = LLMChain( 
             llm=llm, prompt=characteristics_prompt, output_key="characteristics" 
         ) 
         chain = SequentialChain( 
             chains=[characteristics_chain], 
             input_variables=["wine_name", "summary", "country"] + ["response_format"], output_variables=["characteristics"], 
             verbose=False, 
         ) 
         return chain, output_parser, response_format 
     def generate_sequentially(self): 
         """ 
         Generates characteristics sequentially for each wine in the dataframe. 
         The extracted characteristics and costs are added to the dataframe. 
         """ 

         df = self.df 
         chain, output_parser, response_format = self.build_chain() 
         for _, row in df.iterrows(): 
             summary = row["summary"] 
             country = row["country"] 
             unique_id = row["unique_id"] 
             title = row["title"] 
             inputs = { 
                 "summary": summary, 
                 "wine_name": title, 
                 "country": country, 
                 "response_format": response_format, 
              } 
              with get_openai_callback() as cb: 
                  resp = chain.run(inputs) 
                  cost = cb.total_cost 
              characteristics = [ 
                  output_parser.parse(resp)[f"characteristic_{i}"] for i in range(1, 6) 
              ] 
              df.loc[ 
                  df.unique_id == unique_id, 
                  [ 
                      "characteristic_1", 
                      "characteristic_2", 
                      "characteristic_3", 
                      "characteristic_4", 
                      "characteristic_5", 
                      "cost_characteristics", 
                   ], 
               ] = characteristics + [cost] 

######### RUNNING THE CODE #########
llm = ChatOpenAI( 
    temperature=0.0, 
    request_timeout=15, 
    model_name="gpt-3.5-turbo", 
) 
df = pd.read_csv('wine_subset.csv') 

# Summary Chain - Sequential 
s = time.perf_counter() 
summary_chain = SummaryChain(llm=llm,df=df) 
summary_chain.generate_sequentially() 
elapsed = time.perf_counter() - s 
print("\033[1m" + f"Summary Chain (Sequential) executed in {elapsed:0.2f} seconds." + "\033[0m") 

# Characteristics Chain - Sequential 
s = time.perf_counter() 
characteristics_chain = CharacteristicsChain(llm=llm,df=df) 
characteristics_chain.generate_sequentially() 
elapsed = time.perf_counter() - s 
print("\033[1m" + f"Characteristics Chain (Sequential) executed in {elapsed:0.2f} seconds." + "\033[0m")

运行时间(10个示例)：

摘要链(顺序)在22.59秒内执行。

特征链(顺序)在22.85秒内执行。

这段代码的主要内容是链的构建块，如何以顺序方式运行它，以及完成这个循环所花费的时间。重要的是要记住，10个示例大约需要45秒，并且完整的数据集包含130K行。所以异步实现是在合理的时间内运行它的新希望。

有了问题和基线，让我们看看如何优化这段代码，使其运行得更快。

如何使用LangChain运行单个异步链

对于这个，我们将使用一个叫做异步调用的资源。

在我们的示例中，我们遍历数据帧的每一行，从中提取一些信息，将它们添加到提示符中，并调用GPT API以获得响应。响应之后，我们只需解析它并将其添加回数据帧。



这里的主要瓶颈是当我们调用GPT API时，因为我们的计算机必须等待该API的响应(大约3秒)。

因此，如果我们同时将所有调用发送到API，而不是安静地等待响应，会怎么样？这样，我们只需要等待一个响应，然后处理它们。这称为对API的异步调用。



通过这种方式，我们按顺序进行预处理和后处理，但对 API 的调用不必等待最后一个响应返回后再发送下一个响应。

下面是异步链的代码：

from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain
from langchain.chains import SequentialChain
import time
import pandas as pd
import asyncio
from langchain.callbacks import get_openai_callback
from langchain.output_parsers import ResponseSchema
from langchain.output_parsers import StructuredOutputParser

class SummaryChain:
    """
    Class for creating a summary chain for extracting main sentiment and summary from wine reviews.
  
    Attributes

    df : pandas.DataFrame
        The dataframe that contains the wine reviews.
    llm : langchain.chat_models.ChatOpenAI
        The language model for extracting the summary.
 
    Methods

    build_chain():
        Builds a SequentialChain for sentiment extraction.
    generate_concurrently():
        Generates sentiment and summary concurrently for each review in the dataframe.
    async_generate(chain, inputs, unique_id):
        Asynchronous task to extract sentiment and summary from a single review.
    """

    def __init__(self, df, llm):
        self.df = df
        self.llm = llm
    def build_chain(self):
        """
        Builds a SequentialChain for sentiment extraction.
        Returns

        tuple
            A tuple containing the built SequentialChain, the output parser, and the response format.
        """

        llm = self.llm

        sentiment_schema = ResponseSchema(
            name="sentiment",
            description="The main sentiment of the review, limited to 3 words.",
        )
        summary_schema = ResponseSchema(
            name="summary",
            description="Brief Summary of the review, limited to one paragraph.",
        )

        sentiment_response_schemas = [sentiment_schema, summary_schema]

        output_parser = StructuredOutputParser.from_response_schemas(
            sentiment_response_schemas
        )
        response_format = output_parser.get_format_instructions()

        ## sentiment and Summary Chain
        sentiment_prompt = ChatPromptTemplate.from_template(
            """Act like an expert somellier. Your goal is to extract the main sentiment from wine reviews, delimited by triple dashes. Limit the sentiment to 3 words. \

            Review: {review}

            {response_format}
            """
        )

        sentiment_chain = LLMChain(llm=llm, prompt=sentiment_prompt, output_key="sentiment")
        chain = SequentialChain(
            chains=[sentiment_chain],
            input_variables=["review"] + ["response_format"],
            output_variables=["sentiment"],
            verbose=False,
        )

        return chain, output_parser, response_format

    async def generate_concurrently(self):
        """
        Generates sentiment and summary concurrently for each review in the dataframe.
        The extracted sentiments, summaries, and costs are added to the dataframe.
        """

        df = self.df

        chain, output_parser, response_format = self.build_chain()

        tasks = []
        for _, row in df.iterrows():
            review = row["description"]
            unique_id = row["unique_id"]

            inputs={
                       "review": review,
                       "response_format": response_format,
                   }
            tasks.append(self.async_generate(chain, inputs, unique_id))

        results = await asyncio.gather(*tasks)
        for unique_id, response, cost in results:
            summary = output_parser.parse(response)["summary"]
            sentiment = output_parser.parse(response)["sentiment"]

            df.loc[df["unique_id"] == unique_id, ["summary", "sentiment", "sentiment_cost"]] = [summary, sentiment, cost]
    async def async_generate(self, chain, inputs, unique_id):
        """
        Asynchronous task to extract sentiment and summary from a single review.
        Parameters

        chain : SequentialChain
            The SequentialChain used for sentiment extraction.
        inputs : dict
            The inputs for the chain.
        unique_id : any
            The unique identifier for the review.
        Returns

        tuple
            A tuple containing the unique identifier, the extracted sentiment and summary, and the cost.
        """
        with get_openai_callback() as cb:
            resp = await chain.arun(inputs)
        return unique_id, resp, cb.total_cost

class CharacteristicsChain:
    """
    Class for creating a chain for extracting top five main characteristics of the wine.
    Attributes

    df : pandas.DataFrame
        The dataframe that contains the wine reviews.
    llm : langchain.chat_models.ChatOpenAI
        The language model for extracting the characteristics.
    Methods

    build_chain():
        Builds a SequentialChain for characteristic extraction.
    generate_concurrently():
        Generates characteristics concurrently for each wine in the dataframe.
    async_generate(chain, inputs, unique_id):
         Asynchronous task to extract characteristics from a single wine.
    """

    def __init__(self, df, llm):
        self.df = df
        self.llm = llm

    def build_chain(self):
        """
        Builds a SequentialChain for characteristic extraction.
        Returns

        tuple
            A tuple containing the built SequentialChain, the output parser, and the response format.
        """

        llm = self.llm

        characteristics_schema = []
        for i in range(1, 6):
            characteristics_schema.append(
                ResponseSchema(
                    name=f"characteristic_{i}",
                    description=f"The number {i} characteristic. One or two words long.",
                )
             )

         output_parser = StructuredOutputParser.from_response_schemas(characteristics_schema)
         response_format = output_parser.get_format_instructions()

         characteristics_prompt = ChatPromptTemplate.from_template(
         """
         Act like an expert somellier. You will receive the name, the summary of the review and the county of origin of a given wine, delimited by triple dashes.
         Your goal is to extract the top five main characteristics of the wine.

              Wine Name: {wine_name}
              Country: {country}
              Summary Review: {summary}

              {response_format}
              """
         )
         characteristics_chain = LLMChain(
             llm=llm, prompt=characteristics_prompt, output_key="characteristics"
         )

         chain = SequentialChain(
             chains=[characteristics_chain],
             input_variables=["wine_name", "summary", "country"]
             + ["response_format"],
             output_variables=["characteristics"],
             verbose=False,
         )

         return chain, output_parser, response_format

     async def generate_concurrently(self):
         """
         Generates characteristics concurrently for each wine in the dataframe.
         The extracted characteristics and costs are added to the dataframe.
         """

         df = self.df

         chain, output_parser, response_format = self.build_chain()

         asks = []
         for _, row in df.iterrows():
             summary = row["summary"]
             country = row["country"]
             unique_id = row["unique_id"]
             title = row["title"]

             inputs={
                       "summary": summary,
                       "wine_name": title,
                       "country":country,
                       "response_format": response_format,
                    }
             tasks.append(self.async_generate(chain, inputs, unique_id))

         results = await asyncio.gather(*tasks)
         for unique_id, response, cost in results:
             characteristic_1 = output_parser.parse(
                 response)["characteristic_1"]
             characteristic_2 = output_parser.parse(
                 response)["characteristic_2"]
             characteristic_3 = output_parser.parse(
                 response)["characteristic_3"]
             characteristic_4 = output_parser.parse(
                 response)["characteristic_4"]
             characteristic_5 = output_parser.parse(
                 response)["characteristic_5"]
             df.loc[df.unique_id == unique_id, [
                 "characteristic_1",
                 "characteristic_2",
                 "characteristic_3",
                 "characteristic_4",
                 "characteristic_5",
                 "cost_characteristics"
             ]] = [
                  characteristic_1,
                  characteristic_2,
                  characteristic_3,
                  characteristic_4,
                  characteristic_5,
                   cost,
             ]

     async def async_generate(self, chain, inputs, unique_id):
         """
         Asynchronous task to extract characteristics from a single wine.
         Parameters

         chain : SequentialChain
             The SequentialChain used for characteristic extraction.
         inputs : dict
             The inputs for the chain.
         unique_id : any
             The unique identifier for the wine.
         Returns

         tuple
             A tuple containing the unique identifier, the extracted characteristics, and the cost.
         """
         with get_openai_callback() as cb:
             resp = await chain.arun(inputs)
         return unique_id, resp, cb.total_cost
 ######### RUNNING THE CODE #########

llm = ChatOpenAI(
    temperature=0.0,
    request_timeout=15,
    model_name="gpt-3.5-turbo",
)

df = pd.read_csv('wine_subset.csv')

# Issue Chain
s = time.perf_counter()
summary_chain = SummaryChain(llm=llm,df=df)
asyncio.run(summary_chain.generate_concurrently())
elapsed = time.perf_counter() - s
print("\033[1m" + f"Summary Chain (Async) executed in {elapsed:0.2f} seconds." + "\033[0m")

# Characteristics Chain
s = time.perf_counter()
characteristics_chain = CharacteristicsChain(llm=llm,df=df)
asyncio.run(characteristics_chain.generate_concurrently())
elapsed = time.perf_counter() - s
print("\033[1m" + f"Characteristics Chain (Async) executed in {elapsed:0.2f} seconds." + "\033[0m")

df.to_csv('checkpoint.csv')

在这段代码中，我们使用了async和await的Python语法。LangChain还提供了使用arun()函数异步运行链的代码。因此，在开始时，我们首先顺序处理每一行(可以优化)，并创建多个“任务”，这些任务将并行等待API的响应，然后我们依次处理对最终所需格式的响应(也可以优化)。

运行时间(10个示例)：

摘要链(异步)在3.35秒内执行。

特征链(异步)在2.49秒内执行。

与顺序的相比:

摘要链(顺序)在22.59秒内执行。

特征链(顺序)在22.85秒内执行。

我们可以看到在运行时几乎有10倍的改进。因此，对于较大的工作负载，我强烈建议使用这种方法。此外，我的代码充满了for循环，也可以进一步优化以提高性能。

使用异步链进行长工作流的实现技巧

当我不得不运行它时，我遇到了一些限制和障碍，我想和你们分享。

笔记本电脑不适合异步

在Jupyter笔记本上运行异步调用时，你可能会遇到一些问题。然而，只要问一下Chat GPT，它可能会帮你解决这个问题。我构建的代码是为了在.py文件中运行大型工作负载，因此可能需要进行一些更改才能在笔记本中运行。

输出键太多

第一个问题是我的链有多个键作为输出，而当时arun()只接受有一个键作为输出的链。所以为了解决这个问题，我必须把我的链分成两个。

并不是所有的链都是异步的

在我的提示中，我有一个使用矢量数据库进行示例和比较的逻辑，这需要顺序比较示例并将其添加到数据库中。这使得在整个链中对这个链接使用异步是不可行的。

 

结论

LangChain是一个非常强大的工具，用于创建基于LLM的应用程序。

对于运行链的特定主题，对于高工作负载，我们看到了异步调用具有的潜在改进，因此我的建议是花时间了解代码在做什么，并拥有一个样板类(例如我的代码中提供的类)并异步运行它!

对于只需要调用一次API的小型工作负载或应用程序，没有必要异步执行，但是如果你有一个样板类，只需添加一个同步函数，这样就可以轻松地使用。

 

来源：https://towardsdatascience.com/async-calls-for-chains-with-langchain-3818c16062ed