自然语言处理从入门到应用——LangChain：链（Chains）-[通用功能：自定义Chain和Chain的异步API]-Toy模板网

这篇具有很好参考价值的文章主要介绍了自然语言处理从入门到应用——LangChain：链（Chains）-[通用功能：自定义Chain和Chain的异步API]。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

LangChain系列文章：

基础知识
快速入门
- 安装与环境配置
- 链（Chains）、代理（Agent:）和记忆（Memory）
- 快速开发聊天模型
模型（Models）
- 基础知识
- 大型语言模型（LLMs）
  - 基础知识
  - LLM的异步API、自定义LLM包装器、虚假LLM和人类输入LLM（Human Input LLM）
  - 缓存LLM的调用结果
  - 加载与保存LLM类、流式传输LLM与Chat Model响应和跟踪tokens使用情况
- 聊天模型（Chat Models）
  - 基础知识
  - 使用少量示例和响应流式传输
- 文本嵌入模型
  - Aleph Alpha、Amazon Bedrock、Azure OpenAI、Cohere等
  - Embaas、Fake Embeddings、Google Vertex AI PaLM等
提示（Prompts）
- 基础知识
- 提示模板
  - 基础知识
  - 连接到特征存储
  - 创建自定义提示模板和含有Few-Shot示例的提示模板
  - 部分填充的提示模板和提示合成
  - 序列化提示信息
- 示例选择器（Example Selectors）
- 输出解析器（Output Parsers）
记忆（Memory）
- 基础知识
- 记忆的类型
  - 会话缓存记忆、会话缓存窗口记忆和实体记忆
  - 对话知识图谱记忆、对话摘要记忆和会话摘要缓冲记忆
  - 对话令牌缓冲存储器和基于向量存储的记忆
- 将记忆添加到LangChain组件中
- 自定义对话记忆与自定义记忆类
- 聊天消息记录
- 记忆的存储与应用
索引（Indexes）
- 基础知识
- 文档加载器（Document Loaders）
- 文本分割器（Text Splitters）
- 向量存储器（Vectorstores）
- 检索器（Retrievers）
链（Chains）
- 基础知识
- 通用功能
  - 自定义Chain和Chain的异步API
  - LLMChain和RouterChain
  - SequentialChain和TransformationChain
  - 链的保存（序列化）与加载（反序列化）
- 链与索引
  - 文档分析和基于文档的聊天
  - 问答的基础知识
  - 图问答（Graph QA）和带来源的问答（Q&A with Sources）
  - 检索式问答
  - 文本摘要（Summarization）、HyDE和向量数据库的文本生成
代理（Agents）
- 基础知识
- 代理类型
- 自定义代理（Custom Agent）
- 自定义MRKL代理
- 带有ChatModel的LLM聊天自定义代理和自定义多操作代理（Custom MultiAction Agent）
- 工具
  - 基础知识
  - 自定义工具（Custom Tools）
  - 多输入工具和工具输入模式
  - 人工确认工具验证和Tools作为OpenAI函数
- 工具包（Toolkit）
- 代理执行器（Agent Executor）
  - 结合使用Agent和VectorStore
  - 使用Agents的异步API和创建ChatGPT克隆
  - 处理解析错误、访问中间步骤和限制最大迭代次数
  - 为代理程序设置超时时间和限制最大迭代次数和为代理程序和其工具添加共享内存
- 计划与执行
回调函数（Callbacks）

创建自定义Chain

要实现自己的自定义链式连接，我们可以子类化Chain并实现以下方法：

from __future__ import annotations
from typing import Any, Dict, List, Optional
from pydantic import Extra
from langchain.base_language import BaseLanguageModel
from langchain.callbacks.manager import (
    AsyncCallbackManagerForChainRun,
    CallbackManagerForChainRun,
)
from langchain.chains.base import Chain
from langchain.prompts.base import BasePromptTemplate


class MyCustomChain(Chain):
    """
    An example of a custom chain.
    """

    prompt: BasePromptTemplate
    """Prompt object to use."""
    llm: BaseLanguageModel
    output_key: str = "text"  #: :meta private:

    class Config:
        """Configuration for this pydantic object."""

        extra = Extra.forbid
        arbitrary_types_allowed = True

    @property
    def input_keys(self) -> List[str]:
        """Will be whatever keys the prompt expects.

        :meta private:
        """
        return self.prompt.input_variables

    @property
    def output_keys(self) -> List[str]:
        """Will always return text key.

        :meta private:
        """
        return [self.output_key]

    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, str]:
        # 在这里编写你的自定义链逻辑
        # 下面的示例仅模仿了 LLMChain
        prompt_value = self.prompt.format_prompt(**inputs)
        
        # 当调用语言模型或其他链时，应该将回调管理器传递给它。
        # 这样可以让内部运行受到外部运行注册的任何回调的跟踪。
        # 你可以通过调用 `run_manager.get_child()` 获取回调管理器，如下所示。
        response = self.llm.generate_prompt(
            [prompt_value],
            callbacks=run_manager.get_child() if run_manager else None
        )

        # 如果想要记录此次运行的某些信息，可以通过调用 `run_manager` 上的方法来实现。
        # 这将触发为该事件注册的任何回调。
        if run_manager:
            run_manager.on_text("记录此次运行的一些信息")
        
        return {self.output_key: response.generations[0][0].text}

    async def _acall(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
    ) -> Dict[str, str]:
        # 在这里编写你的自定义链逻辑
        # 下面的示例仅模仿了 LLMChain
        prompt_value = self.prompt.format_prompt(**inputs)
        
        # 当调用语言模型或其他链时，应该将回调管理器传递给它。
        # 这样可以让内部运行受到外部运行注册的任何回调的跟踪。
        # 你可以通过调用 `run_manager.get_child()` 获取回调管理器，如下所示。
        response = await self.llm.agenerate_prompt(
            [prompt_value],
            callbacks=run_manager.get_child() if run_manager else None
        )

        # 如果想要记录此次运行的某些信息，可以通过调用 `run_manager` 上的方法来实现。
        # 这将触发为该事件注册的任何回调。
        if run_manager:
            await run_manager.on_text("记录此次运行的一些信息")
        
        return {self.output_key: response.generations[0][0].text}

    @property
    def _chain_type(self) -> str:
        return "my_custom_chain"

from langchain.callbacks.stdout import StdOutCallbackHandler
from langchain.chat_models.openai import ChatOpenAI
from langchain.prompts.prompt import PromptTemplate


chain = MyCustomChain(
    prompt=PromptTemplate.from_template('tell us a joke about {topic}'),
    llm=ChatOpenAI()
)

chain.run({'topic': 'callbacks'}, callbacks=[StdOutCallbackHandler()])

日志输出：

> Entering new MyCustomChain chain...
Log something about this run
> Finished chain.

输出：

Why did the callback function feel lonely? Because it was always waiting for someone to call it back!'

Chain 的异步 API

LangChain通过利用asyncio模块提供了对链式连接的异步支持。目前，LLMChain（通过 arun、apredict和acall方法）、LLMMathChain（通过arun和acall方法）、ChatVectorDBChain和问答链式连接支持异步方法。其他链式连接的异步支持正在计划中。

import asyncio
import time

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain


def generate_serially():
    llm = OpenAI(temperature=0.9)
    prompt = PromptTemplate(
        input_variables=["product"],
        template="What is a good name for a company that makes {product}?",
    )
    chain = LLMChain(llm=llm, prompt=prompt)
    for _ in range(5):
        resp = chain.run(product="toothpaste")
        print(resp)


async def async_generate(chain):
    resp = await chain.arun(product="toothpaste")
    print(resp)


async def generate_concurrently():
    llm = OpenAI(temperature=0.9)
    prompt = PromptTemplate(
        input_variables=["product"],
        template="What is a good name for a company that makes {product}?",
    )
    chain = LLMChain(llm=llm, prompt=prompt)
    tasks = [async_generate(chain) for _ in range(5)]
    await asyncio.gather(*tasks)

s = time.perf_counter()
# If running this outside of Jupyter, use asyncio.run(generate_concurrently())
await generate_concurrently()
elapsed = time.perf_counter() - s
print('\033[1m' + f"Concurrent executed in {elapsed:0.2f} seconds." + '\033[0m')

s = time.perf_counter()
generate_serially()
elapsed = time.perf_counter() - s
print('\033[1m' + f"Serial executed in {elapsed:0.2f} seconds." + '\033[0m')

输出：

BrightSmile Toothpaste Company


BrightSmile Toothpaste Co.


BrightSmile Toothpaste


Gleaming Smile Inc.


SparkleSmile Toothpaste
Concurrent executed in 1.54 seconds.


BrightSmile Toothpaste Co.


MintyFresh Toothpaste Co.


SparkleSmile Toothpaste.


Pearly Whites Toothpaste Co.


BrightSmile Toothpaste.
Serial executed in 6.38 seconds.

参考文献：
[1] LangChain官方网站：https://www.langchain.com/
[2] LangChain 🦜️🔗 中文网，跟着LangChain一起学LLM/GPT开发：https://www.langchain.com.cn/
[3] LangChain中文网 - LangChain 是一个用于开发由语言模型驱动的应用程序的框架：http://www.cnlangchain.com/文章来源地址https://www.toymoban.com/news/detail-663360.html

到了这里，关于自然语言处理从入门到应用——LangChain：链（Chains）-[通用功能：自定义Chain和Chain的异步API]的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！