Coverage for src / quber / core / extractors / camelot / llm / orchestrator.py: 97%

1""" 

2CamelotLLMTableExtractor: the public face of the camelot.llm package. 

3 

4DEPRECATED and dormant: superseded by `SetOfMarkExtractor` (vision-guided, 

5in-region Camelot, grounded correction). Kept importable for comparison only; 

6do not build new work on it. 

7 

8Wires run-scoped dependencies (LLM clients, semaphores, temp dir) into 

9`PipelineDeps` and runs the extraction graph built in `pipeline`. 

10""" 

11 

12from __future__ import annotations 

13 

14import asyncio 

15import tempfile 

16from pathlib import Path 

17from typing import List, Optional 

18 

19from quber.agents.classifier import TableClassifier, get_classifier 

20from quber.agents.llm_client import LLMClient, get_llm_client 

21from quber.agents.unifier import TableUnifier, get_unifier 

22from quber.core.extractors.base import ExtractedTable 

23from quber.core.extractors.camelot.acquire import CAMELOT_FLAVOR_TIMEOUT_S 

24from quber.core.extractors.camelot.llm.pipeline import ( 

25 PipelineDeps, 

26 PipelineState, 

27 build_pipeline_graph, 

28) 

29 

30 

31class CamelotLLMTableExtractor: 

32 llm: LLMClient 

33 classifier: TableClassifier 

34 unifier: TableUnifier 

35 dpi: int 

36 run_llm_correction: bool 

37 max_concurrent: int 

38 flavor_timeout_s: float 

39 

40 def __init__( 

41 self, 

42 llm_client: Optional[LLMClient] = None, 

43 classifier: Optional[TableClassifier] = None, 

44 unifier: Optional[TableUnifier] = None, 

45 dpi: int = 200, 

46 run_llm_correction: bool = True, 

47 max_concurrent: int = 5, 

48 flavor_timeout_s: float = CAMELOT_FLAVOR_TIMEOUT_S, 

49 ) -> None: 

50 self.llm = llm_client or get_llm_client() 

51 self.classifier = classifier or get_classifier() 

52 self.unifier = unifier or get_unifier() 

53 self.dpi = dpi 

54 self.run_llm_correction = run_llm_correction 

55 self.max_concurrent = max_concurrent 

56 self.flavor_timeout_s = flavor_timeout_s 

57 

58 async def extract_tables(self, source: Path) -> List[ExtractedTable]: 

59 source = Path(source) 

60 if not source.exists(): 

61 raise FileNotFoundError(source) 

62 

63 graph = build_pipeline_graph() 

64 with tempfile.TemporaryDirectory(prefix="quber-camelot-") as tmpdir: 

65 deps = PipelineDeps( 

66 llm=self.llm, 

67 classifier=self.classifier, 

68 unifier=self.unifier, 

69 source=source, 

70 tmp_dir=Path(tmpdir), 

71 dpi=self.dpi, 

72 run_llm_correction=self.run_llm_correction, 

73 flavor_timeout_s=self.flavor_timeout_s, 

74 classify_sem=asyncio.Semaphore(self.max_concurrent), 

75 unify_sem=asyncio.Semaphore(self.max_concurrent), 

76 correct_sem=asyncio.Semaphore(self.max_concurrent), 

77 ) 

78 return await graph.run(state=PipelineState(), deps=deps, inputs=source) 

79 

80 def extract_tables_sync(self, source: Path) -> List[ExtractedTable]: 

81 """Sync entry point for callers without an event loop (CLI, 

82 scripts). Wraps the async path with `asyncio.run`. Not callable 

83 from inside an existing event loop — async callers should use 

84 `await extract_tables(...)` directly. 

85 """ 

86 return asyncio.run(self.extract_tables(source))