01. configurable_alternatives 方法与使用技巧

在 LLMOps 项目中,应用编排页面可以在调试的过程中替换大语言模型继续之前的对话进行调试,这就是需要运行时组件替换功能,例如在构建的链应用中,动态替换掉特定的模型、提示词等整个组件本身,而不是替换组件里的参数信息

在 LangChain 中,提供了一个叫 configurable_alternatives() 方法来实现这个功能,所有的 Runnable 组件均支持这个函数。

使用 动态替换 的流程也存在几个步骤:

  1. 为有可能替换的组件定义一个键,这样在链应用中就可以区分出来是哪一个组件;
  2. 接下来为当前的组件选项设置一个默认值,当没有传递任何配置信息时,使用的就是默认值,组件不会发生替换;
  3. 接下来创建所有替换组件的实例,并为备选方案添加上对应的 key 值,以便配置信息知道值和对应组件的关系;
  4. 调用链,并传递配置信息,执行对应的动态替换运行组件。

例如,构建一条链,可以同时选择 gpt-4o、gpt-3.5-turbo-16k、文心一言 等模型,代码如下

import dotenv

from langchain_community.chat_models.baidu_qianfan_endpoint import QianfanChatEndpoint

from langchain_core.output_parsers import StrOutputParser

from langchain_core.prompts import ChatPromptTemplate

from langchain_core.runnables import ConfigurableField

from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

# 1.创建提示模板&定义默认大语言模型

prompt = ChatPromptTemplate.from_template("{query}")

llm = ChatOpenAI(model="gpt-3.5-turbo-16k").configurable_alternatives(

    ConfigurableField(id="llm"),

    gpt4=ChatOpenAI(model="gpt-4o"),

    wenxin=QianfanChatEndpoint(),

)

# 2.构建链应用

chain = prompt | llm | StrOutputParser()

# 3.调用链并传递配置信息

content = chain.invoke(

    {"query": "你好,你是什么模型呢?"},

    # config={"configurable": {"llm": "gpt4"}},

    config={"configurable": {"llm": "wenxin"}}

)

print(content)

输出如下:

您好,我是百度研发的知识增强大语言模型,中文名是文心一言,英文名是ERNIE Bot。我能够与人对话互动,回答问题,协助创作,高效便捷地帮助人们获取信息、知识和灵感。如果您有任何问题,请随时告诉我。

02. configurable_alternatives 运行流程与解析

configurable_alternatives() 的运行流程非常简单,在底层会通过一个字典 alternatives 存储所有替换组件,并且从传递的 configurable 字典中获取当前需要选择的组件 key,根据对应的 key 在 alternatives 中找到对应的组件返回并执行后续的操作。

运行流程图如下

核心代码:

# langchain_core/runnables/configurable.py => RunnableConfigurableAlternatives

def _prepare(

    self, config: Optional[RunnableConfig] = None

) -> Tuple[Runnable[Input, Output], RunnableConfig]:

    config = ensure_config(config)

    which = config.get("configurable", {}).get(self.which.id, self.default_key)

    # remap configurable keys for the chosen alternative

    if self.prefix_keys:

        config = cast(

            RunnableConfig,

            {

                **config,

                "configurable": {

                    _strremoveprefix(k, f"{self.which.id}=={which}/"): v

                    for k, v in config.get("configurable", {}).items()

                },

            },

        )

    # return the chosen alternative

    if which == self.default_key:

        return (self.default, config)

    elif which in self.alternatives:

        alt = self.alternatives[which]

        if isinstance(alt, Runnable):

            return (alt, config)

        else:

            return (alt(), config)

    else:

        raise ValueError(f"Unknown alternative: {which}")

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐