Manages document conversion using configured providers.
In order to not make things super complex, all Converters will be implemented as sync.
The manager will handle async I/O and thread pooling.
Source code in src/llmling_agent/prompts/conversion_manager.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88 | class ConversionManager:
"""Manages document conversion using configured providers.
In order to not make things super complex, all Converters will be implemented as sync.
The manager will handle async I/O and thread pooling.
"""
def __init__(self, config: ConversionConfig | list[DocumentConverter]) -> None:
if isinstance(config, list):
self.config = ConversionConfig()
self._converters = config
else:
self.config = config
self._converters = self._setup_converters()
self._executor = ThreadPoolExecutor(max_workers=3)
def __del__(self) -> None:
self._executor.shutdown(wait=False)
def supports_file(self, path: JoinablePathLike) -> bool:
"""Check if any converter supports the file."""
mime_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
return any(c.supports_mime_type(mime_type) for c in self._converters)
def supports_content(self, content: Any, mime_type: str | None = None) -> bool:
"""Check if any converter supports the file."""
return any(c.supports_mime_type(content) for c in self._converters)
def _setup_converters(self) -> list[DocumentConverter]:
"""Create converter instances from config."""
return [i.get_provider() for i in self.config.providers or []]
# Always add PlainConverter as fallback
# if it gets configured by user, that one gets preference.
async def convert_file(self, path: JoinablePathLike) -> str:
"""Convert file using first supporting converter."""
import mimetypes
loop = asyncio.get_running_loop()
mime_type = mimetypes.guess_type(str(path))[0] or "text/plain"
for converter in self._converters:
# Run support check in thread pool
supports = await loop.run_in_executor(
self._executor, converter.supports_mime_type, mime_type
)
if not supports:
continue
# Run conversion in thread pool
content = await loop.run_in_executor(
self._executor,
converter.convert_file,
path,
)
return str(content)
async def convert_content(self, content: Any, mime_type: str | None = None) -> str:
"""Convert content using first supporting converter."""
loop = asyncio.get_running_loop()
for converter in self._converters:
# Run support check in thread pool
supports = await loop.run_in_executor(
self._executor, converter.supports_mime_type, mime_type or "text/plain"
)
if not supports:
continue
doc = await converter.convert_content(content, mime_type or "text/plain")
return doc.content
return str(content) # Fallback for unsupported content
|
async
convert_content(content: Any, mime_type: str | None = None) -> str
Convert content using first supporting converter.
Source code in src/llmling_agent/prompts/conversion_manager.py
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88 | async def convert_content(self, content: Any, mime_type: str | None = None) -> str:
"""Convert content using first supporting converter."""
loop = asyncio.get_running_loop()
for converter in self._converters:
# Run support check in thread pool
supports = await loop.run_in_executor(
self._executor, converter.supports_mime_type, mime_type or "text/plain"
)
if not supports:
continue
doc = await converter.convert_content(content, mime_type or "text/plain")
return doc.content
return str(content) # Fallback for unsupported content
|
async
convert_file(path: JoinablePathLike) -> str
Convert file using first supporting converter.
Source code in src/llmling_agent/prompts/conversion_manager.py
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71 | async def convert_file(self, path: JoinablePathLike) -> str:
"""Convert file using first supporting converter."""
import mimetypes
loop = asyncio.get_running_loop()
mime_type = mimetypes.guess_type(str(path))[0] or "text/plain"
for converter in self._converters:
# Run support check in thread pool
supports = await loop.run_in_executor(
self._executor, converter.supports_mime_type, mime_type
)
if not supports:
continue
# Run conversion in thread pool
content = await loop.run_in_executor(
self._executor,
converter.convert_file,
path,
)
return str(content)
|
supports_content(content: Any, mime_type: str | None = None) -> bool
Check if any converter supports the file.
Source code in src/llmling_agent/prompts/conversion_manager.py
| def supports_content(self, content: Any, mime_type: str | None = None) -> bool:
"""Check if any converter supports the file."""
return any(c.supports_mime_type(content) for c in self._converters)
|
supports_file(path: JoinablePathLike) -> bool
Check if any converter supports the file.
Source code in src/llmling_agent/prompts/conversion_manager.py
| def supports_file(self, path: JoinablePathLike) -> bool:
"""Check if any converter supports the file."""
mime_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
return any(c.supports_mime_type(mime_type) for c in self._converters)
|