Demonstration of batch indexing capabilities.
| 43 | |
| 44 | |
| 45 | class BatchIndexingDemo: |
| 46 | """Demonstration of batch indexing capabilities.""" |
| 47 | |
| 48 | def __init__(self, config_path: str): |
| 49 | """Initialize the batch indexing demo.""" |
| 50 | self.config_path = config_path |
| 51 | self.config = self._load_config() |
| 52 | self.db = ChatDatabase() |
| 53 | |
| 54 | # Initialize Ollama client |
| 55 | self.ollama_client = OllamaClient() |
| 56 | |
| 57 | # Initialize pipeline with merged configuration |
| 58 | self.pipeline_config = self._merge_configurations() |
| 59 | self.pipeline = IndexingPipeline( |
| 60 | self.pipeline_config, |
| 61 | self.ollama_client, |
| 62 | self.config.get("ollama_config", { |
| 63 | "generation_model": "qwen3:0.6b", |
| 64 | "embedding_model": "qwen3:0.6b" |
| 65 | }) |
| 66 | ) |
| 67 | |
| 68 | def _load_config(self) -> Dict[str, Any]: |
| 69 | """Load batch indexing configuration from file.""" |
| 70 | try: |
| 71 | with open(self.config_path, 'r') as f: |
| 72 | config = json.load(f) |
| 73 | print(f"✅ Loaded configuration from {self.config_path}") |
| 74 | return config |
| 75 | except FileNotFoundError: |
| 76 | print(f"❌ Configuration file not found: {self.config_path}") |
| 77 | sys.exit(1) |
| 78 | except json.JSONDecodeError as e: |
| 79 | print(f"❌ Invalid JSON in configuration file: {e}") |
| 80 | sys.exit(1) |
| 81 | |
| 82 | def _merge_configurations(self) -> Dict[str, Any]: |
| 83 | """Merge batch config with default pipeline config.""" |
| 84 | # Start with default pipeline configuration |
| 85 | merged_config = PIPELINE_CONFIGS.get("default", {}).copy() |
| 86 | |
| 87 | # Override with batch-specific settings |
| 88 | batch_settings = self.config.get("pipeline_settings", {}) |
| 89 | |
| 90 | # Deep merge for nested dictionaries |
| 91 | def deep_merge(base: dict, override: dict) -> dict: |
| 92 | result = base.copy() |
| 93 | for key, value in override.items(): |
| 94 | if key in result and isinstance(result[key], dict) and isinstance(value, dict): |
| 95 | result[key] = deep_merge(result[key], value) |
| 96 | else: |
| 97 | result[key] = value |
| 98 | return result |
| 99 | |
| 100 | return deep_merge(merged_config, batch_settings) |
| 101 | |
| 102 | def validate_documents(self, documents: List[str]) -> List[str]: |