fix(rag): keep practice headings with details

ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py

@@ -23,7 +23,13 @@ from app.models.knowledge import (
     KnowledgeSection,
     KnowledgeVersion,
 )
-from app.services.knowledge_pipeline_service import expand_synonyms, extract_terms
+from app.services.knowledge_pipeline_service import (
+    expand_synonyms,
+    extract_terms,
+    is_numbered_heading_title,
+    is_practice_anchor_title,
+    practice_titles_related,
+)
 from app.services.knowledge_catalog_cache_service import KnowledgeCatalogCacheService
 from app.services.knowledge_service import KnowledgeScope
 from app.services.model_service import _call_configured_model, _system_config_bool
@@ -125,7 +131,10 @@ class KnowledgeAgentService:
                 if practice_overview:
                     selected.sort(key=lambda item: (item.knowledge.name, item.section.sort_order))
                 selected_contents = {
-                    item.section.id: cls._protect_content(cls._read_complete_section(db, item.section))
+                    item.section.id: cls._protect_content(
+                        cls._read_complete_section(db, item.section),
+                        max_chars=900 if practice_overview else 6000,
+                    )
                     for item in selected
                 }
                 chunks = [
@@ -450,6 +459,10 @@ class KnowledgeAgentService:
 
     @staticmethod
     def _is_practice_overview(question: str) -> bool:
+        if "追问：" in question:
+            follow_up = question.rsplit("追问：", 1)[-1]
+            if any(marker in follow_up for marker in ("具体", "详细", "怎么做", "步骤", "方法", "注意", "内容")):
+                return False
         overview_markers = ("有哪些", "是什么", "包括什么", "都有什么", "列出", "汇总", "总结")
         practice_markers = ("作业", "功课", "练习")
         return any(marker in question for marker in practice_markers) and any(marker in question for marker in overview_markers)
@@ -471,19 +484,36 @@ class KnowledgeAgentService:
     @staticmethod
     def _read_complete_section(db: Session, section: KnowledgeSection) -> str:
         """Read a heading together with its lower-level child sections."""
-        level = _heading_level(section.content)
-        if level is None:
-            return section.content
         rows = db.scalars(
             select(KnowledgeSection)
             .where(
                 KnowledgeSection.version_id == section.version_id,
-                KnowledgeSection.sort_order >= section.sort_order,
             )
             .order_by(KnowledgeSection.sort_order)
         ).all()
+        selected_index = next((index for index, row in enumerate(rows) if row.id == section.id), None)
+        if selected_index is None:
+            return section.content
+
+        for anchor_index, row in enumerate(rows):
+            if not is_practice_anchor_title(row.title):
+                continue
+            end_index = anchor_index + 1
+            while end_index < len(rows):
+                next_title = rows[end_index].title
+                if is_practice_anchor_title(next_title):
+                    break
+                if is_numbered_heading_title(next_title) and not practice_titles_related(row.title, next_title):
+                    break
+                end_index += 1
+            if anchor_index <= selected_index < end_index:
+                return "\n\n".join(item.content for item in rows[anchor_index:end_index])
+
+        level = _heading_level(section.content)
+        if level is None:
+            return section.content
         parts: list[str] = []
-        for row in rows:
+        for row in rows[selected_index:]:
             row_level = _heading_level(row.content)
             if row.id != section.id and row_level is not None and row_level <= level:
                 break
@@ -533,8 +563,8 @@ class KnowledgeAgentService:
         return list(dict.fromkeys(base + expand_synonyms(base)))[:40]
 
     @staticmethod
-    def _protect_content(content: str) -> str:
-        return content if len(content) <= 3200 else content[:3200].rstrip() + "\n[章节内容已按保护规则截断]"
+    def _protect_content(content: str, *, max_chars: int = 6000) -> str:
+        return content if len(content) <= max_chars else content[:max_chars].rstrip() + "\n[章节内容已按保护规则截断]"
 
 
 def _dump_trace(trace: list[dict]) -> str:

ai_knowledge_base_v2/apps/backend/app/services/knowledge_pipeline_service.py

@@ -25,10 +25,15 @@ from app.services.external_errors import ExternalServiceError
 from app.services.feishu_async_service import AsyncFeishuKnowledgeService
 from app.services.feishu_service import _feishu_retrieval_config
 
-PROCESSING_RULE_VERSION = "knowledge-pipeline-v3-latest-only"
-MAX_SECTION_CHARS = 2800
+PROCESSING_RULE_VERSION = "knowledge-pipeline-v4-semantic-practice"
+MAX_SECTION_CHARS = 8000
 MAX_CHUNK_CHARS = 650
 
+_PRACTICE_ANCHOR = re.compile(
+    r"(?:^|[、\s])(?:练习|作业|功课)\s*[一二三四五六七八九十百零〇\d]+(?:[：:、.．\s]|$)"
+)
+_NUMBERED_HEADING = re.compile(r"^(?:第)?[一二三四五六七八九十百零〇\d]+[、.．:：]\s*")
+
 
 @dataclass(frozen=True)
 class ParsedSection:
@@ -388,9 +393,25 @@ def parse_sections(content: str) -> list[ParsedSection]:
         boundaries.append((match.start(), raw))
     if not boundaries or boundaries[0][0] > 0:
         boundaries.insert(0, (0, "正文"))
+    spans: list[tuple[int, int, str]] = []
+    index = 0
+    while index < len(boundaries):
+        start, title = boundaries[index]
+        next_index = index + 1
+        if is_practice_anchor_title(title):
+            while next_index < len(boundaries):
+                next_title = boundaries[next_index][1]
+                if is_practice_anchor_title(next_title):
+                    break
+                if is_numbered_heading_title(next_title) and not practice_titles_related(title, next_title):
+                    break
+                next_index += 1
+        end = boundaries[next_index][0] if next_index < len(boundaries) else len(content)
+        spans.append((start, end, title))
+        index = next_index
+
     sections: list[ParsedSection] = []
-    for index, (start, title) in enumerate(boundaries):
-        end = boundaries[index + 1][0] if index + 1 < len(boundaries) else len(content)
+    for start, end, title in spans:
         block = content[start:end].strip()
         if not block:
             continue
@@ -408,6 +429,29 @@ def parse_sections(content: str) -> list[ParsedSection]:
     return sections or [ParsedSection("正文", content, 0, len(content))]
 
 
+def is_practice_anchor_title(title: str) -> bool:
+    return bool(_PRACTICE_ANCHOR.search(title.strip()))
+
+
+def is_numbered_heading_title(title: str) -> bool:
+    return bool(_NUMBERED_HEADING.match(title.strip()))
+
+
+def practice_titles_related(anchor: str, candidate: str) -> bool:
+    """Recognize numbered safety/boundary headings that still belong to a practice."""
+    anchor_topic = _PRACTICE_ANCHOR.sub("", _NUMBERED_HEADING.sub("", anchor)).strip(" ：:、")
+    candidate_text = _NUMBERED_HEADING.sub("", candidate)
+    normalized = re.sub(r"[\s·・，。、“”《》：:（）()\-]", "", anchor_topic)
+    candidate_normalized = re.sub(r"[\s·・，。、“”《》：:（）()\-]", "", candidate_text)
+    ignored = {"练习", "作业", "功课", "静心", "方法", "步骤", "课程"}
+    pairs = {
+        normalized[index : index + 2]
+        for index in range(max(0, len(normalized) - 1))
+        if normalized[index : index + 2] not in ignored
+    }
+    return any(pair in candidate_normalized for pair in pairs)
+
+
 def split_chunks(section: ParsedSection) -> list[tuple[str, int, int]]:
     chunks: list[tuple[str, int, int]] = []
     cursor = 0