基于langchain的长文本多迭代总结

之前我们讲到langchain的rag问答，有兴趣的同学可以做下回顾
langchain基于混元大模型的实时内容的RAG问答
今天我们来了解下如何基于前文的方案实现长文本总结
为什么需要文本总结

通常会议内容是冗长的，如果能够提取关键信息的话，能够帮我们节省大量的时间
模型不能总结吗，为什么单独提出来长文本这个概念

大部分模型都会限制输入长度，如果会议长度超出了模型的限制则无法进行总结
方案

langchain提供了多种方案供我们选择,https://python.langchain.com/v0.1/docs/use_cases/summarization/

• stuff：全文本总结，将整个文本全部投入模型;这样仍然大概会超出模型
  
• MapReduce：将文本拆成n个小段，每个小段分别总结，然后再将最终的内容一起总结;这样虽然能解决问题，但是大概会破坏文本的上下文导致最终的效果不理想
  
• refine：和MapReduce相似的是将文本拆成n个小段，但是会以循环的方式先总结第一段，然后将第一段的总结效果和第二段再总结以此类推，此方法能够更好的保留原文的语义
  

难点

• 代码实现
  
• 流式返回
  
• 如何确定是末了一轮的返回（在流式响应的环境下，每轮都会返回总结效果，那么入会确定是末了一轮并返回个前端）
  

实现

由于langchain的部分实现比力紧凑，导致做二次开辟不是很方便，所以大概有部分修改源码的地方
1.创建文本加载工具，用于加载文本
AttachCode

1. from typing import Dict, Optional
3. from langchain.chains.combine_documents.base import AnalyzeDocumentChain
4. from langchain_community.document_loaders import WebBaseLoader
5. from langchain_core.callbacks import CallbackManagerForChainRun
7. from modules.BasinessException import BusinessException
8. from modules.resultCodeEnum import ResultCodeEnum
9. from service.SubtitleService import SubtitleService
10. from utils import constants
11. from utils.logger import logger
13. class download_summarize_chain(AnalyzeDocumentChain):
14.     def _call(
15.             self,
16.             inputs: Dict[str, str],
17.             run_manager: Optional[CallbackManagerForChainRun] = None,
18.     ) -> dict[str, str]:
20.         docs = self.get_docs(inputs, run_manager)
22.         # Other keys are assumed to be needed for LLM prediction
23.         other_keys: Dict = {k: v for k, v in inputs.items() if k != self.input_key}
24.         other_keys[self.combine_docs_chain.input_key] = docs
26.         _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
27.         return self.combine_docs_chain(
28.             other_keys, return_only_outputs=True, callbacks=_run_manager.get_child()
29.         )
31.     def get_docs(self, inputs, run_manager):
33.         file_download_url = str(inputs[constants.TRANSCRIPTION_FILE_URL])
34.         if file_download_url is not None and file_download_url.startswith("http"):
35.             # 通过下载地址下载文件
36.             loader = WebBaseLoader(file_download_url, None, False)
37.             """Split document into chunks and pass to CombineDocumentsChain."""
38.             document = loader.load()[0].page_content
39.             if len(document) <= 0:
40.                 logger.error(f"file not exists:{file_download_url}")
41.                 raise BusinessException.new_instance_with_rce(400, ResultCodeEnum.EMPTY_CONTENT)
42.         else:
43.             # 通过企业id和会议id获取字幕
44.             enterprise_id: str = run_manager.metadata.get(constants.ENTERPRISE_ID)
45.             meeting_id: str = run_manager.metadata.get(constants.MEETING_ID)
46.             logger.info(f"process task with llm:{enterprise_id}-{meeting_id}")
47.             document = SubtitleService().fetch_subtitles(enterprise_id=enterprise_id, meeting_id=meeting_id)
49.         docs = self.text_splitter.create_documents([document])
50.         logger.info("number of splitting doc parts:{}", len(docs))
51.         return docs

复制代码

3.refine chain
AttachCode

1. """Load summarizing chains."""
3. from typing import Any, Mapping, Optional, Protocol
5. from langchain.chains.combine_documents.base import BaseCombineDocumentsChain
6. from langchain.chains.combine_documents.map_reduce import MapReduceDocumentsChain
7. from langchain.chains.combine_documents.reduce import ReduceDocumentsChain
8. from langchain.chains.combine_documents.stuff import StuffDocumentsChain
9. from langchain.chains.llm import LLMChain
10. from langchain.chains.summarize import map_reduce_prompt, refine_prompts, stuff_prompt
11. from langchain_core.callbacks import Callbacks
12. from langchain_core.language_models import BaseLanguageModel
13. from langchain_core.prompts import BasePromptTemplate
15. from adapters.langchain.chains.refine import RefineDocumentsChain
18. class LoadingCallable(Protocol):
19.     """Interface for loading the combine documents chain."""
21.     def __call__(
22.             self, llm: BaseLanguageModel, **kwargs: Any
23.     ) -> BaseCombineDocumentsChain:
24.         """Callable to load the combine documents chain."""
27. def _load_stuff_chain(
28.         llm: BaseLanguageModel,
29.         prompt: BasePromptTemplate = stuff_prompt.PROMPT,
30.         document_variable_name: str = "text",
31.         verbose: Optional[bool] = None,
32.         **kwargs: Any,
33. ) -> StuffDocumentsChain:
34.     llm_chain = LLMChain(llm=llm, prompt=prompt, verbose=verbose)  # type: ignore[arg-type]
35.     # TODO: document prompt
36.     return StuffDocumentsChain(
37.         llm_chain=llm_chain,
38.         document_variable_name=document_variable_name,
39.         verbose=verbose,  # type: ignore[arg-type]
40.         **kwargs,
41.     )
44. def _load_map_reduce_chain(
45.         llm: BaseLanguageModel,
46.         map_prompt: BasePromptTemplate = map_reduce_prompt.PROMPT,
47.         combine_prompt: BasePromptTemplate = map_reduce_prompt.PROMPT,
48.         combine_document_variable_name: str = "text",
49.         map_reduce_document_variable_name: str = "text",
50.         collapse_prompt: Optional[BasePromptTemplate] = None,
51.         reduce_llm: Optional[BaseLanguageModel] = None,
52.         collapse_llm: Optional[BaseLanguageModel] = None,
53.         verbose: Optional[bool] = None,
54.         token_max: int = 3000,
55.         callbacks: Callbacks = None,
56.         *,
57.         collapse_max_retries: Optional[int] = None,
58.         **kwargs: Any,
59. ) -> MapReduceDocumentsChain:
60.     map_chain = LLMChain(
61.         llm=llm,
62.         prompt=map_prompt,
63.         verbose=verbose,  # type: ignore[arg-type]
64.         callbacks=callbacks,  # type: ignore[arg-type]
65.     )
66.     _reduce_llm = reduce_llm or llm
67.     reduce_chain = LLMChain(
68.         llm=_reduce_llm,
69.         prompt=combine_prompt,
70.         verbose=verbose,  # type: ignore[arg-type]
71.         callbacks=callbacks,  # type: ignore[arg-type]
72.     )
73.     # TODO: document prompt
74.     combine_documents_chain = StuffDocumentsChain(
75.         llm_chain=reduce_chain,
76.         document_variable_name=combine_document_variable_name,
77.         verbose=verbose,  # type: ignore[arg-type]
78.         callbacks=callbacks,
79.     )
80.     if collapse_prompt is None:
81.         collapse_chain = None
82.         if collapse_llm is not None:
83.             raise ValueError(
84.                 "collapse_llm provided, but collapse_prompt was not: please "
85.                 "provide one or stop providing collapse_llm."
86.             )
87.     else:
88.         _collapse_llm = collapse_llm or llm
89.         collapse_chain = StuffDocumentsChain(
90.             llm_chain=LLMChain(
91.                 llm=_collapse_llm,
92.                 prompt=collapse_prompt,
93.                 verbose=verbose,  # type: ignore[arg-type]
94.                 callbacks=callbacks,
95.             ),
96.             document_variable_name=combine_document_variable_name,
97.         )
98.     reduce_documents_chain = ReduceDocumentsChain(
99.         combine_documents_chain=combine_documents_chain,
100.         collapse_documents_chain=collapse_chain,
101.         token_max=token_max,
102.         verbose=verbose,  # type: ignore[arg-type]
103.         callbacks=callbacks,
104.         collapse_max_retries=collapse_max_retries,
105.     )
106.     return MapReduceDocumentsChain(
107.         llm_chain=map_chain,
108.         reduce_documents_chain=reduce_documents_chain,
109.         document_variable_name=map_reduce_document_variable_name,
110.         verbose=verbose,  # type: ignore[arg-type]
111.         callbacks=callbacks,
112.         **kwargs,
113.     )
116. def _load_refine_chain(
117.         llm: BaseLanguageModel,
118.         question_prompt: BasePromptTemplate = refine_prompts.PROMPT,
119.         refine_prompt: BasePromptTemplate = refine_prompts.REFINE_PROMPT,
120.         document_variable_name: str = "text",
121.         initial_response_name: str = "existing_answer",
122.         refine_llm: Optional[BaseLanguageModel] = None,
123.         verbose: Optional[bool] = None,
124.         **kwargs: Any,
125. ) -> RefineDocumentsChain:
126.     initial_chain = LLMChain(llm=llm, prompt=question_prompt, verbose=verbose)  # type: ignore[arg-type]
127.     _refine_llm = refine_llm or llm
128.     refine_chain = LLMChain(llm=_refine_llm, prompt=refine_prompt, verbose=verbose)  # type: ignore[arg-type]
129.     return RefineDocumentsChain(
130.         initial_llm_chain=initial_chain,
131.         refine_llm_chain=refine_chain,
132.         document_variable_name=document_variable_name,
133.         initial_response_name=initial_response_name,
134.         verbose=verbose,  # type: ignore[arg-type]
135.         **kwargs,
136.     )
139. def load_summarize_chain(
140.         llm: BaseLanguageModel,
141.         chain_type: str = "stuff",
142.         verbose: Optional[bool] = None,
143.         **kwargs: Any,
144. ) -> BaseCombineDocumentsChain:
145.     """Load summarizing chain.
147.     Args:
148.         llm: Language Model to use in the chain.
149.         chain_type: Type of document combining chain to use. Should be one of "stuff",
150.             "map_reduce", and "refine".
151.         verbose: Whether chains should be run in verbose mode or not. Note that this
152.             applies to all chains that make up the final chain.
154.     Returns:
155.         A chain to use for summarizing.
156.     """
157.     loader_mapping: Mapping[str, LoadingCallable] = {
158.         "stuff": _load_stuff_chain,
159.         "map_reduce": _load_map_reduce_chain,
160.         "refine": _load_refine_chain,
161.     }
162.     if chain_type not in loader_mapping:
163.         raise ValueError(
164.             f"Got unsupported chain type: {chain_type}. "
165.             f"Should be one of {loader_mapping.keys()}"
166.         )
167.     return loader_mapping[chain_type](llm, verbose=verbose, **kwargs)

复制代码

4.调用chain
AttachCode

1. """Combine documents by doing a first pass and then refining on more documents."""
3. from __future__ import annotations
5. from typing import Any, Dict, List, Tuple
7. from langchain.chains.combine_documents.base import (
8.     BaseCombineDocumentsChain,
9. )
10. from langchain.chains.llm import LLMChain
11. from langchain_core.callbacks import Callbacks, dispatch_custom_event
12. from langchain_core.documents import Document
13. from langchain_core.prompts import BasePromptTemplate, format_document
14. from langchain_core.prompts.prompt import PromptTemplate
15. from pydantic import ConfigDict, Field, model_validator
17. from utils.logger import logger
20. def _get_default_document_prompt() -> PromptTemplate:
21.     return PromptTemplate(input_variables=["page_content"], template="{page_content}")
24. class RefineDocumentsChain(BaseCombineDocumentsChain):
25.     """Combine documents by doing a first pass and then refining on more documents.
27.     This algorithm first calls `initial_llm_chain` on the first document, passing
28.     that first document in with the variable name `document_variable_name`, and
29.     produces a new variable with the variable name `initial_response_name`.
31.     Then, it loops over every remaining document. This is called the "refine" step.
32.     It calls `refine_llm_chain`,
33.     passing in that document with the variable name `document_variable_name`
34.     as well as the previous response with the variable name `initial_response_name`.
36.     Example:
37.         .. code-block:: python
39.             from langchain.chains import RefineDocumentsChain, LLMChain
40.             from langchain_core.prompts import PromptTemplate
41.             from langchain_community.llms import OpenAI
43.             # This controls how each document will be formatted. Specifically,
44.             # it will be passed to `format_document` - see that function for more
45.             # details.
46.             document_prompt = PromptTemplate(
47.                 input_variables=["page_content"],
48.                  template="{page_content}"
49.             )
50.             document_variable_name = "context"
51.             llm = OpenAI()
52.             # The prompt here should take as an input variable the
53.             # `document_variable_name`
54.             prompt = PromptTemplate.from_template(
55.                 "Summarize this content: {context}"
56.             )
57.             initial_llm_chain = LLMChain(llm=llm, prompt=prompt)
58.             initial_response_name = "prev_response"
59.             # The prompt here should take as an input variable the
60.             # `document_variable_name` as well as `initial_response_name`
61.             prompt_refine = PromptTemplate.from_template(
62.                 "Here's your first summary: {prev_response}. "
63.                 "Now add to it based on the following context: {context}"
64.             )
65.             refine_llm_chain = LLMChain(llm=llm, prompt=prompt_refine)
66.             chain = RefineDocumentsChain(
67.                 initial_llm_chain=initial_llm_chain,
68.                 refine_llm_chain=refine_llm_chain,
69.                 document_prompt=document_prompt,
70.                 document_variable_name=document_variable_name,
71.                 initial_response_name=initial_response_name,
72.             )
73.     """
75.     initial_llm_chain: LLMChain
76.     """LLM chain to use on initial document."""
77.     refine_llm_chain: LLMChain
78.     """LLM chain to use when refining."""
79.     document_variable_name: str
80.     """The variable name in the initial_llm_chain to put the documents in.
81.     If only one variable in the initial_llm_chain, this need not be provided."""
82.     initial_response_name: str
83.     """The variable name to format the initial response in when refining."""
84.     document_prompt: BasePromptTemplate = Field(
85.         default_factory=_get_default_document_prompt
86.     )
87.     """Prompt to use to format each document, gets passed to `format_document`."""
88.     return_intermediate_steps: bool = False
89.     """Return the results of the refine steps in the output."""
91.     @property
92.     def output_keys(self) -> List[str]:
93.         """Expect input key.
95.         :meta private:
96.         """
97.         _output_keys = super().output_keys
98.         if self.return_intermediate_steps:
99.             _output_keys = _output_keys + ["intermediate_steps"]
100.         return _output_keys
102.     model_config = ConfigDict(
103.         arbitrary_types_allowed=True,
104.         extra="forbid",
105.     )
107.     @model_validator(mode="before")
108.     @classmethod
109.     def get_return_intermediate_steps(cls, values: Dict) -> Any:
110.         """For backwards compatibility."""
111.         if "return_refine_steps" in values:
112.             values["return_intermediate_steps"] = values["return_refine_steps"]
113.             del values["return_refine_steps"]
114.         return values
116.     @model_validator(mode="before")
117.     @classmethod
118.     def get_default_document_variable_name(cls, values: Dict) -> Any:
119.         """Get default document variable name, if not provided."""
120.         if "initial_llm_chain" not in values:
121.             raise ValueError("initial_llm_chain must be provided")
123.         llm_chain_variables = values["initial_llm_chain"].prompt.input_variables
124.         if "document_variable_name" not in values:
125.             if len(llm_chain_variables) == 1:
126.                 values["document_variable_name"] = llm_chain_variables[0]
127.             else:
128.                 raise ValueError(
129.                     "document_variable_name must be provided if there are "
130.                     "multiple llm_chain input_variables"
131.                 )
132.         else:
133.             if values["document_variable_name"] not in llm_chain_variables:
134.                 raise ValueError(
135.                     f"document_variable_name {values['document_variable_name']} was "
136.                     f"not found in llm_chain input_variables: {llm_chain_variables}"
137.                 )
138.         return values
140.     def combine_docs(
141.             self, docs: List[Document], callbacks: Callbacks = None, **kwargs: Any
142.     ) -> Tuple[str, dict]:
143.         """Combine by mapping first chain over all, then stuffing into final chain.
145.         Args:
146.             docs: List of documents to combine
147.             callbacks: Callbacks to be passed through
148.             **kwargs: additional parameters to be passed to LLM calls (like other
149.                 input variables besides the documents)
151.         Returns:
152.             The first element returned is the single string output. The second
153.             element returned is a dictionary of other keys to return.
154.         """
155.         inputs = self._construct_initial_inputs(docs, **kwargs)
156.         dispatch_custom_event("last_doc_mark", {"chunk": False})
157.         doc_length = len(docs)
158.         if doc_length == 1:
159.             dispatch_custom_event("last_doc_mark", {"chunk": True})
160.         logger.info(f"refine_docs index:1/{doc_length} of {kwargs}")
161.         res = self.initial_llm_chain.predict(callbacks=callbacks, **inputs)
162.         refine_steps = [res]
163.         for index, doc in enumerate(docs[1:], start=1):
164.             logger.info(f"refine_docs index:{index+1}/{doc_length} of {kwargs}")
165.             if index == doc_length - 1:
166.                 dispatch_custom_event("last_doc_mark", {"chunk": True})
167.             base_inputs = self._construct_refine_inputs(doc, res)
168.             inputs = {**base_inputs, **kwargs}
169.             res = self.refine_llm_chain.predict(callbacks=callbacks, **inputs)
170.             refine_steps.append(res)
171.         logger.info(f"refine_docs finished of {kwargs}, result:{res}")
172.         return self._construct_result(refine_steps, res)
174.     async def acombine_docs(
175.             self, docs: List[Document], callbacks: Callbacks = None, **kwargs: Any
176.     ) -> Tuple[str, dict]:
177.         """Async combine by mapping a first chain over all, then stuffing
178.          into a final chain.
180.         Args:
181.             docs: List of documents to combine
182.             callbacks: Callbacks to be passed through
183.             **kwargs: additional parameters to be passed to LLM calls (like other
184.                 input variables besides the documents)
186.         Returns:
187.             The first element returned is the single string output. The second
188.             element returned is a dictionary of other keys to return.
189.         """
190.         inputs = self._construct_initial_inputs(docs, **kwargs)
191.         res = await self.initial_llm_chain.apredict(callbacks=callbacks, **inputs)
192.         refine_steps = [res]
193.         for doc in docs[1:]:
194.             base_inputs = self._construct_refine_inputs(doc, res)
195.             inputs = {**base_inputs, **kwargs}
196.             res = await self.refine_llm_chain.apredict(callbacks=callbacks, **inputs)
197.             refine_steps.append(res)
198.         return self._construct_result(refine_steps, res)
200.     def _construct_result(self, refine_steps: List[str], res: str) -> Tuple[str, dict]:
201.         if self.return_intermediate_steps:
202.             extra_return_dict = {"intermediate_steps": refine_steps}
203.         else:
204.             extra_return_dict = {}
205.         return res, extra_return_dict
207.     def _construct_refine_inputs(self, doc: Document, res: str) -> Dict[str, Any]:
208.         return {
209.             self.document_variable_name: format_document(doc, self.document_prompt),
210.             self.initial_response_name: res,
211.         }
213.     def _construct_initial_inputs(
214.             self, docs: List[Document], **kwargs: Any
215.     ) -> Dict[str, Any]:
216.         base_info = {"page_content": docs[0].page_content}
217.         base_info.update(docs[0].metadata)
218.         document_info = {k: base_info[k] for k in self.document_prompt.input_variables}
219.         base_inputs: dict = {
220.             self.document_variable_name: self.document_prompt.format(**document_info)
221.         }
222.         inputs = {**base_inputs, **kwargs}
223.         return inputs
225.     @property
226.     def _chain_type(self) -> str:
227.         return "refine_documents_chain"

复制代码

5.过滤末了一次迭代
AttachCode

1. def process(tool: BaseTool, prompt_type: QuestionTypeEnum, input_dict: dict,
2.             run_manager: Optional[CallbackManagerForToolRun] = None):
3.     # 获取模型实例
4.     model_type = ModelTypeEnum.from_string(run_manager.metadata.get(constants.MODEL_TYPE))
5.     model_instance = ModelAdapter.get_model_instance(model_type)
7.     # 提示词模板集合
8.     prompt_template = PromptSynchronizer.get_prompt_template(model_type=model_type, questionType=prompt_type)
9.     prompt_map = json.loads(prompt_template)
10.     refine_prompt = PromptTemplate.from_template(prompt_map["refine_template"], template_format="f-string")
11.     question_prompt = PromptTemplate.from_template(prompt_map["prompt"])
13.     logger.info("invoke tool input_dicts:{}",input_dict)
15.     combine_docs_chain=load_summarize_chain(llm=model_instance,
16.                                             chain_type="refine",
17.                                             question_prompt=question_prompt,
18.                                             refine_prompt=refine_prompt,
19.                                             return_intermediate_steps=True,
20.                                             input_key="text",
21.                                             output_key="existing_answer",
22.                                             verbose=True)
24.     res = (tool.pre_handler
25.            | download_summarize_chain(combine_docs_chain=combine_docs_chain,
26.                                   text_splitter=model_instance.get_text_splitter(),
27.                                   verbose=True,
28.                                   input_key="input_document")
29.            | tool.post_handler).invoke(input={"input_document": "", **input_dict}, config=RunnableConfig())
30.     return res

复制代码

此笔记由idea插件辅助生成
idea插件推荐 AnNote - IntelliJ IDEs Plugin | Marketplace 75 折扣头：
MGRYF-TJW4N-WZMSJ-MZDLD-LVGJH
BTKQ8-XZLPH-L3QH3-MPKBH-BP9RR

本文由博客群发一文多发等运营工具平台 OpenWrite 发布

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。