前序
之前使用学习chains的时候，有一个地方，就是很多例子还是使用的LLMChain，但是这个已经是要被遗弃的，改成使用prompt｜llm 这样的格式，然后一直报错RunnableSequence没有input_keys。这就引入了新的概念RunnableSequence以及其它内容。就开始了解一下
langchain_core.runnables
点进源码，可以看到一堆，简单了解几个
也可以参考其他大佬的文档
Runnable的定义和使用
官网文档
Runnable-可运行对象
在前面的学习中，我们曾经学习过LangChain的核心是Chains（链），即使用声明式方法|将多个操作链接在一起进行调用。之前所说的操作实际上就是Runnable对象包含的操作。
Runnable，表示可运行对象，可以理解为“链式可运行对象”或“链对象”，也就是langchain中所谓的“Chain”。
多个Chain（链）使用|组合操作，就是一个chains链条。
LangChain为了让其内部提供的所有组件类都能以LCEL的方式快速简洁的调用，计划将所有组件类都实现Runnable接口。通常情况下，LangChain中的链包含: 大模型(LLMs)、提示词模版(Prompt Template)、工具(Tools)和输出解析器(Output Parsers)。这几部分都继承并实现了Runnable接口，所以他们都是Runnable的实例。
from langchain_core.runnables import Runnable, RunnableSequence
from langchain.chains.llm import LLMChain
from langchain_core.prompts import PromptTemplate
from langchain_deepseek import ChatDeepSeek
import osos.environ['DEEPSEEK_API_KEY'] = "sk-4bxxx"prompt = PromptTemplate.from_template("Tell me a joke about {topic}")llm = ChatDeepSeek(model="deepseek-chat")chain1 = LLMChain(prompt=prompt, llm=llm)print(isinstance(llm, Runnable))
print(isinstance(prompt, Runnable))
print(isinstance(chain1, Runnable))
print(isinstance(StrOutputParser(),Runnable))"""
True
True
True
True
"""
RunnableLambda和RunnableGenerator
RunnableLambda和RunnableGenerator这两个类通常用来自定义Runnable。这两者的主要区别在于：
RunnableLambda:是将Python中的可调用对象包装成Runnable, 这种类型的Runnable无法处理流式数据。
RunnableGenerator:将Python中的生成器包装成Runnable,可以处理流式数据。
RunnableLambda
RunnableLambda可以将Python 可调用对象(例如：回调函数)转换为 Runnable对象，这种转换使得任何函数都可以被看作 LCEL 链的一部分，有了它，我们就可以把自己需要的功能通过自定义函数 + RunnableLambda的方式包装一下成 一个Runnable对象集成到 LCEL 链中，完成一些额外的业务操作流程，但是注意这种类型的Runnable无法处理流式数据。
比如之前在chains中提到的LLMChain改成prmpt｜llm 格式后，无法输出中间内容
import os
import langchain
from langchain_deepseek import ChatDeepSeek
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParseros.environ['DEEPSEEK_API_KEY'] = "sk-4b3xxx"# 在全局范围开启详细模式，能将调用大模型时发送的数据打印到控制台，绿色文本
langchain.verbose = Truellm = ChatDeepSeek(model="deepseek-chat", temperature=0.5)# Chain1 获得内容
prompt1 = ChatPromptTemplate.from_template("请简单介绍一下历史人物: {username}"
)chain1 = prompt1 | llm | StrOutputParser()# Chain2 根据内容，简明扼要
prompt2 = ChatPromptTemplate.from_template("基于给出的人物信息, 简明扼要，整理出不多于40个字符内容: {introduce}"
)chain2 = prompt2 | llm | StrOutputParser()# Chain3 把简介翻译成英文
prompt3 = ChatPromptTemplate.from_template("把给出的内容翻译为英文?: {brief_introduce}"
)chain3 = prompt3 | llm | StrOutputParser()def print_info(input):print(f"introduce：{input}", type(input))return inputdef print_info2(input):print(f"brief_introduce：{input}")return inputdef print_info3(input):print(f"english_brief_introduce：{input}")return input# 因为Runnable内部初始化时会进行自动判断，所以不写Runnaable也默认完成了格式转换
seq_chain = chain1 | RunnableLambda(print_info) | chain2 | print_info2 | chain3 | print_info3 | StrOutputParser()res = seq_chain.invoke({"username": "秦始皇"})
print("最终结果： ", res)运行结果：
这样就获得了运行中的输出结果
RunnableGenerator
llm = ChatDeepSeek(model="deepseek-chat")from langchain_core.prompts import PromptTemplate
prompt=PromptTemplate.from_template("输出《{book}》中{num}个角色名字以及这些角色的背景故事。")from langchain_core.runnables import RunnableGenerator
def stream_add(x):yield f"分别有以下结果：\n"for i in x:yield i.contentstream_chain = prompt | llm | RunnableGenerator(stream_add)
for chunk in stream_chain.stream({"book": "红楼梦", "num": 5}):print(chunk, end='')
RunnableSequence
按顺序调用一系列可运行文件，其中一个Runnable的输出作为下一个的输入。一般通过使用|运算符或可运行项列表来构造
chain2 = prompt | llm
print(isinstance(chain2, RunnableSequence))"""
True
"""
RunnableParallel
RunnableParallel用于并行执行多个Runnable，使用RunnableParallel可以实现部分Runnable或所有Runnable并发执行的业务流程，提高执行效率，也会额外消耗系统性能。
import os
import langchain
from langchain_deepseek import ChatDeepSeek
os.environ['DEEPSEEK_API_KEY'] = "sk-4bxxx"llm = ChatDeepSeek(model="deepseek-chat", temperature=0.5)from langchain_core.prompts import ChatPromptTemplate
prompt_1 = ChatPromptTemplate.from_messages([("system", "{get_format_instructions}"),("human", "列出{city}的{num}个著名景点。"),]
)
prompt_2 = ChatPromptTemplate.from_messages([("system", "{get_format_instructions}"),("human", "列出{city}的{num}个美食。"),]
)# 列表输出解释器
from langchain_core.output_parsers import CommaSeparatedListOutputParser
output_parser = CommaSeparatedListOutputParser()chain_1 = prompt_1 | llm | output_parser
chain_2 = prompt_2 | llm | output_parser# 并行链
from langchain_core.runnables import RunnableParallel
chain_parallel = RunnableParallel(view_spots=chain_1, related_video=chain_2)response = chain_parallel.invoke({"city": "成都", "num": 5, "get_format_instructions": output_parser.get_format_instructions()}
)
print(response)"""
{'view_spots': ['武侯祠', '锦里古街', '宽窄巷子', '杜甫草堂', '青羊宫'],
'related_video': ['火锅', '担担面', '钟水饺', '龙抄手', '夫妻肺片']}
"""
RunnablePassthrough
RunnablePassthrough 主要用在链条中传递数据，可用于格式化上一个子链的输出结果，从而匹配下一个输入子链的需要的格式，一般写用在链的第一个位置，用于接收用户的输入。如果处在中间位置，则用于接收上一步子链的输出，通常与RunnableParallel配合使用。
在之前Chain的那个例子，也用过，如下：
runnable = RunnablePassthrough.assign(username=lambda inputs: inputs['username']) | chain1 | chain2 | chain3
res = runnable.invoke({'username': '秦始皇'})
也可以写在后面，把前面得到的结果相加
# coding=utf-8
import os
import langchain
from langchain_deepseek import ChatDeepSeekos.environ['DEEPSEEK_API_KEY'] = "sk-4xxx"langchain.verbose = True
llm = ChatDeepSeek(model="deepseek-chat", temperature=0.5)from langchain_core.output_parsers import CommaSeparatedListOutputParser
from langchain_core.prompts import ChatPromptTemplateoutput_parser = CommaSeparatedListOutputParser()
prompt_1 = ChatPromptTemplate.from_messages([("system", "{get_format_instructions}"),("human", "列出{city}的{num}个著名景点。"),]
)prompt_2 = ChatPromptTemplate.from_messages([("system", "{get_format_instructions}"),("human", "列出{city}的{num}个相关美食。"),]
)from langchain_core.runnables import RunnablePassthrough, RunnableParallel# 写在开头
# runnable = RunnablePassthrough.assign(num=lambda inputs: inputs['num'] + 1) | RunnableParallel(**{
# 'llm1': prompt_1 | llm | output_parser,
# 'llm2': prompt_2 | llm | output_parser,
# })# 上面的RunnableParallel的调用方法可以简化成dict字典格式，在langchain内部会自动切换花括号为RunnableParallel对象，所以下面的简写方式：
# runnable = RunnablePassthrough.assign(num=lambda inputs: inputs['num'] + 1) | {
# 'llm1': prompt_1 | llm | output_parser,
# 'llm2': prompt_2 | llm | output_parser,
# }# # 写在结尾
# runnable = RunnableParallel(**{
# 'llm1': prompt_1 | llm | output_parser,
# 'llm2': prompt_2 | llm | output_parser,
# }) | RunnablePassthrough.assign(results=lambda inputs: inputs['llm1'] + inputs['llm2']
# )# # 简写如下：
runnable = {'llm1': prompt_1 | llm | output_parser,'llm2': prompt_2 | llm | output_parser,
} | RunnablePassthrough.assign(results=lambda inputs: inputs['llm1'] + inputs['llm2']
)response = runnable.invoke({"city": "成都", "num": 3, "get_format_instructions": output_parser.get_format_instructions()}
)
print(response)
