Map-Reduce 摘要模式

映射-归约模式

第 1 步：映射阶段

import openai

client = openai.OpenAI(api_key='sk-...')

def summarize_chunk(chunk, model='gpt-4o'):
    resp = client.chat.completions.create(
        model=model,
        messages=[
            {'role': 'system', 'content': 'Summarize the following text concisely in 3-5 sentences.'},
            {'role': 'user', 'content': chunk}
        ]
    )
    return resp.choices[0].message.content

def map_phase(chunks):
    return [summarize_chunk(c) for c in chunks]

第 2 步：归约阶段

def reduce_phase(chunk_summaries, model='gpt-4o'):
    combined = '\n\n'.join(
        f'Section {i+1}:\n{s}'
        for i, s in enumerate(chunk_summaries)
    )
    resp = client.chat.completions.create(
        model=model,
        messages=[
            {'role': 'system', 'content': 'You are given summaries of consecutive sections of a document. Write a single coherent summary of the entire document.'},
            {'role': 'user', 'content': combined}
        ]
    )
    return resp.choices[0].message.content

整合应用：原始 API

def map_reduce_summarize(document, chunk_size=1000):
    chunks = fixed_chunk(document, max_tokens=chunk_size)
    print(f'Chunks: {len(chunks)}')

    chunk_summaries = map_phase(chunks)
    print(f'Map phase complete. Summaries: {len(chunk_summaries)}')

    final_summary = reduce_phase(chunk_summaries)
    return final_summary

with open('long_report.txt') as f:
    doc = f.read()

result = map_reduce_summarize(doc)
print(result)

LangChain MapReduceDocumentsChain

from langchain_openai import ChatOpenAI
from langchain.chains.summarize import load_summarize_chain
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.docstore.document import Document

llm = ChatOpenAI(model='gpt-4o', openai_api_key='sk-...')
splitter = RecursiveCharacterTextSplitter(chunk_size=3000, chunk_overlap=200)

with open('long_report.txt') as f:
    text = f.read()

docs = splitter.create_documents([text])
chain = load_summarize_chain(llm, chain_type='map_reduce')
result = chain.invoke(docs)
print(result['output_text'])

并行化映射阶段

from concurrent.futures import ThreadPoolExecutor, as_completed

def map_phase_parallel(chunks, max_workers=10):
    summaries = [None] * len(chunks)
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(summarize_chunk, chunk): i
            for i, chunk in enumerate(chunks)
        }
        for future in as_completed(futures):
            idx = futures[future]
            summaries[idx] = future.result()
    return summaries

递归归约

def recursive_reduce(summaries, batch_size=10):
    while len(summaries) > 1:
        batches = [
            summaries[i:i + batch_size]
            for i in range(0, len(summaries), batch_size)
        ]
        summaries = [reduce_phase(batch) for batch in batches]
        print(f'Reduced to {len(summaries)} summaries')
    return summaries[0]

保留关键细节

MAP_PROMPT = '''Summarize the following section in 5 sentences.
Preserve all key names, numbers, dates, and conclusions.

Section:
{chunk}'''

精炼链：另一种模式

def refine_summarize(chunks):
    current_summary = summarize_chunk(chunks[0])
    for chunk in chunks[1:]:
        prompt = (
            f'Existing summary:\n{current_summary}\n\n'
            f'New section:\n{chunk}\n\n'
            'Update the summary to incorporate the new section.'
        )
        resp = client.chat.completions.create(
            model='gpt-4o',
            messages=[{'role': 'user', 'content': prompt}]
        )
        current_summary = resp.choices[0].message.content
    return current_summary

成本与词元管理

def map_phase_cheap(chunks):
    # Use mini model for map — cheaper, sufficient for chunk summaries
    return [
        summarize_chunk(c, model='gpt-4o-mini')
        for c in chunks
    ]

def reduce_phase_quality(summaries):
    # Use full model for final synthesis
    return reduce_phase(summaries, model='gpt-4o')

何时使用映射-归约

知识检查

回顾：映射-归约摘要