41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
"""RAG 检索服务（Phase 5）。本阶段以罐头桩先行；真实检索后置。"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Protocol, runtime_checkable
|
|
|
|
|
|
@runtime_checkable
|
|
class RagService(Protocol):
|
|
def retrieve_write_rules(self, chapter_id: str) -> list[str]: ...
|
|
def retrieve_design_rules(self, chapter_id: str) -> list[str]: ...
|
|
|
|
|
|
class CannedRagService:
|
|
"""从 samples/ 读入记入规则文档（Markdown），整体作为规则文本返回。"""
|
|
|
|
def __init__(self, samples_dir: str = "samples") -> None:
|
|
self._samples_dir = Path(samples_dir)
|
|
|
|
def _load_rules_text(self) -> list[str]:
|
|
texts: list[str] = []
|
|
for name in ("記入規則.docx", "概要設計做成説明書.docx"):
|
|
p = self._samples_dir / name
|
|
if not p.exists():
|
|
continue
|
|
try:
|
|
from genesis.parsers.rule_doc_parser import RuleDocParser
|
|
|
|
rule_doc = RuleDocParser().parse(str(p))
|
|
if rule_doc.markdown_content:
|
|
texts.append(rule_doc.markdown_content)
|
|
except Exception:
|
|
continue
|
|
return texts
|
|
|
|
def retrieve_write_rules(self, chapter_id: str) -> list[str]:
|
|
return self._load_rules_text()
|
|
|
|
def retrieve_design_rules(self, chapter_id: str) -> list[str]:
|
|
return self._load_rules_text()
|