|
@@ -12,6 +12,7 @@ from __future__ import annotations
|
|
|
|
|
|
|
|
import json
|
|
import json
|
|
|
import os
|
|
import os
|
|
|
|
|
+import re
|
|
|
from pathlib import Path
|
|
from pathlib import Path
|
|
|
|
|
|
|
|
from creation_knowledge.config import Settings
|
|
from creation_knowledge.config import Settings
|
|
@@ -327,6 +328,22 @@ def link_scope(linker: ScopeLinker, scope_type: str, value: str) -> dict:
|
|
|
"top": [{"name": h["name"], "score": h["score"], "path": h.get("path", "")} for h in hits]}
|
|
"top": [{"name": h["name"], "score": h["score"], "path": h.get("path", "")} for h in hits]}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
+_SCOPE_CONN = re.compile(r"\s*[和与、,,//&]\s*") # 作用域值确定性拆分:含连接词必拆成多原子
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _split_scope_items(items: list) -> list:
|
|
|
|
|
+ """连接词拆分守卫:把"社会共识与个体真实矛盾"这种拆成多原子,平铺。LLM 漏拆时兜底。"""
|
|
|
|
|
+ out = []
|
|
|
|
|
+ for it in (items or []):
|
|
|
|
|
+ st, v = it.get("scope_type"), it.get("value")
|
|
|
|
|
+ if not (st and v):
|
|
|
|
|
+ continue
|
|
|
|
|
+ parts = [p.strip() for p in _SCOPE_CONN.split(v) if p.strip()]
|
|
|
|
|
+ for p in (parts if len(parts) > 1 else [v]):
|
|
|
|
|
+ out.append({"scope_type": st, "value": p})
|
|
|
|
|
+ return out
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
def apply_scopes(knowledges: list[dict], scopes: list, linker: ScopeLinker) -> None:
|
|
def apply_scopes(knowledges: list[dict], scopes: list, linker: ScopeLinker) -> None:
|
|
|
by_k = {k.get("id"): k for k in knowledges}
|
|
by_k = {k.get("id"): k for k in knowledges}
|
|
|
for sc in scopes:
|
|
for sc in scopes:
|
|
@@ -334,7 +351,7 @@ def apply_scopes(knowledges: list[dict], scopes: list, linker: ScopeLinker) -> N
|
|
|
if not k:
|
|
if not k:
|
|
|
continue
|
|
continue
|
|
|
linked = [link_scope(linker, it["scope_type"], it["value"])
|
|
linked = [link_scope(linker, it["scope_type"], it["value"])
|
|
|
- for it in (sc.get("items") or []) if it.get("scope_type") and it.get("value")]
|
|
|
|
|
|
|
+ for it in _split_scope_items(sc.get("items"))]
|
|
|
if k.get("type") == "how" and sc.get("step_id"):
|
|
if k.get("type") == "how" and sc.get("step_id"):
|
|
|
for s in k.get("steps", []):
|
|
for s in k.get("steps", []):
|
|
|
if s.get("id") == sc["step_id"]:
|
|
if s.get("id") == sc["step_id"]:
|