1+ import asyncio
12import http .cookiejar as cj
23import json
34import logging
2728 "max_tokens" : 200 ,
2829}
2930
31+ # Process-wide async semaphore that bounds the number of concurrent
32+ # in-flight requests to the docling-serve backend. Initialized lazily
33+ # so it binds to the running event loop on first use; tests that
34+ # spin up fresh loops should reset this back to None between cases
35+ # (see ``tests/conftest.py``).
36+ _docling_sem : asyncio .Semaphore | None = None
37+
38+
39+ def get_docling_sem () -> asyncio .Semaphore :
40+ global _docling_sem
41+ if _docling_sem is None :
42+ _docling_sem = asyncio .Semaphore (get_settings ().docling_concurrency )
43+ return _docling_sem
44+
3045
3146def do_repl (data ):
3247 if isinstance (data , dict ):
@@ -39,7 +54,14 @@ def do_repl(data):
3954
4055
4156def is_html (file_bytes : bytes ) -> bool :
42- return (file_bytes .startswith (b"<!DOCTYPE html>" ) or b"<html" in file_bytes [:100 ]) and b"<body" in file_bytes
57+ """Detect HTML from the leading bytes only.
58+
59+ The ``<html`` check stays at the first 100 bytes (matches the
60+ original behavior). The ``<body`` check is bounded to the first
61+ 8 KB so non-HTML payloads (e.g. multi-MB PDFs) do not trigger a
62+ full-buffer ``bytes.find``.
63+ """
64+ return (file_bytes .startswith (b"<!DOCTYPE html>" ) or b"<html" in file_bytes [:100 ]) and b"<body" in file_bytes [:8192 ]
4365
4466
4567@retry (stop = stop_after_attempt (4 ), wait = wait_exponential_jitter (), reraise = True )
@@ -52,17 +74,53 @@ async def docling_convert(
5274) -> dict :
5375 """Convert a document via the docling-serve HTTP backend.
5476
55- Concurrency for ``parse`` steps is now bounded at the worker
56- level by the ``parse`` consumer pool count
57- (``Worker(consumers={"parse": N, ...})``); the library does no
58- rate-limiting of its own. To preserve the previous default,
59- operators should set ``consumers["parse"]`` equal to
60- ``settings.docling_concurrency``.
77+ Process-wide concurrency to docling-serve is bounded by
78+ ``settings.docling_concurrency`` via the module-global
79+ ``_docling_sem``. The semaphore wraps only the HTTP / websocket
80+ / result-GET round trip — large-document post-processing
81+ (recursive image-placeholder substitution and whole-tree
82+ re-serialization) runs outside the slot and on a worker thread,
83+ so CPU work cannot starve docling-serve capacity.
84+
85+ The semaphore is acquired *inside* the tenacity retry so a
86+ failing attempt releases its slot while it waits to retry.
87+
88+ Note: the gate is per-process. If multiple worker processes
89+ share one docling-serve, the server sees
90+ ``N_workers * docling_concurrency`` aggregate parallelism; in
91+ that topology, rate-limit at the proxy or in docling-serve's
92+ own queue instead.
93+ """
94+ async with get_docling_sem ():
95+ res , parameters = await _docling_request (
96+ file_bytes ,
97+ mime_type ,
98+ source_uri ,
99+ config_dict ,
100+ output_formats ,
101+ )
102+ return await asyncio .to_thread (_process_result , res , parameters , source_uri , output_formats )
103+
104+
105+ async def _docling_request (
106+ file_bytes : bytes ,
107+ mime_type : str ,
108+ source_uri : str ,
109+ config_dict : dict [str , str | int | bool ],
110+ output_formats : list [str ],
111+ ) -> tuple [dict , dict ]:
112+ """POST + websocket wait + result GET against docling-serve.
113+
114+ Returns ``(res, parameters)``. ``res`` is the parsed result
115+ document; ``parameters`` is the request body as sent (the
116+ caller needs ``image_export_mode`` from it to decide on
117+ placeholder replacement). Raises ``ValueError`` on protocol
118+ errors, which the outer ``@retry`` retries.
61119 """
62120 env = get_settings ()
63121 local_jar = cj .CookieJar ()
64122 async_url = f"{ env .docling_server_url }
65- parameters = {
123+ parameters : dict = {
66124 "from_formats" : [
67125 "docx" ,
68126 "pptx" ,
@@ -102,7 +160,6 @@ async def docling_convert(
102160 parameters ["do_picture_description" ] = False
103161
104162 file_name = source_uri .split ("/" )[- 1 ]
105-
106163 if mime_type and "markdown" in mime_type and not file_name .endswith (".md" ):
107164 file_name = file_name + ".md"
108165 # docling requires some special handling for html
@@ -112,9 +169,7 @@ async def docling_convert(
112169
113170 f = BytesIO (file_bytes )
114171 try :
115- files = {
116- "files" : (file_name , f , mime_type ),
117- }
172+ files = {"files" : (file_name , f , mime_type )}
118173 logger .debug (f"using { parameters } { file_name } )
119174 async with httpx .AsyncClient (timeout = env .docling_http_timeout , cookies = local_jar ) as _async_client :
120175 response = await _async_client .post (async_url , files = files , data = parameters )
@@ -143,28 +198,48 @@ async def docling_convert(
143198 logger .error (f"no errors in response: { payload } )
144199 result_url = f"{ env .docling_server_url } { task_id }
145200 response = await _async_client .get (result_url )
146- res = response .json ()
201+ # The result body carries the full Docling document, which
202+ # can be many MB. ``response.json()`` runs stdlib
203+ # ``json.loads`` on the calling thread — offload so the
204+ # event loop stays responsive while it parses.
205+ res = await asyncio .to_thread (response .json )
206+ logger .info (f"{ task_id } { res .get ('status' )} { res .get ('processing_time' )} )
147207 finally :
148208 f .close ()
209+ return res , parameters
210+
211+
212+ def _process_result (
213+ res : dict ,
214+ parameters : dict ,
215+ source_uri : str ,
216+ output_formats : list [str ],
217+ ) -> dict :
218+ """Validate the docling-serve result and re-serialize each
219+ requested output format.
220+
221+ Pure Python. ``do_repl`` walks the entire JSON tree and
222+ ``json.dumps`` re-serializes it — seconds of work on a large
223+ document. Designed to run on a worker thread (see the
224+ ``asyncio.to_thread`` call in :func:`docling_convert`).
225+ """
149226 if "status" not in res :
150227 raise ValueError (f"no status in response: { res } )
151- logger .info (f"{ task_id } { res ['status' ]} { res ['processing_time' ]} )
152-
153- if res ["status" ] == "success" :
154- parsed = {}
155- for output_format in output_formats :
156- output_content = res ["document" ][f"{ output_format } ]
157- if output_format == "json" :
158- if "image_export_mode" in parameters and parameters ["image_export_mode" ] == "placeholder" :
159- logger .info (f" doing placeholder replacement for { source_uri } )
160- output_content = do_repl (output_content )
161- parsed [output_format ] = json .dumps (output_content ).encode ("utf-8" )
162- else :
163- parsed [output_format ] = str (output_content ).encode ("utf-8" )
164- return parsed
165- else :
228+ if res ["status" ] != "success" :
166229 raise ValueError (str (res ["errors" ]))
167230
231+ parsed : dict [str , bytes ] = {}
232+ for output_format in output_formats :
233+ output_content = res ["document" ][f"{ output_format } ]
234+ if output_format == "json" :
235+ if parameters .get ("image_export_mode" ) == "placeholder" :
236+ logger .info (f" doing placeholder replacement for { source_uri } )
237+ output_content = do_repl (output_content )
238+ parsed [output_format ] = json .dumps (output_content ).encode ("utf-8" )
239+ else :
240+ parsed [output_format ] = str (output_content ).encode ("utf-8" )
241+ return parsed
242+
168243
169244def get_docling_schema_version () -> str :
170245 import docling_core .types .doc .document as dd
0 commit comments