求助：LangChain/LlamaIndex对接私有LLM处理超512token文档的总结与主题提取

一、长文档处理核心策略 #

处理超过模型token上限的单文档，核心是分块处理+结果聚合，LangChain提供了成熟的工具链：

• 文档分块：用RecursiveCharacterTextSplitter按字符/语义拆分文档，确保每个块的token数低于模型上限（比如设为450，留冗余给prompt）
• 结果聚合：针对总结/主题提取场景，常用两种模式：
  • Map-Reduce：先对每个分块生成独立结果，再将所有结果合并成最终输出
  • Refine：基于前一个块的结果，迭代优化后续块的输出，适合需要上下文连贯的场景

二、LangChain对接私有LLM的包装器实现 #

LangChain的LLM基类允许自定义对接私有模型，只需实现两个核心方法：_call（处理prompt调用）和_identifying_params（返回模型标识参数）。

示例代码（基于私有API的包装器） #

from langchain.llms.base import LLM
from typing import Optional, List, Mapping, Any
import requests

class PrivateLLM(LLM):
    # 私有LLM的API地址
    api_url: str
    # 模型参数，比如温度、最大token数
    temperature: float = 0.1
    max_tokens: int = 512

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        return {
            "api_url": self.api_url,
            "temperature": self.temperature,
            "max_tokens": self.max_tokens
        }

    @property
    def _llm_type(self) -> str:
        return "private-llm"

    def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str:
        # 调用私有LLM的API
        payload = {
            "prompt": prompt,
            "temperature": self.temperature,
            "max_tokens": self.max_tokens,
            "stop": stop or []
        }
        response = requests.post(self.api_url, json=payload)
        response.raise_for_status()
        # 假设API返回的json里有"response"字段
        return response.json()["response"].strip()

三、文档总结+主题提取完整示例 #

结合长文档分块和私有LLM，实现端到端的处理流程：

步骤1：加载并拆分长文档 #

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 加载长文档（示例用字符串模拟，实际可从文件/数据库读取）
long_document = """这里替换成你的长文档内容，长度超过512token..."""

# 初始化分块器，按字符拆分，确保每个块token数不超过450
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=450,
    chunk_overlap=50,  # 块间重叠，保证上下文连贯
    length_function=lambda x: len(x.split()),  # 简单用单词数估算token，也可用tiktoken精确计算
)
chunks = text_splitter.split_text(long_document)

步骤2：初始化私有LLM实例 #

# 替换为你的私有LLM API地址
private_llm = PrivateLLM(api_url="http://your-private-llm-api:port/generate")

步骤3：文档总结（Map-Reduce模式） #

from langchain.chains.summarize import load_summarize_chain
from langchain.docstore.document import Document

# 将分块转为LangChain的Document对象
docs = [Document(page_content=chunk) for chunk in chunks]

# 加载Map-Reduce总结链
summary_chain = load_summarize_chain(
    llm=private_llm,
    chain_type="map_reduce",
    # 自定义Map阶段的prompt（针对单块总结）
    map_prompt="请总结以下文本的核心内容：\n{text}",
    # 自定义Reduce阶段的prompt（合并所有块的总结）
    combine_prompt="请将以下多个文本片段的总结合并成一篇连贯的完整总结：\n{text}"
)

# 执行总结
final_summary = summary_chain.run(docs)
print("文档总结：", final_summary)

步骤4：主题提取（分块提取+聚合） #

from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

# 定义主题提取的prompt模板
topic_prompt = PromptTemplate(
    input_variables=["text"],
    template="请提取以下文本的核心主题，用3-5个关键词或短语概括：\n{text}"
)

# 初始化单块主题提取链
topic_chain = LLMChain(llm=private_llm, prompt=topic_prompt)

# 提取每个块的主题
chunk_topics = [topic_chain.run(chunk) for chunk in chunks]

# 聚合所有块的主题，去重并生成最终主题列表
aggregated_topics = list(set(", ".join(chunk_topics).split(", ")))
print("文档核心主题：", aggregated_topics)

注意事项 #

• token计数：建议用模型对应的tokenizer（比如tiktoken、sentencepiece）精确计算块的token数，避免超过模型上限
• API容错：在_call方法中添加重试、超时处理，提升稳定性
• prompt优化：根据私有LLM的特性调整prompt措辞，提升输出质量

内容的提问来源于stack exchange，提问作者Ja4H3ad