2.3 四种组合文档链方式在LangChain开发中的作用及源码分析

假设阅读一本书，作者和智华合作写的一本关于Spark方面的书籍，一共1300多页，如果从第一页开始，逐渐去遍历检索信息，无论从速度、还是从性价比的角度，这是一个问题，LangChain给我们提供了一些比较经典的方式，例如map_reduce、map_rerank、refine、stuff等四种方式。

Map Reduce方式把一个很大的文件分成多个文件，如果大家做Hadoop、Spark或Flink，对这个概念应该很清楚，这是一种分而治之的思想。

map_reduce.py的代码实现：

1. """通过先在文档上映射一个链来组合文档，然后再组合结果."""
2. from __future__ import annotations
3. from typing import Any, Callable, Dict, List, Optional, Protocol, Tuple
4. from pydantic import Extra, root_validator
5. from langchain.callbacks.manager import Callbacks
6. from langchain.chains.combine_documents.base import BaseCombineDocumentsChain
7. from langchain.chains.llm import LLMChain
8. from langchain.docstore.document import Document
9. class CombineDocsProtocol(Protocol):
10. """ combine_docs方法的接口."""
11. def __call__(self, docs: List[Document], **kwargs: Any) -> str:
12. """combine_docs方法的接口."""
13. def _split_list_of_docs(
14. docs: List[Document], length_func: Callable, token_max: int, **kwargs: Any
15. ) -> List[List[Document]]:
16. new_result_doc_list = []
17. _sub_result_docs = []
18. for doc in docs:
19. _sub_result_docs.append(doc)
20. _num_tokens = length_func(_sub_result_docs, **kwargs)
21. if _num_tokens > token_max:
22. if len(_sub_result_docs) == 1:
23. raise ValueError(
24. "A single document was longer than the context length,"
25. " we cannot handle this."
26. )
27. if len(_sub_result_docs) == 2:
28. raise ValueError(
29. "A single document was so long it could not be combined "
30. "with another document, we cannot handle this."
31. )
32. new_result_doc_list.append(_sub_result_docs[:-1])
33. _sub_result_docs = _sub_result_docs[-1:]
34. new_result_doc_list.append(_sub_result_docs)
35. return new_result_doc_list
36. def _collapse_docs(
37. docs: List[Document],
38. combine_document_func: CombineDocsProtocol,
39. **kwargs: Any,
40. ) -> Document:
41. result = combine_document_func(docs, **kwargs)
42. combined_metadata = {k: str(v) for k, v in docs[0].metadata.items()}
43. for doc in docs[1:]:
44. for k, v in doc.metadata.items():
45. if k in combined_metadata:
46. combined_metadata[k] += f", {v}"
47. else:
48. combined_metadata[k] = str(v)
49. return Document(page_content=result, metadata=combined_metadata)
50. class MapReduceDocumentsChain(BaseCombineDocumentsChain):
51. """通过在文档上映射一个链来组合文档，然后组合结果."""
52. llm_chain: LLMChain
53. """链单独应用于每个文档"""
54. combine_document_chain: BaseCombineDocumentsChain
55. """用于组合将llm_chain应用于文档的结果."""
56. collapse_document_chain: Optional[BaseCombineDocumentsChain] = None
57. """ 链用于在需要时折叠中间结果。如果None，将使用combine_document_chain."""
58. document_variable_name: str
59. """ llm_chain中用于放置文档的变量名。如果llm_chain中只有一个变量，则不需要提供这个."""
60. return_intermediate_steps: bool = False
61. """返回输出中映射步骤的结果."""
62. @property
63. def output_keys(self) -> List[str]:
64. """期望输入键
65. :meta private:
66. """
67. _output_keys = super().output_keys
68. if self.return_intermediate_steps:
69. _output_keys = _output_keys + ["intermediate_steps"]
70. return _output_keys
71. class Config:
72. """pydantic对象的配置"""
73. extra = Extra.forbid
74. arbitrary_types_allowed = True
75. @root_validator(pre=True)
76. def get_return_intermediate_steps(cls, values: Dict) -> Dict:
77. """为了向后兼容"""
78. if "return_map_steps" in values:
79. values["return_intermediate_steps"] = values["return_map_steps"]
80. del values["return_map_steps"]
81. return values
82. @root_validator(pre=True)
83. def get_default_document_variable_name(cls, values: Dict) -> Dict:
84. """获取默认文档变量名(如果没有提供)"""
85. if "document_variable_name" not in values:
86. llm_chain_variables = values["llm_chain"].prompt.input_variables
87. if len(llm_chain_variables) == 1:
88. values["document_variable_name"] = llm_chain_variables[0]
89. else:
90. raise ValueError(
91. "document_variable_name must be provided if there are "
92. "multiple llm_chain input_variables"
93. )
94. else:
95. llm_chain_variables = values["llm_chain"].prompt.input_variables
96. if values["document_variable_name"] not in llm_chain_variables:
97. raise ValueError(
98. f"document_variable_name {values['document_variable_name']} was "
99. f"not found in llm_chain input_variables: {llm_chain_variables}"
100. )
101. return values
102. @property
103. def _collapse_chain(self) -> BaseCombineDocumentsChain:
104. if self.collapse_document_chain is not None:
105. return self.collapse_document_chain
106. else:
107. return self.combine_document_chain
108. def combine_docs(
109. self,
110. docs: List[Document],
111. token_max: int = 3000,
112. callbacks: Callbacks = None,
113. **kwargs: Any,
114. ) -> Tuple[str, dict]:
115. """ 以map reduce方式组合文档。
116. 通过在所有文档上映射第一个链，然后合并结果来进行组合。
117. 如果需要(如果有很多文档)，这种合并可以递归地完成。
118. """
119. results = self.llm_chain.apply(
120. # FYI -这是并行化的，所以速度很快
121. [{self.document_variable_name: d.page_content, **kwargs} for d in docs],
122. callbacks=callbacks,
123. )
124. return self._process_results(
125. results, docs, token_max, callbacks=callbacks, **kwargs
126. )
127. async def acombine_docs(
128. self, docs: List[Document], callbacks: Callbacks = None, **kwargs: Any
129. ) -> Tuple[str, dict]:
130. """ 以map reduce方式组合文档。
131. 通过在所有文档上映射第一个链，然后合并结果来进行组合。
132. 如果需要(如果有很多文档)，这种合并可以递归地完成。
133. """
134. results = await self.llm_chain.aapply(
135. # FYI - this is parallelized and so it is fast.
136. [{**{self.document_variable_name: d.page_content}, **kwargs} for d in docs],
137. callbacks=callbacks,
138. )
139. return self._process_results(results, docs, callbacks=callbacks, **kwargs)
140. def _process_results(
141. self,
142. results: List[Dict],
143. docs: List[Document],
144. token_max: int = 3000,
145. callbacks: Callbacks = None,
146. **kwargs: Any,
147. ) -> Tuple[str, dict]:
148. question_result_key = self.llm_chain.output_key
149. result_docs = [
150. Document(page_content=r[question_result_key], metadata=docs[i].metadata)
151. # 它使用来自文档的元数据，以及来自' results '的文本结果
152. for i, r in enumerate(results)
153. ]
154. length_func = self.combine_document_chain.prompt_length
155. num_tokens = length_func(result_docs, **kwargs)
156. def _collapse_docs_func(docs: List[Document], **kwargs: Any) -> str:
157. return self._collapse_chain.run(
158. input_documents=docs, callbacks=callbacks, **kwargs
159. )
160. while num_tokens is not None and num_tokens > token_max:
161. new_result_doc_list = _split_list_of_docs(
162. result_docs, length_func, token_max, **kwargs
163. )
164. result_docs = []
165. for docs in new_result_doc_list:
166. new_doc = _collapse_docs(docs, _collapse_docs_func, **kwargs)
167. result_docs.append(new_doc)
168. num_tokens = self.combine_document_chain.prompt_length(
169. result_docs, **kwargs
170. )
171. if self.return_intermediate_steps:
172. _results = [r[self.llm_chain.output_key] for r in results]
173. extra_return_dict = {"intermediate_steps": _results}
174. else:
175. extra_return_dict = {}
176. output = self.combine_document_chain.run(
177. input_documents=result_docs, callbacks=callbacks, **kwargs
178. )
179. return output, extra_return_dict
180. @property
181. def _chain_type(self) -> str:
182. return "map_reduce_documents_chain"

我们对文档进行切分，把一个大文件分成很多小部分，然后把这小部分再进行合并，这是Map Reduce的方式。在map_reduce.py代码文件中，MapReduceDocumentsChain类用于将多个文档进行处理和合并。该类通过将一个链映射到每个文档上，然后将结果组合起来来实现文档的合并。MapReduceDocumentsChain类包含了一个LLMChain类型的语言模型链，一个BaseCombineDocumentsChain类型的文档合并链，一个可选的BaseCombineDocumentsChain类型的文档折叠链、一个文档变量名，以及一个返回中间结果的布尔值。MapReduceDocumentsChain类的combine_docs方法将语言模型链应用于每个文档，并使用文档合并链将结果组合起来。如果需要，可以使用文档折叠链折叠中间结果。如果结果文档的长度超过了token_max的限制，方法将会递归地将文档拆分、折叠和合并，直到结果文档的长度小于或等于token_max。map_reduce.py文件还实现了一个CombineDocsProtocol接口，用于定义合并文档的方法。该接口定义了一个接受文档列表和其他关键字参数的方法，并返回一个字符串。该方法可以根据需要进行自定义，以适应不同的应用场景。 map_reduce.py代码文件提供了一种方便的方式来将多个文档进行处理和合并。它支持使用不同的语言模型链和文档合并链来适应不同的应用场景，并提供了可选的文档折叠链来折叠中间结果。该类还支持递归拆分、折叠和合并结果文档，以处理长度超过限制的情况。

……..

LangChain组合文档链一共有4种方式，也可以使用自定义的方式，实现自定义子类，BaseCombineDocumentsChain 类继承至ABC类。

base.py的代码实现：

1. class BaseCombineDocumentsChain(Chain, ABC):
2. """用于组合文档的链的基本接口."""
3. input_key: str = "input_documents" #: :meta private:
4. output_key: str = "output_text" #: :meta private:
5. @property
6. def input_keys(self) -> List[str]:
7. """期望输入键
8. :meta private:
9. """
10. return [self.input_key]
11. @property
12. def output_keys(self) -> List[str]:
13. """返回输出键
14. :meta private:
15. """
16. return [self.output_key]
17. def prompt_length(self, docs: List[Document], **kwargs: Any) -> Optional[int]:
18. """ 返回给定传入文档的提示长度。
19. 如果方法不依赖于提示符长度，则返回None。
20. """
21. return None
22. @abstractmethod
23. def combine_docs(self, docs: List[Document], **kwargs: Any) -> Tuple[str, dict]:
24. """将文档组合成单个字符串."""
25. @abstractmethod
26. async def acombine_docs(
27. self, docs: List[Document], **kwargs: Any
28. ) -> Tuple[str, dict]:
29. """将文档异步地组合成单个字符串."""
30. def _call(
31. self,
32. inputs: Dict[str, List[Document]],
33. run_manager: Optional[CallbackManagerForChainRun] = None,
34. ) -> Dict[str, str]:
35. _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
36. docs = inputs[self.input_key]
37. # 其他键被认为是LLM预测所需要的
38. other_keys = {k: v for k, v in inputs.items() if k != self.input_key}
39. output, extra_return_dict = self.combine_docs(
40. docs, callbacks=_run_manager.get_child(), **other_keys
41. )
42. extra_return_dict[self.output_key] = output
43. return extra_return_dict
44. async def _acall(
45. self,
46. inputs: Dict[str, List[Document]],
47. run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
48. ) -> Dict[str, str]:
49. _run_manager = run_manager or AsyncCallbackManagerForChainRun.get_noop_manager()
50. docs = inputs[self.input_key]
51. # 其他键被认为是LLM预测所需要的
52. other_keys = {k: v for k, v in inputs.items() if k != self.input_key}
53. output, extra_return_dict = await self.acombine_docs(
54. docs, callbacks=_run_manager.get_child(), **other_keys
55. )
56. extra_return_dict[self.output_key] = output
57. return extra_return_dict

base.py代码文件中，定义了一个BaseCombineDocumentsChain类，它是 Chain 和 ABC 的子类，用于通过将多个文档组合成单个字符串来处理文档。

BaseCombineDocumentsChain类具有以下属性：

• input_key： 接收文档输入键的字符串名称。
• output_key：输出的字符串键的字符串名称。

BaseCombineDocumentsChain类具有以下方法：

• prompt_length()：接收一个 Document 对象的列表以及关键字参数，并返回所需的提示长度。如果该方法不依赖于提示长度，则返回 None。
• combine_docs()：是一个抽象方法，接收一个 Document 对象的列表以及关键字参数，并将其组合成单个字符串。
• acombine_docs()：combine_docs() 方法的异步版本。
• _call()：该方法接收一个由文档列表组成的字典输入。它调用 combine_docs() 方法来组合文档，并将其输出作为字典返回。
• _acall()：_call() 方法的异步版本。

BaseCombineDocumentsChain 类提供了一个接口，用于处理文档并将其组合成单个字符串，提供一个prompt_length方法来确定所需的提示长度，并通combine_docs抽象方法实现文档组合的逻辑。BaseCombineDocumentsChain是一个抽象基类，需要子类实现其抽象方法。

ABC类是一个通用的抽象接口，abc.py的代码实现：

1. class ABC(metaclass=ABCMeta):
2. """ Helper类，它提供了使用继承创建ABC的标准方法.
3. """
4. __slots__ = ()

如果做ChatGDP或者大模型的开发，基于历史数据和当前的输入信息，检索出本地的信息，例如Notion里面的数据，为什么还需要跟大模型进行交互？这是一个很核心的问题，也是一个很好的面试题，如果对这个问题能有很好理解的话，通过技术面试的第一轮，不会有太大的问题。

《企业级ChatGPT开发入门实战直播21课》报名课程请联系：

Gavin老师：NLP_Matrix_Space

Sam老师：NLP_ChatGPT_LLM

我们的两本最新书籍年底即将出版：

• 《企业级Transformer&ChatGPT解密：原理、源码及案例》
• 《企业级Transformer&Rasa解密：原理、源码及案例》

《企业级Transformer&ChatGPT解密：原理、源码及案例》本书以Transformer和ChatGPT技术为主线，系统剖析了Transformer架构的理论基础、模型设计与实现，Transformer语言模型GPT与BERT，ChatGPT技术及其开源实现，以及相关应用案例。内容涉及贝叶斯数学、注意力机制、语言模型、最大似然与贝叶斯推理等理论，和Transformer架构设计、GPT、BERT、ChatGPT等模型的实现细节，以及OpenAI API、ChatGPT提示工程、类ChatGPT大模型等应用。第一卷介绍了Transformer的Bayesian Transformer思想、架构设计与源码实现，Transformer语言模型的原理与机制，GPT自回归语言模型和BERT自编码语言模型的设计与实现。第二卷深入解析ChatGPT技术，包括ChatGPT发展历史、基本原理与项目实践，OpenAI API基础与高级应用，ChatGPT提示工程与多功能应用，类ChatGPT开源大模型技术与项目实践。

ChatGPT 技术：从基础应用到进阶实践涵盖了ChatGPT技术和OpenAI API的基础和应用，分为8个章节，从ChatGPT技术概述到类ChatGPT开源大模型技术的进阶项目实践。

1. ChatGPT技术概述：主要介绍了GPT-1、GPT-2、GPT-3、GPT-3.5和GPT-4的发展历程和技术特点，以及ChatGPT技术的基本原理和项目案例实战。

2. OpenAI API基础应用实践：主要介绍了OpenAI API模型及接口概述，以及如何使用OpenAI API进行向量检索和文本生成。

3. OpenAI API进阶应用实践：主要介绍了如何使用OpenAI API基于嵌入式向量检索实现问答系统，如何使用OpenAI API对特定领域模型进行微调。

4. ChatGPT提示工程基础知识：主要介绍了如何构建优质提示的两个关键原则，以及如何迭代快速开发构建优质提示。

5. ChatGPT提示工程实现多功能应用：主要介绍了如何使用ChatGPT提示工程实现概括总结、推断任务、文本转换和扩展功能。

6. ChatGPT提示工程构建聊天机器人：主要介绍了聊天机器人的应用场景，以及如何使用ChatGPT提示工程构建聊天机器人和订餐机器人。

7. 类ChatGPT开源大模型技术概述：主要介绍了类ChatGPT开源大模型的发展历程和技术特点，以及ChatGLM项目案例实践和LMFlow项目案例实践。

8. 类ChatGPT开源大模型进阶项目实践：主要介绍了类ChatGPT开源大模型的进阶项目实践，包括基于LoRA SFT+RM+RAFT技术进行模型微调、基于P-Tuning等技术对特定领域数据进行模型微调、基于LLama Index和Langchain技术的全面实践，以及使用向量检索技术对特定领域数据进行模型微调。

本书适用于NLP工程师、AI研究人员以及对Transformer和ChatGPT技术感兴趣的读者。通过学习，读者能够系统掌握Transformer理论基础，模型设计与训练推理全过程，理解ChatGPT技术内幕，并能运用OpenAI API、ChatGPT提示工程等技术进行项目实践。

Transformer作为目前NLP领域最为主流和成功的神经网络架构，ChatGPT作为Transformer技术在对话系统中的典型应用，本书内容涵盖了该领域的最新进展与技术。通过案例实践，使理论知识变成技能，这也是本书的独特之处。

《企业级Transformer&Rasa解密：原理、源码及案例》：是一本深入介绍Rasa对话机器人框架的实战开发指南。本书分为两卷，第一卷主要介绍基于Transformer的Rasa Internals解密，详细介绍了DIETClassifier和TED在Rasa架构中的实现和源码剖析。第二卷主要介绍Rasa 3.X硬核对话机器人应用开发，介绍了基于Rasa Interactive Learning和ElasticSearch的实战案例，以及通过Rasa Interactive Learning发现和解决对话机器人的Bugs案例实战。

第一卷中介绍了Rasa智能对话机器人中的Retrieval Model和Stateful Computations，解析了Rasa中去掉对话系统的Intent的内幕，深入研究了End2End Learning，讲解了全新一代可伸缩的DAG图架构的内幕，介绍了如何定制Graph NLU及Policies组件，讨论了自定义GraphComponent的内幕，从Python角度分析了GraphComponent接口，详细解释了自定义模型的create和load内幕，并讲述了自定义模型的languages及Packages支持。深入剖析了自定义组件Persistence源码，包括自定义对话机器人组件代码示例分析、Resource源码逐行解析、以及ModelStorage、ModelMetadata等逐行解析等。介绍了自定义组件Registering源码的内幕，包括采用Decorator进行Graph Component注册内幕源码分析、不同NLU和Policies组件Registering源码解析、以及手工实现类似于Rasa注册机制的Python Decorator全流程实现。讨论了自定义组件及常见组件源码的解析，包括自定义Dense Message Featurizer和Sparse Message Featurizer源码解析、Rasa的Tokenizer及WhitespaceTokenizer源码解析、以及CountVectorsFeaturizer及SpacyFeaturizer源码解析。深入剖析了框架核心graph.py源码，包括GraphNode源码逐行解析及Testing分析、GraphModelConfiguration、ExecutionContext、GraphNodeHook源码解析以及GraphComponent源码回顾及其应用源码。

第二卷主要介绍了基于Rasa Interactive Learning和ElasticSearch的实战案例，以及通过Rasa Interactive Learning发现和解决对话机器人的Bugs案例实战。介绍了使用Rasa Interactive Learning来调试nlu和prediction的案例实战，使用Rasa Interactive Learning来发现和解决对话机器人的Bugs案例实战介绍了使用Rasa Interactive Learning透视Rasa Form的NLU和Policies的内部工作机制案例实战，使用ElasticSearch来实现对话机器人的知识库功能，并介绍了相关的源码剖析和最佳实践，介绍了Rasa微服务和ElasticSearch整合中的代码架构分析，使用Rasa Interactive Learning对ConcertBot进行源码、流程及对话过程的内幕解密，介绍了使用Rasa来实现Helpdesk Assistant功能，并介绍了如何使用Debug模式进行Bug调试，使用Rasa Interactive Learning纠正Helpdesk Assistant中的NLU和Prediction错误，逐行解密Domain和Action微服务的源码。

本书适合对Rasa有一定了解的开发人员和研究人员，希望通过本书深入了解Rasa对话机器人的内部工作原理及其源代码实现方式。无论您是想要深入了解Rasa的工作原理还是想要扩展和定制Rasa，本书都将为您提供有价值的参考和指导。

《企业级Transformer&ChatGPT解密：原理、源码及案例》、《企业级Transformer&Rasa解密：原理、源码及案例》，是您深入学习的好选择，年底即将重磅出版，欢迎购买！