【橘子大模型】Runnable和Chain以及串行和并行

一、Runnable

前面我们实现了一些关于如何和大模型进行交互的操作。那么我们此时来回顾一下我们当前进行的结构。

我们已经很清楚这些操作的具体含义了，所以我这里就不在多介绍了。我们来看其中的几个点
1、用户那边就是客户，没啥说的。
2、langchain，这个是中间的框架。
3、template，关键就在这里，我们大部分业务编码其实都在这里到目前为止。
我们使用各种template来处理一些业务。

promptTemplate = ChatPromptTemplate([("system", "假如你是一个研究{cityMajor}的专家。"),("human", "介绍一下当前的经济环境。"),
])prompt = promptTemplate.invoke({"cityMajor": "山西经济"})
content = llm.invoke(prompt).content
print(f"大模型回答的结果正文是{content}")

我们构建起来一个模板之后，关键的一步就是执行invoke操作，包括llm也是执行invoke操作。而且除了invoke我们还可以执行stream，batch等操作。
我们先来看一下模板的文档，我们随便找一个，比如说BasePromptTemplate，他作为基类，文档比较全，下面那些子类忽略了一些文档，可能是他们不想重复写吧。

我们看到他说模板实现了标准的Runnable接口，该接口具有一系列方法，我们来看一下这个Runnable接口。

许多组件基本都基于这个抽象，我们来看一下这个模板类的文档

class BasePromptTemplate(RunnableSerializable[dict, PromptValue], Generic[FormatOutputType], ABC
):

我们看到他是实现了RunnableSerializable这个接口的。而RunnableSerializable的文档如下介绍。

RunnableSerializable implements the standard Runnable Interface. 🏃
他是实现了标准的Runnable接口的，所以这些组件实现了RunnableSerializable基本就是实现了Runnable接口

而Runnable接口是有这些方法的，所以其实我们可以大胆的下一个结论，你能看到的能调这几个方法的基本都是Runnable的实现类。

所以我们再来看我们的代码：

llm = ChatOllama(base_url = "http://127.0.0.1:11434",model = "deepseek-r1:8b",temperature = 0.5,num_predict = -2,max_tokens = 250
)promptTemplate = ChatPromptTemplate([("system", "假如你是一个研究{cityMajor}的专家。"),("human", "介绍一下当前的经济环境。"),
])prompt = promptTemplate.invoke({"cityMajor": "山西经济"})
content = llm.invoke(prompt).content
print(f"大模型回答的结果正文是{content}")

这里面promptTemplate.invoke， llm.invoke 其实promptTemplate和llm都是Runnable了。你去看他的类就知道，他们都是实现了RunnableSerializable的。所以我们的结构图中这部分就是Runnable。

ok，我们到这里为止，引入了一个概念叫做Runnable。那我们接下来看看这个Runnable

Runnable概念官方文档位于Runnable
我就不一个字一个字的去翻译这个文档了，这是翻译工具的活。我挑其中一段来说一下。

Runnable 方式定义了一个标准接口，允许 Runnable 组件：Invoked，Batched，Streamed等等操作。这些我们也都说过了。
但是他还有一个比较不错的功能，就是组合。
组合：可以使用 LangChain 表达式语言 （LCEL） 组合多个 Runnable 以协同工作，以创建复杂的管道。
这就是我们今天要说的第二个概念，LCEL组合Runnable

1、LCEL组合Runnable

他的功能很强大，而且非常多，具体可以去看文档。我们这里来做个分析操作来改造一下我们前面的代码。
我们说每一个组件现在都可以被我们认为成一个Runnable。其实就是一个任务，作为javaer可以理解为一个方法。
作为方法，必然就有输入(构建参数)，输出(返回值)。那么我们多个Runnable也就是多个方法的时候，方法1的输出就可以作为方法2的输入。这种就可以形成一个链路。而LCEL就是做这个的，他可以整合我们的Runnable形成一个链路，然后顺着链路去调用。
我们先看一个没有LCEL的我们咋做的。

1.1常规做法

from langchain_core.prompts import ChatPromptTemplatepromptTemplate = ChatPromptTemplate([("system", "假如你是一个研究{cityMajor}的专家。"),("human", "介绍一下当前的经济环境。"),
])prompt = promptTemplate.invoke({"cityMajor": "山西经济"})
content = llm.invoke(prompt).content
print(f"大模型回答的结果正文是{content}")

常规做法我们是以前写烂了。那我们再来看看使用LCEL咋做。

1.2、LCEL

from langchain_core.prompts import ChatPromptTemplate
# llm是Runnable1# promptTemplate 是Runnable2
promptTemplate = ChatPromptTemplate([("system", "假如你是一个研究{cityMajor}的专家。"),("human", "介绍一下当前的经济环境。"),
])# LCEL使用管道符号串联两个Runnable
chain = promptTemplate | llm
content = chain.invoke({"cityMajor":"山西经济"})
print(f"大模型回答的结果正文是{content}")

我们看到我们用LCEL使用管道符号串联两个Runnable也就是promptTemplate | llm
这个操作其实就是等于promptTemplate执行的结果传递给llm执行，也就是我们之前的。

prompt = promptTemplate.invoke({"cityMajor": "山西经济"})
content = llm.invoke(prompt)

我想你大概知道这个操作的含义了，下面我们再串联一个组件进去玩一玩。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field, model_validator
from typing import List# 构建一个类，表示单个观点结构
class Opinion(BaseModel):num: str = Field(description="观点的标号")opinion: str = Field(description="回答的观点")# 构建一个集合类，表示多个观点结构
class OpinionsList(BaseModel):opinions: List[Opinion] = Field(description="包含多个观点的列表")# Runnable1
promptTemplate = ChatPromptTemplate([("system", "假如你是一个研究{cityMajor}的专家。"),("human", "介绍一下当前的经济环境，不要思考过程。请以JSON格式输出多条观点，注意一定是多条观点，而不是一个，格式如下：{{'opinions': [{{'num': '1', 'opinion': '观点1'}}, {{'num': '2', 'opinion': '观点2'}}]}}"),
])
# llm是Runnable2
# Runnable3
parser = JsonOutputParser(pydantic_object=OpinionsList)
# 串联起来，等于先运行一个promptTemplate得到prompt,然后输出给llm执行得到结果。把结果最后给JsonOutputParser按照结构解析成json输出
chain = promptTemplate | llm | parser
try:content = chain.invoke({"cityMajor": "山西经济"})print(f"大模型回答的结果正文是{content}")
except Exception as e:print(f"解析失败：{e}")

最后就得到的结果就是json结构的。

{'opinions': [{'num': '1', 'opinion': '山西近年来经济保持稳定增长，传统制造业和高新技术产业是主要驱动力。'}, {'num': '2', 'opinion': '山西吸引外资和优化营商环境取得进展，为企业发展提供了支持。'}, {'num': '3', 'opinion': '山西在纺织、食品加工等传统行业保持竞争力，同时高新技术产业快速崛起。'}, {'num': '4', 'opinion': '面对内地市场竞争和国际环境变化，山西需加强产品质量和创新能力。'}, {'num': '5', 'opinion': '山西应加强区域协作，与周边省份合作，以推动经济发展。'}]}

还有很多高级操作，可以按照文档来学习，后面我们争取多用用。

我们现在完成了多个Runnable的链路，得到的是一个我们称之为chain(链)的东西。我们看一下这个chain是啥类型。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field, model_validator
from typing import List# 单个观点结构
class Opinion(BaseModel):num: str = Field(description="观点的标号")opinion: str = Field(description="回答的观点")# 多个观点结构
class OpinionsList(BaseModel):opinions: List[Opinion] = Field(description="包含多个观点的列表")promptTemplate = ChatPromptTemplate([("system", "假如你是一个研究{cityMajor}的专家。"),("human", "介绍一下当前的经济环境，不要思考过程。请以JSON格式输出多条观点，注意一定是多条观点，而不是一个，格式如下：{{'opinions': [{{'num': '1', 'opinion': '观点1'}}, {{'num': '2', 'opinion': '观点2'}}]}}"),
])parser = JsonOutputParser(pydantic_object=OpinionsList)chain = promptTemplate | llm | parser
# python中可以用type函数来输出类型
type(chain)

结果为langchain_core.runnables.base.RunnableSequence
他的构造函数为:

class RunnableSequence(RunnableSerializable[Input, Output]):

我们看到他也实现了RunnableSerializable，所以多个Runnable链路运行的结果其实还是一个Runnable。那么既然是一个Runnable，是不是我们还可以继续组合。我们可以把一个chain当成一个完整的Runnable继续串。
也就是这样的结构:

我们的两个chain甚至可以是完全不一样的，背后的大模型都可以不一样，这样就可以任意组合了。比如有些模型比较擅长推理，有些适合做文字生成。那我们就可以让文字生成的生成一个好的描述，交给推理的模型。最后给我们一个好的结果。good。
why not，接下来我们就来看如何操作chain的组合。

2、LCEL组合chain

我们来构建一个需求:

1、我们需要两个chain
2、第一个分析山西经济学家视角下的经济环境，得到经济环境的描述。
3、第二个chain，拿到山西经济环境学家的分析结果，把结果中每一点的标题提取出来。这样就结束。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from pydantic import BaseModel, Field, model_validator
from typing import List# 1、构建第一个chain的模板，根据这个模板让大模型为我们输出一个他作为cityMajor的专家的一个当前经济环境的结果。
getRespPromptTemplate = ChatPromptTemplate([("system", "假如你是一个研究{cityMajor}的专家。"),("human", "介绍一下当前的经济环境，不要思考过程。"),
])# 2、构建第一个chain，先获取模板，发给llm，最后结果通过一个字符串的parser解析输出为字符串
getRespChain =  getRespPromptTemplate | llm | StrOutputParser()# 3、构建第二个chain的模板，这个模板有一个参数，就是上一个chain的输出作为输入，分析其中的内容，我们只要结果的标题小点的总结
getHeadingPromptTemplate = ChatPromptTemplate.from_template("""分析{resp}，并且仅仅获取其中的每一点的标题,注意只获取每一点的标题,不要后面的描述，我只要标题点，并且不要思考过程""")
"""
4、构建第二个chain，第二个chain首先我们要把第一个chain的结果当成第一步传递给getHeadingPromptTemplate
因为getHeadingPromptTemplate需要第一个的结果去填充{resp},所以我们构建一个{}的map结构，得到第一个chain的输出给getHeadingPromptTemplate
getHeadingPromptTemplate填充完了发给llm
最后以字符串输出解析器解析为字符串。
"""
getHeadingChain = {"resp":getRespChain} | getHeadingPromptTemplate | llm | StrOutputParser()# 执行最后的chain，然后把第一个chain的参数可以传进去，此时他们是一个chain了，可以这么传
resp = getHeadingChain.invoke({"cityMajor":"山西经济"} )
print(resp)

输出结果如下:

1. 山西经济环境呈现多元化发展态势  
2. 产业结构不断优化，制造业、科技服务业和绿色经济成为主要驱动力  
3. 新兴产业如电子信息、生物医药等快速发展  
4. 传统优势产业在升级改造后依然占据重要地位  
5. 政策支持力度大，吸引了大量外资和内资  
6. 基础设施建设不断完善，为经济发展提供了坚实支撑  
7. 面临国内市场机遇与国际环境挑战，山西正积极推进数字经济和绿色低碳发展  
8. 展现出较强的经济韧性和活力

我在这个代码中踩了一个坑，如下记录:

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from pydantic import BaseModel, Field, model_validator
from typing import ListgetRespPromptTemplate = ChatPromptTemplate([("system", "假如你是一个研究{cityMajor}的专家。"),("human", "介绍一下当前的经济环境，不要思考过程。"),
])
getRespChain = {"cityMajor":"山西经济"} | getRespPromptTemplate | llm | StrOutputParser()
getHeadingPromptTemplate = ChatPromptTemplate.from_template("""分析{resp}，并且仅仅获取其中的每一点的总结的标题,注意只获取每一点的总结的标题，并且不要思考过程""")
getHeadingChain = {"resp":getRespChain} | getHeadingPromptTemplate | llm | StrOutputParser()
resp = getHeadingChain.invoke()
print(resp)

其实改动也不大，我只是在第一个chain的最前面模仿第二个chain把参数放前面了，这样是不对的，你要严格遵守一个原则，就是这种链路的每一个元素都是一个Runnable才行。getHeadingChain = {“resp”:getRespChain} | getHeadingPromptTemplate | llm | StrOutputParser()之所以可以是因为第一个{“resp”:getRespChain} 他是一个runnable结果，因为有getRespChain的存在。
而我把{“cityMajor”:“山西经济”}放前面这个只是一个map结构，啥也不是，所以不对。

ok，至此我们就完成了chain的串联。接下来我们再来看看如何让他们并行联动。

3、RunnableParallel

我们有时候有一些需求是需要同时执行然后一起返回的。这时候需要我们并行来运行多个Runnable或者Chain。这样可以提高效率。
但是注意这种需求不要运行那种有依赖的Chain，这样虽然是并行的操作，但是因为依赖，还是会先执行一个再根据输出执行另一个。其实还是串行执行。
我们这里来搞一个需求。

我们的ollama中有两个大模型，一个是deepseek-r1:8b，一个llama3.2:latest。我们构造两个chain，然后同时各自发去一个模型中，并行执行最后输出结果。
这个就需要用到RunnableParallel

此时我们需要构建两个llm，我们就使用另一个模型llama3.2:latest。而且ollama其实不用run模型，直接就能调。

from langchain_ollama import ChatOllamallm = ChatOllama(base_url = "http://127.0.0.1:11434",model = "deepseek-r1:8b",temperature = 0.5,num_predict = 10000,max_tokens = 250
)llmllama32 = ChatOllama(base_url = "http://127.0.0.1:11434",model = "llama3.2:latest",temperature = 0.5,num_predict = 10000,max_tokens = 250
)

然后我们来编写并行调用llm的操作。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel# 1、构建第一个chain的模板，去调用deepseek模型
dsPromptTemplate = ChatPromptTemplate([("system", "假如你是一个研究{dsCityMajor}的专家。"),("human", "介绍一下当前的经济环境，不要思考过程。"),
])# 2、构建第一个chain，先获取模板，发给deepseek llm，最后结果通过一个字符串的parser解析输出为字符串
dsRespChain =  dsPromptTemplate | llm | StrOutputParser()# 3、构建第二个chain的模板，去调用llama模型
llamaPromptTemplate = ChatPromptTemplate([("system", "假如你是一个研究{llamaCityMajor}的专家。"),("human", "介绍一下当前的经济环境，不要思考过程。"),
])# 4、构建第二个chain，先获取模板，发给llama llm，最后结果通过一个字符串的parser解析输出为字符串
llamaRespChain = llamaPromptTemplate | llmllama32 | StrOutputParser()# 5、构建并行执行的Runnable 并行运行两个没有依赖关系的chain
parallelRunnable = RunnableParallel(dsChain=dsRespChain, llamaChain=llamaRespChain)# 6、执行Runnable
resp = parallelRunnable.invoke({"dsCityMajor" : "山西经济","llamaCityMajor" :  "北京经济"})
print("deepseek:")
print(resp['dsChain'])print("llama:")
print(resp['llamaChain'])

输出结果如下:

deepseek:
当前，山西省经济正处于转型升级阶段，面临着传统产业转型与新兴产业发展双重任务。山西作为内陆开放枢纽，通过“双循环”新发展格局，推动区域经济一体化，在高科技、数字经济、新能源等领域取得了一定进展。同时，绿色低碳发展成为新时代的重要方向，山西省在可再生能源和环保治理方面也在积极作为。此外，山西还通过优化营商环境，加强基础设施建设，促进就业市场向高质量发展转型，为经济高质量发展奠定了坚实基础。llama:
另一方面，北京面临着一些挑战。例如，人口流动和城市化问题，导致了交通拥挤、住房短缺等问题。同时，环境保护和空气污染问题也在获得关注。
总体而言，北京的经济环境虽然有其挑战，但仍处于相对稳定的状态下，并且具有很大的发展潜力。

我们先不管输出质量，输出结构是符合我们预期的。此时我们再去langsmith中看一下。我们发现并行的已经监控到了。

我们看一下详细信息。基本是和我们的预期一样的。说明我们的并行请求成功被执行了。

ok，至此我们的串行和并行就都完成了。