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"(? 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)