如何在LangChain加载器中传入文档二进制数据替代文件路径？

1. PDF文档：替代PyPDFLoader用二进制加载 #

PyPDFLoader本身仅支持传入文件路径，无法直接接收二进制数据，但可以通过底层PDF处理库读取二进制内容，手动构建LangChain的Document对象：

from langchain.docstore.document import Document
from PyPDF2 import PdfReader
import io

# 假设pdf_bytes是你的PDF二进制数据（可从网络请求、内存对象等获取）
pdf_bytes = b"..."

# 将二进制数据转为类文件对象
pdf_file = io.BytesIO(pdf_bytes)
reader = PdfReader(pdf_file)

# 提取每页内容并生成Document列表
docs = []
for page_num, page in enumerate(reader.pages):
    text = page.extract_text()
    docs.append(
        Document(
            page_content=text,
            metadata={"page": page_num + 1, "source": "binary_pdf"}
        )
    )

# 如需拆分文本，使用RecursiveCharacterTextSplitter
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
split_docs = splitter.split_documents(docs)

2. JSON文档：读取二进制数据并处理 #

JSON二进制数据本质是UTF-8编码的字节流，先解码为字符串再解析，按需转为Document：

import json
from langchain.docstore.document import Document

# 假设json_bytes是JSON文件的二进制数据
json_bytes = b'{"key1": "value1", "key2": "value2"}'

# 解码并解析JSON内容
json_str = json_bytes.decode("utf-8")
json_data = json.loads(json_str)

# 格式化内容并生成Document
doc_content = json.dumps(json_data, indent=2)
doc = Document(page_content=doc_content, metadata={"source": "binary_json"})

# 若为JSON数组，可循环生成多个Document
# docs = [Document(page_content=json.dumps(item), metadata={"index": i}) for i, item in enumerate(json_data)]

3. Excel文档：读取二进制流 #

可通过pandas或openpyxl读取二进制流，提取内容后转为Document：

方法1：用pandas快速处理 #

import pandas as pd
from langchain.docstore.document import Document
import io

# 假设excel_bytes是Excel文件的二进制数据
excel_bytes = b"..."

# 读取二进制流为DataFrame并转为CSV格式文本
df = pd.read_excel(io.BytesIO(excel_bytes))
doc_content = df.to_csv(index=False)
doc = Document(page_content=doc_content, metadata={"source": "binary_excel", "sheet_name": df.index.name})

方法2：用openpyxl逐页读取 #

from openpyxl import load_workbook
from langchain.docstore.document import Document
import io

excel_bytes = b"..."
wb = load_workbook(filename=io.BytesIO(excel_bytes), read_only=True)

docs = []
for sheet_name in wb.sheetnames:
    sheet = wb[sheet_name]
    # 提取表格内容为文本（每行单元格用逗号分隔）
    content = "\n".join([",".join([str(cell.value) for cell in row]) for row in sheet.iter_rows(values_only=True)])
    docs.append(
        Document(
            page_content=content,
            metadata={"source": "binary_excel", "sheet": sheet_name}
        )
    )

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