admin/PlantAI-dev
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

轻量拓展分析自然语言触发：nl_intent 模块、后端意图回退与短路、两步向导与服务器路径导入

Browse Source 
- 新增 gfa_core/extension_stage_lightweight/nl_intent.py：粗筛+DeepSeek 解析+sanitize，client_factory 可注入
- main.py：_match_lightweight_extension_intent 与三个空返回点回退、/api/agent/plan 守卫、_run_agent_chat 预闸短路、POST /api/nl-import-paths
- index.html：lightweightNlRun 两步向导（拖拽上传+服务器路径导入）、executeLightweightNlRun、poll 回话
- 一并提交工作区既有改动：api.py store 默认路径 .runtime/lightweight-extension/store、index.html stop 404 容错
This commit is contained in:
potaser
2026-08-26 23:46:42 +08:00
parent 90f51ed607
commit a60f457541
4 changed files with 813 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gfa_core/extension_stage_lightweight/api.py
+4 -1
View File
@@ -100,7 +100,10 @@ class LightweightApiSettings:
return cls(
project_root=root,
store_root=_path_from_env(
str(env.get("BBAV5_LIGHTWEIGHT_EXTENSION_STORE") or "extension_stages_lightweight"),
str(
env.get("BBAV5_LIGHTWEIGHT_EXTENSION_STORE")
or ".runtime/lightweight-extension/store"
),
project_root=root,
),
input_root=_path_from_env(
gfa_core/extension_stage_lightweight/nl_intent.py
+346
View File
@@ -0,0 +1,346 @@
"""轻量版拓展分析的自然语言触发解析。
从已发布的轻量拓展记录（store_root/extensions/*/extension.json）中，用 DeepSeek
把自然语言消息解析为一次运行意图：
- 粗筛（_rough_trigger/_rough_candidates）先用关键词判断消息是否可能命中任一已发布拓展，
未命中时完全不发起 LLM 调用；
- 命中后把候选目录与 job 文件提示交给 LLM，要求返回严格的 JSON resolution；
- 所有外部输入经过 _sanitize_resolution 白名单校验，本模块任何函数都不对外抛异常。
对外入口：resolve_lightweight_extension_intent(message, published_records, ...)。
测试可注入 client_factory(config) 以获得不依赖真实环境的假客户端。
"""
from __future__ import annotations
from dataclasses import dataclass
import json
import os
import re
from pathlib import Path
from typing import Any, Callable, Mapping
PREFILL_THRESHOLD = 0.6
MAX_CATALOG_CHARS = 12000
MAX_JOB_FILE_HINTS = 200
_ENGLISH_STOPWORDS = {"analysis", "data", "file", "run", "input", "output", "files", "using", "tool"}
_CHINESE_STOPWORDS = {"分析", "序列", "文件", "数据", "蛋白", "拓展", "扩展", "工具", "进行", "搜索"}
_RESPONSE_SYSTEM_PROMPT = (
"你是植物基因组分析平台的自然语言意图解析器。"
"用户消息会引用一个轻量拓展分析（lightweight extension analysis）。"
"你必须只返回一个严格 JSON 对象（不要 markdown、不要多余文字），格式：\n"
'{"extension_id": 字符串或 null, "input_overrides": [{"path": "...", "value": "..."}], '
'"parameter_overrides": [{"fact_id": "...", "value": "..."}], "explicit_prefix": 字符串或 null, '
'"confidence": 0到1之间的数字}\n\n'
"规则：\n"
"1. extension_id 只能从给定 extensions 列表里选；无法确定时填 null。\n"
"2. input_overrides 里的 path 只能引用该拓展 inputs[].path；parameter_overrides 里的 fact_id "
"只能引用该拓展 parameters[].fact_id。\n"
"3. 参数值只能来自用户消息原文；用户没提到的参数不要出现。\n"
"4. 用户提到的文件优先与 job_files 匹配，统一写成 \"input/<job_id>/<文件名>\" 或 "
"\"workspace/<job_id>/<文件名>\" 的 job 相对形式；禁止绝对路径。\n"
"5. 禁止编造任何目录、路径、参数或数值。\n"
'6. 输出必须能被 JSON.parse 直接解析。'
)
@dataclass(frozen=True)
class LightweightIntentResolution:
extension_id: str | None
extension_name: str | None
input_overrides: dict[str, str]
parameter_overrides: dict[str, Any]
explicit_prefix: str | None
confidence: float
prefill: bool
reason: str | None
candidates_considered: list[str]
def load_published_lightweight_records(store_root: Path) -> list[dict]:
"""加载 store_root/extensions/*/extension.json 中 status=="published" 的记录；坏文件跳过。"""
try:
candidates = sorted(Path(store_root).glob("extensions/*/extension.json"))
except (OSError, TypeError, ValueError):
return []
records: list[dict] = []
for path in candidates:
try:
data = json.loads(path.read_text(encoding="utf-8", errors="ignore"))
except (OSError, json.JSONDecodeError, ValueError) as exc:
print(f"[LightweightNLIntent] skip bad record {path}: {exc}")
continue
if not isinstance(data, dict):
print(f"[LightweightNLIntent] skip bad record {path}: not an object")
continue
if data.get("status") == "published":
records.append(data)
return records
def _function_summary_text(record: Mapping[str, Any]) -> str:
candidate = record.get("semantic_candidate")
if not isinstance(candidate, Mapping):
return ""
return str(candidate.get("function_summary") or "")
def _summary_keywords(record: Mapping[str, Any]) -> tuple[list[str], list[str]]:
"""从 function_summary 抽英文 token（len>=3）与中文子串（len>=2），过滤停用词。"""
summary = _function_summary_text(record).lower()
english = [
token for token in re.split(r"[^A-Za-z0-9]+", summary)
if len(token) >= 3 and token not in _ENGLISH_STOPWORDS
]
chinese_segments = re.findall(r"[一-鿿]+", summary)
chinese: list[str] = []
for segment in chinese_segments:
length = len(segment)
for size in range(2, min(4, length) + 1):
for start in range(length - size + 1):
sub = segment[start:start + size]
if sub not in _CHINESE_STOPWORDS and sub not in chinese:
chinese.append(sub)
return english, chinese
def _rough_candidates(message: str, records: list[dict]) -> list[str]:
"""按关键词粗筛可能命中的记录 id；不含 LLM 调用。"""
text = str(message or "").lower()
if any(token in text for token in ("hmmsearch", "hmmer", "blast", "blastp")):
return [str(record.get("extension_id")) for record in records if record.get("extension_id")]
if any(token in text for token in ("拓展分析", "扩展分析", "extension")):
return [str(record.get("extension_id")) for record in records if record.get("extension_id")]
matched: list[str] = []
for record in records:
english, chinese = _summary_keywords(record)
if any(token in text for token in english) or any(sub in text for sub in chinese):
matched.append(str(record.get("extension_id")))
return matched
def _rough_trigger(message: str, records: list[dict]) -> bool:
"""消息是否可能命中任一已发布轻量拓展（粗筛布尔版）。"""
return bool(_rough_candidates(message, records))
def _build_catalog(records: list[dict]) -> dict:
"""构造发给 LLM 的拓展目录；序列化后总长超过 MAX_CATALOG_CHARS 时截断。"""
items: list[dict] = []
total = 0
truncated = False
for record in records:
extension_id = str(record.get("extension_id") or "").strip()
if not extension_id:
continue
contract = record.get("runtime_binding_contract")
if not isinstance(contract, Mapping):
continue
inputs = [
{
"path": str(item.get("path") or ""),
"role": str(item.get("role") or ""),
"description": str(item.get("description") or ""),
}
for item in contract.get("inputs") or []
if isinstance(item, Mapping) and item.get("replaceable")
]
parameters = [
{
"fact_id": str(item.get("fact_id") or ""),
"flag": str(item.get("flag") or ""),
"display_name": str(item.get("display_name") or ""),
"description": str(item.get("description") or ""),
}
for item in contract.get("parameters") or []
if isinstance(item, Mapping)
]
entry = {
"extension_id": extension_id,
"function_summary": _function_summary_text(record)[:300],
"inputs": inputs,
"parameters": parameters,
}
size = len(json.dumps(entry, ensure_ascii=False))
if total + size > MAX_CATALOG_CHARS:
truncated = True
break
items.append(entry)
total += size
return {"extensions": items, "truncated": truncated}
def _build_messages(message: str, catalog: Mapping[str, Any], job_file_hints: list[str]) -> list[dict]:
"""构造 chat messages：system 规则 + user 里放消息、目录与 job 文件提示。"""
return [
{"role": "system", "content": _RESPONSE_SYSTEM_PROMPT},
{
"role": "user",
"content": json.dumps(
{
"user_message": str(message or ""),
"extensions": catalog.get("extensions") or [],
"job_files": list(job_file_hints or []),
},
ensure_ascii=False,
),
},
]
def _sanitize_resolution(raw: Any, catalog: Mapping[str, Any]) -> tuple[dict | None, str | None]:
"""白名单校验 LLM 返回的 resolution；任何违规返回 (None, 原因)。
返回的 dict 含 extension_id（已从 raw 校验并取出），
以及 input_overrides/parameter_overrides/explicit_prefix/confidence。
"""
if not isinstance(raw, dict):
return None, "LLM 返回的不是 JSON 对象"
extension_id = raw.get("extension_id")
if not extension_id or not isinstance(extension_id, str):
return None, "extension_id 缺失或不是字符串"
extensions = catalog.get("extensions") or []
selected = next(
(item for item in extensions if isinstance(item, Mapping) and item.get("extension_id") == extension_id),
None,
)
if selected is None:
return None, f"extension_id 不在可选目录中: {extension_id}"
allowed_paths = {str(item.get("path") or "") for item in selected.get("inputs") or []}
allowed_fact_ids = {str(item.get("fact_id") or "") for item in selected.get("parameters") or []}
raw_overrides = raw.get("input_overrides")
if raw_overrides is not None and not isinstance(raw_overrides, list):
return None, "input_overrides 必须是数组"
input_overrides: dict[str, str] = {}
for item in raw_overrides or []:
if not isinstance(item, Mapping):
return None, "input_overrides 元素必须是对象"
path = item.get("path")
value = item.get("value")
if not isinstance(path, str) or path not in allowed_paths:
return None, f"input_overrides 引用了不存在的 path: {path}"
if not isinstance(value, str) or not value.strip():
return None, f"input_overrides[{path}] 的 value 必须是非空字符串"
input_overrides[path] = value.strip()
raw_parameters = raw.get("parameter_overrides")
if raw_parameters is not None and not isinstance(raw_parameters, list):
return None, "parameter_overrides 必须是数组"
parameter_overrides: dict[str, Any] = {}
for item in raw_parameters or []:
if not isinstance(item, Mapping):
return None, "parameter_overrides 元素必须是对象"
fact_id = item.get("fact_id")
value = item.get("value")
if not isinstance(fact_id, str) or fact_id not in allowed_fact_ids:
return None, f"parameter_overrides 引用了不存在的 fact_id: {fact_id}"
if not isinstance(value, str) or not value.strip():
return None, f"parameter_overrides[{fact_id}] 的 value 必须是非空字符串"
parameter_overrides[fact_id] = value.strip()
explicit_prefix = raw.get("explicit_prefix")
if explicit_prefix is not None and not isinstance(explicit_prefix, str):
return None, "explicit_prefix 必须是字符串或 null"
confidence = raw.get("confidence")
try:
confidence = float(confidence) if confidence is not None else 0.0
except (TypeError, ValueError):
return None, "confidence 不是数字"
if not (0.0 <= confidence <= 1.0):
return None, "confidence 超出 [0,1]"
return (
{
"extension_id": extension_id,
"input_overrides": input_overrides,
"parameter_overrides": parameter_overrides,
"explicit_prefix": explicit_prefix.strip() if isinstance(explicit_prefix, str) else None,
"confidence": confidence,
},
None,
)
def resolve_lightweight_extension_intent(
message: str,
published_records: list[dict],
environ: Mapping[str, str] | None = None,
client_factory: Callable[[Any], Any] | None = None,
job_file_hints: list[str] | None = None,
) -> LightweightIntentResolution:
"""把自然语言消息解析为轻量拓展运行意图；绝不抛异常。
client_factory(config) 注入时即使 config 为 None 也会被调用，由工厂自行容错
（测试用它注入不依赖 config 的假客户端）。
"""
candidates = _rough_candidates(message, published_records)
if not published_records:
return LightweightIntentResolution(None, None, {}, {}, None, 0.0, False, "no_published_records", [])
if not candidates:
return LightweightIntentResolution(None, None, {}, {}, None, 0.0, False, "no_trigger", [])
catalog = _build_catalog(
[record for record in published_records if str(record.get("extension_id") or "") in candidates]
)
if not catalog["extensions"]:
return LightweightIntentResolution(None, None, {}, {}, None, 0.0, False, "no_trigger", candidates)
hints = list(job_file_hints or [])[:MAX_JOB_FILE_HINTS]
try:
from .deepseek_semantics import DeepSeekSemanticError, load_deepseek_config
config = None
try:
config = load_deepseek_config(environ if environ is not None else os.environ)
except DeepSeekSemanticError:
config = None
if client_factory is not None:
client = client_factory(config)
else:
from .deepseek_semantics import _default_client
if config is None:
raise DeepSeekSemanticError("no deepseek config for default client", code="configuration_error")
client = _default_client(config)
model = getattr(config, "model", None) if config is not None else None
response = client.chat.completions.create(
model=model or "deepseek-chat",
messages=_build_messages(message, catalog, hints),
temperature=0,
response_format={"type": "json_object"},
max_tokens=4000,
)
content = response.choices[0].message.content
sanitized, reason = _sanitize_resolution(json.loads(content), catalog)
if sanitized is None:
return LightweightIntentResolution(
None, None, {}, {}, None, 0.0, False, f"invalid_resolution: {reason}", candidates
)
except Exception as exc:
print(f"[LightweightNLIntent] resolve failed: {exc}")
return LightweightIntentResolution(
None, None, {}, {}, None, 0.0, False, "llm_unavailable_or_invalid", candidates
)
extension_id = sanitized["extension_id"]
selected = next(
(item for item in catalog["extensions"] if item.get("extension_id") == extension_id),
None,
)
extension_name = extension_id
if selected is not None:
first_sentence = re.split(r"[。.!?；;]", str(selected.get("function_summary") or ""))[0].strip()
extension_name = first_sentence[:60] or extension_id
return LightweightIntentResolution(
extension_id=extension_id,
extension_name=extension_name,
input_overrides=sanitized["input_overrides"],
parameter_overrides=sanitized["parameter_overrides"],
explicit_prefix=sanitized["explicit_prefix"],
confidence=sanitized["confidence"],
prefill=bool(
sanitized["confidence"] >= PREFILL_THRESHOLD
and (sanitized["input_overrides"] or sanitized["parameter_overrides"])
),
reason="ok",
candidates_considered=candidates,
)