大模型从入门到应用——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的异步APILLMChain和RouterChainSequentialChain和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 annotationsfrom typing import Any, Dict, List, Optionalfrom pydantic import Extrafrom langchain.base_language import BaseLanguageModelfrom langchain.callbacks.manager import (AsyncCallbackManagerForChainRun,CallbackManagerForChainRun,)from langchain.chains.base import Chainfrom langchain.prompts.base import BasePromptTemplateclass MyCustomChain(Chain):"""An example of a custom chain."""prompt: BasePromptTemplate"""Prompt object to use."""llm: BaseLanguageModeloutput_key: str = "text" #: :meta private:class Config:"""Configuration for this pydantic object."""extra = Extra.forbidarbitrary_types_allowed = True@propertydef input_keys(self) -> List[str]:"""Will be whatever keys the prompt expects.:meta private:"""return self.prompt.input_variables@propertydef 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]:# 在这里编写你的自定义链逻辑# 下面的示例仅模仿了 LLMChainprompt_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]:# 在这里编写你的自定义链逻辑# 下面的示例仅模仿了 LLMChainprompt_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}@propertydef _chain_type(self) -> str:return "my_custom_chain"from langchain.callbacks.stdout import StdOutCallbackHandlerfrom langchain.chat_models.openai import ChatOpenAIfrom langchain.prompts.prompt import PromptTemplatechain = 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 asyncioimport timefrom langchain.llms import OpenAIfrom langchain.prompts import PromptTemplatefrom langchain.chains import LLMChaindef 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() - sprint('\033[1m' + f"Concurrent executed in {elapsed:0.2f} seconds." + '\033[0m')s = time.perf_counter()generate_serially()elapsed = time.perf_counter() - sprint('\033[1m' + f"Serial executed in {elapsed:0.2f} seconds." + '\033[0m')

输出：

BrightSmile Toothpaste CompanyBrightSmile Toothpaste Co.BrightSmile ToothpasteGleaming Smile Inc.SparkleSmile ToothpasteConcurrent 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官方网站：/

[2] LangChain 🦜️🔗 中文网，跟着LangChain一起学LLM/GPT开发：/

[3] LangChain中文网 - LangChain 是一个用于开发由语言模型驱动的应用程序的框架：/