Deploy a Document AI Service
Build a production API service that extracts data from freight Bills of Lading.
Introduction
30 Second Summary
Every day, millions of freight shipments move across the world accompanied by Bills of Lading. These dense, multi-column documents contain critical logistics data that companies still extract by hand or through brittle template-based tools.
In this project, you will deploy a production-grade Document AI service that processes Bills of Lading using state-of-the-art layout detection and Vision Language Models, then build a custom extraction microservice that returns structured logistics data as typed JSON.
What You'll Build
A local document intelligence system consisting of a production API service with a professional web UI, REST API scripts for pipeline comparison, and a custom FastAPI extraction microservice for freight logistics data.
By the end of this project, you'll have:
- A running Docling Serve instance with a professional web UI for uploading and processing freight documents through both standard and VLM pipelines.
- API scripts that process the same document through both a layout model pipeline (RT-DETR + OCR) and a Vision Language Model (GraniteDocling) to compare results programmatically.
- A custom FastAPI extraction service that accepts Bill of Lading uploads, calls Docling Serve for processing, and returns structured logistics fields (shipper, consignee, commodities) in typed Pydantic models with confidence scoring.
- Secret Mission: Add async batch processing that handles multiple documents concurrently with graceful error recovery.
Are there any prerequisites?
You should have basic Python familiarity (variables, functions, running scripts). You also need an Apple Silicon Mac with Python 3.10+ and Cursor already installed.
No cloud accounts or paid APIs are needed. Everything in this project runs locally on your machine at zero cost.
Before We Start
Before We Start...
Deploy Docling Serve with the Production UI
Production document AI systems need a stable API layer between the processing engine and any downstream services. Docling Serve is the official production deployment tool for Docling. It provides a REST API, interactive Swagger docs, and a professional web UI out of the box.
In this step, you will create a reusable launch script for Docling Serve, verify that all three endpoints are running, and upload your sample Bill of Lading through the web UI to confirm the processing pipeline works end-to-end.
In this step, get ready to:
- Create a launch script that starts Docling Serve with the production UI enabled.
- Verify the API, Swagger docs, and web UI are all accessible.
- Upload and process a sample Bill of Lading through the UI.
Create the server launch script
A launch script keeps the startup command repeatable and documents the endpoints for anyone on your team. Instead of memorizing the command each time, you run a single script.
- Create a file called run_server.sh in your project folder by pasting the following code:
#!/bin/bash
# Launch Docling Serve with UI enabled
echo "Starting Docling Serve..."
echo "API: http://127.0.0.1:5001"
echo "Docs: http://127.0.0.1:5001/docs"
echo "UI: http://127.0.0.1:5001/ui"
echo ""
docling-serve run --enable-uiWhat does this script do?
- The echo lines print the three endpoint URLs to your terminal so you always know where to find the API, docs, and UI.
- docling-serve run --enable-ui starts the Docling Serve process with the web UI enabled at the /ui path.
- Save run_server.sh.
- Make the script executable by running this command in your terminal:
chmod +x run_server.sh- Confirm the file exists and is executable by running:
ls -la run_server.shYou should see -rwxr-xr-x (or similar) at the start of the line, confirming execute permissions are set.
Not seeing the execute permission?
Make sure you ran chmod +x run_server.sh in the same directory where the file lives. If you see a "No such file or directory" message, confirm you are inside your project folder.
Help me troubleshoot file permissions
✔️ Awesome, I've got everything!
Great. Double check you have saved run_server.sh and it has execute permissions.
ⓧ I'd like to double check the full code
Here is the complete run_server.sh file:
#!/bin/bash
# Launch Docling Serve with UI enabled
echo "Starting Docling Serve..."
echo "API: http://127.0.0.1:5001"
echo "Docs: http://127.0.0.1:5001/docs"
echo "UI: http://127.0.0.1:5001/ui"
echo ""
docling-serve run --enable-uiLaunch the server and verify the endpoints
With the script ready, you can now start Docling Serve and confirm all three endpoints respond.
- Start Docling Serve by running:
./run_server.shFirst launch downloads model weights
The very first time you run Docling Serve, it downloads layout detection and OCR model weights from Hugging Face. This is a one-time download that may take a few minutes depending on your connection speed.
Subsequent launches will be much faster because the models are cached locally.
Once you see the server output in your terminal, the API is ready.
- Open your browser and navigate to http://127.0.0.1:5001/docs to view the Swagger API documentation.
- Navigate to http://127.0.0.1:5001/ui to view the Docling Serve web UI.
- Navigate to http://127.0.0.1:5001 to confirm the base API responds.
You should see the Swagger interface at /docs listing all available endpoints, and a professional web UI at /ui with tabs for URL and file conversion.
Connection refused in the browser?
Check that ./run_server.sh is still running in your terminal. The server process must stay active for the endpoints to respond.
If you see an error about port 5001 already being in use, another process is occupying that port. Stop the other process or check for a previous Docling Serve instance.
Help me troubleshoot the connection
Upload a sample Bill of Lading through the UI
Now you will use the web UI to process your sample freight document. This confirms the full pipeline works before you build automation on top of it.
- In your browser, make sure you are on http://127.0.0.1:5001/ui.
- Click the Convert File tab at the top of the UI.
- Drag your sample Bill of Lading PDF into the Upload File area, or click the area to browse for the file.
- Click Process File.
What should I see?
After a few seconds of processing, the UI displays the extracted content in multiple output tabs. You will see the document converted to Markdown and JSON formats, with tables, text blocks, and layout regions identified.
The Markdown output shows the document content in a readable format. The JSON output contains the full structural representation including bounding boxes and element types.
Processing seems stuck or errors out?
The first document processed after launch takes longer because the layout model runs initial setup. Give it up to 60 seconds for the first document.
If you see an error, check the terminal where Docling Serve is running for detailed error messages.
Help me debug document processing
Your document processing service is live. Next up, you will use the REST API directly to process documents programmatically and compare the standard and VLM pipelines.
Process Documents via the REST API
You confirmed that Docling Serve is running and healthy through the UI. Now it's time to interact with the same processing engine programmatically.
Production document AI systems are driven by APIs, not manual UI uploads. In this step, you'll write a Python client that talks to Docling Serve's REST API directly, processing the same Bill of Lading through both the standard layout pipeline and the VLM pipeline.
In this step, get ready to:
- Process a freight document via URL using the /v1/convert/source endpoint.
- Upload and process a local PDF using the /v1/convert/file endpoint.
- Compare standard and VLM pipeline outputs side by side.
Use the /v1/convert/source endpoint to process a BoL via URL
Docling Serve exposes two main conversion endpoints. The /v1/convert/source endpoint accepts a JSON payload with document URLs, so you can point the API at any publicly accessible PDF without downloading it first.
- Create a new file called api_client.py in your project folder.
- Add the imports, base URL constant, and the process_via_url function by copying this code into the file:
"""Demonstrate processing documents via Docling Serve's REST API."""
import json
from pathlib import Path
import httpx
DOCLING_BASE_URL = "http://127.0.0.1:5001"
def process_via_url(document_url: str, pipeline: str = "standard") -> dict:
"""Process a document by URL using Docling Serve API."""
endpoint = f"{DOCLING_BASE_URL}/v1/convert/source"
payload = {
"sources": [{"kind": "http", "url": document_url}],
"to_formats": ["md", "json"],
}
# Add VLM pipeline option if requested
if pipeline == "vlm":
payload["pipeline"] = "vlm"
payload["vlm_model"] = "granite_docling"
response = httpx.post(
endpoint,
json=payload,
headers={"accept": "application/json", "Content-Type": "application/json"},
timeout=300.0,
)
response.raise_for_status()
return response.json()What does this code do?
- DOCLING_BASE_URL points at your running Docling Serve instance on port 5001.
- process_via_url sends a POST request to /v1/convert/source with a JSON body containing the document URL and desired output formats (md and json).
- When pipeline="vlm" is passed, it tells Docling to use the GraniteDocling vision language model instead of the default layout detection approach.
- Save api_client.py.
You won't run this yet. The next two sections add the remaining functions before you execute the full script.
Use the /v1/convert/file endpoint to upload a local PDF
The second endpoint, /v1/convert/file, accepts multipart form data. This is what you'd use when documents live on disk rather than at a public URL.
- Add the process_via_file function below process_via_url in api_client.py:
def process_via_file(file_path: str, pipeline: str = "standard") -> dict:
"""Process a local file via Docling Serve API."""
endpoint = f"{DOCLING_BASE_URL}/v1/convert/file"
with open(file_path, "rb") as f:
files = {"files": (Path(file_path).name, f, "application/pdf")}
data = {"to_formats": "md", "do_ocr": "true"}
if pipeline == "vlm":
data["pipeline"] = "vlm"
data["vlm_model"] = "granite_docling"
response = httpx.post(
endpoint,
files=files,
data=data,
timeout=300.0,
)
response.raise_for_status()
return response.json()What does this code do?
- process_via_file opens a local PDF in binary mode and sends it as multipart form data to /v1/convert/file.
- The data dictionary carries form fields like to_formats and do_ocr alongside the uploaded file.
- The timeout is set to 300 seconds because VLM processing on complex documents can take a minute or more depending on page count.
- Save api_client.py.
You won't see a change yet. The next section ties both functions together and runs the full script.
Switch between pipelines and compare outputs
Now you'll tie both functions together in a main() function that processes the same VICS Bill of Lading through both the standard pipeline and the VLM pipeline, saving results for comparison.
- Add the main() function and entry point at the bottom of api_client.py:
def main():
print("\ud83d\ude80 DOCLING SERVE API CLIENT")
print("=" * 60)
# Process via URL with standard pipeline
bol_url = "https://assets.ctfassets.net/kjv8gs2ccggb/gomUPTy0eW68cAUuG2O4g/d22b23c5448d126190abc81110f2a86b/vics_bol.pdf"
print("\n\u25b6\ufe0f Processing BoL via URL (Standard Pipeline)...")
try:
result = process_via_url(bol_url, pipeline="standard")
print(f" \u2705 Success! Response keys: {list(result.keys())}")
# Save output
output_dir = Path("output")
output_dir.mkdir(exist_ok=True)
(output_dir / "api_standard_result.json").write_text(
json.dumps(result, indent=2, default=str)
)
print(f" \ud83d\udcc4 Saved to: output/api_standard_result.json")
except httpx.ConnectError:
print(" \u274c Connection failed. Is Docling Serve running?")
print(" Start it with: docling-serve run --enable-ui")
return
except httpx.HTTPStatusError as e:
print(f" \u274c HTTP error: {e.response.status_code}")
return
# Process via URL with VLM pipeline
print("\n\u25b6\ufe0f Processing BoL via URL (VLM Pipeline)...")
try:
result_vlm = process_via_url(bol_url, pipeline="vlm")
print(f" \u2705 Success! Response keys: {list(result_vlm.keys())}")
(output_dir / "api_vlm_result.json").write_text(
json.dumps(result_vlm, indent=2, default=str)
)
print(f" \ud83d\udcc4 Saved to: output/api_vlm_result.json")
except httpx.HTTPStatusError as e:
print(f" \u274c VLM pipeline error: {e.response.status_code}")
print("\n\u2705 API demonstration complete!")
print(" Open http://127.0.0.1:5001/ui to try the web interface")
if __name__ == "__main__":
main()What does this code do?
- main() processes the same VICS standard Bill of Lading PDF twice. First with the standard pipeline (RT-DETR layout detection + OCR), then with the VLM pipeline (GraniteDocling vision language model).
- Each result is saved as a JSON file in the output/ directory so you can compare the structural differences between the two approaches.
- Error handling catches connection failures (server not running) and HTTP errors (bad requests) gracefully.
- Save api_client.py.
- Make sure Docling Serve is still running in your other terminal.
- Run the API client by executing this command:
python api_client.py- You should see output confirming both pipelines processed successfully, with results saved to output/api_standard_result.json and output/api_vlm_result.json.
Why two pipelines?
The standard pipeline uses the Heron layout model (based on RT-DETR) to detect document regions, then runs OCR on each detected area. It is fast and works well on clean, single-column documents.
The VLM pipeline sends entire page images through the GraniteDocling-258M vision language model, which understands complex layouts (multi-column, dense tables, nested forms) without needing separate layout detection. On Apple Silicon with MLX, it runs with hardware acceleration.
Connection failed or timeout error?
If you see a connection error, switch to the terminal running Docling Serve and confirm it is still active. If it crashed, restart it with docling-serve run --enable-ui.
If the VLM pipeline times out, this is likely the first time GraniteDocling model weights are being loaded into memory. The first VLM request can take 2-3 minutes on initial model load. Subsequent requests will be much faster.
Help me debug a connection or timeout error when calling the Docling Serve API.
✔️ Awesome, I've got everything!
Great. Double check you have saved api_client.py and both JSON files exist in your output/ folder.
ⓧ I'd like to double check the full code
"""Demonstrate processing documents via Docling Serve's REST API."""
import json
from pathlib import Path
import httpx
DOCLING_BASE_URL = "http://127.0.0.1:5001"
def process_via_url(document_url: str, pipeline: str = "standard") -> dict:
"""Process a document by URL using Docling Serve API."""
endpoint = f"{DOCLING_BASE_URL}/v1/convert/source"
payload = {
"sources": [{"kind": "http", "url": document_url}],
"to_formats": ["md", "json"],
}
# Add VLM pipeline option if requested
if pipeline == "vlm":
payload["pipeline"] = "vlm"
payload["vlm_model"] = "granite_docling"
response = httpx.post(
endpoint,
json=payload,
headers={"accept": "application/json", "Content-Type": "application/json"},
timeout=300.0,
)
response.raise_for_status()
return response.json()
def process_via_file(file_path: str, pipeline: str = "standard") -> dict:
"""Process a local file via Docling Serve API."""
endpoint = f"{DOCLING_BASE_URL}/v1/convert/file"
with open(file_path, "rb") as f:
files = {"files": (Path(file_path).name, f, "application/pdf")}
data = {"to_formats": "md", "do_ocr": "true"}
if pipeline == "vlm":
data["pipeline"] = "vlm"
data["vlm_model"] = "granite_docling"
response = httpx.post(
endpoint,
files=files,
data=data,
timeout=300.0,
)
response.raise_for_status()
return response.json()
def main():
print("\ud83d\ude80 DOCLING SERVE API CLIENT")
print("=" * 60)
# Process via URL with standard pipeline
bol_url = "https://assets.ctfassets.net/kjv8gs2ccggb/gomUPTy0eW68cAUuG2O4g/d22b23c5448d126190abc81110f2a86b/vics_bol.pdf"
print("\n\u25b6\ufe0f Processing BoL via URL (Standard Pipeline)...")
try:
result = process_via_url(bol_url, pipeline="standard")
print(f" \u2705 Success! Response keys: {list(result.keys())}")
# Save output
output_dir = Path("output")
output_dir.mkdir(exist_ok=True)
(output_dir / "api_standard_result.json").write_text(
json.dumps(result, indent=2, default=str)
)
print(f" \ud83d\udcc4 Saved to: output/api_standard_result.json")
except httpx.ConnectError:
print(" \u274c Connection failed. Is Docling Serve running?")
print(" Start it with: docling-serve run --enable-ui")
return
except httpx.HTTPStatusError as e:
print(f" \u274c HTTP error: {e.response.status_code}")
return
# Process via URL with VLM pipeline
print("\n\u25b6\ufe0f Processing BoL via URL (VLM Pipeline)...")
try:
result_vlm = process_via_url(bol_url, pipeline="vlm")
print(f" \u2705 Success! Response keys: {list(result_vlm.keys())}")
(output_dir / "api_vlm_result.json").write_text(
json.dumps(result_vlm, indent=2, default=str)
)
print(f" \ud83d\udcc4 Saved to: output/api_vlm_result.json")
except httpx.HTTPStatusError as e:
print(f" \u274c VLM pipeline error: {e.response.status_code}")
print("\n\u2705 API demonstration complete!")
print(" Open http://127.0.0.1:5001/ui to try the web interface")
if __name__ == "__main__":
main()You now have a working API client that can process documents through both pipelines programmatically. Next up, you'll build a custom extraction microservice that sits on top of this API and pulls structured logistics data out of the raw processing output.
Build the Extraction Microservice
You now have Docling Serve processing documents and returning structured output through its REST API. Real production document AI systems add a second layer on top: domain-specific extraction logic that transforms raw processing output into typed, validated business data.
In this step, you'll build a custom FastAPI microservice that sits on top of Docling Serve. It accepts Bill of Lading uploads, sends them to Docling for processing, then extracts structured logistics fields into a typed Pydantic model with confidence scoring.
In this step, get ready to:
- Create a FastAPI extraction service with typed Pydantic models.
- Implement field extraction logic with confidence scoring.
- Launch and test the service via its Swagger docs.
Create the FastAPI service scaffold
Your extraction microservice needs its own web framework and server to handle incoming requests independently from Docling Serve. FastAPI provides automatic request validation, Swagger docs, and async support out of the box.
- Install the extraction service dependencies by running:
pip install fastapi uvicorn python-multipartWhat do these packages do?
- fastapi is the web framework for building your API endpoints.
- uvicorn is the ASGI server that runs FastAPI applications.
- python-multipart enables file upload handling in FastAPI endpoints.
Install failed?
Make sure your virtual environment is active (the same one you used for httpx and docling-serve).
help me troubleshoot a pip install failure for fastapi, uvicorn, or python-multipart
- Create a new file called extraction_service.py in your project folder with the following imports and app setup:
"""Production extraction microservice built on top of Docling Serve."""
import json
from pathlib import Path
from typing import Optional
import httpx
from fastapi import FastAPI, File, UploadFile
from pydantic import BaseModel, Field
app = FastAPI(
title="Freight Document Extraction API",
description="Extracts structured logistics data from Bills of Lading using Docling AI pipelines",
version="1.0.0",
)
DOCLING_SERVE_URL = "http://127.0.0.1:5001"What does this code do?
- The FastAPI() constructor creates your web application with metadata that appears in the auto-generated Swagger docs.
- DOCLING_SERVE_URL points to the Docling Serve instance already running on port 5001. Your extraction service calls Docling Serve as its processing backend.
- Save extraction_service.py.
You won't see a change yet. The next chunk adds the data models your endpoints will use.
Define the Pydantic models
Your service needs typed data structures that define what a Bill of Lading looks like as structured data. Pydantic models validate the shape of your output and auto-generate JSON schema for the API docs.
- In extraction_service.py, add the domain models below the DOCLING_SERVE_URL line:
# --- Pydantic Models ---
class CommodityItem(BaseModel):
description: Optional[str] = None
weight: Optional[str] = None
pieces: Optional[str] = None
nmfc_class: Optional[str] = None
class BillOfLading(BaseModel):
shipper_name: Optional[str] = None
shipper_address: Optional[str] = None
consignee_name: Optional[str] = None
consignee_address: Optional[str] = None
carrier_name: Optional[str] = None
bill_of_lading_number: Optional[str] = None
pro_number: Optional[str] = None
freight_terms: Optional[str] = None
commodity_items: list[CommodityItem] = Field(default_factory=list)
extraction_confidence: float = Field(default=0.0, ge=0.0, le=1.0)What does this code do?
- CommodityItem represents a single line item on the BoL (description, weight, number of pieces, NMFC class).
- BillOfLading holds all the key logistics fields: shipper, consignee, carrier, BoL number, PRO number, and freight terms.
- Every field is Optional because document extraction is imperfect. Some fields may not be found in every document.
- extraction_confidence is a float between 0 and 1 that tells the caller how many fields were successfully extracted.
- Now add the API response wrapper model directly below the BillOfLading class:
class ExtractionResponse(BaseModel):
success: bool
document_name: str
pipeline_used: str
processing_time_seconds: Optional[float] = None
extracted_data: Optional[BillOfLading] = None
raw_markdown: Optional[str] = None
error: Optional[str] = NoneWhat does this code do?
ExtractionResponse wraps every API response with metadata: whether the extraction succeeded, which pipeline was used, how long it took, and optionally the raw markdown output for debugging. This consistent response shape makes it easy for downstream systems to handle both success and failure cases.
- Save extraction_service.py.
You won't see a change yet. The next chunk adds the extraction logic that populates these models.
Implement extraction logic and API endpoints
The extraction function is the core intelligence of your microservice. It scans Docling's markdown output line by line, looking for logistics keywords like "SHIP FROM", "CONSIGNEE", and "CARRIER NAME", then captures the values that follow.
- In extraction_service.py, add the extraction function below the ExtractionResponse class:
# --- Extraction Logic ---
def extract_bol_fields(markdown_text: str) -> BillOfLading:
"""Extract Bill of Lading fields from Docling markdown output."""
bol = BillOfLading()
lines = markdown_text.split("\n")
fields_found = 0
total_fields = 8
for i, line in enumerate(lines):
line_upper = line.upper().strip()
if "SHIP FROM" in line_upper or "SHIPPER" in line_upper:
for j in range(i + 1, min(i + 4, len(lines))):
next_line = lines[j].strip()
if next_line and not any(
kw in next_line.upper()
for kw in ["NAME:", "ADDRESS:", "SHIP TO", "CARRIER", "---", "|"]
):
if bol.shipper_name is None:
bol.shipper_name = next_line
fields_found += 1
break
if "SHIP TO" in line_upper or "CONSIGNEE" in line_upper:
for j in range(i + 1, min(i + 4, len(lines))):
next_line = lines[j].strip()
if next_line and not any(
kw in next_line.upper()
for kw in ["NAME:", "ADDRESS:", "CARRIER", "THIRD", "---", "|"]
):
if bol.consignee_name is None:
bol.consignee_name = next_line
fields_found += 1
break
if "CARRIER NAME" in line_upper or "CARRIER:" in line_upper:
for j in range(i + 1, min(i + 3, len(lines))):
next_line = lines[j].strip()
if next_line and "---" not in next_line and "|" not in next_line:
if bol.carrier_name is None:
bol.carrier_name = next_line
fields_found += 1
break
if "BILL OF LADING NUMBER" in line_upper or "B/L NO" in line_upper:
parts = line.split(":")
if len(parts) > 1 and parts[1].strip():
bol.bill_of_lading_number = parts[1].strip()
fields_found += 1
if "PRO NUMBER" in line_upper or "PRO NO" in line_upper:
parts = line.split(":")
if len(parts) > 1 and parts[1].strip():
bol.pro_number = parts[1].strip()
fields_found += 1
if "PREPAID" in line_upper and bol.freight_terms is None:
bol.freight_terms = "Prepaid"
fields_found += 1
elif (
"COLLECT" in line_upper
and "CHECK" not in line_upper
and bol.freight_terms is None
):
bol.freight_terms = "Collect"
fields_found += 1
bol.extraction_confidence = round(fields_found / total_fields, 2)
return bolWhat does this code do?
- The function iterates through every line of the markdown looking for known BoL section headers (SHIP FROM, SHIP TO, CARRIER NAME, etc.).
- When it finds a header, it scans the next few lines for the actual value, skipping formatting characters like --- and | that Docling uses for table borders.
- The extraction_confidence score is calculated as the ratio of fields found to total expected fields (8). A score of 0.75 means 6 out of 8 fields were extracted.
- Save extraction_service.py.
You won't see a change yet. The next chunk adds the API endpoints that use this function.
- Add the API endpoints below the extract_bol_fields function:
# --- API Endpoints ---
@app.get("/health")
def health_check():
"""Health check endpoint."""
return {"status": "healthy", "service": "freight-extraction-api"}
@app.post("/extract", response_model=ExtractionResponse)
def extract_from_file(
file: UploadFile = File(..., description="PDF Bill of Lading to process"),
pipeline: str = "vlm",
):
"""Upload a Bill of Lading PDF and extract structured logistics fields."""
import time
start_time = time.time()
# Send file to Docling Serve for processing
try:
files = {"files": (file.filename, file.file, file.content_type)}
data = {"to_formats": "md", "do_ocr": "true"}
if pipeline == "vlm":
data["pipeline"] = "vlm"
data["vlm_model"] = "granite_docling"
response = httpx.post(
f"{DOCLING_SERVE_URL}/v1/convert/file",
files=files,
data=data,
timeout=300.0,
)
response.raise_for_status()
except httpx.ConnectError:
return ExtractionResponse(
success=False,
document_name=file.filename or "unknown",
pipeline_used=pipeline,
error="Cannot connect to Docling Serve. Is it running on port 5001?",
)
except httpx.HTTPStatusError as e:
return ExtractionResponse(
success=False,
document_name=file.filename or "unknown",
pipeline_used=pipeline,
error=f"Docling Serve returned HTTP {e.response.status_code}",
)
elapsed = time.time() - start_time
# Parse the Docling response and extract markdown
result = response.json()
markdown = result.get("md_content", "") or result.get("content", "")
if not markdown:
# Try to find markdown in nested structure
documents = result.get("documents", [])
if documents:
markdown = documents[0].get("md_content", "")
# Extract structured fields
bol = extract_bol_fields(markdown)
return ExtractionResponse(
success=True,
document_name=file.filename or "unknown",
pipeline_used=pipeline,
processing_time_seconds=round(elapsed, 1),
extracted_data=bol,
raw_markdown=markdown[:2000] if markdown else None,
)What does this code do?
- /health is a simple status endpoint that downstream services use to verify the extraction API is running.
- /extract accepts a PDF file upload, forwards it to Docling Serve's /v1/convert/file endpoint, then runs the extraction logic on the returned markdown.
- Error handling catches both connection failures (Docling Serve not running) and HTTP errors (bad request, server error), returning a structured error response instead of crashing.
- The raw_markdown field in the response is truncated to 2000 characters for debugging without overwhelming the response payload.
- Save extraction_service.py.
You won't see a change yet. The next chunk adds the URL-based extraction endpoint.
- Add the URL extraction endpoint at the bottom of extraction_service.py:
@app.post("/extract/url", response_model=ExtractionResponse)
def extract_from_url(document_url: str, pipeline: str = "vlm"):
"""Process a Bill of Lading from a URL and extract structured logistics fields."""
import time
start_time = time.time()
try:
payload = {
"sources": [{"kind": "http", "url": document_url}],
"to_formats": ["md"],
}
if pipeline == "vlm":
payload["pipeline"] = "vlm"
payload["vlm_model"] = "granite_docling"
response = httpx.post(
f"{DOCLING_SERVE_URL}/v1/convert/source",
json=payload,
headers={"accept": "application/json", "Content-Type": "application/json"},
timeout=300.0,
)
response.raise_for_status()
except httpx.ConnectError:
return ExtractionResponse(
success=False,
document_name=document_url,
pipeline_used=pipeline,
error="Cannot connect to Docling Serve. Is it running on port 5001?",
)
except httpx.HTTPStatusError as e:
return ExtractionResponse(
success=False,
document_name=document_url,
pipeline_used=pipeline,
error=f"Docling Serve returned HTTP {e.response.status_code}",
)
elapsed = time.time() - start_time
result = response.json()
markdown = result.get("md_content", "") or result.get("content", "")
if not markdown:
documents = result.get("documents", [])
if documents:
markdown = documents[0].get("md_content", "")
bol = extract_bol_fields(markdown)
return ExtractionResponse(
success=True,
document_name=document_url.split("/")[-1],
pipeline_used=pipeline,
processing_time_seconds=round(elapsed, 1),
extracted_data=bol,
raw_markdown=markdown[:2000] if markdown else None,
)What does this code do?
/extract/url follows the same pattern as /extract, but instead of receiving an uploaded file, it passes a document URL to Docling Serve's /v1/convert/source endpoint. This is useful for processing documents hosted online without downloading them first.
- Save extraction_service.py.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
Here is the complete extraction_service.py file:
"""Production extraction microservice built on top of Docling Serve."""
import json
from pathlib import Path
from typing import Optional
import httpx
from fastapi import FastAPI, File, UploadFile
from pydantic import BaseModel, Field
app = FastAPI(
title="Freight Document Extraction API",
description="Extracts structured logistics data from Bills of Lading using Docling AI pipelines",
version="1.0.0",
)
DOCLING_SERVE_URL = "http://127.0.0.1:5001"
# --- Pydantic Models ---
class CommodityItem(BaseModel):
description: Optional[str] = None
weight: Optional[str] = None
pieces: Optional[str] = None
nmfc_class: Optional[str] = None
class BillOfLading(BaseModel):
shipper_name: Optional[str] = None
shipper_address: Optional[str] = None
consignee_name: Optional[str] = None
consignee_address: Optional[str] = None
carrier_name: Optional[str] = None
bill_of_lading_number: Optional[str] = None
pro_number: Optional[str] = None
freight_terms: Optional[str] = None
commodity_items: list[CommodityItem] = Field(default_factory=list)
extraction_confidence: float = Field(default=0.0, ge=0.0, le=1.0)
class ExtractionResponse(BaseModel):
success: bool
document_name: str
pipeline_used: str
processing_time_seconds: Optional[float] = None
extracted_data: Optional[BillOfLading] = None
raw_markdown: Optional[str] = None
error: Optional[str] = None
# --- Extraction Logic ---
def extract_bol_fields(markdown_text: str) -> BillOfLading:
"""Extract Bill of Lading fields from Docling markdown output."""
bol = BillOfLading()
lines = markdown_text.split("\n")
fields_found = 0
total_fields = 8
for i, line in enumerate(lines):
line_upper = line.upper().strip()
if "SHIP FROM" in line_upper or "SHIPPER" in line_upper:
for j in range(i + 1, min(i + 4, len(lines))):
next_line = lines[j].strip()
if next_line and not any(
kw in next_line.upper()
for kw in ["NAME:", "ADDRESS:", "SHIP TO", "CARRIER", "---", "|"]
):
if bol.shipper_name is None:
bol.shipper_name = next_line
fields_found += 1
break
if "SHIP TO" in line_upper or "CONSIGNEE" in line_upper:
for j in range(i + 1, min(i + 4, len(lines))):
next_line = lines[j].strip()
if next_line and not any(
kw in next_line.upper()
for kw in ["NAME:", "ADDRESS:", "CARRIER", "THIRD", "---", "|"]
):
if bol.consignee_name is None:
bol.consignee_name = next_line
fields_found += 1
break
if "CARRIER NAME" in line_upper or "CARRIER:" in line_upper:
for j in range(i + 1, min(i + 3, len(lines))):
next_line = lines[j].strip()
if next_line and "---" not in next_line and "|" not in next_line:
if bol.carrier_name is None:
bol.carrier_name = next_line
fields_found += 1
break
if "BILL OF LADING NUMBER" in line_upper or "B/L NO" in line_upper:
parts = line.split(":")
if len(parts) > 1 and parts[1].strip():
bol.bill_of_lading_number = parts[1].strip()
fields_found += 1
if "PRO NUMBER" in line_upper or "PRO NO" in line_upper:
parts = line.split(":")
if len(parts) > 1 and parts[1].strip():
bol.pro_number = parts[1].strip()
fields_found += 1
if "PREPAID" in line_upper and bol.freight_terms is None:
bol.freight_terms = "Prepaid"
fields_found += 1
elif (
"COLLECT" in line_upper
and "CHECK" not in line_upper
and bol.freight_terms is None
):
bol.freight_terms = "Collect"
fields_found += 1
bol.extraction_confidence = round(fields_found / total_fields, 2)
return bol
# --- API Endpoints ---
@app.get("/health")
def health_check():
"""Health check endpoint."""
return {"status": "healthy", "service": "freight-extraction-api"}
@app.post("/extract", response_model=ExtractionResponse)
def extract_from_file(
file: UploadFile = File(..., description="PDF Bill of Lading to process"),
pipeline: str = "vlm",
):
"""Upload a Bill of Lading PDF and extract structured logistics fields."""
import time
start_time = time.time()
try:
files = {"files": (file.filename, file.file, file.content_type)}
data = {"to_formats": "md", "do_ocr": "true"}
if pipeline == "vlm":
data["pipeline"] = "vlm"
data["vlm_model"] = "granite_docling"
response = httpx.post(
f"{DOCLING_SERVE_URL}/v1/convert/file",
files=files,
data=data,
timeout=300.0,
)
response.raise_for_status()
except httpx.ConnectError:
return ExtractionResponse(
success=False,
document_name=file.filename or "unknown",
pipeline_used=pipeline,
error="Cannot connect to Docling Serve. Is it running on port 5001?",
)
except httpx.HTTPStatusError as e:
return ExtractionResponse(
success=False,
document_name=file.filename or "unknown",
pipeline_used=pipeline,
error=f"Docling Serve returned HTTP {e.response.status_code}",
)
elapsed = time.time() - start_time
result = response.json()
markdown = result.get("md_content", "") or result.get("content", "")
if not markdown:
documents = result.get("documents", [])
if documents:
markdown = documents[0].get("md_content", "")
bol = extract_bol_fields(markdown)
return ExtractionResponse(
success=True,
document_name=file.filename or "unknown",
pipeline_used=pipeline,
processing_time_seconds=round(elapsed, 1),
extracted_data=bol,
raw_markdown=markdown[:2000] if markdown else None,
)
@app.post("/extract/url", response_model=ExtractionResponse)
def extract_from_url(document_url: str, pipeline: str = "vlm"):
"""Process a Bill of Lading from a URL and extract structured logistics fields."""
import time
start_time = time.time()
try:
payload = {
"sources": [{"kind": "http", "url": document_url}],
"to_formats": ["md"],
}
if pipeline == "vlm":
payload["pipeline"] = "vlm"
payload["vlm_model"] = "granite_docling"
response = httpx.post(
f"{DOCLING_SERVE_URL}/v1/convert/source",
json=payload,
headers={"accept": "application/json", "Content-Type": "application/json"},
timeout=300.0,
)
response.raise_for_status()
except httpx.ConnectError:
return ExtractionResponse(
success=False,
document_name=document_url,
pipeline_used=pipeline,
error="Cannot connect to Docling Serve. Is it running on port 5001?",
)
except httpx.HTTPStatusError as e:
return ExtractionResponse(
success=False,
document_name=document_url,
pipeline_used=pipeline,
error=f"Docling Serve returned HTTP {e.response.status_code}",
)
elapsed = time.time() - start_time
result = response.json()
markdown = result.get("md_content", "") or result.get("content", "")
if not markdown:
documents = result.get("documents", [])
if documents:
markdown = documents[0].get("md_content", "")
bol = extract_bol_fields(markdown)
return ExtractionResponse(
success=True,
document_name=document_url.split("/")[-1],
pipeline_used=pipeline,
processing_time_seconds=round(elapsed, 1),
extracted_data=bol,
raw_markdown=markdown[:2000] if markdown else None,
)- Create a launcher script called run_extraction_service.sh with the following content:
#!/bin/bash
# Launch the extraction microservice (requires Docling Serve running on port 5001)
echo "Starting Freight Document Extraction Service..."
echo "Extraction API: http://127.0.0.1:8000"
echo "Swagger Docs: http://127.0.0.1:8000/docs"
echo ""
echo "NOTE: Docling Serve must be running on port 5001"
echo " Start it with: docling-serve run --enable-ui"
echo ""
uvicorn extraction_service:app --host 127.0.0.1 --port 8000 --reloadWhat does this script do?
The script launches your extraction service using uvicorn with --reload enabled for development. The --reload flag automatically restarts the server whenever you save changes to extraction_service.py.
The import string extraction_service:app tells Uvicorn to import the app object from extraction_service.py.
- Save run_extraction_service.sh.
- Make the script executable and launch it by running:
chmod +x run_extraction_service.sh
./run_extraction_service.sh- Open http://127.0.0.1:8000/docs in your browser.
You should see the Swagger UI for your Freight Document Extraction API with three endpoints listed: /health, /extract, and /extract/url.
Server won't start?
- If you see an "Address already in use" error, another process is already on port 8000. Stop it or change the port in the script.
- If you see a ModuleNotFoundError, make sure you ran the pip install command earlier in this step.
- Make sure Docling Serve is still running in its own terminal on port 5001.
help me troubleshoot my FastAPI extraction service that won't start with uvicorn
Why a microservice on top of Docling Serve?
Docling Serve is a general-purpose document processing engine. It converts documents to markdown, JSON, or other formats, but it has no concept of what a Bill of Lading is.
Your extraction microservice adds the domain intelligence: it knows what fields to look for, how to score confidence, and returns a clean typed schema that downstream logistics systems can consume directly. This separation keeps each service focused on one responsibility.
Your extraction microservice is live. You now have a full production architecture: Docling Serve handles the heavy document processing, and your custom FastAPI service handles domain-specific field extraction with typed output and confidence scoring.
Secret mission
Add Async Batch Processing
Your extraction service handles one document at a time. In production, logistics teams need to process entire shipments worth of Bills of Lading in a single request. Can you extend your service with concurrent batch processing using Docling Serve's async API?
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Docling Serve process (running on port 5001).
- Extraction microservice process (running on port 8000, if started).
- Project files (api_client.py, extraction_service.py, extraction_service_batch.py, run_server.sh, run_extraction_service.sh, output/ directory).
- Python packages (docling-serve, docling, mlx, mlx-lm, httpx, fastapi, uvicorn, python-multipart).
- Cached model weights in ~/.cache/huggingface/hub/ (GraniteDocling VLM and layout model files).
Keep everything running
No action needed. Choose this if you are still actively building or experimenting with different documents.
- Docling Serve stays running at http://127.0.0.1:5001 and you can continue uploading documents via the UI.
- Your project files, Python packages, and cached model weights remain on disk.
- The model weights in ~/.cache/huggingface/hub/ can be several hundred MB. They stay cached until you manually remove them.
Pause - I'll come back to this later
Shut down running processes to free up memory and CPU. Your files and packages stay intact for next time.
- Switch to the terminal running Docling Serve and press Ctrl+C to stop it.
- If your extraction service is running, switch to that terminal and press Ctrl+C to stop it too.
- Your project files, installed packages, and cached models stay on disk. Next time, run docling-serve run --enable-ui to start up again.
Delete - I don't want to use this again
Remove all project resources and cached models to reclaim disk space.
- Stop any running processes by pressing Ctrl+C in each terminal window (Docling Serve and the extraction service).
- Delete your project files by running:
cd ~/Desktop
rm -rf api_client.py extraction_service.py extraction_service_batch.py run_server.sh run_extraction_service.sh output/- Uninstall the Python packages by running:
pip uninstall -y docling-serve docling mlx mlx-lm httpx fastapi uvicorn python-multipart- Remove the cached GraniteDocling model weights by running:
rm -rf ~/.cache/huggingface/hub/models--ibm-granite--granite-docling*What do these commands remove?
The first rm -rf deletes your project scripts and the output folder. The pip uninstall command removes the packages you installed for this project.
The final cache cleanup removes the GraniteDocling VLM weights that Docling downloaded on first run. You may also see additional model directories under ~/.cache/huggingface/hub/ for the layout detection model. You can inspect that folder and remove any other models-- directories related to Docling.
Nice Work!
Nice Work!
Nice work! You've deployed a production-grade document AI system that processes complex freight logistics documents through state-of-the-art layout models and Vision Language Models.
You've learned how to:
- Deploy Docling Serve as a production API service with a professional web UI, Swagger docs, and both standard (RT-DETR layout) and VLM (GraniteDocling) pipelines accessible via REST API.
- Build a custom FastAPI extraction microservice that wraps Docling's processing engine and returns structured logistics data (shipper, consignee, commodities, freight terms) as typed Pydantic models with confidence scoring.
- Architect a production document AI system using the layered pattern: a general-purpose processing engine with an API, fronted by a domain-specific extraction service.
- Secret Mission: Add async batch processing with concurrent document handling via Docling Serve's async API and graceful error recovery for partial failures.
Ready to quiz yourself?