From 3feb0a3e0df5b49dcd1d61c75b6438f2f486e5e5 Mon Sep 17 00:00:00 2001 From: sladro Date: Fri, 30 Jan 2026 15:01:38 +0800 Subject: [PATCH] update --- .../controller/ragflow_controller.py | 21 ++++++ .../module_admin/service/kb_es_service.py | 30 +++++++- .../sub_applications/kb_es/create_index.py | 3 + .../sub_applications/kb_es/ingest_folder.py | 69 +++++++++++++++++++ 4 files changed, 121 insertions(+), 2 deletions(-) diff --git a/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py b/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py index da065d7..483492e 100644 --- a/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py +++ b/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py @@ -538,6 +538,27 @@ async def stream_kb_es_response( ping_message = 'event: ping\ndata: {"status": "connected"}\n\n' yield ping_message + # 结构化Q&A命中时，直接返回 answer（最快，不走 LLM） + if chunks: + top = chunks[0] + top_answer = (top.get('answer') or '').strip() + if top_answer: + chunk_size = 12 + for i in range(0, len(top_answer), chunk_size): + if i == 0: + latency = time.perf_counter() - start_time + logger.info( + f"[KB_ES_STREAM {time.time():.3f}] 首token到达(抽取式)，ttft={latency:.3f}s | retrieve_ms={retrieve_ms} | chat_id={chat_id}" + ) + yield format_sse({'answer': top_answer[i : i + chunk_size], 'source': 'kb_es'}) + yield format_sse({'status': 'completed', 'source': 'kb_es'}, event='end') + if cache_store_func and len(top_answer.strip()) >= 10: + try: + await cache_store_func(top_answer) + except Exception as e: + logger.warning(f"[KB_ES_STREAM] 缓存存储失败: {e}") + return + context_parts = [] for i, c in enumerate(chunks[: KBConfig.KB_TOP_K], 1): title = (c.get('title') or '').strip() diff --git a/ruoyi-fastapi-backend/module_admin/service/kb_es_service.py b/ruoyi-fastapi-backend/module_admin/service/kb_es_service.py index bbaf94e..eb9826b 100644 --- a/ruoyi-fastapi-backend/module_admin/service/kb_es_service.py +++ b/ruoyi-fastapi-backend/module_admin/service/kb_es_service.py @@ -72,7 +72,18 @@ class KBESService: retrieve_ms = (time.perf_counter() - t0) * 1000 match_service = get_match_service() - combined = "\n".join([(c.get("title") or "") + "\n" + (c.get("content") or "") for c in chunks]) + combined = "\n".join( + [ + (c.get("question") or "") + + "\n" + + (c.get("answer") or "") + + "\n" + + (c.get("title") or "") + + "\n" + + (c.get("content") or "") + for c in chunks + ] + ) coverage = match_service.calculate_keyword_coverage(question, combined) top1 = metrics_raw.get("top1_score", 0.0) @@ -133,6 +144,9 @@ class KBESService: "_source": [ "doc_id", "chunk_id", + "qa_id", + "question", + "answer", "title", "content", "source", @@ -147,11 +161,20 @@ class KBESService: { "multi_match": { "query": question, - "fields": ["title^2", "content", "tags^1.2", "category^1.1"], + "fields": [ + "question^3", + "answer^1.5", + "title^2", + "content", + "tags^1.2", + "category^1.1", + ], "type": "best_fields", "minimum_should_match": "60%", } }, + {"match_phrase": {"question": {"query": question, "boost": 5}}}, + {"match_phrase": {"answer": {"query": question, "boost": 2}}}, {"match_phrase": {"title": {"query": question, "boost": 3}}}, {"match_phrase": {"content": {"query": question, "boost": 2}}}, ], @@ -181,6 +204,9 @@ class KBESService: { "doc_id": src.get("doc_id"), "chunk_id": src.get("chunk_id"), + "qa_id": src.get("qa_id"), + "question": src.get("question"), + "answer": src.get("answer"), "title": src.get("title"), "content": src.get("content"), "source": src.get("source"), diff --git a/ruoyi-fastapi-backend/sub_applications/kb_es/create_index.py b/ruoyi-fastapi-backend/sub_applications/kb_es/create_index.py index bfd3d66..2db001e 100644 --- a/ruoyi-fastapi-backend/sub_applications/kb_es/create_index.py +++ b/ruoyi-fastapi-backend/sub_applications/kb_es/create_index.py @@ -41,6 +41,9 @@ def _mapping() -> Dict[str, Any]: "properties": { "doc_id": {"type": "keyword"}, "chunk_id": {"type": "keyword"}, + "qa_id": {"type": "keyword"}, + "question": {"type": "text", "analyzer": "kb_default"}, + "answer": {"type": "text", "analyzer": "kb_default"}, "title": {"type": "text", "analyzer": "kb_default"}, "content": {"type": "text", "analyzer": "kb_default"}, "source": {"type": "keyword"}, diff --git a/ruoyi-fastapi-backend/sub_applications/kb_es/ingest_folder.py b/ruoyi-fastapi-backend/sub_applications/kb_es/ingest_folder.py index 67284f3..6963346 100644 --- a/ruoyi-fastapi-backend/sub_applications/kb_es/ingest_folder.py +++ b/ruoyi-fastapi-backend/sub_applications/kb_es/ingest_folder.py @@ -63,12 +63,81 @@ def chunk_text(text: str, chunk_size: int, overlap: int) -> List[str]: return out +_QA_Q_RE = re.compile(r"^\s*(\d+)\s*[\.．、)]\s*(.+?)\s*$") +_QA_A_RE = re.compile(r"^\s*[•\-*]\s*答案\s*[:：]\s*(.*)$") + + +def parse_markdown_qa(text: str) -> List[Dict[str, str]]: + """解析形如： + 1.xxx？\n•答案：yyy + 的 Q&A 列表。 + """ + + lines = [ln.rstrip() for ln in (text or "").splitlines()] + items: List[Dict[str, str]] = [] + + cur_q: str | None = None + cur_a_lines: List[str] = [] + + def flush(): + nonlocal cur_q, cur_a_lines + if cur_q: + ans = "\n".join([a for a in cur_a_lines if a.strip()]).strip() + if ans: + items.append({"question": cur_q.strip(), "answer": ans}) + cur_q = None + cur_a_lines = [] + + for ln in lines: + if not ln.strip(): + continue + m_q = _QA_Q_RE.match(ln) + if m_q: + flush() + cur_q = m_q.group(2) + continue + m_a = _QA_A_RE.match(ln) + if m_a and cur_q: + first = m_a.group(1).strip() + if first: + cur_a_lines.append(first) + continue + if cur_q: + # 支持答案多行（没有“•答案：”前缀的续行） + cur_a_lines.append(ln.strip()) + + flush() + return items + + def build_docs(file_path: Path, root: Path, *, chunk_size: int, overlap: int) -> List[Dict[str, Any]]: rel = str(file_path.relative_to(root)).replace("\\", "/") content = read_text_file(file_path) title = file_path.stem doc_id = _slug_id(rel) + # 优先按 Q&A 结构化入库（更准、更快）；解析不到再退回 chunk + qa_items = parse_markdown_qa(content) if file_path.suffix.lower() == ".md" else [] + if qa_items: + docs: List[Dict[str, Any]] = [] + for i, qa in enumerate(qa_items, 1): + qa_id = f"{doc_id}_qa_{i}" + q = qa["question"].strip() + a = qa["answer"].strip() + docs.append( + { + "doc_id": doc_id, + "chunk_id": qa_id, + "qa_id": qa_id, + "title": title, + "question": q, + "answer": a, + "content": f"{q}\n{a}", + "source": rel, + } + ) + return docs + parts = chunk_text(content, chunk_size=chunk_size, overlap=overlap) docs = [] for idx, part in enumerate(parts):