Load knowledge files from a directory with change detection and metadata enhancement. This function now includes enhanced error handling and compatibility with the intelligent memory consolidation system.
(
log_item: LogItem | None,
knowledge_dir: str,
index: Dict[str, KnowledgeImport],
metadata: dict[str, Any] = {},
filename_pattern: str = "**/*",
recursive: bool = True,
)
| 31 | |
| 32 | |
| 33 | def load_knowledge( |
| 34 | log_item: LogItem | None, |
| 35 | knowledge_dir: str, |
| 36 | index: Dict[str, KnowledgeImport], |
| 37 | metadata: dict[str, Any] = {}, |
| 38 | filename_pattern: str = "**/*", |
| 39 | recursive: bool = True, |
| 40 | ) -> Dict[str, KnowledgeImport]: |
| 41 | """ |
| 42 | Load knowledge files from a directory with change detection and metadata enhancement. |
| 43 | |
| 44 | This function now includes enhanced error handling and compatibility with the |
| 45 | intelligent memory consolidation system. |
| 46 | """ |
| 47 | |
| 48 | # Mapping file extensions to corresponding loader classes |
| 49 | # Note: Using TextLoader for JSON and MD to avoid parsing issues with consolidation |
| 50 | file_types_loaders = { |
| 51 | "txt": TextLoader, |
| 52 | "pdf": PyPDFLoader, |
| 53 | "csv": CSVLoader, |
| 54 | "html": UnstructuredHTMLLoader, |
| 55 | "json": TextLoader, # Use TextLoader for better consolidation compatibility |
| 56 | "md": TextLoader, # Use TextLoader for better consolidation compatibility |
| 57 | } |
| 58 | |
| 59 | cnt_files = 0 |
| 60 | cnt_docs = 0 |
| 61 | |
| 62 | # Validate and create knowledge directory if needed |
| 63 | if not knowledge_dir: |
| 64 | if log_item: |
| 65 | log_item.stream(progress="\nNo knowledge directory specified") |
| 66 | PrintStyle(font_color="yellow").print("No knowledge directory specified") |
| 67 | return index |
| 68 | |
| 69 | if not os.path.exists(knowledge_dir): |
| 70 | try: |
| 71 | os.makedirs(knowledge_dir, exist_ok=True) |
| 72 | # Verify the directory was actually created and is accessible |
| 73 | if not os.path.exists(knowledge_dir) or not os.access(knowledge_dir, os.R_OK): |
| 74 | error_msg = f"Knowledge directory {knowledge_dir} was created but is not accessible" |
| 75 | if log_item: |
| 76 | log_item.stream(progress=f"\n{error_msg}") |
| 77 | PrintStyle(font_color="red").print(error_msg) |
| 78 | return index |
| 79 | |
| 80 | if log_item: |
| 81 | log_item.stream(progress=f"\nCreated knowledge directory: {knowledge_dir}") |
| 82 | PrintStyle(font_color="green").print(f"Created knowledge directory: {knowledge_dir}") |
| 83 | except (OSError, PermissionError) as e: |
| 84 | error_msg = f"Failed to create knowledge directory {knowledge_dir}: {e}" |
| 85 | if log_item: |
| 86 | log_item.stream(progress=f"\n{error_msg}") |
| 87 | PrintStyle(font_color="red").print(error_msg) |
| 88 | return index |
| 89 | |
| 90 | # Final accessibility check for existing directories |
nothing calls this directly
no test coverage detected