贵州小城镇建设网站,金融代理平台代理加盟,附近广告设计与制作门店电话,深圳市信用网分类目录#xff1a;《自然语言处理从入门到应用》总目录 创建自定义Chain
要实现自己的自定义链式连接#xff0c;我们可以子类化Chain并实现以下方法#xff1a;
from __future__ import annotations
from typing import Any, Dict, List, Optional
from pydantic impor…分类目录《自然语言处理从入门到应用》总目录 创建自定义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 BasePromptTemplateclass MyCustomChain(Chain):An example of a custom chain.prompt: BasePromptTemplatePrompt object to use.llm: BaseLanguageModeloutput_key: str text #: :meta private:class Config:Configuration for this pydantic object.extra Extra.forbidarbitrary_types_allowed Truepropertydef input_keys(self) - List[str]:Will be whatever keys the prompt expects.:meta private:return self.prompt.input_variablespropertydef 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],callbacksrun_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],callbacksrun_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_chainfrom langchain.callbacks.stdout import StdOutCallbackHandler
from langchain.chat_models.openai import ChatOpenAI
from langchain.prompts.prompt import PromptTemplatechain MyCustomChain(promptPromptTemplate.from_template(tell us a joke about {topic}),llmChatOpenAI()
)chain.run({topic: callbacks}, callbacks[StdOutCallbackHandler()])日志输出 Entering new MyCustomChain chain...
Log something about this runFinished 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 timefrom langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChaindef generate_serially():llm OpenAI(temperature0.9)prompt PromptTemplate(input_variables[product],templateWhat is a good name for a company that makes {product}?,)chain LLMChain(llmllm, promptprompt)for _ in range(5):resp chain.run(producttoothpaste)print(resp)async def async_generate(chain):resp await chain.arun(producttoothpaste)print(resp)async def generate_concurrently():llm OpenAI(temperature0.9)prompt PromptTemplate(input_variables[product],templateWhat is a good name for a company that makes {product}?,)chain LLMChain(llmllm, promptprompt)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 fConcurrent executed in {elapsed:0.2f} seconds. \033[0m)s time.perf_counter()
generate_serially()
elapsed time.perf_counter() - s
print(\033[1m fSerial executed in {elapsed:0.2f} seconds. \033[0m)输出
BrightSmile Toothpaste CompanyBrightSmile Toothpaste Co.BrightSmile ToothpasteGleaming 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/
