当前位置：首页 > news >正文

langchain链中的高级组件

news 2025/9/23 11:19:35

1.RunnableLambda定义链:将自己写的函数加入链，该函数只支持一个传递一个参数（键值对传参）

（1）注意传入链值的类型
<class 'langchain_core.runnables.base.RunnableSequence'>
<class 'int'>
（2）#itemgetter("foo")：用于从输入的字典中提取键为 "foo" 的值，转为runnable对象

（3）如果是其它静态值，需要先转换为Runnable对象

# 创建返回静态值的 Runnable
static_3 = RunnableLambda(lambda x: 3)
static_5 = RunnableLambda(lambda x: 5)

from operator import itemgetterfrom langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import chain, RunnableLambdaimport utils
import langchain
#开启该参数，会输出调试信息
# langchain.debug = True
#获得访问大模型客户端
client = utils.get_client()#定义提示模版工具
chat_template = ChatPromptTemplate.from_template(" {a} + {b}是多少？")
# chat_template = ChatPromptTemplate.from_template(" {a}是多少？")
#获得字符串的长度
def length_function(text):return len(text)
#将两个字符串长度的数量相乘
def _multiple_length_function(text1, text2):return len(text1) * len(text2)#任何可以转为Lambda的函数，它接受一个参数
#装饰器 @chain是RunnableLambda另外一种写法
@chain
def multiple_length_function(_dict):return _multiple_length_function(_dict["text1"], _dict["text2"])#使用RunnableLambda将函数转换为与LCEL兼容的组件
#itemgetter("foo")：用于从输入的字典中提取键为 "foo" 的值。chain2 = ({"a": itemgetter("foo") | RunnableLambda(length_function),"b": ( {"text1": itemgetter("foo"), "text2": itemgetter("bar")}| multiple_length_function )}| chat_template | client)print(chain2.invoke({"foo": "abc", "bar": "abcd"}))from operator import itemgetterfrom langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import chain, RunnableLambdaimport utils
import langchain
#开启该参数，会输出调试信息
langchain.debug = True
#获得访问大模型客户端
client = utils.get_client()#定义提示模版工具
chat_template = ChatPromptTemplate.from_template("{a}+{b}的值是多少")#定义函数
@chain
def label(x):return x+1# 转为runnablelambda表达式
chain = {"a":(RunnableLambda(lambda x: 3)| label ), "b":(RunnableLambda(lambda x: 5)| label ) }| chat_template | client
print(chain.invoke("计算"))

2.RunnableSequence串行链

里面是一个runnable对象传递

def mul_three(x: int) -> int:
return x * 3
runnable_3 = RunnableLambda(mul_three)#chain = prompt | client | output_parser 和下面的等价
chain = RunnableSequence(chat_template,client,parser)

3.RunableParallelb并行链

里面是一个runnable对象传递

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnableParallel
import utilsdef add_one(x: int) -> int:return x + 1def mul_two(x: int) -> int:return x * 2def mul_three(x: int) -> int:return x * 3runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
runnable_3 = RunnableLambda(mul_three)client = utils.get_client()# 创建提示模板
pmt1 = ChatPromptTemplate.from_template("{mul_two}+1是多少")
pmt2 = ChatPromptTemplate.from_template("{mul_three}+1是多少")# 正确的链式结构
chain2 = (runnable_1 |RunnableParallel(mul_two=runnable_2, mul_three=runnable_3) |RunnableParallel(result1=(lambda z: pmt1.invoke({"mul_two": z["mul_two"]})) | client,result2=(lambda z: pmt2.invoke({"mul_three": z["mul_three"]})) | client)
)print(chain2.invoke(1))

4.RunablePassthrough数据增强链

1）用户输入什么，就传递什么，
2）通过assign对数据进行增强，在往后传，原始值和传递后的值

#RunnablePassthrough的使用
#RunnablePassthrough的两种用法都将在我们后面的课程中看到
from langchain_core.runnables import RunnableParallel, RunnablePassthrough# RunnablePassthrough对数据增强后传递
runnable = RunnableParallel(passed = RunnablePassthrough().assign(query=lambda x: x["num"] + 2),modified = lambda x: x["num"] + 1,
)
print(runnable.invoke({"num": 1}))#结果
{'passed': {'num': 1, 'query': 3}, 'modified': 2}

5.ChatMessageHistory：存储历史消息，聊天记录本

单纯存储所有聊天记录的数据库。
没有智能。它不知道如何利用这些历史记录，也不会主动把历史记录发送给AI模型。它只是被动地存储数据。

chat_history = ChatMessageHistory()

（1）存储历史消息

chat_template = ChatPromptTemplate.from_messages(
[
SystemMessagePromptTemplate.from_template("你是人工智能助手"),
("placeholder","{messages}")
]
)

（2）添加消息

chat_history.add_user_message

chat_history.add_ai_message

（3）取出历史消息

chat_history.messages

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate
from langchain_community.chat_message_histories import ChatMessageHistory
import utils#想让大模型知道我们聊天中每次的会话内容，提示词中有专门的地方存放我们的聊天的历史
chat_template = ChatPromptTemplate.from_messages([SystemMessagePromptTemplate.from_template("你是人工智能助手"),("placeholder","{messages}")]
)client = utils.get_client()
parser = StrOutputParser()
chain = chat_template | client | parser#引入消息历史组件
chat_history = ChatMessageHistory()
chat_history.add_user_message("你好，我是云帆，我爱吃苹果")
res=chain.invoke({"messages":chat_history.messages})
print(res)
chat_history.add_ai_message(res)
print("chat_history:",chat_history.messages)chat_history.add_user_message("你好，我是谁？我爱吃什么") #把用户的提问加入消息历史
print(chain.invoke({"messages": chat_history.messages}))