vent_ai_platform/app/services/knowledge_hub_service.py

from __future__ import annotations

import asyncio
import re
import time
from collections import Counter
from collections.abc import AsyncIterator
from functools import lru_cache
from typing import Any
from urllib.parse import quote

import httpx
from sqlalchemy.orm import Session

from app.agents.prompts import (
    KNOWLEDGE_ASSISTANT_CHAT_PROMPT,
    KNOWLEDGE_ASSISTANT_CHAT_SYSTEM_PROMPT,
    KNOWLEDGE_RAG_PROMPT,
    KNOWLEDGE_RAG_SYSTEM_PROMPT,
)
from app.core.config import settings
from app.core.errors import APIError
from app.db.repositories import knowledge as knowledge_repo
from app.schemas.knowledge import (
    KnowledgeAskRequest,
    KnowledgeAskResponse,
    KnowledgeCatalogItem,
    KnowledgeCatalogResponse,
    KnowledgeCategoryStat,
    KnowledgeGraphEdge,
    KnowledgeGraphNode,
    KnowledgeGraphResponse,
    KnowledgeNoteChunk,
    KnowledgeNoteDetail,
    KnowledgeOverviewResponse,
    KnowledgeServiceStatus,
    KnowledgeSource,
)

CATEGORY_DEFS: tuple[tuple[str, str, str], ...] = (
    ("standards", "规范标准", "规程、标准、办法及技术要求"),
    ("cases", "事故案例", "典型事故与隐患处置案例"),
    ("expert", "专家经验", "专家规则、判识经验与处置建议"),
    ("management", "管理流程", "业务流程、责任岗位与闭环规则"),
)

CATEGORY_NAMES = {key: name for key, name, _ in CATEGORY_DEFS}

DOMAIN_TERMS: tuple[str, ...] = (
    "局部通风机",
    "局部通风",
    "主要通风机",
    "煤与瓦斯突出",
    "煤矿安全规程",
    "矿井防灭火",
    "矿井防尘",
    "监测监控",
    "掘进通风",
    "通风系统",
    "通风设施",
    "停风",
    "瓦斯超限",
    "瓦斯抽采",
    "粉尘防治",
    "自然发火",
    "一氧化碳",
    "二氧化碳",
    "循环风",
    "风筒",
    "风门",
    "密闭",
    "漏风",
    "风量",
    "风速",
    "阻力",
    "瓦斯",
    "甲烷",
    "防尘",
    "防火",
    "火灾",
    "传感器",
    "隐患",
    "处置",
    "标准",
    "规程",
    "措施",
    "判定",
)

GENERIC_QUERY_TERMS = {"处置", "标准", "规程", "措施", "判定"}

SECTION_RANGE_RE = re.compile(
    r"(\d+(?:\.\d+){1,5})\s*[—–~～至到\-一]\s*(\d+(?:\.\d+){1,5})"
)
SECTION_NUMBER_RE = re.compile(r"(?<![\d.])(\d{1,3}(?:\.\d{1,3}){2,5})(?![\d.])")
SECTION_HEADING_RE = re.compile(r"^\s*(\d+(?:\.\d+){1,5})(?:[\s、.．:：-]|$)")

CONVERSATIONAL_TERMS = (
    "你好",
    "您好",
    "您好呀",
    "在吗",
    "早上好",
    "上午好",
    "下午好",
    "晚上好",
    "谢谢",
    "感谢",
    "你是谁",
    "你叫什么",
    "你能做什么",
    "你可以做什么",
    "怎么使用",
    "帮助",
    "hello",
    "hi",
)

ACTION_WORDS: tuple[str, ...] = (
    "应当",
    "必须",
    "严禁",
    "不得",
    "立即",
    "停止",
    "撤出",
    "切断",
    "检查",
    "确认",
    "设置",
    "恢复",
    "复测",
    "处置",
    "排查",
    "调整",
    "关闭",
    "启动",
    "监测",
)


class KnowledgeHubClient:
    def __init__(
        self,
        base_url: str | None = None,
        api_key: str | None = None,
        timeout: float | None = None,
    ) -> None:
        configured_base = base_url if base_url is not None else settings.knowledge_service_base
        configured_key = api_key if api_key is not None else settings.knowledge_service_api_key
        self.base_url = configured_base.strip().rstrip("/")
        self.api_key = configured_key.strip()
        self.timeout = timeout if timeout is not None else settings.knowledge_service_timeout

    @property
    def configured(self) -> bool:
        return bool(self.base_url)

    async def _request(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, Any] | None = None,
        json: dict[str, Any] | None = None,
    ) -> Any:
        if not self.configured:
            raise APIError(503, "knowledge_service_not_configured", "知识库服务地址未配置")

        headers = {"Accept": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"

        timeout = httpx.Timeout(self.timeout, connect=min(self.timeout, 10.0))
        try:
            async with httpx.AsyncClient(
                base_url=self.base_url,
                headers=headers,
                timeout=timeout,
                follow_redirects=True,
            ) as client:
                response = await client.request(method, path, params=params, json=json)
                response.raise_for_status()
                if response.status_code == 204:
                    return None
                return response.json()
        except httpx.HTTPStatusError as exc:
            detail = exc.response.text[:500]
            raise APIError(
                502,
                "knowledge_service_http_error",
                f"知识库服务返回异常状态：{exc.response.status_code}",
                detail=detail,
            ) from exc
        except (httpx.RequestError, ValueError) as exc:
            raise APIError(
                503,
                "knowledge_service_unavailable",
                "知识库服务暂时不可用",
                detail=f"{type(exc).__name__}: {exc}",
            ) from exc

    async def status(self) -> KnowledgeServiceStatus:
        if not self.configured:
            return KnowledgeServiceStatus(
                base_url="",
                error="知识库服务地址未配置",
            )
        try:
            payload = await self._request("GET", "/api/admin/status")
        except APIError as exc:
            return KnowledgeServiceStatus(
                base_url=self.base_url,
                error=exc.message,
            )

        stats = payload.get("stats") or {}
        embedding_model = str(payload.get("embedding_model") or "")
        return KnowledgeServiceStatus(
            available=True,
            base_url=self.base_url,
            mode=str(payload.get("mode") or ""),
            status=str(payload.get("status") or ""),
            total_notes=int(stats.get("total_notes") or 0),
            total_chunks=int(stats.get("total_chunks") or 0),
            embedding_model=embedding_model,
            semantic_enabled=embedding_model.lower() not in {"", "disabled", "none"},
            last_indexed=stats.get("last_indexed"),
        )

    async def tags(self) -> list[dict[str, Any]]:
        payload = await self._request("GET", "/api/tags")
        return list(payload.get("tags") or [])

    async def notes_page(self, page: int, page_size: int = 200) -> dict[str, Any]:
        return await self._request(
            "GET",
            "/api/notes",
            params={"page": page, "page_size": page_size},
        )

    async def note(self, note_id: str) -> dict[str, Any]:
        return await self._request("GET", f"/api/notes/{quote(note_id, safe='/')}")

    async def note_full(self, note_id: str) -> dict[str, Any]:
        return await self._request("POST", "/api/note_full", json={"note_id": note_id})

    async def search(
        self,
        query: str,
        *,
        mode: str = "keyword",
        top_k: int = 5,
        tag: str | None = None,
        min_score: float = 0.0,
        min_match_ratio: float = 0.0,
    ) -> dict[str, Any]:
        params: dict[str, Any] = {
            "q": query,
            "mode": mode,
            "top_k": top_k,
            "min_score": min_score,
            "min_match_ratio": min_match_ratio,
        }
        if tag:
            params["tag"] = tag
        return await self._request("GET", "/api/search", params=params)


@lru_cache(maxsize=1)
def get_knowledge_hub_client() -> KnowledgeHubClient:
    return KnowledgeHubClient()


_STATUS_CACHE: tuple[float, KnowledgeServiceStatus] | None = None
_CATALOG_CACHE: tuple[float, list[dict[str, Any]]] | None = None
_CATALOG_LOCKS: dict[int, asyncio.Lock] = {}


def reset_knowledge_hub_state() -> None:
    global _STATUS_CACHE, _CATALOG_CACHE, _CATALOG_LOCKS
    _STATUS_CACHE = None
    _CATALOG_CACHE = None
    _CATALOG_LOCKS = {}
    get_knowledge_hub_client.cache_clear()


async def fetch_status(*, force: bool = False) -> KnowledgeServiceStatus:
    global _STATUS_CACHE
    now = time.monotonic()
    if not force and _STATUS_CACHE and now - _STATUS_CACHE[0] < 30:
        return _STATUS_CACHE[1]
    status = await get_knowledge_hub_client().status()
    _STATUS_CACHE = (now, status)
    return status


async def fetch_catalog(*, force: bool = False) -> list[dict[str, Any]]:
    global _CATALOG_CACHE
    now = time.monotonic()
    if (
        not force
        and _CATALOG_CACHE
        and now - _CATALOG_CACHE[0] < max(settings.knowledge_catalog_cache_ttl, 1)
    ):
        return list(_CATALOG_CACHE[1])

    lock = _CATALOG_LOCKS.setdefault(id(asyncio.get_running_loop()), asyncio.Lock())
    async with lock:
        now = time.monotonic()
        if (
            not force
            and _CATALOG_CACHE
            and now - _CATALOG_CACHE[0] < max(settings.knowledge_catalog_cache_ttl, 1)
        ):
            return list(_CATALOG_CACHE[1])

        client = get_knowledge_hub_client()
        first = await client.notes_page(page=1, page_size=200)
        notes = list(first.get("notes") or [])
        total = int(first.get("total") or len(notes))
        page = 1
        while len(notes) < total and page < 20:
            page += 1
            payload = await client.notes_page(page=page, page_size=200)
            batch = list(payload.get("notes") or [])
            if not batch:
                break
            notes.extend(batch)

        _CATALOG_CACHE = (now, notes)
        return list(notes)


def classify_note(note: dict[str, Any]) -> tuple[str, str]:
    note_id = str(note.get("note_id") or "")
    title = str(note.get("title") or "")
    tags = [str(item) for item in note.get("tags") or []]
    haystack = " ".join([note_id, title, *tags])

    if note_id.startswith(("queries/", "references/")) or any(
        tag == "workflow" or tag.startswith("workflow/") for tag in tags
    ):
        return "management", CATEGORY_NAMES["management"]
    if re.search(r"案例|典型事故|事故复盘", title) and "事故隐患判定" not in title:
        return "cases", CATEGORY_NAMES["cases"]
    if note_id.startswith(("concepts/", "entities/")):
        return "expert", CATEGORY_NAMES["expert"]
    if (
        note_id.startswith("raw/")
        or any(
            tag in {"source/standard", "source/regulation"}
            or tag.startswith(("regulation/", "standard/"))
            for tag in tags
        )
        or re.search(r"规程|规范|标准|细则|办法|技术条件|技术要求|MT/?T|AQ/?|GB/?", haystack, re.I)
    ):
        return "standards", CATEGORY_NAMES["standards"]
    return "other", "其他"


def catalog_item(note: dict[str, Any]) -> KnowledgeCatalogItem:
    category, category_name = classify_note(note)
    return KnowledgeCatalogItem(
        note_id=str(note.get("note_id") or ""),
        title=str(note.get("title") or note.get("note_id") or ""),
        category=category,
        category_name=category_name,
        tags=[str(item) for item in note.get("tags") or []],
        aliases=[str(item) for item in note.get("aliases") or []],
        file_path=str(note.get("file_path") or ""),
        updated=str(note.get("updated") or note.get("created") or ""),
        chunk_count=int(note.get("chunk_count") or 0),
        confidence=str(note.get("confidence") or ""),
        status=str(note.get("status") or ""),
        version=str(note.get("version") or ""),
    )


def _category_filter(note: dict[str, Any], category: str | None) -> bool:
    if not category or category == "all":
        return True
    return classify_note(note)[0] == category


def _keyword_filter(note: dict[str, Any], keyword: str) -> bool:
    needle = keyword.casefold()
    values = [
        str(note.get("note_id") or ""),
        str(note.get("title") or ""),
        str(note.get("file_path") or ""),
        *[str(item) for item in note.get("tags") or []],
        *[str(item) for item in note.get("aliases") or []],
    ]
    return needle in " ".join(values).casefold()


def _is_index_note(note: dict[str, Any]) -> bool:
    note_id = str(note.get("note_id") or "").casefold()
    title = str(note.get("title") or "").strip()
    return note_id.endswith("-index") or title.startswith("Source:")


async def build_overview() -> KnowledgeOverviewResponse:
    status = await fetch_status()
    try:
        notes = await fetch_catalog()
    except APIError:
        notes = []

    category_counts = Counter(classify_note(note)[0] for note in notes)
    categories = [
        KnowledgeCategoryStat(
            key=key,
            name=name,
            count=category_counts.get(key, 0),
            description=description,
            available=category_counts.get(key, 0) > 0,
        )
        for key, name, description in CATEGORY_DEFS
    ]
    return KnowledgeOverviewResponse(
        service=status,
        total_notes=status.total_notes or len(notes),
        total_chunks=status.total_chunks,
        category_counts=categories,
        search_mode=settings.knowledge_search_mode,
        min_score=settings.knowledge_search_min_score,
        min_match_ratio=settings.knowledge_search_min_match_ratio,
        last_indexed=status.last_indexed,
    )


async def build_catalog(
    *,
    page: int,
    page_size: int,
    category: str | None = None,
    keyword: str | None = None,
    tag: str | None = None,
) -> KnowledgeCatalogResponse:
    status = await fetch_status()
    notes = await fetch_catalog()
    filtered = [
        note
        for note in notes
        if _category_filter(note, category)
        and (not tag or tag in {str(item) for item in note.get("tags") or []})
        and (not keyword or _keyword_filter(note, keyword))
    ]
    filtered.sort(
        key=lambda note: (
            str(note.get("updated") or ""),
            float(note.get("mtime") or 0),
        ),
        reverse=True,
    )
    filtered.sort(key=_is_index_note)
    total = len(filtered)
    start = (page - 1) * page_size
    return KnowledgeCatalogResponse(
        items=[catalog_item(note) for note in filtered[start : start + page_size]],
        total=total,
        page=page,
        page_size=page_size,
        service=status,
    )


async def build_note_detail(note_id: str) -> KnowledgeNoteDetail:
    notes = await fetch_catalog()
    note = next((item for item in notes if str(item.get("note_id")) == note_id), None)
    if note is None:
        raise APIError(404, "knowledge_note_not_found", "知识条目不存在")

    client = get_knowledge_hub_client()
    full_payload, detail_payload = await asyncio.gather(
        client.note_full(note_id),
        client.note(note_id),
        return_exceptions=True,
    )
    if isinstance(full_payload, Exception):
        raise full_payload
    detail = note if isinstance(detail_payload, Exception) else {**note, **detail_payload}
    item = catalog_item(detail)
    chunks = [
        KnowledgeNoteChunk(
            chunk_id=str(chunk.get("chunk_id") or ""),
            heading=str(chunk.get("heading") or ""),
            content=clean_knowledge_text(str(chunk.get("content") or "")),
        )
        for chunk in full_payload.get("chunks") or []
    ]
    links = detail.get("backlinks") or detail.get("links") or []
    return KnowledgeNoteDetail(
        **item.model_dump(),
        full_content=clean_knowledge_text(str(full_payload.get("full_content") or "")),
        links=list(links),
        chunks=chunks,
    )


def extract_query_terms(question: str) -> list[str]:
    matched = [term for term in DOMAIN_TERMS if term in question]
    if matched:
        matched.sort(key=lambda term: (question.find(term), -len(term)))
        return list(dict.fromkeys(matched))[:4]

    cleaned = re.sub(
        r"请问|如何|应该|应该怎样|怎么|怎样|什么|哪些|是否|可以|需要|处置|处理|规定|标准|要求|后",
        " ",
        question,
    )
    segments = re.findall(r"[\u4e00-\u9fffA-Za-z0-9]{2,12}", cleaned)
    return list(dict.fromkeys(segments))[:4]


def _normalize_search_mode(requested_mode: str | None, status: KnowledgeServiceStatus) -> str:
    mode = (requested_mode or settings.knowledge_search_mode or "keyword").strip().lower()
    if mode not in {"keyword", "semantic", "hybrid"}:
        mode = "keyword"
    if not status.semantic_enabled and mode != "keyword":
        return "keyword"
    return mode


def _candidate_from_result(result: dict[str, Any]) -> dict[str, Any]:
    category, category_name = classify_note(
        {
            "note_id": result.get("note_id"),
            "title": result.get("title"),
            "tags": [],
        }
    )
    return {
        "chunk_id": str(result.get("chunk_id") or ""),
        "note_id": str(result.get("note_id") or ""),
        "title": str(result.get("title") or result.get("note_id") or ""),
        "file_path": str(result.get("file_path") or ""),
        "content": clean_knowledge_text(str(result.get("content") or "")),
        "heading": clean_knowledge_text(str(result.get("heading") or "")),
        "score": float(result.get("score") or 0),
        "category": category,
        "category_name": category_name,
    }


async def search_candidates(
    question: str,
    *,
    requested_mode: str | None = None,
    top_k: int = 5,
    tag: str | None = None,
) -> tuple[list[dict[str, Any]], str, bool]:
    status = await fetch_status()
    mode = _normalize_search_mode(requested_mode, status)
    client = get_knowledge_hub_client()
    terms = extract_query_terms(question)
    specific_terms = [term for term in terms if term not in GENERIC_QUERY_TERMS]
    queries = list(dict.fromkeys([question, *(specific_terms or terms)[:3]]))

    responses = await asyncio.gather(
        *[
            client.search(
                query,
                mode=mode,
                top_k=max(top_k, 5),
                tag=tag,
                min_score=settings.knowledge_search_min_score,
                min_match_ratio=settings.knowledge_search_min_match_ratio,
            )
            for query in queries
        ],
        return_exceptions=True,
    )

    candidates: dict[str, dict[str, Any]] = {}
    unavailable = False
    for response in responses:
        if isinstance(response, Exception):
            unavailable = True
            continue
        for result in response.get("results") or []:
            candidate = _candidate_from_result(result)
            key = candidate["chunk_id"] or f"{candidate['note_id']}:{candidate['content'][:40]}"
            current = candidates.get(key)
            if current is None or candidate["score"] > current["score"]:
                candidates[key] = candidate

    ranked = sorted(
        candidates.values(),
        key=lambda item: (float(item.get("score") or 0), len(item.get("content") or "")),
        reverse=True,
    )
    return ranked[: max(top_k * 3, top_k)], mode, unavailable


def _local_search_candidates(
    db: Session,
    question: str,
    *,
    top_k: int,
) -> list[dict[str, Any]]:
    candidates: dict[str, dict[str, Any]] = {}
    for keyword in list(dict.fromkeys([question, *extract_query_terms(question)]))[:4]:
        items, _ = knowledge_repo.list_knowledge(
            db,
            keyword=keyword,
            page=1,
            page_size=top_k,
        )
        for item in items:
            score = 0.0
            if keyword and keyword.casefold() in item.title.casefold():
                score = 0.8
            elif keyword and keyword.casefold() in (item.summary or "").casefold():
                score = 0.65
            elif keyword and keyword.casefold() in (item.content or "").casefold():
                score = 0.5
            candidate = {
                "chunk_id": item.id,
                "note_id": item.id,
                "title": item.title,
                "file_path": item.source_ref,
                "content": clean_knowledge_text(item.content or item.summary or ""),
                "heading": item.category,
                "score": score,
                "category": "other",
                "category_name": item.category or "其他",
            }
            current = candidates.get(item.id)
            if current is None or candidate["score"] > current["score"]:
                candidates[item.id] = candidate
    return sorted(
        candidates.values(),
        key=lambda item: float(item.get("score") or 0),
        reverse=True,
    )[:top_k]


def clean_knowledge_text(value: str) -> str:
    text = str(value or "")
    text = re.sub(r"---\s*PAGE\s*\d+\s*---", "\n", text, flags=re.I)
    text = re.sub(r"\[\[([^|\]]+)\|([^\]]+)\]\]", r"\2", text)
    text = re.sub(r"\[\[([^\]]+)\]\]", r"\1", text)
    text = re.sub(r"(?m)^\s{0,3}#{1,6}\s*", "", text)
    text = re.sub(r"\s+(?=[，。；：！？、])", "", text)
    text = re.sub(r"(?<=[\u4e00-\u9fff])\s+(?=[\u4e00-\u9fff])", "", text)
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()


def _split_sentences(text: str) -> list[str]:
    parts = re.split(r"(?<=[。！？；])|\n+", clean_knowledge_text(text))
    return [part.strip(" -;\t\r\n") for part in parts if len(part.strip()) >= 8]


def _rank_sentences(
    text: str,
    terms: list[str],
    *,
    formula_mode: bool = False,
) -> list[str]:
    ranked: list[tuple[float, str]] = []
    for sentence in _split_sentences(text):
        score = 0.0
        score += sum(2.5 for term in terms if term and term in sentence)
        score += sum(1.5 for word in ACTION_WORDS if word in sentence)
        score += min(len(sentence) / 80, 1.2)
        if formula_mode:
            score += 2.5 * len(
                re.findall(
                    r"公式|式\s*[（(]?\d+|m3/min|m³/min|Q[_\u4e00-\u9fffA-Za-z]",
                    sentence,
                    flags=re.I,
                )
            )
            score += 1.2 * len(re.findall(r"\d+(?:\.\d+)?", sentence))
        ranked.append((score, sentence))
    ranked.sort(key=lambda item: item[0], reverse=True)
    return list(dict.fromkeys(sentence for _, sentence in ranked))


def _extract_steps(answer: str, terms: list[str]) -> list[str]:
    steps: list[str] = []
    for line in answer.splitlines():
        cleaned = re.sub(r"^\s*(?:[-*]|\d+[.、)])\s*", "", line).strip()
        if len(cleaned) >= 8 and any(word in cleaned for word in ACTION_WORDS):
            steps.append(cleaned[:160])
    if len(steps) >= 2:
        return list(dict.fromkeys(steps))[:4]
    for sentence in _rank_sentences(answer, terms):
        if any(word in sentence for word in ACTION_WORDS):
            steps.append(sentence[:160])
        if len(steps) >= 3:
            break
    return list(dict.fromkeys(steps))


def extractive_answer(candidates: list[dict[str, Any]], question: str) -> tuple[str, list[str]]:
    if not candidates:
        return (
            "暂未在知识库中检索到与该问题直接相关的内容，"
            "请补充标准编号、设备名称或业务场景后重试。",
            [],
        )

    terms = extract_query_terms(question)
    sentences: list[str] = []
    for candidate in candidates[:4]:
        sentences.extend(_rank_sentences(candidate.get("content") or "", terms)[:3])
    sentences = list(dict.fromkeys(sentences))[:5]
    if not sentences:
        sentences = [str(candidates[0].get("content") or "")[:300]]
    answer = "".join(sentences)[:900]
    return answer, _extract_steps(answer, terms)


def _build_rag_context(
    question: str,
    candidates: list[dict[str, Any]],
) -> tuple[str, int]:
    terms = extract_query_terms(question)
    formula_mode = any(word in question for word in ("公式", "计算", "参数", "系数", "方法"))
    max_total = max(settings.knowledge_rag_context_max_chars, 2000)
    max_source = max(settings.knowledge_rag_source_max_chars, 500)
    blocks: list[str] = []
    total_chars = 0
    for index, item in enumerate(candidates):
        content = clean_knowledge_text(str(item.get("content") or ""))
        ranked = _rank_sentences(content, terms, formula_mode=formula_mode)
        source_text = "".join(ranked[:20])[:max_source] or content[:max_source]
        block = (
            f"[{index + 1}] 标题：{item.get('title') or item.get('note_id')}\n"
            f"来源：{item.get('file_path') or item.get('note_id')}\n"
            f"内容：{source_text}"
        )
        remaining = max_total - total_chars
        if remaining < 300:
            break
        if len(block) > remaining:
            block = block[:remaining]
        blocks.append(block)
        total_chars += len(block)
    return "\n\n".join(blocks), total_chars


def _section_tuple(value: str) -> tuple[int, ...]:
    return tuple(int(part) for part in value.split("."))


def _extract_section_ranges(text: str) -> list[tuple[str, str]]:
    ranges = [
        (match.group(1), match.group(2))
        for match in SECTION_RANGE_RE.finditer(text)
    ]
    if ranges:
        return list(dict.fromkeys(ranges))[:4]
    numbers = list(dict.fromkeys(SECTION_NUMBER_RE.findall(text)))
    return [(number, number) for number in numbers[:4]]


def _extract_section_blocks(text: str, start_ref: str, end_ref: str) -> str:
    content = clean_knowledge_text(text)
    start_tuple = _section_tuple(start_ref)
    end_tuple = _section_tuple(end_ref)
    selected: list[str] = []
    in_range = False
    for line in content.splitlines():
        heading = SECTION_HEADING_RE.match(line)
        if heading:
            current_tuple = _section_tuple(heading.group(1))
            if not in_range and start_tuple <= current_tuple <= end_tuple:
                in_range = True
            elif in_range and current_tuple > end_tuple:
                break
        if in_range:
            selected.append(line)
    if selected:
        return "\n".join(selected)[:8000]

    start_index = content.find(start_ref)
    if start_index < 0:
        return ""
    return content[start_index : start_index + 3000]


def _ranges_for_anchor(
    anchor_content: str,
    full_content: str,
) -> list[tuple[str, str]]:
    anchor_numbers = [
        _section_tuple(number)
        for number in SECTION_NUMBER_RE.findall(anchor_content)
    ]
    if not anchor_numbers:
        return _extract_section_ranges(anchor_content)

    matched: list[tuple[str, str]] = []
    for match in SECTION_RANGE_RE.finditer(full_content):
        start_ref, end_ref = match.group(1), match.group(2)
        start_tuple = _section_tuple(start_ref)
        end_tuple = _section_tuple(end_ref)
        for number in anchor_numbers:
            if start_tuple <= number <= end_tuple:
                matched.append((start_ref, end_ref))
                break
            if number[:2] == start_tuple[:2] == end_tuple[:2]:
                matched.append((start_ref, end_ref))
                break
    if matched:
        return list(dict.fromkeys(matched))[:4]
    return _extract_section_ranges(anchor_content)


async def _expand_section_context(
    candidates: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    references: dict[str, str] = {}
    for item in candidates:
        note_id = str(item.get("note_id") or "")
        if not note_id:
            continue
        content = str(item.get("content") or "")
        if _extract_section_ranges(content) or SECTION_NUMBER_RE.search(content):
            references[note_id] = content
    if not references:
        return candidates

    client = get_knowledge_hub_client()
    note_ids = list(references)
    payloads = await asyncio.gather(
        *[client.note_full(note_id) for note_id in note_ids],
        return_exceptions=True,
    )
    expanded_blocks: dict[str, str] = {}
    for note_id, payload in zip(note_ids, payloads, strict=False):
        if isinstance(payload, Exception):
            continue
        full_content = str(payload.get("full_content") or "")
        ranges = _ranges_for_anchor(references[note_id], full_content)
        blocks = [
            _extract_section_blocks(full_content, start_ref, end_ref)
            for start_ref, end_ref in ranges
        ]
        content = "\n\n".join(block for block in blocks if block)
        if content:
            expanded_blocks[note_id] = content[:8000]

    if not expanded_blocks:
        return candidates
    expanded: list[dict[str, Any]] = []
    for item in candidates:
        note_id = str(item.get("note_id") or "")
        section_content = expanded_blocks.get(note_id)
        if not section_content:
            expanded.append(item)
            continue
        expanded.append(
            {
                **item,
                "content": (
                    f"{item.get('content') or ''}\n\n"
                    f"[章节展开]\n{section_content}"
                ),
                "heading": f"{item.get('heading') or ''} · 条款展开",
            }
        )
    return expanded


async def _generated_answer(
    question: str,
    candidates: list[dict[str, Any]],
) -> tuple[str, list[str], str, int] | None:
    if not candidates:
        return None

    context, context_chars = _build_rag_context(question, candidates)
    if not context:
        return None
    prompt = f"{KNOWLEDGE_RAG_PROMPT}\n\n用户问题：{question}\n\n检索资料：\n{context}"
    answer = await _model_text(
        prompt,
        system_prompt=KNOWLEDGE_RAG_SYSTEM_PROMPT,
    )
    if not answer:
        return None
    return (
        answer,
        _extract_steps(answer, extract_query_terms(question)),
        settings.deepagent_model,
        context_chars,
    )


def is_conversational_question(question: str) -> bool:
    normalized = re.sub(r"[\s，。！？!?,.、]+", "", question).lower()
    if not normalized or len(normalized) > 24:
        return False
    if any(term in normalized for term in DOMAIN_TERMS):
        return False
    return any(term in normalized for term in CONVERSATIONAL_TERMS)


async def _conversational_answer(question: str) -> tuple[str, str] | None:
    answer = await _model_text(
        f"{KNOWLEDGE_ASSISTANT_CHAT_PROMPT}\n\n用户：{question}",
        system_prompt=KNOWLEDGE_ASSISTANT_CHAT_SYSTEM_PROMPT,
    )
    if not answer:
        return None
    return answer, settings.deepagent_model


def _conversational_fallback() -> str:
    return (
        "你好，我是通防知识助手。你可以询问规程标准、通风、瓦斯、防灭火、"
        "防尘、监测监控或隐患处置等通防业务问题。"
    )


def _sources_from_candidates(
    candidates: list[dict[str, Any]],
) -> list[KnowledgeSource]:
    return [
        KnowledgeSource(
            citation_index=index + 1,
            note_id=item["note_id"],
            title=item["title"],
            file_path=item["file_path"],
            score=round(float(item["score"] or 0), 4),
            heading=item["heading"],
            snippet=item["content"][:320],
            category=item["category"],
            category_name=item["category_name"],
        )
        for index, item in enumerate(candidates)
    ]


async def _prepare_business_answer(
    db: Session,
    payload: KnowledgeAskRequest,
) -> dict[str, Any]:
    requested_mode = payload.search_mode or settings.knowledge_search_mode
    retrieval_started = time.perf_counter()
    candidates, mode, unavailable = await search_candidates(
        payload.question,
        requested_mode=requested_mode,
        top_k=payload.top_k,
        tag=payload.tag,
    )
    if payload.category:
        candidates = [
            item
            for item in candidates
            if classify_note({"note_id": item["note_id"], "title": item["title"]})[0]
            == payload.category
        ]

    degraded = unavailable
    if not candidates:
        candidates = _local_search_candidates(
            db,
            payload.question,
            top_k=payload.top_k,
        )
        degraded = bool(candidates)
    if payload.generate_answer:
        candidates = await _expand_section_context(candidates)
        candidates.sort(
            key=lambda item: (
                "[章节展开]" in str(item.get("content") or ""),
                float(item.get("score") or 0),
            ),
            reverse=True,
        )
    return {
        "requested_mode": requested_mode,
        "search_mode": mode,
        "degraded": degraded,
        "candidates": candidates,
        "context_candidates": candidates[: payload.top_k],
        "retrieval_ms": round((time.perf_counter() - retrieval_started) * 1000, 1),
    }


async def _model_text(
    prompt: str,
    *,
    system_prompt: str = KNOWLEDGE_RAG_SYSTEM_PROMPT,
) -> str:
    if not settings.model_configured:
        return ""

    from openai import AsyncOpenAI

    client = AsyncOpenAI(
        api_key=settings.openai_api_key,
        base_url=settings.openai_base_url or None,
        timeout=settings.openai_timeout,
    )
    response = await client.chat.completions.create(
        model=settings.deepagent_model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt},
        ],
        stream=False,
    )
    if not response.choices:
        return ""
    return str(response.choices[0].message.content or "").strip()


async def _stream_model_deltas(
    prompt: str,
    *,
    system_prompt: str = KNOWLEDGE_RAG_SYSTEM_PROMPT,
) -> AsyncIterator[tuple[str, str]]:
    if not settings.model_configured:
        return

    from openai import AsyncOpenAI

    client = AsyncOpenAI(
        api_key=settings.openai_api_key,
        base_url=settings.openai_base_url or None,
        timeout=settings.openai_timeout,
    )
    stream = await client.chat.completions.create(
        model=settings.deepagent_model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt},
        ],
        stream=True,
    )
    try:
        async for chunk in stream:
            if not chunk.choices:
                continue
            delta = chunk.choices[0].delta
            reasoning = getattr(delta, "reasoning_content", None)
            if reasoning:
                yield "thinking", str(reasoning)
            content = delta.content
            if content:
                yield "answer", str(content)
    finally:
        await stream.close()


async def stream_answer(
    db: Session,
    payload: KnowledgeAskRequest,
) -> AsyncIterator[dict[str, Any]]:
    started = time.perf_counter()
    if is_conversational_question(payload.question):
        yield {
            "event": "meta",
            "data": {
                "question": payload.question,
                "answer_status": "conversational",
                "search_mode": "conversation",
                "sources": [],
            },
        }
        answer = ""
        generation_started = time.perf_counter()
        if payload.generate_answer:
            try:
                async for event_type, delta in _stream_model_deltas(
                    f"{KNOWLEDGE_ASSISTANT_CHAT_PROMPT}\n\n用户：{payload.question}",
                    system_prompt=KNOWLEDGE_ASSISTANT_CHAT_SYSTEM_PROMPT,
                ):
                    if event_type == "thinking":
                        yield {"event": "thinking", "data": {"content": delta}}
                    else:
                        answer += delta
                        yield {"event": "delta", "data": {"content": delta}}
            except Exception as exc:
                yield {
                    "event": "warning",
                    "data": {"message": f"大模型流式生成失败：{type(exc).__name__}"},
                }
        if not answer:
            answer = _conversational_fallback()
            yield {"event": "delta", "data": {"content": answer}}
        yield {
            "event": "done",
            "data": {
                "answer": answer,
                "answer_status": "conversational",
                "answer_model": settings.deepagent_model if payload.generate_answer else "",
                "sources": [],
                "steps": [],
                "generation_ms": round((time.perf_counter() - generation_started) * 1000, 1),
                "latency_ms": round((time.perf_counter() - started) * 1000, 1),
            },
        }
        return

    prepared = await _prepare_business_answer(db, payload)
    candidates = prepared["candidates"]
    context_candidates = prepared["context_candidates"]
    sources = _sources_from_candidates(context_candidates)
    yield {
        "event": "meta",
        "data": {
            "question": payload.question,
            "answer_status": "generating" if payload.generate_answer else "extractive",
            "requested_mode": str(prepared["requested_mode"] or "keyword"),
            "search_mode": prepared["search_mode"],
            "retrieval_ms": prepared["retrieval_ms"],
            "total_candidates": len(candidates),
            "degraded": prepared["degraded"],
            "sources": [source.model_dump() for source in sources],
        },
    }

    answer = ""
    answer_model = ""
    context_chars = 0
    generation_started = time.perf_counter()
    if payload.generate_answer and context_candidates:
        context, context_chars = _build_rag_context(payload.question, context_candidates)
        prompt = (
            f"{KNOWLEDGE_RAG_PROMPT}\n\n用户问题：{payload.question}"
            f"\n\n检索资料：\n{context}"
        )
        try:
            async for event_type, delta in _stream_model_deltas(
                prompt,
                system_prompt=KNOWLEDGE_RAG_SYSTEM_PROMPT,
            ):
                if event_type == "thinking":
                    yield {"event": "thinking", "data": {"content": delta}}
                else:
                    answer += delta
                    yield {"event": "delta", "data": {"content": delta}}
            if answer:
                answer_model = settings.deepagent_model
        except Exception as exc:
            yield {
                "event": "warning",
                "data": {"message": f"大模型流式生成失败：{type(exc).__name__}"},
            }

    if not answer:
        answer, steps = extractive_answer(candidates, payload.question)
        yield {"event": "delta", "data": {"content": answer}}
        answer_status = "no_result" if not candidates else "extractive"
    else:
        steps = _extract_steps(answer, extract_query_terms(payload.question))
        answer_status = "generated"

    yield {
        "event": "done",
        "data": {
            "answer": answer,
            "answer_status": answer_status,
            "answer_model": answer_model,
            "sources": [source.model_dump() for source in sources],
            "steps": steps,
            "retrieval_ms": prepared["retrieval_ms"],
            "context_chars": context_chars,
            "generation_ms": round((time.perf_counter() - generation_started) * 1000, 1),
            "latency_ms": round((time.perf_counter() - started) * 1000, 1),
            "total_candidates": len(candidates),
            "degraded": prepared["degraded"],
        },
    }


async def build_answer(
    db: Session,
    payload: KnowledgeAskRequest,
) -> KnowledgeAskResponse:
    started = time.perf_counter()
    if is_conversational_question(payload.question):
        generation_started = time.perf_counter()
        conversational = None
        if payload.generate_answer:
            conversational = await _conversational_answer(payload.question)
        generation_ms = round((time.perf_counter() - generation_started) * 1000, 1)
        answer, answer_model = (
            conversational
            if conversational is not None
            else (_conversational_fallback(), "")
        )
        return KnowledgeAskResponse(
            question=payload.question,
            answer=answer,
            answer_status="conversational",
            answer_model=answer_model,
            generation_enabled=payload.generate_answer,
            requested_mode="conversation",
            search_mode="conversation",
            generation_ms=generation_ms,
            latency_ms=round((time.perf_counter() - started) * 1000, 1),
        )

    prepared = await _prepare_business_answer(db, payload)
    candidates = prepared["candidates"]
    mode = prepared["search_mode"]
    degraded = prepared["degraded"]
    retrieval_ms = prepared["retrieval_ms"]

    answer_model = ""
    answer: str
    steps: list[str]
    answer_status = "extractive"
    generation_ms = 0.0
    context_chars = 0
    context_candidates = prepared["context_candidates"]
    if payload.generate_answer:
        generation_started = time.perf_counter()
        generated = await _generated_answer(payload.question, context_candidates)
        generation_ms = round((time.perf_counter() - generation_started) * 1000, 1)
        if generated is None:
            answer, steps = extractive_answer(candidates, payload.question)
        else:
            answer, steps, answer_model, context_chars = generated
            answer_status = "generated"
    else:
        answer, steps = extractive_answer(candidates, payload.question)

    if not candidates:
        answer_status = "no_result"

    sources = _sources_from_candidates(context_candidates)
    return KnowledgeAskResponse(
        question=payload.question,
        answer=answer,
        answer_status=answer_status,
        answer_model=answer_model,
        generation_enabled=payload.generate_answer,
        steps=steps,
        sources=sources,
        requested_mode=str(prepared["requested_mode"] or "keyword"),
        search_mode=mode,
        retrieval_ms=retrieval_ms,
        generation_ms=generation_ms,
        context_chars=context_chars,
        latency_ms=round((time.perf_counter() - started) * 1000, 1),
        total_candidates=len(candidates),
        degraded=degraded,
    )


def _matching_notes(
    notes: list[dict[str, Any]],
    *,
    tags: set[str],
    keywords: tuple[str, ...] = (),
) -> list[dict[str, Any]]:
    matches: list[dict[str, Any]] = []
    for note in notes:
        note_tags = {str(tag) for tag in note.get("tags") or []}
        title = str(note.get("title") or "")
        note_id = str(note.get("note_id") or "")
        if note_tags & tags or any(keyword in title or keyword in note_id for keyword in keywords):
            matches.append(note)
    return matches


async def build_graph(topic: str = "瓦斯超限风险事件") -> KnowledgeGraphResponse:
    try:
        notes = await fetch_catalog()
    except APIError:
        notes = []

    groups = {
        "risk_gas": _matching_notes(notes, tags={"safety/gas"}, keywords=("瓦斯",)),
        "local_fan": _matching_notes(
            notes,
            tags={"ventilation/local-fan"},
            keywords=("局部通风机", "循环风", "停风"),
        ),
        "door": _matching_notes(
            notes,
            tags={"ventilation/airflow"},
            keywords=("风门", "漏风", "通风设施"),
        ),
        "airflow": _matching_notes(notes, tags={"ventilation/airflow"}, keywords=("风量",)),
        "monitoring": _matching_notes(
            notes,
            tags={"safety/monitoring", "safety/监控"},
            keywords=("监测监控", "传感器"),
        ),
    }

    def samples(group: list[dict[str, Any]]) -> list[str]:
        return [
            str(note.get("title") or note.get("note_id") or "")
            for note in group[:3]
        ]

    nodes = [
        KnowledgeGraphNode(
            id="risk_gas",
            label="瓦斯超限",
            sub_label="风险事件",
            node_type="risk",
            value=len(groups["risk_gas"]),
            sample_notes=samples(groups["risk_gas"]),
            x=280,
            y=180,
        ),
        KnowledgeGraphNode(
            id="local_fan",
            label="局部通风机",
            sub_label="停风/循环风",
            node_type="cause",
            value=len(groups["local_fan"]),
            sample_notes=samples(groups["local_fan"]),
            x=126,
            y=88,
        ),
        KnowledgeGraphNode(
            id="door",
            label="风门",
            sub_label="漏风/失效",
            node_type="cause",
            value=len(groups["door"]),
            sample_notes=samples(groups["door"]),
            x=438,
            y=92,
        ),
        KnowledgeGraphNode(
            id="airflow",
            label="风量分配",
            sub_label="风量不足",
            node_type="cause",
            value=len(groups["airflow"]),
            sample_notes=samples(groups["airflow"]),
            x=118,
            y=280,
        ),
        KnowledgeGraphNode(
            id="monitoring",
            label="监测系统",
            sub_label="数据异常",
            node_type="evidence",
            value=len(groups["monitoring"]),
            sample_notes=samples(groups["monitoring"]),
            x=438,
            y=282,
        ),
    ]
    edges = [
        KnowledgeGraphEdge(
            source="local_fan",
            target="risk_gas",
            relation="导致",
            weight=len(groups["local_fan"]),
        ),
        KnowledgeGraphEdge(
            source="door",
            target="risk_gas",
            relation="导致",
            weight=len(groups["door"]),
        ),
        KnowledgeGraphEdge(
            source="airflow",
            target="risk_gas",
            relation="导致",
            weight=len(groups["airflow"]),
        ),
        KnowledgeGraphEdge(
            source="monitoring",
            target="risk_gas",
            relation="监测",
            weight=len(groups["monitoring"]),
        ),
        KnowledgeGraphEdge(
            source="local_fan",
            target="door",
            relation="关联",
            weight=min(len(groups["local_fan"]), len(groups["door"])),
        ),
        KnowledgeGraphEdge(
            source="airflow",
            target="monitoring",
            relation="印证",
            weight=min(len(groups["airflow"]), len(groups["monitoring"])),
        ),
    ]
    return KnowledgeGraphResponse(topic=topic, nodes=nodes, edges=edges)