AI大模型应用开发入门-LangChain实现文档总结

一、整体思路

长网页文本往往超过 LLM 单次处理的 token 限制，我们需要设计一个 map-reduce 流水线来拆分、局部总结、归并：

• 加载网页内容
  
• 拆分成可控大小的 chunk
  
• 对每个 chunk 做初步总结 (map)
  
• 汇总所有初步总结 (reduce)
  
• 如有需要递归 reduce 直到满足 token 限制
  
• 输出最终总结
  

接下来我们用代码实现！
二、准备工作

1. 初始化 LLM

首先我们通过 init_chat_model 加载 LLM：

1. # llm_env.py
2. from langchain.chat_models import init_chat_model
4. llm = init_chat_model("gpt-4o-mini", model_provider="openai")

复制代码

三、主程序 main.py

1. 导入依赖 & 初始化

1. import os
2. import sys
4. sys.path.append(os.getcwd())
6. from langchain_community.document_loaders import WebBaseLoader
7. from langchain.chains.combine_documents import create_stuff_documents_chain
8. from langchain.chains.llm import LLMChain
9. from langchain_core.prompts import ChatPromptTemplate
10. from langchain_text_splitters import CharacterTextSplitter
11. import operator
12. from typing import Annotated, List, Literal, TypedDict
13. from langchain.chains.combine_documents.reduce import collapse_docs, split_list_of_docs
14. from langchain_core.documents import Document
15. from langgraph.constants import Send
16. from langgraph.graph import END, START, StateGraph
18. from llm_set import llm_env
20. llm = llm_env.llm

复制代码

2. 加载网页

1. loader = WebBaseLoader("https://en.wikipedia.org/wiki/Artificial_intelligence")
2. docs = loader.load()

复制代码

通过 WebBaseLoader 可以轻松加载网页文本到 docs 列表中。
3. 定义 Prompt 模板

- Map 阶段 Prompt

1. map_prompt = ChatPromptTemplate.from_messages(
2.     [("system", "Write a concise summary of the following: \\n\\n{context}")]
3. )

复制代码

- Reduce 阶段 Prompt

1. reduce_template = """
2. The following is a set of summaries:
3. {docs}
4. Take these and distill it into a final, consolidated summary
5. of the main themes.
6. """
8. reduce_prompt = ChatPromptTemplate([("human", reduce_template)])

复制代码

4. 拆分文档 chunk

1. text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
2. split_docs = text_splitter.split_documents(docs)
3. print(f"Split into {len(split_docs)} chunks")

复制代码

将网页内容拆分成多个 chunk，chunk 大小设置 1000 tokens，便于单次处理。
5. 定义 Token 长度计算

1. token_max = 1000
3. def length_function(documents: List[Document]) -> int:
4.     return sum(llm.get_num_tokens(d.page_content) for d in documents)

复制代码

计算输入文档 token 总量，用于判断是否需要继续 collapse。
6. 定义状态

主状态：

1. class OverallState(TypedDict):
2.     contents: List[str]
3.     summaries: Annotated[list, operator.add]
4.     collapsed_summaries: List[Document]
5.     final_summary: str

复制代码

Map 阶段状态：

1. class SummaryState(TypedDict):
2.     content: str

复制代码

7. 生成初步 summary (Map 阶段)

1. def generate_summary(state: SummaryState):
2.     prompt = map_prompt.invoke(state["content"])
3.     response = llm.invoke(prompt)
4.     return {"summaries": [response.content]}

复制代码

8. Map 调度逻辑

1. def map_summaries(state: OverallState):
2.     return [
3.         Send("generate_summary", {"content": content}) for content in state["contents"]
4.     ]

复制代码

9. 收集 summary

1. def collect_summaries(state: OverallState):
2.     return {
3.         "collapsed_summaries": [Document(summary) for summary in state["summaries"]]
4.     }

复制代码

10. Reduce 逻辑

- 内部 reduce 函数

1. def _reduce(input: dict) -> str:
2.     prompt = reduce_prompt.invoke(input)
3.     response = llm.invoke(prompt)
4.     return response.content

复制代码

- Collapse summaries

1. def collapse_summaries(state: OverallState):
2.     docs_lists = split_list_of_docs(
3.         state["collapsed_summaries"],
4.         length_function,
5.         token_max,
6.     )
8.     results = []
9.     for doc_list in docs_lists:
10.         combined = collapse_docs(doc_list, _reduce)
11.         results.append(combined)
13.     return {"collapsed_summaries": results}

复制代码

11. 是否继续 collapse

1. def should_collapse(state: OverallState):
2.     num_tokens = length_function(state["collapsed_summaries"])
3.     if num_tokens > token_max:
4.         return "collapse_summaries"
5.     else:
6.         return "generate_final_summary"

复制代码

12. 生成最终 summary

1. def generate_final_summary(state: OverallState):
2.     response = _reduce(state["collapsed_summaries"])
3.     return {"final_summary": response}

复制代码

四、构建流程图 (StateGraph)

1. graph = StateGraph(OverallState)
3. graph.add_node("generate_summary", generate_summary)
4. graph.add_node("collect_summaries", collect_summaries)
5. graph.add_node("collapse_summaries", collapse_summaries)
6. graph.add_node("generate_final_summary", generate_final_summary)
8. graph.add_conditional_edges(START, map_summaries, ["generate_summary"])
9. graph.add_edge("generate_summary", "collect_summaries")
10. graph.add_conditional_edges("collect_summaries", should_collapse)
11. graph.add_conditional_edges("collapse_summaries", should_collapse)
12. graph.add_edge("generate_final_summary", END)
14. app = graph.compile()

复制代码

五、执行总结流程

1. for step in app.stream(
2.     {"contents": [doc.page_content for doc in split_docs]},
3.     {"recursion_limit": 10},
4. ):
5.     print(list(step.keys()))

复制代码

通过 .stream() 启动整个流水线，传入切片后的 contents，流式输出每步结果，直到最终汇总完成。
六、总结

通过这个示例，你可以看到： 
✅ 使用 LangChain + LLM 轻松实现 网页总结
✅ 设计了 自动 map-reduce 流程，支持长文本拆分和递归 reduce
✅ 通过 StateGraph 灵活编排流程、

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！