LangChain(五)——Chanis

LangChain模块架构图 [1]

一、Chains

1.1 LLM链

将大语言模型(LLM)和提示（Prompt）组合成链。这个大语言模型链非常简单，可以让我们以一种顺序的方式去通过运行提示并且结合到大语言模型中。

1
2
3
4
5
6
7
8
9
10
11
12
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate  
from langchain.chains import LLMChain  

llm = ChatOpenAI(temperature=0.0)  

prompt = ChatPromptTemplate.from_template(   
    "描述制造{product}的一个公司的最佳名称是什么?"
)
chain = LLMChain(llm=llm, prompt=prompt)
product = "大号床单套装"
chain.invoke(product)

输出

1
{'product': '大号床单套装', 'text': '“豪华床品集团”'}

1.2 顺序链

顺序链（SequentialChains）是按预定义顺序执行其链接的链。

1.2.1 简单顺序链

使用简单顺序链（SimpleSequentialChain），这是顺序链的最简单类型，其中每个步骤都有一个输入/输出，一个步骤的输出是下一个步骤的输入。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from langchain.chains import SimpleSequentialChain

first_prompt = ChatPromptTemplate.from_template(   
    "描述制造{product}的一个公司的最好的名称是什么"
)
chain_one = LLMChain(llm=llm, prompt=first_prompt)

second_prompt = ChatPromptTemplate.from_template(   
    "写一个20字的描述对于下面这个\
    公司：{company_name}的"
)
chain_two = LLMChain(llm=llm, prompt=second_prompt)


overall_simple_chain = SimpleSequentialChain(chains=[chain_one, chain_two],verbose=False)
product = "大号床单套装"
overall_simple_chain.invoke(product)

输出

1
2
3
4
5
6
> Entering new SimpleSequentialChain chain...
“豪华床品集团”
提供高品质、舒适的床上用品，打造豪华睡眠体验，让您每晚都能享受极致舒适。

> Finished chain.
{'input': '大号床单套装', 'output': '提供高品质、舒适的床上用品，打造豪华睡眠体验，让您每晚都能享受极致舒适。'}

1.2.2 顺序链

当只有一个输入和一个输出时，简单顺序链（SimpleSequentialChain）即可实现。当有多个输入或多个输出时，我们则需要使用顺序链（SequentialChain）来实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
from langchain.chains import SequentialChain


llm = ChatOpenAI(temperature=0.9)

#子链1
# prompt模板 1: 翻译成英语（把下面的review翻译成英语）
first_prompt = ChatPromptTemplate.from_template(
    "把下面的评论review翻译成英文:"
    "\n\n{Review}"
)
# chain 1: 输入：Review    输出：英文的 Review
chain_one = LLMChain(llm=llm, prompt=first_prompt, output_key="English_Review")

#子链2
# prompt模板 2: 用一句话总结下面的 review
second_prompt = ChatPromptTemplate.from_template(
    "请你用一句话来总结下面的评论review:"
    "\n\n{English_Review}"
)
# chain 2: 输入：英文的Review   输出：总结
chain_two = LLMChain(llm=llm, prompt=second_prompt, output_key="summary")


#子链3
# prompt模板 3: 下面review使用的什么语言
third_prompt = ChatPromptTemplate.from_template(
    "下面的评论review使用的什么语言:\n\n{Review}"
)
# chain 3: 输入：Review  输出：语言
chain_three = LLMChain(llm=llm, prompt=third_prompt, output_key="language")


#子链4
# prompt模板 4: 使用特定的语言对下面的总结写一个后续回复
fourth_prompt = ChatPromptTemplate.from_template(
    "使用特定的语言对下面的总结写一个后续回复:"
    "\n\n总结: {summary}\n\n语言: {language}"
)
# chain 4: 输入： 总结, 语言    输出： 后续回复
chain_four = LLMChain(llm=llm, prompt=fourth_prompt,
                      output_key="followup_message"
                     )


# 对四个子链进行组合
#输入：review    输出：英文review，总结，后续回复 

overall_chain = SequentialChain(
    chains=[chain_one, chain_two, chain_three, chain_four],
    input_variables=["Review"],
    output_variables=["English_Review", "summary","language","followup_message"],
    verbose=False
)


review = "Je trouve le goût médiocre. La mousse ne tient pas, c'est bizarre. J'achète les mêmes dans le commerce et le goût est bien meilleur...\nVieux lot ou contrefaçon !?"
overall_chain.invoke(review)
1
2
3
4
5
{'Review': "Je trouve le goût médiocre. La mousse ne tient pas, c'est bizarre. J'achète les mêmes dans le commerce et le goût est bien meilleur...\nVieux lot ou contrefaçon !?",
 'English_Review': "I find the taste mediocre. The foam doesn't last, it's strange. I buy the same ones in stores and the taste is much better... Old batch or counterfeit!?",
 'summary': 'The reviewer is disappointed with the taste and quality of the product, suspecting it may be an old batch or counterfeit.',
 'language': "C'est en français.",
 'followup_message': "Merci pour vos commentaires. Nous sommes désolés d'apprendre que vous n'avez pas été satisfait de notre produit. Nous vous assurons que nous prenons la qualité de nos produits très au sérieux et nous allons enquêter sur cette question pour comprendre ce qui s'est passé. Nous vous remercions de nous en avoir informé et nous ferons tout notre possible pour rectifier la situation."}

1.2.3 路由链

一个相当常见但基本的操作是根据输入将其路由到一条链，具体取决于该输入到底是什么。如果你有多个子链，每个子链都专门用于特定类型的输入，那么可以组成一个路由链，它首先决定将它传递给哪个子链，然后将它传递给那个链。

路由器由两个组件组成：

  • 路由链（Router Chain）：路由器链本身，负责选择要调用的下一个链
  • destination_chains：路由器链可以路由到的链
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
from langchain.chains.router import MultiPromptChain  #导入多提示链
from langchain.chains.router.llm_router import LLMRouterChain,RouterOutputParser
from langchain.prompts import PromptTemplate

llm = ChatOpenAI(temperature=0)


#第一个提示适合回答物理问题
physics_template = """你是一个非常聪明的物理专家。 \
你擅长用一种简洁并且易于理解的方式去回答问题。\
当你不知道问题的答案时，你承认\
你不知道.

这是一个问题:
{input}"""


#第二个提示适合回答数学问题
math_template = """你是一个非常优秀的数学家。 \
你擅长回答数学问题。 \
你之所以如此优秀， \
是因为你能够将棘手的问题分解为组成部分，\
回答组成部分，然后将它们组合在一起，回答更广泛的问题。

这是一个问题：
{input}"""


#第三个适合回答历史问题
history_template = """你是以为非常优秀的历史学家。 \
你对一系列历史时期的人物、事件和背景有着极好的学识和理解\
你有能力思考、反思、辩证、讨论和评估过去。\
你尊重历史证据，并有能力利用它来支持你的解释和判断。

这是一个问题:
{input}"""


#第四个适合回答计算机问题
computerscience_template = """ 你是一个成功的计算机科学专家。\
你有创造力、协作精神、\
前瞻性思维、自信、解决问题的能力、\
对理论和算法的理解以及出色的沟通技巧。\
你非常擅长回答编程问题。\
你之所以如此优秀，是因为你知道  \
如何通过以机器可以轻松解释的命令式步骤描述解决方案来解决问题，\
并且你知道如何选择在时间复杂性和空间复杂性之间取得良好平衡的解决方案。

这还是一个输入：
{input}"""

#对提示模版进行命名和描述
prompt_infos = [
    {
        "名字": "物理学", 
        "描述": "擅长回答关于物理学的问题", 
        "提示模板": physics_template
    },
    {
        "名字": "数学", 
        "描述": "擅长回答数学问题", 
        "提示模板": math_template
    },
    {
        "名字": "历史", 
        "描述": "擅长回答历史问题", 
        "提示模板": history_template
    },
    {
        "名字": "计算机科学", 
        "描述": "擅长回答计算机科学问题", 
        "提示模板": computerscience_template
    }
]

#基于提示模版信息创建相应目标链
destination_chains = {}
for p_info in prompt_infos:
    name = p_info["名字"]
    prompt_template = p_info["提示模板"]
    prompt = ChatPromptTemplate.from_template(template=prompt_template)
    chain = LLMChain(llm=llm, prompt=prompt)
    destination_chains[name] = chain  
    
destinations = [f"{p['名字']}: {p['描述']}" for p in prompt_infos]
destinations_str = "\n".join(destinations)


#创建默认目标链，这是一个当路由器无法决定使用哪个子链时调用的链
default_prompt = ChatPromptTemplate.from_template("{input}")
default_chain = LLMChain(llm=llm, prompt=default_prompt)

# 多提示路由模板
MULTI_PROMPT_ROUTER_TEMPLATE = 
"""给语言模型一个原始文本输入，\
让其选择最适合输入的模型提示。\
系统将为您提供可用提示的名称以及最适合改提示的描述。\
如果你认为修改原始输入最终会导致语言模型做出更好的响应，\
你也可以修改原始输入。


<< 格式 >>
返回一个带有JSON对象的markdown代码片段，该JSON对象的格式如下：

{% raw %}
```json
{{{{
    "destination": 字符串 \ 使用的提示名字或者使用 "DEFAULT"
    "next_inputs": 字符串 \ 原始输入的改进版本
}}}}
```
{% endraw %}

记住：“destination”必须是下面指定的候选提示名称之一，\
或者如果输入不太适合任何候选提示，\
则可以是 “DEFAULT” 。
记住：如果您认为不需要任何修改，\
则 “next_inputs” 可以只是原始输入。

<< 候选提示 >>
{destinations}

<< 输入 >>
{{input}}

<< 输出 (记得要包含 ```json)>>

样例:
<< 输入 >>
"什么是黑体辐射?"
<< 输出 >>

{% raw %}
```json
{{{{
    "destination": 字符串 \ 使用的提示名字或者使用 "DEFAULT"
    "next_inputs": 字符串 \ 原始输入的改进版本
}}}}
```
{% endraw %}
"""

# router_chain
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(
    destinations=destinations_str
)
router_prompt = PromptTemplate(
    template=router_template,
    input_variables=["input"],
    output_parser=RouterOutputParser(),
)

router_chain = LLMRouterChain.from_llm(llm, router_prompt)


#多提示链
chain = MultiPromptChain(router_chain=router_chain,    #路由链路
                         destination_chains=destination_chains,   #目标链路
                         default_chain=default_chain,      #默认链路
                         verbose=True   
                        )

print(chain.invoke("什么是黑体辐射？"))

print("====")

print(chain.invoke("你知道李白是谁嘛?"))

print("====")

print(chain.invoke("2 + 2 等于多少"))

print("====")

print(chain.invoke("电影肖申克的救赎讲了什么"))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
> Entering new MultiPromptChain chain...
物理学: {'input': '请解释一下黑体辐射是什么？'}
> Finished chain.
{'input': '请解释一下黑体辐射是什么？', 'text': '黑体辐射是指一个理想化的物体，它能够吸收所有入射到它表面的辐射能量，并且以最高效率地辐射出来。根据黑体辐射定律，黑体辐射的辐射强度与波长和温度有关，遵循普朗克辐射定律和维恩位移定律。黑体辐射在研究热辐射和量子力学等领域有着重要的应用。希望这个解释对你有所帮助。'}
====


> Entering new MultiPromptChain chain...
历史: {'input': '你知道李白是哪位诗人吗?'}
> Finished chain.
{'input': '你知道李白是哪位诗人吗?', 'text': '是的，李白是中国唐代著名的诗人，被誉为“诗仙”。他的诗作以豪放、奔放、豪情和浪漫著称，对后世诗人产生了深远的影响。他的诗作多以描绘自然风光、抒发豪情壮志和饮酒作乐为主题，被后人称为“酒仙诗派”的代表。李白的诗歌在中国文学史上占有重要地位，被誉为中国古代文学的瑰宝之一。'}
====


> Entering new MultiPromptChain chain...
数学: {'input': '求解 2 + 2'}
> Finished chain.
{'input': '求解 2 + 2', 'text': '2 + 2 = 4.'}
====


> Entering new MultiPromptChain chain...
None: {'input': '电影肖申克的救赎讲了什么'}
> Finished chain.
{'input': '电影肖申克的救赎讲了什么', 'text': '电影《肖申克的救赎》是一部由弗兰克·达拉邦特执导，蒂姆·罗宾斯和摩根·弗里曼主演的经典影片。影片讲述了银行家安迪·杜佛兰因被错误指控谋杀妻子和情人而被判无期徒刑，被送往肖申克监狱服刑的故事。\n\n在监狱里，安迪遭受各种虐待和不公正对待，但他始终保持着乐观和坚韧的态度。他结识了监狱里的一些狱友，包括摩根·弗里曼饰演的瑞德。通过他们的帮助和支持，安迪逐渐在监狱里建立了自己的地位，并展现出了非凡的智慧和毅力。\n\n最终，安迪通过自己的努力和智慧成功逃脱了监狱，并揭露了监狱长的腐败和罪行。影片通过安迪的故事展现了人性的善良和坚韧，以及对自由和正义的追求。它深刻地揭示了监狱制度中的黑暗面和人性的复杂性，同时也传达了希望和勇气的力量。'}

1.3 构造检索式问答连

基于 LangChain，我们可以构造一个使用 GPT3.5 进行问答的检索式问答链，这是一种通过检索步骤进行问答的方法。我们可以通过传入一个语言模型和一个向量数据库来创建它作为检索器。然后，我们可以用问题作为查询调用它，得到一个答案。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from langchain_openai import OpenAIEmbeddings
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import PyPDFLoader
from langchain_community.embeddings.dashscope import DashScopeEmbeddings

# 加载 PDF
loader = PyPDFLoader("MachineLearning-Lecture01.pdf")
docs = []
docs.extend(loader.load())

# 分割文本

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size = 1500,  # 每个文本块的大小。这意味着每次切分文本时，会尽量使每个块包含 1500 个字符。
    chunk_overlap = 150  # 每个文本块之间的重叠部分。
)

texts = text_splitter.split_documents(docs)

llm = ChatOpenAI(model="gpt-3.5-turbo",temperature=0)  # 默认是gpt-3.5-turbo

#embeddings = OpenAIEmbeddings()

# 灌库 
#阿里的向量模型
embeddings = DashScopeEmbeddings(
    model="text-embedding-v2",
)


vectordb =  Chroma.from_documents(texts, embeddings)

# 声明一个检索式问答链
qa_chain = RetrievalQA.from_chain_type(
    llm,
    retriever=vectordb.as_retriever()
)


# # 可以以该方式进行检索问答
question = "What are major topics for this class?"
result = qa_chain({"query": question})
print(result["result"])
1
2
3
4
5
6
The major topics for this class, as outlined by the instructor, include:

1. Machine learning fundamentals covered in the main lectures.
2. Optional discussion sections covering extensions to the main lecture materials, such as convex optimization and hidden Markov models.
3. The use of MATLAB for assignments, with the instructor strongly advising against using R due to potential compatibility issues.
4. Flexibility in group sizes for the term project, with options for groups of three, two, or individual work, with grading being the same regardless of group size.

1.4 深入探究检索式问答链

1.4.1 基于模板的检索式问答链

流程如图:

首先定义了一个提示模板。它包含一些关于如何使用下面的上下文片段的说明，然后有一个上下文变量的占位符。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from langchain.prompts import PromptTemplate

# Build prompt
template = """使用以下上下文片段来回答最后的问题。如果你不知道答案，只需说不知道，不要试图编造答案。答案最多使用三个句子。尽量简明扼要地回答。在回答的最后一定要说"感谢您的提问！"
{context}
问题：{question}
有用的回答："""
QA_CHAIN_PROMPT = PromptTemplate.from_template(template)

# Run chain
qa_chain = RetrievalQA.from_chain_type(
    llm,
    retriever=vectordb.as_retriever(),
    return_source_documents=True,
    chain_type_kwargs={"prompt": QA_CHAIN_PROMPT}
)

question = "机器学习是其中一节的话题吗"
result = qa_chain({"query": question})
result["result"]

输出

1
'是的，机器学习是这门课的一个话题。\n感谢您的提问！'

LangSmith流程展现:

这种方法非常好，因为它只涉及对LLM的一次调用。然而，它也有局限性，即如果文档太多，可能无法将它们全部适配到上下文窗口中。我们可以使用另一种技术来对文档进行问答，即MapReduce技术。

1.4.2 基于 MapReduce 的检索式问答链

流程如图:

在 MapReduce 技术中，首先将每个独立的文档单独发送到LLM以获取原始答案。然后，这些答案通过最终对LLM的一次调用组合成最终的答案。虽然这样涉及了更多对语言模型的调用，但它的优势在于可以处理任意数量的文档。

1
2
3
4
5
6
7
8
9
qa_chain_mr = RetrievalQA.from_chain_type(
    llm,
    retriever=vectordb.as_retriever(),
    chain_type="map_reduce"
)
# 中文版
question = "概率论是其中一节的话题吗"
result = qa_chain_mr({"query": question})
result["result"]

输出

1
'根据提供的文本片段，似乎没有提及与概率论直接相关的内容。文档主要讨论了参数的可能性以及数据的概率，特别是在参数化高斯密度的背景下。因此，文档似乎并未专注于概率论作为一个整体的主题。\n\n所以，根据提供的信息，似乎这个文档并没有涉及概率论作为一个独立的话题。'

LangSmith流程展现:

这种方法的两个问题。第一，速度要慢得多。第二，结果实际上更差。根据给定文档的这一部分，对这个问题并没有明确的答案。这可能是因为它是基于每个文档单独回答的。因此，如果信息分布在两个文档之间，它并没有在同一上下文中获取到所有的信息。

1.4.3 基于 Refine 的检索式问答链

流程如图:

Refine是一种新的链式类型。Refine 文档链类似于 MapReduce 链，对于每一个文档，会调用一次 LLM，但有所改进的是，每次发送给 LLM 的最终提示是一个序列，这个序列会将先前的响应与新数据结合在一起，并请求得到改进后的响应。因此，这是一种类似于 RNN 的概念，增强了上下文，从而解决信息分布在不同文档的问题。

  • 对于每个文档，它将所有非文档输入、当前文档以及最新的中间答案传递给 LLM 链以获得新的答案
  • 由于 Refine 链一次仅将单个文档传递给 LLM，因此很适合需要分析的文档数量多于模型上下文的任务
  • 明显的缺点是：将比 Stuff 文档链进行更多的 LLM 调用；当文档频繁地互相交叉引用时很可能表现不佳
1
2
3
4
5
6
7
8
qa_chain_mr = RetrievalQA.from_chain_type(
    llm,
    retriever=vectordb.as_retriever(),
    chain_type="refine"
)
question = "概率论是其中一节的话题吗"
result = qa_chain_mr({"query": question})
result["result"]

输出

1
'根据新的上下文信息，我们可以得出结论，概率论是其中一节的话题。在这段对话中，讲师提到了关于概率的假设和观点，以及在统计学中的频率学派观点。他还提到了关于条件概率的符号表示，以及在theta不被视为随机变量时的表示方法。因此，概率论是这节课的一个话题。'

LangSmith流程展现:

这个结果比MapReduce链的结果要好。这是因为使用Refined Chain允许你逐个地组合信息，实际上比MapReduce链鼓励更多的信息传递。

1.4.5 更多链

  • ConversationalRetrievalChain：建立在 Retrieval QA Chain 的基础上，并接入 Memory 组件
  • APIChain:使用 LLM 与 API 交互以检索相关信息，通过提供与提供的 API 文档相关的问题来构建链
  • MultiRetrievalQAChain：该链选择与给定问题最相关的 Retrieval QA Chain，然后使用它回答问题
  • MultiPromptChain：使用多提示链创建一个问答链，选择与给定问题最相关的提示并进行回答
  • ConstitutionalChain：自我批判链是一条确保语言模型的输出遵守一组预定义准则的链
  • 更多请参考 https://python.langchain.com/docs/modules/chains/

二、LangChain Expression Language (LCEL)

LangChain Expression Language（LCEL）是一种声明式语言，可轻松组合不同的调用顺序构成 Chain。LCEL 自创立之初就被设计为能够支持将原型投入生产环境，无需代码更改，从最简单的“提示+LLM”链到最复杂的链（已有用户成功在生产环境中运行包含数百个步骤的 LCEL Chain）。

LCEL 的一些亮点包括：

  1. 流支持：使用 LCEL 构建 Chain 时，你可以获得最佳的首个令牌时间（即从输出开始到首批输出生成的时间）。对于某些 Chain，这意味着可以直接从 LLM 流式传输令牌到流输出解析器，从而以与 LLM 提供商输出原始令牌相同的速率获得解析后的、增量的输出。

  2. 异步支持：任何使用 LCEL 构建的链条都可以通过同步 API（例如，在 Jupyter 笔记本中进行原型设计时）和异步 API（例如，在 LangServe 服务器中）调用。这使得相同的代码可用于原型设计和生产环境，具有出色的性能，并能够在同一服务器中处理多个并发请求。

  3. 优化的并行执行：当你的 LCEL 链条有可以并行执行的步骤时（例如，从多个检索器中获取文档），我们会自动执行，无论是在同步还是异步接口中，以实现最小的延迟。

  4. 重试和回退：为 LCEL 链的任何部分配置重试和回退。这是使链在规模上更可靠的绝佳方式。目前我们正在添加重试/回退的流媒体支持，因此你可以在不增加任何延迟成本的情况下获得增加的可靠性。

  5. 访问中间结果：对于更复杂的链条，访问在最终输出产生之前的中间步骤的结果通常非常有用。这可以用于让最终用户知道正在发生一些事情，甚至仅用于调试链条。你可以流式传输中间结果，并且在每个 LangServe 服务器上都可用。

  6. 输入和输出模式：输入和输出模式为每个 LCEL 链提供了从链的结构推断出的 Pydantic 和 JSONSchema 模式。这可以用于输入和输出的验证，是 LangServe 的一个组成部分。

  7. 无缝 LangSmith 跟踪集成：随着链条变得越来越复杂，理解每一步发生了什么变得越来越重要。通过 LCEL，所有步骤都自动记录到 LangSmith，以实现最大的可观察性和可调试性。

  8. 无缝 LangServe 部署集成：任何使用 LCEL 创建的链都可以轻松地使用 LangServe 进行部署。

原文：https://python.langchain.com/docs/expression_language/

官方给出使用LCEL前后对比：https://python.langchain.com/docs/expression_language/why/

LangChain Runnable Chain [2]

LCEL旨在简化组合链的过程，使从原型设计到生产的轻松过程成为可能。LCEL 通过标准的 Runnable 协议使创建自定义链路并以标准方式调用它们变得非常容易。

2.1 LCEL 例子

符号类似于 unix 管道操作符，它将不同的组件链接在一起，将一个组件的输出作为下一个组件的输入。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser

prompt = ChatPromptTemplate.from_template("tell me a joke about {foo}")
model = ChatOpenAI()

functions = [
    {
        "name": "joke",
        "description": "A joke",
        "parameters": {
            "type": "object",
            "properties": {
                "setup": {"type": "string", "description": "The setup for the joke"},
                "punchline": {
                    "type": "string",
                    "description": "The punchline for the joke",
                },
            },
            "required": ["setup", "punchline"],
        },
    }
]

#LCEL
chain = (
    prompt
    | model.bind(function_call={"name": "joke"}, functions=functions) #LCEL 支持将函数调用信息附加到链上
    | JsonOutputFunctionsParser() #指定要返回的函数时，可以使用 LCEL 直接解析它
)
result = chain.invoke({"foo": "bears"})
print(result)

输出

1
{'setup': "Why don't bears wear shoes?", 'punchline': 'Because they have bear feet!'}

三、Runnable Sequence

LangChain Expression Language (LCEL) 的核心

3.1 Runnable 对象简介

把一个个 Runnable 对象串联起来，就构成了 Runnable Sequence

Runnable对象的定义：标准接口包含:

  • invoke：对输入直接进行链式调用
  • batch：对输入列表进行批量的链式调用
  • stream：（流式）分块返回响应（需要 Model、Parser 等支持）

异步的方法，需要使用 await 语法实现异步等待：

  • astream：异步的流式输出模式
  • ainvoke：异步的使用单个输入去调用链
  • abatch：异步的使用一个数组的输入去调用链
  • astream_log:中间步骤发生时进行回调
  • astream_events：beta 异步流式返回链中发生的事件（在 langchain-core 0.1.14 中引入）

输入类型和输出类型因组件而异:

组件 输入类型 输出类型
Prompt 字典 PromptValue
ChatModel 单个字符串、聊天消息列表或 PromptValue ChatMessage
LLM 单个字符串、聊天消息列表或 PromptValue 字符串
OutputParser LLM 或 ChatModel 的输出 取决于解析器
Retriever 单个字符串 文档列表
Tool 单个字符串或字典，取决于工具 取决于工具

3.2 RunnableParallel

RunnableParallel可以并行运行多个Runnable对象

1
2
3
4
5
6
7
8
9
10
11
12
13
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI

model = ChatOpenAI()
joke_chain = ChatPromptTemplate.from_template("tell me a joke about {topic}") | model
poem_chain = (
    ChatPromptTemplate.from_template("write a 2-line poem about {topic}") | model
)

map_chain = RunnableParallel(joke=joke_chain, poem=poem_chain)

map_chain.invoke({"topic": "bear"})
1
2
{'joke': AIMessage(content="Why did the bear break up with his girlfriend? \nBecause he couldn't bear the relationship anymore!", response_metadata={'token_usage': {'completion_tokens': 20, 'prompt_tokens': 13, 'total_tokens': 33}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_3b956da36b', 'finish_reason': 'stop', 'logprobs': None}, id='run-b56a3628-fc44-49d6-ae3a-97254f8dcd84-0'),
 'poem': AIMessage(content='In the forest deep and wide,\nThe bear roams, majestic stride.', response_metadata={'token_usage': {'completion_tokens': 15, 'prompt_tokens': 15, 'total_tokens': 30}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_3b956da36b', 'finish_reason': 'stop', 'logprobs': None}, id='run-b42e1898-a757-4495-a468-813ba998b19c-0')}

LangSmith流程展现:

当一个Runnable对象接收的输入需要被预处理时，也可用RunnableParallel进行处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
from langchain.schema.output_parser import StrOutputParser
from langchain_core.runnables import RunnableParallel
from langchain_core.runnables import RunnablePassthrough
from operator import itemgetter

llm = ChatOpenAI()
prompt1 = ChatPromptTemplate.from_template("""
What is the city {person} is from? Only respond with city name.
""")
  
prompt2 = ChatPromptTemplate.from_template("""
What country is the city {city} in? Respond in {language}.
""")

chain1 = prompt1|llm|StrOutputParser()

rp =RunnableParallel(
    city=chain1,
    language=itemgetter("language")
)

chain = (
    rp
    | prompt2
    | llm
    | StrOutputParser()
)

print(chain.invoke({"person": "李白", "language": "Chinese" }))


chain = (
    {"city":chain1,"language":itemgetter("language")} #RunnableParallel可以写成这样
    | prompt2
    | llm
    | StrOutputParser()
)

print(chain.invoke({"person": "李白", "language": "Chinese" }))

输出

1
2
长安市位于中国。
长安市位于中国。

3.2 RunnablePassthrough

RunnablePassthrough允许传递输入数据，可以保持不变或添加额外的键。通常与RunnableParallel一起使用，将数据分配给映射中的新键。使用assign参数调用RunnablePassthrough，将接收输入，并添加传递给assign函数的额外参数。

1
2
3
4
5
6
7
8
9
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

runnable = RunnableParallel(
    passed=RunnablePassthrough(),
    extra=RunnablePassthrough.assign(mult=lambda x: x["num"] * 3),
    modified=lambda x: x["num"] + 1,
)

runnable.invoke({"num": 1})

LangSmith流程展现:

用 LCEL 实现 RAG

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from langchain_community.vectorstores import Chroma
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_community.embeddings.dashscope import DashScopeEmbeddings

embeddings = DashScopeEmbeddings(
    model="text-embedding-v2",
)


vectorstore = Chroma.from_texts(
    ["harrison worked at kensho"], embedding=embeddings
)
retriever = vectorstore.as_retriever()
template = """Answer the question based only on the following context:
{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()

retrieval_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

retrieval_chain.invoke("where did harrison work?")

输出

1
'Harrison worked at Kensho.'

3.3 RunnableLambda

RunnableLambda可以让我们运行自定义函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from operator import itemgetter

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI


def length_function(text):
    return len(text)


def _multiple_length_function(text1, text2):
    return len(text1) * len(text2)


def multiple_length_function(_dict):
    return _multiple_length_function(_dict["text1"], _dict["text2"])


prompt = ChatPromptTemplate.from_template("what is {a} + {b}")
model = ChatOpenAI()

chain1 = prompt | model

chain = (
    {
        "a": itemgetter("foo") | RunnableLambda(length_function),
        "b": {"text1": itemgetter("foo"), "text2": itemgetter("bar")}
        | RunnableLambda(multiple_length_function),
    }
    | prompt
    | model
)
chain.invoke({"foo": "bar", "bar": "gah"})

输出

1
AIMessage(content='3 + 9 is equal to 12.', response_metadata={'token_usage': {'completion_tokens': 10, 'prompt_tokens': 14, 'total_tokens': 24}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_3b956da36b', 'finish_reason': 'stop', 'logprobs': None}, id='run-08d1e13f-1fe1-486a-8d76-2c24b08ff1e9-0')

LangSmith流程展现:

3.4 RunnableBranch

为 Runnable Sequence 提供条件路由，路由允许您创建非确定性的 Chain，其中上一步的输出定义了下一步的路由方向，路由会依次匹配条件直至选择默认 Runnable 返回

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableBranch


chain = (
    PromptTemplate.from_template(
        """Given the user question below, classify it as either being about `LangChain`, `Anthropic`, or `Other`.

Do not respond with more than one word.

<question>
{question}
</question>

Classification:"""
    )
    | ChatOpenAI(model_name="gpt-4-turbo")
    | StrOutputParser()
)

langchain_chain = PromptTemplate.from_template(
    """You are an expert in langchain. \
Always answer questions starting with "As Harrison Chase told me". \
Respond to the following question:

Question: {question}
Answer:"""
) | ChatOpenAI(model_name="gpt-4-turbo")

anthropic_chain = PromptTemplate.from_template(
    """You are an expert in anthropic. \
Always answer questions starting with "As Dario Amodei told me". \
Respond to the following question:

Question: {question}
Answer:"""
) | ChatOpenAI(model_name="gpt-4-turbo")

general_chain = PromptTemplate.from_template(
    """Respond to the following question:

Question: {question}
Answer:"""
) | ChatOpenAI(model_name="gpt-4-turbo")



branch = RunnableBranch(
    (lambda x: "anthropic" in x["topic"].lower(), anthropic_chain),
    (lambda x: "langchain" in x["topic"].lower(), langchain_chain),
    general_chain,
)

full_chain = {"topic": chain, "question": lambda x: x["question"]} | branch
full_chain.invoke({"question": "how do I use LangChain?"})
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
As Harrison Chase told me, using LangChain effectively involves understanding its capabilities as a language model interaction tool. Here’s a step-by-step guide on how to get started:

1. **Installation**: First, you need to install the LangChain library. This can typically be done via pip:
   ```bash
   pip install langchain
   ```

2. **Setup**: After installation, you should familiarize yourself with the key components of LangChain, such as the LLM (Large Language Models), Chains (which are sequences of operations involving language models), and Adapters (which allow LangChain to interact with external systems).

3. **Creating a Chain**: Begin by creating a simple chain. This involves defining what you want the language model to do. For example, you can create a chain that takes a user's input, processes it through a language model, and then outputs a response.

4. **Testing and Iteration**: Test your chain to see how it performs. Based on the output, you may need to tweak your setup or the sequence of operations in your chain to improve the interaction or the quality of the responses.

5. **Integration**: Finally, integrate LangChain with your application. This might involve setting up interfaces for users to interact with or automating certain tasks using LangChain in the backend.

By following these steps, you can effectively utilize LangChain to enhance language model interactions in your projects or applications.

LangSmith流程展现:

3.5 Tool 也是一种 Runnable 对象

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
from langchain_core.tools import tool
from typing import Union
from operator import itemgetter
from langchain_core.runnables import (
    Runnable,
    RunnableLambda,
    RunnableMap,
    RunnablePassthrough,
)
import json


@tool
def multiply(first_int: int, second_int: int) -> int:
    """两个整数相乘"""
    return first_int * second_int


@tool
def add(first_int: int, second_int: int) -> int:
    "Add two integers."
    return first_int + second_int

@tool
def exponentiate(base: int, exponent: int) -> int:
    "Exponentiate the base to the exponent power."
    return base**exponent

tools = [multiply, add, exponentiate]

# 名称到函数的映射
tool_map = {tool.name: tool for tool in tools}

llm = ChatOpenAI()

def call_tool(tool_invocation: dict) -> Union[str, Runnable]:
    """Function for dynamically constructing the end of the chain based on the model-selected tool."""
    tool = tool_map[tool_invocation["type"]]
    return RunnablePassthrough.assign(
        output=itemgetter("args") | tool
    )


# .map() allows us to apply a function to a list of inputs.
call_tool_list = RunnableLambda(call_tool).map()

def route(response):
    if len(response["functions"]) > 0:
        return response["functions"]
    else:
        return response["text"]


chain = llm.bind_tools(tools) | {
    "functions": JsonOutputToolsParser() | call_tool_list,
    "text": StrOutputParser()
} | RunnableLambda(route)

result = chain.invoke("1024的平方是多少")
print(result)

result = chain.invoke("你好")
print(result)

输出

1
2
[{'args': {'base': 1024, 'exponent': 2}, 'type': 'exponentiate', 'output': 1048576}]
你好！有什么我可以帮助你的吗？

LangSmith流程展现:

3.6 在 Runnable Sequence 中引入 Memory

Memory 可以被添加到任何 Runnable Sequence 中，本质和就是生成上下文文本填充到提示词中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai.chat_models import ChatOpenAI
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory


model = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You're an assistant who's good at {ability}. Respond in 20 words or fewer",
        ),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)
runnable = prompt | model


store = {}


def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]


with_message_history = RunnableWithMessageHistory(
    runnable,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

result = with_message_history.invoke(
    {"ability": "math", "input": "What does cosine mean?"},
    config={"configurable": {"session_id": "abc123"}},
)

with_message_history.invoke(
    {"ability": "math", "input": "What?"},
    config={"configurable": {"session_id": "abc123"}},
)

print(get_session_history("abc123"))

输出

1
2
3
4
Human: What does cosine mean?
AI: Cosine is a trigonometric function that gives the ratio of the adjacent side to the hypotenuse in a right triangle.
Human: What?
AI: Cosine is a way to relate the angles and sides of a right triangle using ratios.

LangSmith流程展现:

3.7 Runnable 的 Fallback 机制

Fallback 机制可以提供优雅的异常处理，通过“以次充优”的方式尽可能保障链路执行继续执行，Fallback不仅可以支撑 Runnable 对象，还可以直接 Fallback 整个 Runnable Sequence：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

from langchain_community.chat_models import ChatTongyi

Tongyi_chat = ChatTongyi()

model = ChatOpenAI(model="fake-model",max_retries=0)

badChain = model | StrOutputParser()
goodChain = Tongyi_chat | StrOutputParser()

chain = badChain.with_fallbacks([goodChain])

print(chain.invoke("你是谁"))

输出

1
我是通义千问，由阿里云开发的AI助手。我被设计用来回答各种问题、提供信息和与用户进行对话。有什么我可以帮助你的吗？

LangSmith流程展现:

更多例子：https://python.langchain.com/docs/expression_language/why/

参考

  1. https://www.bilibili.com/video/BV1214y1X7Aq ↩︎
  2. https://www.bilibili.com/read/cv25953108/ ↩︎
LangChain
#LLM #LLM学习笔记 #LangChain
LangChain(五)——Chanis
https://mztchaoqun.com.cn/posts/D29_LangChain_Chains/
作者
mztchaoqun
发布于
2024年7月22日
许可协议