在 LangChain 框架中,Runnable 组件是构建灵活、可配置的 AI 应用的核心。本文将深入探讨 Runnable 组件的高级特性,包括动态参数配置、组件替换、错误处理机制以及生命周期管理。通过掌握这些特性,开发者可以构建更加健壮和可维护的 AI 应用。
1. Runnable 组件动态添加默认调用参数
1.1 bind 函数的用途与使用技巧
在 LangChain 开发中,我们经常需要在 Runnable 可运行队列中调用另一个 Runnable,并传递一些常量参数。这些参数可能不是前一个 Runnable 的输出,也不是用户输入的一部分,而是某个 Runnable 组件的特定参数。这时,我们可以使用 Runnable.bind()
方法来传递这些默认参数。
例如,我们可以创建一个 ChatOpenAI 的 LLM 大语言模型,并用它构建两条链,分别设置不同的 temperature 值:
import dotenv from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI dotenv.load_dotenv() prompt = ChatPromptTemplate.from_messages([ ("system", "你正在执行一项测试,请重复用户传递的内容,除了重复其他均不要操作"), ("human", "{query}") ]) llm = ChatOpenAI(model="gpt-4o") chain = prompt | llm.bind(stop="world") | StrOutputParser() content = chain.invoke({"query": "Hello world"}) print(content)
输出:
Hello
bind()
函数用于修改 Runnable 底层的默认调用参数,并在调用时自动传递该参数,无需手动传递。这使得在构建 Runnable 链应用时可以更灵活地设置参数。
1.2 解决多参 RunnableLambda 函数传参
在 LangChain 中,如果要将一个函数变成 Runnable 组件,可以通过 RunnableLambda 函数进行包装。但是封装后,所有的 Runnable 组件的 invoke 函数只能传递一个参数。使用 bind()
函数可以巧妙地解决这个问题,实现 RunnableLambda 组件接收多个参数。
2. Runnable 组件配置运行时链内部
2.1 configurable_fields 方法使用技巧
configurable_fields()
方法允许在链运行时为链中的给定步骤指定参数,比 bind()
更灵活。使用这个方法可以在链运行时动态调整温度、停止词、传递自定义参数,甚至动态替换模型。
例如,在链的调用过程中将 temperature 温度设置为 0:
import dotenv 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() prompt = ChatPromptTemplate.from_template("请生成一个小于{x}的随机整数") llm = ChatOpenAI(model="gpt-3.5-turbo-16k").configurable_fields( temperature=ConfigurableField( id="llm_temperature", name="大语言模型温度", description="用于调整大语言模型生成内容的随机性" ), ) chain = prompt | llm | StrOutputParser() content = chain.invoke( {"x": 1000}, config={"configurable": {"llm_temperature": 0}} ) print(content)
3. Runnable 组件动态替换运行时组件
3.1 configurable_alternatives 方法与使用技巧
configurable_alternatives()
方法允许在运行时动态替换链中的特定组件,如模型或提示词。这在 LLMOps 项目中特别有用,可以在调试过程中替换大语言模型继续之前的对话。
例如,构建一条链,可以同时选择 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() 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(), ) chain = prompt | llm | StrOutputParser() content = chain.invoke( {"query": "你好,你是什么模型呢?"}, config={"configurable": {"llm": "wenxin"}} ) print(content)
4. Runnable 组件重试与回退机制降低程序错误率
4.1 Runnable 重试机制
LangChain 提供了 with_retry()
方法来实现 Runnable 组件的重试机制。当 Runnable 组件出现异常时,可以针对特定的异常或所有异常进行重试,并配置重试次数和时间间隔。
例如,让一个 Runnable 组件最多重试 2 次:
from langchain_core.runnables import RunnableLambda counter = -1 def func(x): global counter counter += 1 print(f"当前的值为 {counter=}") return x / counter chain = RunnableLambda(func).with_retry(stop_after_attempt=2) resp = chain.invoke(2) print(resp)
4.2 Runnable 回退机制
with_fallback()
方法提供了回退机制,当 Runnable 组件出错时,可以执行特定的备份/回退方案。例如,当 OpenAI 的 LLM 大模型出现异常时,自动切换到文心一言的模型:
import dotenv from langchain_community.chat_models import QianfanChatEndpoint from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI dotenv.load_dotenv() prompt = ChatPromptTemplate.from_template("{query}") llm = ChatOpenAI(model="gpt-3.5-turbo-18k").with_fallbacks([QianfanChatEndpoint()]) chain = prompt | llm | StrOutputParser() content = chain.invoke({"query": "你好,你是?"}) print(content)
5. Runnable 组件生命周期监听器与使用场景
5.1 Runnable 生命周期监听器
LangChain 提供了 with_listeners()
方法来监听 Runnable 组件的开始、结束和出错三个常见的生命周期事件。这种方法比 CallbackHandler 更简洁、更统一。
例如,为 RunnableLambda 添加生命周期监听器:
import time from langchain_core.runnables import RunnableLambda, RunnableConfig from langchain_core.tracers.schemas import Run def on_start(run_obj: Run, config: RunnableConfig) -> None: print("on_start") print("run_obj:", run_obj.inputs) print("config:", config) print("========================") def on_end(run_obj: Run, config: RunnableConfig) -> None: print("on_end") print("run_obj:", run_obj) print("config:", config) print("========================") def on_error(run_obj: Run, config: RunnableConfig) -> None: print("on_error") print("run_obj:", run_obj) print("config:", config) print("========================") runnable = RunnableLambda(lambda x: time.sleep(x)) chain = runnable.with_listeners(on_start=on_start, on_end=on_end, on_error=on_error) chain.invoke(2)
6. 基于 Runnable 封装记忆链实现记忆自动管理
在 Runnable 链应用中,可以通过 config+configurable 的形式传递 memory 给链。在链的执行函数中,可以通过第二个参数获取到 memory 实例,从而获取记忆历史。同时,可以为链添加 on_end 函数,在生命周期结束时将对话信息存储到记忆系统中。
这种方法allows为 Runnable 组件实现自动的记忆管理,提高了 AI 应用的上下文理解能力和连贯性。
结论:
通过深入理解和灵活运用 LangChain 的 Runnable 组件的这些高级特性,开发者可以构建出更加强大、灵活和可靠的 AI 应用。从动态参数配置到错误处理,再到生命周期管理,这些功能为开发复杂的 AI 系统提供了强大的工具和抽象。在实际应用中,合理利用这些特性可以大大提高开发效率和应用质量。
标签:core,Runnable,langchain,print,LangChain,组件,import,错误处理 From: https://www.cnblogs.com/muzinan110/p/18539601