本文介绍在 langchain 0.1.9 环境下，通过预处理文档分块与动态上下文裁剪相结合的方式，有效规避大模型上下文长度超限（如 openai 的 8192 token 限制）问题，确保检索增强对话链稳定运行。
本文介绍在 langchain 0.1.9 环境下，通过预处理文档分块与动态上下文裁剪相结合的方式，有效规避大模型上下文长度超限（如 openai 的 8192 token 限制）问题，确保检索增强对话链稳定运行。
在使用 create_retrieval_chain 构建 RAG（检索增强生成）对话系统时，一个常见但易被忽视的瓶颈是：检索返回的文档片段过大，导致拼接后 Context 超出 LLM 的最大上下文窗口（例如 gpt-3.5-turbo-16k 实际受限于 prompt + context + response 总 token 数）。你遇到的 context_length_exceeded 错误，根源往往不在链路逻辑本身，而在于向量库中存储的文档粒度不合理——将整篇长文档（如 PDF 全文、日志文件）未经切分直接嵌入，导致单次检索返回数 KB 甚至 MB 级别的 page_content，远超 token 预算。
✅ 根本解法：前置文档分块（Chunking），而非链中截断
LangChain 的 create_retrieval_chain 及其配套组件（如 create_stuff_documents_chain）不提供内置的 token 截断机制，也不支持对 context 字段做动态 token 限长（尤其在 0.1.9 版本中，ReduceDocumentsChain 和 ConversationalRetrievalChain 均已弃用或不兼容 ChatOpenAI）。因此，最稳健、高效且符合设计哲学的方案是：在文档入库前完成语义合理、长度可控的分块（chunking）。
推荐使用 CharacterTextSplitter（适用于通用文本）或更智能的 RecursiveCharacterTextSplitter（优先按段落/句子切分）：
from langchain.text_splitter import RecursiveCharacterTextSplitter
def chunk_document(text: str, chunk_size: int = 800, chunk_overlap: int = 80) -> list:
text_splitter = RecursiveCharacterTextSplitter(
separators=["\n\n", "\n", " ", ""], # 优先按空行、换行、空格切分
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len
)
return text_splitter.split_text(text)
# ✅ 正确入库方式：分块后批量嵌入
chunked_docs = [Document(page_content=chunk) for chunk in chunk_document(file_contents)]
vector_ids = qdrant_collection.add_documents(chunked_docs) # 注意：非 add_texts()
⚠️ 关键提示：
- chunk_size 建议设为 500–1000 字符（对应约 150–300 tokens），确保单个 chunk 在嵌入和检索时均轻量；
- chunk_overlap（如 50–150）可缓解语义断裂，提升检索召回质量；
- 使用 add_documents()（传入 Document 对象列表）而非 add_texts()，便于后续保留元数据（如 source、page）。
? 进阶控制：检索后动态 Token 裁剪（可选补充）
若业务场景要求更高灵活性（如允许部分长文档存在、需按需压缩 context），可在 retriever 后添加轻量级 token 截断层。以下是一个兼容 0.1.9 的 Runnable 封装示例（基于 tiktoken）：
import tiktoken
def truncate_context_by_tokens(docs, max_tokens: int = 2000, model_name: str = "gpt-3.5-turbo"):
encoder = tiktoken.encoding_for_model(model_name)
truncated_docs = []
total_tokens = 0
for doc in docs:
content = doc.page_content
tokens = len(encoder.encode(content))
if total_tokens + tokens 0:
truncated_text = encoder.decode(encoder.encode(content)[:remaining])
truncated_docs.append(Document(page_content=truncated_text, metadata=doc.metadata))
break
return truncated_docs
# 在 retrieval_chain 中集成
retriever = existing_vector_store.as_retriever(search_kwargs={"k": 5})
# 替换原 retriever：先检索，再按 token 截断
safe_retriever = lambda query: truncate_context_by_tokens(retriever.invoke(query), max_tokens=1800)
retrieval_chain = RunnablePassthrough.assign(
context=lambda x: safe_retriever(x["messages"][-1].content),
).assign(
answer=document_chain,
)
? 总结与最佳实践
| 措施 | 作用 | 推荐程度 |
|---|---|---|
| 入库前分块（必做） | 从源头控制单个向量单元大小，提升检索精度与效率 | ⭐⭐⭐⭐⭐ |
| 设置 search_kwargs={"k": N} | 限制检索返回文档数量（如 k=3），避免冗余 | ⭐⭐⭐⭐ |
| Prompt 中明确指令 | 在 system prompt 加入 "请严格基于以下上下文回答，总输出不超过 500 字" 等约束 | ⭐⭐⭐ |
| 链内 token 截断（可选） | 应对不可控长文档或动态预算场景，增加复杂度 | ⭐⭐ |
? 最终建议：不要试图在 create_retrieval_chain 中“打补丁式”限长，而应将分块作为数据治理标准流程。这不仅解决 token 超限，更能显著提升检索相关性与响应速度——因为小而精的 chunk 比大而泛的全文更易匹配用户 query。
通过以上方法，你无需升级 LangChain 版本，即可在现有技术栈上构建鲁棒、可扩展的 RAG 对话系统。
