项目简介
Kreuzberg
是一个现代 Python 库,用于从文档中提取文本,旨在简洁高效。它提供统一的异步接口,用于从包括 PDF、图片、办公文档等多种文件格式中提取文本。
特点
Kreuzberg
是为了解决 RAG(检索增强生成)应用中的文本提取需求而创建的,但它适用于任何文本提取用例。与许多需要 API 调用或复杂设置的商用解决方案不同,库尔茨贝格专注于本地处理,依赖性最小。
功能
安装
-
安装 Python 包
2. 安装系统依赖项
Kreuzberg
需要两个系统级依赖项:
请使用各自的安装指南进行安装。
架构
Kreuzberg
被设计为一个在现有开源工具之上的高级异步抽象。它集成了:
-
PDF 处理:
-
pdfium2
为可搜索的 PDF 文件
-
Tesseract OCR 用于扫描内容
-
文档转换:
-
文本处理:
支持格式
文档格式
标记和文本格式
数据和研究成果格式
-
Excel 电子表格(
.xlsx
)
-
CSV(
.csv
)和 TSV(
.tsv
)文件
-
Jupyter Notebooks (
.ipynb
)
-
BibTeX(
.bib
)和 BibLaTeX(
.bib
)
-
CSL-JSON (
.json
)
-
EndNote XML (
.xml
)
-
RIS (
.ris
)
-
JATS XML(
.xml
)
图片格式
-
JPEG (
.jpg
,
.jpeg
,
.pjpeg
)
-
PNG (
.png
)
-
TIFF (
.tiff
,
.tif
)
-
BMP (
.bmp
)
-
GIF (
.gif
)
-
WebP(
.webp
)
-
JPEG 2000(
.jp2
,
.jpx
,
.jpm
,
.mj2
)
-
便携式 Anymap(
.pnm
)
-
便携式位图(
.pbm
)
-
便携式灰度图(
.pgm
)
-
便携式像素图(
.ppm
)
使用
库克斯堡提供了一个简单、以异步为主的文本提取 API。该库导出两个主要功能:
快速开始
from pathlib import Path
from kreuzberg import extract_file, extract_bytes
async def extract_document():
pdf_result = await extract_file("document.pdf")
print(f"PDF text: {pdf_result.content}")
img_result = await extract_file("scan.png")
print(f"Image text: {img_result.content}")
docx_result = await extract_file(Path("document.docx"))
print(f"Word text: {docx_result.content}")
处理上传的文件
from kreuzberg import extract_bytes
async def process_upload(file_content: bytes, mime_type: str):
"""Process uploaded file content with known MIME type."""
result = await extract_bytes(file_content, mime_type=mime_type)
return result.content
async def handle_uploads():
pdf_result = await extract_bytes(pdf_bytes, mime_type="application/pdf")
img_result = await extract_bytes(image_bytes, mime_type="image/jpeg")
docx_result = await extract_bytes(docx_bytes,
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document")
高级功能
PDF 处理选项
from kreuzberg import extract_file
async def process_pdf():
result = await extract_file("document.pdf", force_ocr=True)
scanned = await extract_file("scanned.pdf")
提取结果对象
所有提取函数返回一个包含:
from kreuzberg import ExtractionResult
async def process_document(path: str) -> tuple[str, str]:
result: ExtractionResult = await extract_file(path)
print(f"Content: {result.content}")
print(f"Format: {result.mime_type}")
content, mime_type = await extract_file(path)
return content, mime_type