""" ベクトルストア操作用のドキュメント読み込みユーティリティ。 """ import json import logging from typing import List from langchain_core.documents import Document logger = logging.getLogger(__name__) def load_chunks_from_json(json_file_path: str) -> List[Document]: """ JSONファイルからドキュメントチャンクを読み込み、LangChainのDocumentに変換する。 期待されるJSONフォーマット: [ { "metadata": { "title": "...", "file_name": "...", "source_url": "...", "chunk_index": 0 }, "page_content": "..." }, ... ] """ try: with open(json_file_path, 'r', encoding='utf-8') as f: chunks_data = json.load(f) documents = [] for chunk in chunks_data: # メタデータ付きでDocumentを作成 doc = Document( page_content=chunk['page_content'], metadata=chunk['metadata'] ) documents.append(doc) logger.info(f"{len(documents)}個のドキュメントチャンクを正常に読み込みました") return documents except FileNotFoundError: logger.error(f"エラー: {json_file_path} が見つかりません!") logger.error("カレントディレクトリにchunks.jsonが存在するか確認してください。") return [] except json.JSONDecodeError as e: logger.error(f"エラー: {json_file_path} のJSON形式が不正です") logger.error(f"JSONエラー: {e}") return [] except Exception as e: logger.error(f"{json_file_path} の読み込み中にエラーが発生しました: {e}") return [] def display_document_info(docs: List[Document], max_display: int = 5) -> None: """ 読み込んだドキュメントの情報を表示する。 Args: docs: 表示するドキュメントのリスト max_display: 表示する最大ドキュメント数(デフォルト: 5) """ logger.info(f"\n=== ドキュメント情報 ===") logger.info(f"総ドキュメント数: {len(docs)}") if not docs: logger.info("ドキュメントが見つかりませんでした。") return # 最大max_display件まで表示 for i, doc in enumerate(docs[:max_display]): meta = doc.metadata logger.info(f"\n--- ドキュメント {i+1} ---") logger.info(f"タイトル: {meta.get('title', 'N/A')}") logger.info(f"ファイル名: {meta.get('file_name', 'N/A')}") logger.info(f"URL: {meta.get('source_url', 'N/A')}") logger.info(f"チャンク番号: {meta.get('chunk_index', 'N/A')}") # 内容のプレビュー(最初の100文字) preview = doc.page_content[:100].replace('\n', ' ') logger.info(f"内容プレビュー: {preview}{'...' if len(doc.page_content) > 100 else ''}") # 残りのドキュメント数を表示 if len(docs) > max_display: logger.info(f"\n...他 {len(docs) - max_display} 件省略")