feat(ontology): 仅放行财务业务相关问题的信号校验

- 新增 _has_supported_business_signal，在加载目录前拦截非财务问题并抛错
- 同步重构 ontology 服务测试覆盖业务信号判定分支

server/src/app/services/ontology.py

@@ -103,11 +103,14 @@ class SemanticOntologyService(
         if not query:
             raise ValueError("query 不能为空。")
 
-        AgentFoundationService(self.db).ensure_foundation_ready()
         context_json = normalize_ontology_context_json(payload.context_json or {})
         payload = payload.model_copy(update={"context_json": context_json})
-        reference = self._load_reference_catalog()
         compact_query = self._compact(query)
+        if not self._has_supported_business_signal(compact_query, context_json):
+            raise ValueError("当前系统仅支持财务业务相关问题。")
+
+        AgentFoundationService(self.db).ensure_foundation_ready()
+        reference = self._load_reference_catalog()
         entities = self._extract_entities(query, compact_query, reference, context_json=context_json)
         rule_scenario, scenario_score = self._detect_scenario(compact_query)
         time_range, _time_score = self._extract_time_range(

server/src/app/services/ontology_detection.py

@@ -92,6 +92,92 @@ class OntologyDetectionMixin:
     def _looks_like_expense_application(compact_query: str) -> bool:
         return looks_like_expense_application_signal(compact_query)
 
+    def _has_supported_business_signal(self, compact_query: str, context_json: dict[str, Any]) -> bool:
+        has_business_context = (
+            self._is_expense_application_context(context_json)
+            or self._resolve_session_type_scenario(context_json) == "knowledge"
+            or self._resolve_context_scenario(context_json) is not None
+        )
+
+        if self._looks_like_expense_application(compact_query):
+            return True
+
+        domain_keywords = [
+            keyword
+            for keywords in SCENARIO_KEYWORDS.values()
+            for keyword, _weight in keywords
+        ]
+        if any(keyword in compact_query for keyword in domain_keywords):
+            return True
+        if any(keyword in compact_query for keyword in EXPENSE_NARRATIVE_KEYWORDS):
+            return True
+        knowledge_keywords = (
+            "制度",
+            "规则",
+            "办法",
+            "依据",
+            "政策",
+            "知识库",
+            "规定",
+            "流程",
+            "口径",
+            "标准",
+            "上限",
+            "额度",
+            "补贴",
+            "票据要求",
+        )
+        if any(keyword in compact_query for keyword in knowledge_keywords):
+            return True
+
+        approval_keywords = (
+            "待我审核",
+            "待审",
+            "审核",
+            "审批",
+            "审核意见",
+            "审批意见",
+            "审批通过",
+            "审批驳回",
+            "驳回",
+            "退回",
+            "审核中心",
+            "审批中心",
+            "领导审批",
+            "财务审核",
+            "处理意见",
+        )
+        if any(keyword in compact_query for keyword in approval_keywords):
+            return True
+        if has_business_context and self._looks_like_contextual_business_follow_up(compact_query):
+            return True
+
+        return False
+
+    @staticmethod
+    def _looks_like_contextual_business_follow_up(compact_query: str) -> bool:
+        if not compact_query:
+            return False
+        if compact_query in {
+            "好",
+            "好的",
+            "行",
+            "可以",
+            "嗯",
+            "继续",
+            "下一步",
+            "确认",
+            "确定",
+            "补充",
+            "再补充",
+            "再看看",
+            "没问题",
+        }:
+            return True
+        if any(keyword in compact_query for keyword in DRAFT_FOLLOW_UP_KEYWORDS):
+            return True
+        return compact_query.startswith(("那", "这", "它", "这个", "那个"))
+
     def _detect_scenario(self, compact_query: str) -> tuple[str, float]:
         scores = {key: 0.0 for key in SCENARIO_KEYWORDS}
         for scenario, keywords in SCENARIO_KEYWORDS.items():
@@ -126,6 +212,7 @@ class OntologyDetectionMixin:
 
         return best_scenario, round(min(best_score, 0.34), 2)
 
+
     def _detect_intent(
         self,
         compact_query: str,

server/tests/test_ontology_service.py

@@ -1115,6 +1115,21 @@ def test_parse_ontology_endpoint_returns_eight_fields_and_writes_trace() -> None
     assert run_payload["semantic_parse"]["intent"] == "risk_check"
 
 
+def test_parse_ontology_endpoint_blocks_non_business_input() -> None:
+    client, _ = build_client()
+
+    response = client.post(
+        "/api/v1/ontology/parse",
+        json={
+            "query": "你好",
+            "user_id": "pytest",
+        },
+    )
+
+    assert response.status_code == 400
+    assert "财务业务相关问题" in response.json()["detail"]
+
+
 def test_parse_ontology_endpoint_returns_forbidden_for_unprivileged_payment_request() -> None:
     client, _ = build_client()