AI大模型应用开发入门-LangChain实现文档总结
<h4>一、整体思路</h4><p>长网页文本往往超过 LLM 单次处理的 token 限制，我们需要设计一个 <strong>map-reduce</strong> 流水线来拆分、局部总结、归并：</p>
<ol>
<li>
<p><strong>加载网页内容</strong></p>
</li>
<li>
<p><strong>拆分成可控大小的 chunk</strong></p>
</li>
<li>
<p><strong>对每个 chunk 做初步总结 (map)</strong></p>
</li>
<li>
<p><strong>汇总所有初步总结 (reduce)</strong></p>
</li>
<li>
<p><strong>如有需要递归 reduce 直到满足 token 限制</strong></p>
</li>
<li>
<p><strong>输出最终总结</strong></p>
</li>
</ol>
<p>接下来我们用代码实现！</p>
<h4>二、准备工作</h4>
<h6>1. 初始化 LLM</h6>
<p>首先我们通过 <code>init_chat_model</code> 加载 LLM：</p>
<pre><code class="language-python"># llm_env.py
from langchain.chat_models import init_chat_model
llm = init_chat_model("gpt-4o-mini", model_provider="openai")</code></pre>
<h4>三、主程序 main.py</h4>
<h6>1. 导入依赖 & 初始化</h6>
<pre><code class="language-python">import os
import sys
sys.path.append(os.getcwd())
from langchain_community.document_loaders import WebBaseLoader
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.llm import LLMChain
from langchain_core.prompts import ChatPromptTemplate
from langchain_text_splitters import CharacterTextSplitter
import operator
from typing import Annotated, List, Literal, TypedDict
from langchain.chains.combine_documents.reduce import collapse_docs, split_list_of_docs
from langchain_core.documents import Document
from langgraph.constants import Send
from langgraph.graph import END, START, StateGraph
from llm_set import llm_env
llm = llm_env.llm
</code></pre>
<h6>2. 加载网页</h6>
<pre><code class="language-python">loader = WebBaseLoader("https://en.wikipedia.org/wiki/Artificial_intelligence")
docs = loader.load()
</code></pre>
<p>通过 WebBaseLoader 可以轻松加载网页文本到 <code>docs</code> 列表中。</p>
<h6>3. 定义 Prompt 模板</h6>
<p>- Map 阶段 Prompt</p>
<pre><code class="language-python">map_prompt = ChatPromptTemplate.from_messages(
[("system", "Write a concise summary of the following: \\n\\n{context}")]
)
</code></pre>
<p>- Reduce 阶段 Prompt</p>
<pre><code class="language-python">reduce_template = """
The following is a set of summaries:
{docs}
Take these and distill it into a final, consolidated summary
of the main themes.
"""
reduce_prompt = ChatPromptTemplate([("human", reduce_template)])</code></pre>
<h6>4. 拆分文档 chunk</h6>
<pre><code class="language-python">text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
split_docs = text_splitter.split_documents(docs)
print(f"Split into {len(split_docs)} chunks")
</code></pre>
<p>将网页内容拆分成多个 chunk，chunk 大小设置 1000 tokens，便于单次处理。</p>
<h6>5. 定义 Token 长度计算</h6>
<pre><code class="language-python">token_max = 1000
def length_function(documents: List) -> int:
return sum(llm.get_num_tokens(d.page_content) for d in documents)
</code></pre>
<p>计算输入文档 token 总量，用于判断是否需要继续 collapse。</p>
<h6>6. 定义状态</h6>
<p>主状态：</p>
<pre><code class="language-python">class OverallState(TypedDict):
contents: List
summaries: Annotated
collapsed_summaries: List
final_summary: str
</code></pre>
<p>Map 阶段状态：</p>
<pre><code class="language-python">class SummaryState(TypedDict):
content: str</code></pre>
<h6>7. 生成初步 summary (Map 阶段)</h6>
<pre><code class="language-python">def generate_summary(state: SummaryState):
prompt = map_prompt.invoke(state["content"])
response = llm.invoke(prompt)
return {"summaries": }</code></pre>
<h6>8. Map 调度逻辑</h6>
<pre><code class="language-python">def map_summaries(state: OverallState):
return [
Send("generate_summary", {"content": content}) for content in state["contents"]
]</code></pre>
<h6>9. 收集 summary</h6>
<pre><code class="language-python">def collect_summaries(state: OverallState):
return {
"collapsed_summaries": ]
}</code></pre>
<h6>10. Reduce 逻辑</h6>
<p>- 内部 reduce 函数</p>
<pre><code class="language-python">def _reduce(input: dict) -> str:
prompt = reduce_prompt.invoke(input)
response = llm.invoke(prompt)
return response.content
</code></pre>
<p>- Collapse summaries</p>
<pre><code class="language-python">def collapse_summaries(state: OverallState):
docs_lists = split_list_of_docs(
state["collapsed_summaries"],
length_function,
token_max,
)
results = []
for doc_list in docs_lists:
combined = collapse_docs(doc_list, _reduce)
results.append(combined)
return {"collapsed_summaries": results}</code></pre>
<h6>11. 是否继续 collapse</h6>
<pre><code class="language-python">def should_collapse(state: OverallState):
num_tokens = length_function(state["collapsed_summaries"])
if num_tokens > token_max:
return "collapse_summaries"
else:
return "generate_final_summary"</code></pre>
<h6>12. 生成最终 summary</h6>
<pre><code class="language-python">def generate_final_summary(state: OverallState):
response = _reduce(state["collapsed_summaries"])
return {"final_summary": response}</code></pre>
<h4>四、构建流程图 (StateGraph)</h4>
<pre><code class="language-python">graph = StateGraph(OverallState)
graph.add_node("generate_summary", generate_summary)
graph.add_node("collect_summaries", collect_summaries)
graph.add_node("collapse_summaries", collapse_summaries)
graph.add_node("generate_final_summary", generate_final_summary)
graph.add_conditional_edges(START, map_summaries, ["generate_summary"])
graph.add_edge("generate_summary", "collect_summaries")
graph.add_conditional_edges("collect_summaries", should_collapse)
graph.add_conditional_edges("collapse_summaries", should_collapse)
graph.add_edge("generate_final_summary", END)
app = graph.compile()</code></pre>
<h6>五、执行总结流程</h6>
<pre><code class="language-python">for step in app.stream(
{"contents": },
{"recursion_limit": 10},
):
print(list(step.keys()))
</code></pre>
<p>通过 <code>.stream()</code> 启动整个流水线，传入切片后的 <code>contents</code>，流式输出每步结果，直到最终汇总完成。</p>
<h6>六、总结</h6>
<p>通过这个示例，你可以看到： </p>
<p>✅ 使用 LangChain + LLM 轻松实现 <strong>网页总结</strong><br>✅ 设计了 <strong>自动 map-reduce</strong> 流程，支持长文本拆分和递归 reduce<br>✅ 通过 StateGraph 灵活编排流程、</p><br><br>
来源：https://www.cnblogs.com/chenyishi/p/18945790
頁:
[1]