Skip to content

conversion_manager

Class info

Classes

Name Children Inherits
ConversionConfig
llmling_agent_config.converters
Global conversion configuration.
    ConversionManager
    llmling_agent.prompts.conversion_manager
    Manages document conversion using configured providers.
      MarkItDownConfig
      llmling_agent_config.converters
      Configuration for MarkItDown-based converter.
        • BaseConverterConfig
        PlainConverterConfig
        llmling_agent_config.converters
        Configuration for plain text fallback converter.
          • BaseConverterConfig

          🛈 DocStrings

          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.

          Source code in src/llmling_agent/prompts/conversion_manager.py 
           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
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
          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]):
        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):
        self._executor.shutdown(wait=False)

    def supports_file(self, path: StrPath) -> bool:
        """Check if any converter supports the file."""
        return any(c.supports_file(path) 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_content(content, mime_type) for c in self._converters)

    def _setup_converters(self) -> list[DocumentConverter]:
        """Create converter instances from config."""
        from llmling_agent_converters.plain_converter import PlainConverter

        converters: list[DocumentConverter] = []
        for cfg in self.config.providers or []:
            if not cfg.enabled:
                continue
            converter = cfg.get_converter()
            converters.append(converter)
        # Always add PlainConverter as fallback
        # if it gets configured by user, that one gets preference.
        converters.append(PlainConverter())
        return converters

    async def convert_file(self, path: StrPath) -> str:
        """Convert file using first supporting converter."""
        loop = asyncio.get_running_loop()
        content = await read_path(path, "rb")

        for converter in self._converters:
            # Run support check in thread pool
            supports = await loop.run_in_executor(
                self._executor, converter.supports_file, path
            )
            if not supports:
                continue
            # Run conversion in thread pool
            import mimetypes

            typ = mimetypes.guess_type(str(path))[0]
            return await loop.run_in_executor(
                self._executor,
                converter.convert_content,
                content,
                typ,
            )
        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_content, content, mime_type
            )
            if not supports:
                continue

            # Run conversion in thread pool
            return await loop.run_in_executor(
                self._executor, converter.convert_content, content, mime_type
            )

        return str(content)  # Fallback for unsupported content

    def convert_file_sync(self, path: StrPath) -> str:
        """Sync wrapper for convert_file."""
        try:
            loop = asyncio.get_running_loop()
            return loop.run_until_complete(self.convert_file(path))
        except RuntimeError:
            # No running loop - create new one
            return asyncio.run(self.convert_file(path))

    def convert_content_sync(self, content: Any, mime_type: str | None = None) -> str:
        """Sync wrapper for convert_content."""
        try:
            loop = asyncio.get_running_loop()
            return loop.run_until_complete(self.convert_content(content, mime_type))
        except RuntimeError:
            # No running loop - create new one
            return asyncio.run(self.convert_content(content, mime_type))

          convert_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 
           87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
          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_content, content, mime_type
        )
        if not supports:
            continue

        # Run conversion in thread pool
        return await loop.run_in_executor(
            self._executor, converter.convert_content, content, mime_type
        )

    return str(content)  # Fallback for unsupported content

          convert_content_sync 

          convert_content_sync(content: Any, mime_type: str | None = None) -> str

          Sync wrapper for convert_content.

          Source code in src/llmling_agent/prompts/conversion_manager.py 
          115
116
117
118
119
120
121
122
          def convert_content_sync(self, content: Any, mime_type: str | None = None) -> str:
    """Sync wrapper for convert_content."""
    try:
        loop = asyncio.get_running_loop()
        return loop.run_until_complete(self.convert_content(content, mime_type))
    except RuntimeError:
        # No running loop - create new one
        return asyncio.run(self.convert_content(content, mime_type))

          convert_file async 

          convert_file(path: StrPath) -> str

          Convert file using first supporting converter.

          Source code in src/llmling_agent/prompts/conversion_manager.py 
          63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
          async def convert_file(self, path: StrPath) -> str:
    """Convert file using first supporting converter."""
    loop = asyncio.get_running_loop()
    content = await read_path(path, "rb")

    for converter in self._converters:
        # Run support check in thread pool
        supports = await loop.run_in_executor(
            self._executor, converter.supports_file, path
        )
        if not supports:
            continue
        # Run conversion in thread pool
        import mimetypes

        typ = mimetypes.guess_type(str(path))[0]
        return await loop.run_in_executor(
            self._executor,
            converter.convert_content,
            content,
            typ,
        )
    return str(content)

          convert_file_sync 

          convert_file_sync(path: StrPath) -> str

          Sync wrapper for convert_file.

          Source code in src/llmling_agent/prompts/conversion_manager.py 
          106
107
108
109
110
111
112
113
          def convert_file_sync(self, path: StrPath) -> str:
    """Sync wrapper for convert_file."""
    try:
        loop = asyncio.get_running_loop()
        return loop.run_until_complete(self.convert_file(path))
    except RuntimeError:
        # No running loop - create new one
        return asyncio.run(self.convert_file(path))

          supports_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 
          44
45
46
          def supports_content(self, content: Any, mime_type: str | None = None) -> bool:
    """Check if any converter supports the file."""
    return any(c.supports_content(content, mime_type) for c in self._converters)

          supports_file 

          supports_file(path: StrPath) -> bool

          Check if any converter supports the file.

          Source code in src/llmling_agent/prompts/conversion_manager.py 
          40
41
42
          def supports_file(self, path: StrPath) -> bool:
    """Check if any converter supports the file."""
    return any(c.supports_file(path) for c in self._converters)