ภาพรวม Python SDK

ชุดพัฒนาซอฟต์แวร์ (SDK) NextPDF สำหรับ Python ออกแบบมาสำหรับแอปพลิเคชัน Python ที่ต้องการแยกข้อมูล Portable Document Format (PDF) พร้อมข้อมูลแหล่งที่มา โดยจะคืนค่าเป็นบล็อกแบบมีโครงสร้างพร้อมจุดยึดสำหรับการอ้างอิง รวมถึงดัชนีหน้า ค่าความเชื่อมั่น กล่องขอบเขต (bounding box) แบบเลือกได้ และตัวระบุโหนดเชิงความหมายเมื่อ PDF ต้นทางเปิดเผยโครงสร้างนั้น

ใช้ SDK นี้เมื่อไปป์ไลน์ของคุณต้องตอบคำถามอย่าง “ข้อความนี้มาจากหน้าใด” “ตารางใดสนับสนุนค่านี้” หรือ “อะไรเปลี่ยนแปลงไประหว่าง PDF สองไฟล์นี้” โดยไม่ปฏิบัติต่อผลลัพธ์การแยกข้อมูลจาก PDF เสมือนเป็นข้อความธรรมดาที่ไม่มีแหล่งที่มา

สิ่งที่ SDK มอบให้

ไคลเอนต์ NextPDF แบบซิงโครนัสสำหรับสคริปต์ งานแบบแบตช์ และโน้ตบุ๊ก
ไคลเอนต์ AsyncNextPDF แบบอะซิงโครนัสสำหรับ asyncio FastAPI และรันไทม์อะซิงโครนัสอื่น ๆ
อินเทอร์เฟซบรรทัดคำสั่ง (CLI) nextpdf สำหรับการแยกข้อมูลแบบครั้งเดียวจากเส้นทางไฟล์หรืออินพุตมาตรฐาน และเขียนผลลัพธ์ไปยังเอาต์พุตมาตรฐานหรือไฟล์
เซิร์ฟเวอร์ Model Context Protocol (MCP) แบบเลือกได้ เพื่อให้เอเจนต์ปัญญาประดิษฐ์ (AI) เรียกใช้เครื่องมือแยกข้อมูล PDF ได้โดยตรง
แบ็กเอนด์ระยะไกลสำหรับการใช้งานจริงร่วมกับ NextPDF Connect
แบ็กเอนด์ในเครื่องสำหรับการแยกข้อมูลแบบออฟไลน์โดยใช้ไลบรารีเท่านั้นผ่าน pypdf

ตัวเลือกแบ็กเอนด์

แบ็กเอนด์ระยะไกลจะส่งไบต์ของ PDF ไปยังเซิร์ฟเวอร์ NextPDF Connect นี่คือเส้นทางที่แนะนำสำหรับการใช้งานจริง เนื่องจากรวมศูนย์พฤติกรรมการแยกข้อมูล การพิสูจน์ตัวตน โควตา และการควบคุมเชิงปฏิบัติการไว้ในที่เดียว

แบ็กเอนด์ในเครื่องทำงานภายในกระบวนการ Python และอ่าน PDF ผ่าน pypdf เหมาะสำหรับการพัฒนาแบบออฟไลน์และ PDF ที่มีแท็ก แต่ไม่สามารถให้กล่องขอบเขตที่แม่นยำได้ และใช้การแยกข้อมูลระดับย่อหน้าแบบฮิวริสติกสำหรับ PDF ที่ไม่มีแท็ก แบ็กเอนด์ในเครื่องใช้ได้เฉพาะในฐานะไลบรารี โดยต้องฉีด LocalBackend เข้าไปใน AsyncNextPDF เพื่อใช้งาน ส่วน CLI nextpdf และเซิร์ฟเวอร์ MCP ไม่สามารถใช้แบ็กเอนด์นี้ได้ ดู เมทริกซ์ตัวเลือกแบ็กเอนด์ สำหรับการเปรียบเทียบฉบับเต็ม

ข้อจำกัด

SDK ไม่ทำการรู้จำอักขระด้วยแสง (OCR) PDF ที่สแกนหรือมีเฉพาะรูปภาพจำเป็นต้องผ่านขั้นตอน OCR ก่อนที่ NextPDF จะแยกข้อความที่ฝังอยู่ได้ เค้าโครงที่ซับซ้อน ข้อความที่ซ้อนทับกัน และโปรแกรมสร้าง PDF ที่ผิดปกติอาจลดคุณภาพของการแยกข้อมูลได้เช่นกัน

CLI nextpdf ใช้งานได้แบบระยะไกลเท่านั้นและไม่ใช่อินเทอร์เฟซแบบสตรีม แต่ละคำสั่งจะอ่าน PDF ทั้งไฟล์เข้าสู่หน่วยความจำ (จากเส้นทางไฟล์หรืออินพุตมาตรฐาน) ส่งไปยังเซิร์ฟเวอร์ NextPDF Connect สร้างผลลัพธ์ทั้งหมดในหน่วยความจำ และซีเรียลไลซ์ด้วยการเขียนเพียงครั้งเดียว คุณสามารถเปลี่ยนเส้นทางเอาต์พุตนั้นไปยังไฟล์ด้วย --output (หรือ -o) หรือไปยังเอาต์พุตมาตรฐาน แต่ผลลัพธ์จะถูกบัฟเฟอร์ไว้ทั้งหมด ไม่ได้สร้างขึ้นทีละส่วน CLI ไม่สามารถใช้แบ็กเอนด์ในเครื่อง pypdf ได้

การเลือกไคลเอนต์: ซิงค์เทียบกับอะซิงค์

ไคลเอนต์ทั้งสองใช้เนมสเปซเมท็อด ast ร่วมกันสำหรับการดำเนินการกับโครงสร้างไวยากรณ์เชิงนามธรรม (AST) และคืนค่าโมเดล Pydantic เดียวกัน ความแตกต่างมีเพียงโมเดลการทำงานพร้อมกันเท่านั้น

บริบทของคุณ	ใช้	เหตุผล
สคริปต์และงานแบบแบตช์	`NextPDF` (ซิงค์)	โฟลว์การควบคุมเชิงเส้น ไม่มี event loop ให้ต้องจัดการ
โน้ตบุ๊ก Jupyter	`NextPDF` (ซิงค์)	`run_sync` จะตรวจหา event loop ที่กำลังทำงานอยู่และส่งงานไปยังเทรดผู้ทำงาน เพื่อให้การเรียกแบบบล็อกทำงานได้ภายในเซลล์
เครื่องมือ `nextpdf` CLI	`NextPDF` (ซิงค์ ภายใน)	CLI สร้างไคลเอนต์ซิงค์ให้คุณ
`asyncio` บริการ	`AsyncNextPDF`	แบบเนทีฟ `await` ไม่มีการส่งต่อไปยังเทรด
เฟรมเวิร์ก FastAPI, Starlette, Asynchronous Server Gateway Interface (ASGI)	`AsyncNextPDF`	ใช้ event loop ของคำขอและพูลการเชื่อมต่อชุดเดียวกันร่วมกัน
การกระจายงานแบบมีการทำงานพร้อมกันสูง	`AsyncNextPDF`	รันการแยกข้อมูลจำนวนมากพร้อมกันด้วย `asyncio.gather` บนไคลเอนต์ที่ใช้พูลเดียว

NextPDF ห่อหุ้ม AsyncNextPDF ไว้ภายใน และรันการเรียกแต่ละครั้งผ่าน run_sync เมื่ออยู่ใน event loop ที่กำลังทำงานอยู่ เช่นในโน้ตบุ๊ก run_sync จะส่งโครูทีนไปยังเทรดผู้ทำงานเดี่ยวที่มี loop ของตนเอง คุณจึงไม่พบข้อผิดพลาดจากการซ้อน asyncio.run ในบริการ asyncio หรือ ASGI ให้เรียก AsyncNextPDF โดยตรง เพื่อหลีกเลี่ยงต้นทุนจากการส่งต่อไปยังเทรดนั้นในทุกการเรียก

ไคลเอนต์อะซิงค์เป็นเจ้าของ httpx.AsyncClient สำหรับการพูลการเชื่อมต่อ ดังนั้นให้ใช้อินสแตนซ์ AsyncNextPDF เดียวซ้ำและปิดเพียงครั้งเดียว ไคลเอนต์ NextPDF แบบซิงค์ไม่เปิดเผยเมท็อด close() สำหรับภาระงานอะซิงค์ที่ทำงานต่อเนื่องเป็นเวลานาน ให้เลือกใช้ AsyncNextPDF และจัดการวงจรชีวิตอย่างชัดเจน (ดู โมเดลการดำเนินงานในระดับการใช้งานจริง)

เมทริกซ์ตัวเลือกแบ็กเอนด์

แบ็กเอนด์จะใช้งานโปรโตคอล PdfBackend แบ็กเอนด์ระยะไกล (RemoteBackend) จะถูกเลือกโดยอัตโนมัติเมื่อคุณส่ง base_url และ api_key ส่วนแบ็กเอนด์ในเครื่อง (LocalBackend) ต้องฉีดเข้าไปอย่างชัดเจนผ่านพารามิเตอร์ backend= ของ AsyncNextPDF โดยจะไม่ส่งออกจากแพ็กเกจระดับบนสุด nextpdf และไม่สามารถเข้าถึงได้จาก CLI หรือเซิร์ฟเวอร์ MCP

ความสามารถ	ระยะไกล (`RemoteBackend`)	ในเครื่อง (`LocalBackend`)
เลือกโดย	`base_url` + `api_key`	`AsyncNextPDF(backend=LocalBackend(...))`
เครือข่าย	NextPDF Connect ผ่าน Hypertext Transfer Protocol Secure (HTTPS)	ไม่มี ทำงานภายในกระบวนการ
การพิสูจน์ตัวตน โควตา การวัดปริมาณการใช้งาน	รวมศูนย์ไว้ที่เซิร์ฟเวอร์	ไม่มี
ความสามารถในการสังเกตการณ์และการควบคุมเชิงปฏิบัติการ	ฝั่งเซิร์ฟเวอร์	ไม่มี
การแยกข้อมูล PDF ที่มีแท็ก (StructTree)	ได้	ได้
การแยกข้อมูล PDF ที่ไม่มีแท็ก	เอนจินบนเซิร์ฟเวอร์	การแบ่งย่อหน้าแบบฮิวริสติก ค่าความเชื่อมั่น `0.5`
กล่องขอบเขต (bounding box)	ได้ (เมื่อเซิร์ฟเวอร์ให้มา)	ไม่ได้ (`bbox` เป็น `None`)
การแยกข้อมูลตารางบน PDF ที่ไม่มีแท็ก	เอนจินบนเซิร์ฟเวอร์	ไม่คืนค่าตารางใด ๆ
เข้าถึงได้จาก CLI / เซิร์ฟเวอร์ MCP	ได้	ไม่ได้ (ใช้ได้เฉพาะไลบรารี)
แนะนำสำหรับ	การใช้งานจริง	การพัฒนาแบบออฟไลน์ การทดสอบกับ PDF ที่มีแท็ก

ใช้แบ็กเอนด์ระยะไกลสำหรับการใช้งานจริง เนื่องจากเป็นเส้นทางเดียวที่มีการพิสูจน์ตัวตน การบังคับใช้โควตา การวัดปริมาณการใช้งาน และความสามารถในการสังเกตการณ์แบบรวมศูนย์ ใช้แบ็กเอนด์ในเครื่องสำหรับการพัฒนาแบบออฟไลน์และการทดสอบกับ PDF ที่มีแท็ก โดยยอมรับว่าผลลัพธ์เป็นแบบฮิวริสติก ไม่มีกล่องขอบเขต และไม่มีตารางสำหรับอินพุตที่ไม่มีแท็ก

"""Inject the local backend for offline, library-only extraction."""

from nextpdf import AsyncNextPDF
from nextpdf.backends.local import LocalBackend


async def extract_offline(pdf_bytes: bytes) -> None:
    """Extract cited text without a NextPDF Connect server."""
    async with AsyncNextPDF(backend=LocalBackend()) as client:
        blocks = await client.ast.extract_cited_text(pdf_bytes)
        for block in blocks:
            # Heuristic blocks on untagged PDFs report confidence 0.5.
            print(block.citation.confidence, block.text)

โมเดลการดำเนินงานในระดับการใช้งานจริง

ในการใช้งานจริง ให้รันแบ็กเอนด์ระยะไกลกับ NextPDF Connect รูปแบบด้านล่างครอบคลุมการใช้ไคลเอนต์ซ้ำ การจัดการข้อผิดพลาด การลองใหม่ การจัดการโควตา และการหมดเวลา สัญลักษณ์ทุกตัวที่ใช้ในที่นี้มีอยู่จริงใน SDK SDK จะไม่ลองใหม่ให้คุณโดยอัตโนมัติ ดังนั้นลูปการลองใหม่จึงเป็นความรับผิดชอบของคุณ

ใช้ไคลเอนต์ซ้ำและพูลการเชื่อมต่อ

RemoteBackend จะคง httpx.AsyncClient ที่ทำงานต่อเนื่องตัวเดียวไว้สำหรับการพูลการเชื่อมต่อ สร้าง AsyncNextPDF เพียงครั้งเดียว ใช้ร่วมกันในทุกคำขอ และปิดเมื่อปิดระบบ อย่าสร้างไคลเอนต์ใหม่ในแต่ละคำขอ

"""Reuse one pooled async client for the lifetime of the process."""

import asyncio
import os
from pathlib import Path

from nextpdf import AsyncNextPDF


async def main() -> None:
    """Run several extractions over a single pooled client."""
    base_url = os.environ["NEXTPDF_BASE_URL"]
    # Treat the API key as a secret; read it from the environment, never hard-code it.
    api_key = os.environ["NEXTPDF_API_KEY"]

    async with AsyncNextPDF(base_url=base_url, api_key=api_key) as client:
        pdf_paths = (Path("a.pdf"), Path("b.pdf"), Path("c.pdf"))
        tasks = [
            client.ast.get_document_ast(path.read_bytes())
            for path in pdf_paths
        ]
        documents = await asyncio.gather(*tasks)
        for document in documents:
            print(document.page_count, document.estimated_tokens)


if __name__ == "__main__":
    asyncio.run(main())

ตัวจัดการบริบทแบบอะซิงค์จะเรียก close() เมื่อออกจากบริบท ซึ่งจะปิดทรานสปอร์ตที่อยู่เบื้องล่าง หากไม่มีตัวจัดการบริบท ให้เรียก await client.close() ด้วยตนเอง

จัดการข้อผิดพลาดด้วยลำดับชั้นของข้อยกเว้น

SDK จะยกข้อยกเว้นที่มีลำดับชั้นและกำหนดชนิดไว้ ข้อผิดพลาดทั้งหมดสืบทอดมาจาก NextPDFError ความล้มเหลวระดับ Hypertext Transfer Protocol (HTTP) สืบทอดมาจาก NextPDFAPIError และมี status_code ติดมาด้วย ให้ดักจับชนิดเฉพาะที่คุณสามารถจัดการได้ และใช้ชนิดฐานเป็นทางสำรอง

ข้อยกเว้น	ยกขึ้นเมื่อ	แอตทริบิวต์สำคัญ
`NextPDFError`	ชนิดฐานสำหรับทุกข้อผิดพลาดของ SDK	`status_code`
`NextPDFAPIError`	ข้อผิดพลาด HTTP ใด ๆ จากเซิร์ฟเวอร์	`status_code`, `error_code`
`NextPDFLicenseError`	HTTP 402 คุณสมบัตินี้ต้องใช้ระดับเซิร์ฟเวอร์ที่สูงขึ้น	`status_code` (402)
`QuotaExceededError`	HTTP 429 เกินขีดจำกัดอัตราหรือโควตา	`retry_after`
`AstNoStructTreeError`	HTTP 422 PDF ที่ไม่มีแท็กโดยปิดโหมดฮิวริสติก	`status_code` (422)
`AstBuildTimeoutError`	HTTP 504 การสร้าง AST หมดเวลา	`status_code` (504)

"""Map SDK exceptions to caller-facing outcomes."""

from nextpdf import (
    AstBuildTimeoutError,
    AstNoStructTreeError,
    AsyncNextPDF,
    NextPDFAPIError,
    NextPDFError,
    NextPDFLicenseError,
    QuotaExceededError,
)


async def safe_extract(client: AsyncNextPDF, pdf_bytes: bytes) -> str:
    """Extract text, translating known failures into a stable status string."""
    try:
        blocks = await client.ast.extract_cited_text(pdf_bytes)
    except QuotaExceededError as exc:
        # exc.retry_after holds the server Retry-After value in seconds, or None.
        return f"rate-limited; retry after {exc.retry_after}s"
    except NextPDFLicenseError:
        return "feature requires a higher server tier"
    except AstNoStructTreeError:
        return "untagged PDF; enable heuristic mode or use a tagged PDF"
    except AstBuildTimeoutError:
        return "build timed out; reduce the page range"
    except NextPDFAPIError as exc:
        return f"server error (status {exc.status_code})"
    except NextPDFError:
        return "extraction failed"
    return "\n".join(block.text for block in blocks)

ลองใหม่เมื่อเกิดความล้มเหลวชั่วคราวด้วยการหน่วงเวลา

SDK จะไม่ลองใหม่โดยอัตโนมัติ ให้ห่อหุ้มการเรียกไว้ในลูปของคุณเองที่ลองใหม่เมื่อเกิดความล้มเหลว HTTP ชั่วคราว และเคารพค่า Retry-After ของเซิร์ฟเวอร์ ซึ่ง QuotaExceededError เปิดเผยเป็น retry_after (จำนวนวินาทีแบบจำนวนเต็ม หรือ None) ใช้การหน่วงเวลาแบบเอ็กซ์โพเนนเชียลสำหรับสถานะชั่วคราวอื่น ๆ และอย่าลองใหม่กับ NextPDFLicenseError

"""Retry transient failures with exponential backoff and Retry-After support."""

import asyncio
from collections.abc import Awaitable, Callable
from typing import TypeVar

from nextpdf import NextPDFAPIError, QuotaExceededError

_RETRYABLE_STATUS = frozenset({500, 502, 503, 504})

_T = TypeVar("_T")


async def with_retry(
    coro_factory: Callable[[], Awaitable[_T]],
    *,
    max_attempts: int = 4,
) -> _T:
    """Call coro_factory() with bounded retries on transient server errors.

    Args:
        coro_factory: A zero-argument callable returning a fresh awaitable.
        max_attempts: Maximum number of attempts before giving up.

    Returns:
        The awaited result of the first successful attempt.

    Raises:
        NextPDFAPIError: When all attempts fail or the error is not retryable.
    """
    delay = 1.0
    for attempt in range(1, max_attempts + 1):
        try:
            return await coro_factory()
        except QuotaExceededError as exc:
            if attempt == max_attempts:
                raise
            await asyncio.sleep(exc.retry_after if exc.retry_after is not None else delay)
            delay *= 2.0
        except NextPDFAPIError as exc:
            if attempt == max_attempts or exc.status_code not in _RETRYABLE_STATUS:
                raise
            await asyncio.sleep(delay)
            delay *= 2.0
    raise RuntimeError("unreachable")

จัดการโควตา ขีดจำกัดอัตรา และการหมดเวลา

การบังคับใช้โควตาและขีดจำกัดอัตราอยู่ที่เซิร์ฟเวอร์ เมื่อเกิด HTTP 429 SDK จะยก QuotaExceededError และแยกวิเคราะห์ส่วนหัว Retry-After เป็น retry_after แบ็กเอนด์ระยะไกลยังเปิดเผยส่วนหัว X-RateLimit-* ในการตอบสนองของการเรนเดอร์ด้วย คุณจึงสามารถควบคุมปริมาณงานล่วงหน้าได้ก่อนถึงขีดจำกัดสูงสุด

การหมดเวลาของคำขอใช้ค่าเริ่มต้นแบบตายตัวที่ 60 วินาทีโดยรวม พร้อมการหมดเวลาการเชื่อมต่อ 10 วินาที (httpx.Timeout(60.0, connect=10.0)) หากต้องการจำกัดการสร้าง AST ที่ใช้เวลานาน ให้จำกัดขอบเขตงานด้วย page_range_start, page_range_end หรือ token_budget แทนที่จะพึ่งพาการหมดเวลาเพียงอย่างเดียว การสร้างที่ใช้เวลานานเกินไปจะคืนค่า AstBuildTimeoutError (HTTP 504)

ตัวอย่างสถาปัตยกรรม

งานแบบแบตช์

ตัวประมวลผลแบบแบตช์จะอ่าน PDF แยกข้อความที่อ้างอิงแหล่งที่มา และเขียนผลลัพธ์แบบมีโครงสร้าง ใช้ไคลเอนต์ที่ใช้พูลเดียวซ้ำ จำกัดการทำงานพร้อมกันด้วยเซมาฟอร์ และใช้ตัวช่วยลองใหม่ข้างต้น

"""Batch-extract a directory of PDFs over one pooled async client."""

import asyncio
import os
from pathlib import Path

from nextpdf import AsyncNextPDF


async def run_batch(input_dir: Path, concurrency: int = 8) -> None:
    """Extract cited text for every PDF in input_dir, bounded by concurrency."""
    semaphore = asyncio.Semaphore(concurrency)

    async def worker(client: AsyncNextPDF, path: Path) -> None:
        async with semaphore:
            blocks = await client.ast.extract_cited_text(path.read_bytes())
            out = path.with_suffix(".txt")
            out.write_text("\n".join(b.text for b in blocks), encoding="utf-8")

    async with AsyncNextPDF(
        base_url=os.environ["NEXTPDF_BASE_URL"],
        api_key=os.environ["NEXTPDF_API_KEY"],
    ) as client:
        await asyncio.gather(*(worker(client, p) for p in input_dir.glob("*.pdf")))

บริการเว็บ

บริการ FastAPI จะใช้ AsyncNextPDF ตัวเดียวร่วมกันในทุกคำขอตลอดอายุการทำงานของแอปพลิเคชัน ดังนั้นทุกคำขอจึงใช้พูลการเชื่อมต่อซ้ำ อ่านข้อมูลรับรองจากสภาพแวดล้อมและปฏิบัติต่อ API key เสมือนเป็นความลับ

"""FastAPI service that shares one pooled NextPDF client across requests."""

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, UploadFile

from nextpdf import AsyncNextPDF


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Create the pooled client on startup and close it on shutdown."""
    app.state.client = AsyncNextPDF(
        base_url=os.environ["NEXTPDF_BASE_URL"],
        api_key=os.environ["NEXTPDF_API_KEY"],
    )
    try:
        yield
    finally:
        await app.state.client.close()


app = FastAPI(lifespan=lifespan)


@app.post("/extract")
async def extract(file: UploadFile) -> dict[str, list[str]]:
    """Return cited text blocks for an uploaded PDF."""
    pdf_bytes = await file.read()
    blocks = await app.state.client.ast.extract_cited_text(pdf_bytes)
    return {"text": [block.text for block in blocks]}

เครื่องมือสำหรับเอเจนต์

สำหรับเอเจนต์ AI ให้รันเซิร์ฟเวอร์ MCP โดยจะเปิดเผยเครื่องมือ PDF (ตัวอย่างเช่น nextpdf_extract_text, nextpdf_extract_tables, nextpdf_get_ast, nextpdf_info, nextpdf_search, nextpdf_get_outline, nextpdf_diff และ nextpdf_health) ผ่านอินพุตและเอาต์พุตมาตรฐาน เซิร์ฟเวอร์จะอ่าน NEXTPDF_BASE_URL และ NEXTPDF_API_KEY จากสภาพแวดล้อม จึงรองรับด้วยแบ็กเอนด์ระยะไกล เช่นเดียวกับ CLI และไม่สามารถใช้แบ็กเอนด์ในเครื่องได้ ติดตั้งส่วนเสริมแบบเลือกได้และรันโมดูล

pip install "nextpdf[mcp]"
python -m nextpdf.mcp

ดู เซิร์ฟเวอร์ Python MCP สำหรับคำแนะนำการผสานรวมเอเจนต์ Python CLI สำหรับการใช้งานในเทอร์มินัล และ เอกสารอ้างอิง Python API สำหรับพื้นผิวทั้งหมดของไคลเอนต์ โมเดล และข้อยกเว้น