Spaces:
Sleeping
Sleeping
File size: 2,753 Bytes
994a408 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 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 | """
Knowledge Base Service
ローカルファイルからKnowledge Baseを読み込む
"""
import os
import logging
from pathlib import Path
from typing import Optional
logger = logging.getLogger(__name__)
# 教科コードとファイル名のマッピング
SUBJECT_TO_FILE = {
"jp": "japanese.md",
"math": "math.txt",
"sci": "science.md",
"soc": "social.md"
}
# 教科コードと日本語名のマッピング
SUBJECT_TO_NAME = {
"jp": "国語",
"math": "算数",
"sci": "理科",
"soc": "社会"
}
class KnowledgeService:
"""Knowledge Base読み込みサービス"""
def __init__(self, knowledge_dir: str = "knowledge"):
self.knowledge_dir = Path(knowledge_dir)
self._cache = {} # ファイル内容キャッシュ
self._load_all()
def _load_all(self):
"""全教科のKBを読み込んでキャッシュ"""
for subject, filename in SUBJECT_TO_FILE.items():
filepath = self.knowledge_dir / filename
if filepath.exists():
try:
with open(filepath, "r", encoding="utf-8") as f:
self._cache[subject] = f.read()
logger.info(f"Loaded KB for {subject}: {len(self._cache[subject])} chars")
except Exception as e:
logger.error(f"Failed to load KB for {subject}: {e}")
self._cache[subject] = ""
else:
logger.warning(f"KB file not found: {filepath}")
self._cache[subject] = ""
def is_available(self) -> bool:
"""サービス利用可能かチェック"""
return any(len(content) > 0 for content in self._cache.values())
def get_knowledge(self, subject: str) -> str:
"""
指定教科のKnowledge Base内容を取得
Args:
subject: 教科コード (jp, math, sci, soc)
Returns:
Knowledge Baseの内容
"""
return self._cache.get(subject, "")
def get_subject_name(self, subject: str) -> str:
"""教科コードから日本語名を取得"""
return SUBJECT_TO_NAME.get(subject, subject)
def get_all_subjects(self) -> list:
"""利用可能な教科リストを取得"""
return [s for s, content in self._cache.items() if len(content) > 0]
def get_kb_stats(self) -> dict:
"""KB統計情報を取得"""
return {
subject: {
"chars": len(content),
"lines": content.count("\n") + 1 if content else 0
}
for subject, content in self._cache.items()
}
|