908bfe151f
- KB RAG 전체 파이프라인: 업로드, 파싱(xlsx/pdf/docx/text), 임베딩, Qdrant 인덱싱 - KB 관리 UI(14번 탭): 로그인, 문서 목록, 업로드, 삭제, 재인덱스 - OllamaController: 한글 시스템 프롬프트, plant_context.md 외부 파일화, SSE tool_start/tool_result 이벤트 - 프론트: 툴 실행 카드, KB 인용 링크, 표 자동 렌더, 추천 질문 칩 - nl2sql_worker: history_table.recorded_at 사용, tag_metadata 응답 개선 - DB: KB 테이블 5개 DDL + 시드, pgcrypto 확장
57 lines
1.4 KiB
Python
57 lines
1.4 KiB
Python
"""md / txt 청킹 — md는 # 헤딩 단위, txt는 빈 줄 두 개 단위."""
|
|
from __future__ import annotations
|
|
import os
|
|
|
|
|
|
def parse(path: str) -> list[dict]:
|
|
ext = os.path.splitext(path)[1].lower()
|
|
with open(path, "r", encoding="utf-8", errors="ignore") as f:
|
|
content = f.read()
|
|
|
|
if ext == ".md":
|
|
return _parse_md(content)
|
|
return _parse_txt(content)
|
|
|
|
|
|
def _parse_md(text: str) -> list[dict]:
|
|
chunks: list[dict] = []
|
|
lines = text.split("\n")
|
|
|
|
cur_heading = "preface"
|
|
buf: list[str] = []
|
|
section_idx = 0
|
|
|
|
def flush():
|
|
nonlocal section_idx
|
|
body = "\n".join(buf).strip()
|
|
if body:
|
|
section_idx += 1
|
|
chunks.append({
|
|
"text": body,
|
|
"chunk_kind": "heading",
|
|
"locator": f"heading={cur_heading}",
|
|
})
|
|
|
|
for ln in lines:
|
|
s = ln.lstrip()
|
|
if s.startswith("#"):
|
|
flush()
|
|
buf = []
|
|
cur_heading = s.lstrip("#").strip() or "section"
|
|
else:
|
|
buf.append(ln)
|
|
flush()
|
|
return chunks
|
|
|
|
|
|
def _parse_txt(text: str) -> list[dict]:
|
|
chunks: list[dict] = []
|
|
parts = [p.strip() for p in text.split("\n\n") if p.strip()]
|
|
for i, p in enumerate(parts, start=1):
|
|
chunks.append({
|
|
"text": p,
|
|
"chunk_kind": "paragraph",
|
|
"locator": f"paragraph={i}",
|
|
})
|
|
return chunks
|