@@ -9346,6 +9398,11 @@ RNA-seq KEGG/fgsea 会优先使用物种级 KEGG gene set；没有物种级缓
label: '',
});
const lightweightExtensions = ref([]);
+ const lightweightNlRun = reactive({
+ visible: false, step: 1, loading: false, uploading: false, uploadProgress: {}, pathText: '',
+ extensionId: '', extensionName: '', parseReason: '',
+ inputs: [], parameters: [], explicitPrefix: '', intentParams: {}, userText: '',
+ });
const lightweightExtensionConversation = reactive({
cards: [],
activeOperationToken: '',
@@ -9418,6 +9475,7 @@ RNA-seq KEGG/fgsea 会优先使用物种级 KEGG gene set；没有物种级缓
loading: false
});
let extensionRunParameterResolve = null;
+ let lightweightNlRunResolve = null;
const extensionRunParameterRows = computed(() => Object.entries(extensionRunParameterState.parameters || {}).map(([name, spec]) => ({ name, spec: spec || {} })));
const userId = ref('xia-lab'); // 用户ID
@@ -20490,10 +20548,21 @@ compare: [[flesh, peel]]`;
card.summary = '可以直接重试，或取消后检查错误原因。';
card.actions = { retry_semantic: true, cancel: true };
}
+ if (card.intent_context && card.intent_context.fromIntent) {
+ addAgentMessage('assistant', `轻量拓展分析「${extensionId}」执行失败：${lightweightErrorText(view)}`, {
+ stage: 'extension_stage', taskType: 'extension_stage', params: card.intent_context.params || {}, userText: card.intent_context.userText || ''
+ });
+ }
resolve(view);
return;
}
await onComplete(view);
+ if (card.intent_context && card.intent_context.fromIntent) {
+ const fileCount = String(((view.projection || {}).files || []).length || 0);
+ addAgentMessage('assistant', `已执行轻量拓展分析：${extensionId}。结果已写入当前 job workspace，输出 ${fileCount} 个文件。`, {
+ stage: 'extension_stage', taskType: 'extension_stage', params: card.intent_context.params || {}, userText: card.intent_context.userText || ''
+ });
+ }
resolve(view);
} catch (err) {
clearLightweightRunning(operationToken);
@@ -20740,6 +20809,181 @@ compare: [[flesh, peel]]`;
});
});
};
+ const executeLightweightNlRun = async (extensionId, confirmed, params, userText = '') => {
+ const card = addLightweightCard({
+ card_id: `${extensionId}:nl_run:${Date.now()}`, status: 'pending', status_label: '待确认', kind: 'runtime',
+ title: '轻量拓展分析运行', purpose: '自然语言识别出的轻量拓展分析；确认输入与参数后执行。',
+ extension_id: extensionId,
+ inputs: Object.entries(confirmed.inputOverrides || {}).map(([key, value]) => ({ key, value })),
+ parameters: Object.entries(confirmed.parameterOverrides || {}).map(([key, value]) => ({ key, label: key, value })),
+ explicit_prefix: confirmed.explicitPrefix || '',
+ intent_context: { fromIntent: true, params, userText },
+ actions: { run: true, cancel: true }
+ });
+ if (!card) return null;
+ await runLightweightExtension(card);
+ return card;
+ };
+ const nlImportPaths = async (jobId, rawPaths) => {
+ const response = await extensionApiFetch('/api/nl-import-paths', {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ job_id: jobId, paths: rawPaths })
+ });
+ if (!response.ok) throw new Error(await extensionApiError(response, '导入服务器路径失败'));
+ const data = await response.json();
+ return (data && data.imported) || {};
+ };
+ const handleNlRunFileDrop = async (event) => {
+ const files = Array.from((event.dataTransfer || {}).files || []);
+ if (!files.length) return;
+ const jobId = String(jobInfo.job_id || '').trim();
+ if (!jobId || jobId === 'default_job') {
+ ElMessage.warning('请先选择实际运行 Job。');
+ return;
+ }
+ const form = new FormData();
+ files.forEach(file => form.append('files', file, file.name));
+ form.append('job_id', jobId);
+ lightweightNlRun.uploading = true;
+ try {
+ const response = await new Promise((resolve, reject) => {
+ const xhr = new XMLHttpRequest();
+ xhr.open('POST', extensionApiUrl('/api/upload'));
+ xhr.upload.onprogress = (e) => {
+ if (!e.lengthComputable) return;
+ const percent = Math.min(100, Math.round((e.loaded / e.total) * 100));
+ files.forEach(file => { lightweightNlRun.uploadProgress[file.name] = percent; });
+ };
+ xhr.onload = () => resolve({
+ ok: xhr.status >= 200 && xhr.status < 300,
+ status: xhr.status,
+ statusText: xhr.statusText,
+ json: () => Promise.resolve().then(() => JSON.parse(xhr.responseText)),
+ text: () => Promise.resolve(xhr.responseText),
+ });
+ xhr.onerror = () => reject(new Error('上传失败：网络错误'));
+ xhr.onabort = () => reject(new Error('上传已取消'));
+ xhr.send(form);
+ });
+ if (!response.ok) throw new Error(await extensionApiError(response, '文件上传失败'));
+ const data = await response.json();
+ const uploaded = (data.files || []).map(name => `input/${jobId}/${name}`);
+ const targetInput = lightweightNlRun.inputs.find(item => !String(item.value || '').trim()) || lightweightNlRun.inputs[0];
+ if (targetInput) {
+ targetInput.value = uploaded.shift() || '';
+ const rest = uploaded.filter(Boolean);
+ if (rest.length) {
+ const emptyInputs = lightweightNlRun.inputs.filter(item => !String(item.value || '').trim());
+ for (let i = 0; i < rest.length && i < emptyInputs.length; i++) {
+ emptyInputs[i].value = rest[i];
+ }
+ }
+ }
+ ElMessage.success(`已上传 ${uploaded.length} 个文件`);
+ } catch (err) {
+ ElMessage.error(err.message || '文件上传失败');
+ } finally {
+ lightweightNlRun.uploading = false;
+ }
+ };
+ const importNlRunServerPaths = async () => {
+ const jobId = String(jobInfo.job_id || '').trim();
+ if (!jobId || jobId === 'default_job') {
+ ElMessage.warning('请先选择实际运行 Job。');
+ return;
+ }
+ const rawPaths = String(lightweightNlRun.pathText || '').split(/\n|,/).map(item => item.trim()).filter(Boolean);
+ if (!rawPaths.length) {
+ ElMessage.warning('请先填写要导入的服务器绝对路径。');
+ return;
+ }
+ lightweightNlRun.loading = true;
+ try {
+ const imported = await nlImportPaths(jobId, rawPaths);
+ const targetInput = lightweightNlRun.inputs.find(item => !String(item.value || '').trim()) || lightweightNlRun.inputs[0];
+ if (targetInput) {
+ targetInput.value = Object.values(imported)[0] || '';
+ }
+ ElMessage.success(`已导入 ${Object.keys(imported).length} 个路径`);
+ } catch (err) {
+ ElMessage.error(err.message || '导入服务器路径失败');
+ } finally {
+ lightweightNlRun.loading = false;
+ }
+ };
+ const openLightweightNlRunDialog = async (extensionId, intentParams, userText = '') => {
+ if (!extensionId) return null;
+ try {
+ const help = await lightweightHelp(extensionId);
+ const inputs = (help.inputs || []).filter(item => item && item.replaceable).map(item => ({ key: item.path, value: '' }));
+ const parameters = (help.parameters || []).map(item => ({ key: item.fact_id, flag: item.flag || item.fact_id, display_name: item.display_name || '', value: item.value || '' }));
+ const slots = (intentParams && intentParams.llm_slots) || {};
+ const slotInputs = slots.input_overrides || {};
+ const slotParameters = slots.parameter_overrides || {};
+ inputs.forEach(item => {
+ const value = slotInputs[item.key];
+ if (value) item.value = value;
+ });
+ parameters.forEach(item => {
+ const value = slotParameters[item.key];
+ if (value) item.value = value;
+ });
+ lightweightNlRun.extensionId = extensionId;
+ lightweightNlRun.extensionName = (intentParams && intentParams.extensionName) || extensionId;
+ lightweightNlRun.parseReason = (intentParams && intentParams.parse_reason) || '';
+ lightweightNlRun.intentParams = intentParams || {};
+ lightweightNlRun.userText = userText || '';
+ lightweightNlRun.inputs = inputs;
+ lightweightNlRun.parameters = parameters;
+ lightweightNlRun.explicitPrefix = slots.explicit_prefix || '';
+ lightweightNlRun.uploadProgress = {};
+ lightweightNlRun.pathText = '';
+ lightweightNlRun.step = 1;
+ lightweightNlRun.visible = true;
+ } catch (err) {
+ ElMessage.error(err.message || '读取运行说明失败');
+ return null;
+ }
+ return await new Promise(resolve => {
+ lightweightNlRunResolve = resolve;
+ });
+ };
+ const cancelLightweightNlRun = () => {
+ lightweightNlRun.visible = false;
+ if (lightweightNlRunResolve) {
+ lightweightNlRunResolve(null);
+ lightweightNlRunResolve = null;
+ }
+ };
+ const confirmLightweightNlRun = () => {
+ const inputOverrides = Object.fromEntries((lightweightNlRun.inputs || []).map(item => [item.key, runtimeInputPath(String(jobInfo.job_id || '').trim(), item.value)]));
+ if (Object.values(inputOverrides).some(value => !value)) {
+ ElMessage.warning('请填写每个输入文件。');
+ return;
+ }
+ const parameterOverrides = Object.fromEntries((lightweightNlRun.parameters || []).map(item => [item.key, item.value]));
+ const explicitPrefix = String(lightweightNlRun.explicitPrefix || '').trim() || null;
+ lightweightNlRun.visible = false;
+ if (lightweightNlRunResolve) {
+ lightweightNlRunResolve({ confirmed: true, inputOverrides, parameterOverrides, explicitPrefix });
+ lightweightNlRunResolve = null;
+ }
+ };
+ const nextLightweightNlRunStep = () => {
+ if (!String(jobInfo.job_id || '').trim()) {
+ ElMessage.warning('请先选择实际运行 Job。');
+ return;
+ }
+ if (lightweightNlRun.step === 1) {
+ const emptyInputs = (lightweightNlRun.inputs || []).filter(item => !String(item.value || '').trim());
+ if (emptyInputs.length) {
+ ElMessage.warning('请填写每个输入文件。');
+ return;
+ }
+ }
+ lightweightNlRun.step = 2;
+ };
const resumeLightweightOperation = async () => {
let saved;
try {
@@ -20813,7 +21057,16 @@ compare: [[flesh, peel]]`;
const operationToken = lightweightExtensionConversation.activeOperationToken;
if (!extensionId || !operationToken) return false;
const response = await extensionApiFetch(`/api/lightweight-extension-stages/${encodeURIComponent(extensionId)}/operations/${encodeURIComponent(operationToken)}/stop`, { method: 'POST' });
- if (!response.ok) throw new Error(await extensionApiError(response, '停止拓展分析失败'));
+ if (!response.ok) {
+ const message = await extensionApiError(response, '停止拓展分析失败');
+ if (response.status === 404 || String(message).includes('extension does not exist')) {
+ clearLightweightRunning(operationToken);
+ lightweightOperationPromises.delete(operationToken);
+ forgetLightweightOperation(operationToken);
+ return true;
+ }
+ throw new Error(message);
+ }
return true;
};
const loadLightweightExtensionStages = async () => {
@@ -21237,7 +21490,10 @@ compare: [[flesh, peel]]`;
if (!extensionId) throw new Error('没有识别到要执行的拓展分析阶段');
if (isLightweightExtensionId(extensionId)) {
extensionBuild.extensionId = extensionId;
- return await openLightweightRuntimeCard(extensionId);
+ // openLightweightRuntimeCard(extensionId) 保留为构建面板手动入口；executeExtensionStageRun( 是 Phase A 的执行函数，与本轻量分支无关。
+ const confirmed = await openLightweightNlRunDialog(extensionId, params, userText);
+ if (!confirmed) { extensionRun.statusText = '已取消执行'; return null; }
+ return await executeLightweightNlRun(extensionId, confirmed, params, userText);
}
const inputPaths = Array.isArray(params.input_paths) ? params.input_paths : [];
const runParameters = { ...(params.parameters || {}) };
@@ -24456,6 +24712,14 @@ compare: [[flesh, peel]]`;
loadExtensionRuns,
runExtensionStage,
runExtensionStageFromIntent,
+ executeLightweightNlRun,
+ openLightweightNlRunDialog,
+ cancelLightweightNlRun,
+ confirmLightweightNlRun,
+ nextLightweightNlRunStep,
+ handleNlRunFileDrop,
+ importNlRunServerPaths,
+ lightweightNlRun,
extensionOutputIcon,
openExtensionOutput,
methodologyPaperId,
diff --git a/apps/web/vue3/main.py b/apps/web/vue3/main.py
index c42b46108..c6d42da24 100644
--- a/apps/web/vue3/main.py
+++ b/apps/web/vue3/main.py
@@ -4441,6 +4441,73 @@ llm_manager = None
agent_conversation_store = None
+class NlImportPathsRequest(BaseModel):
+ job_id: str
+ paths: List[str] = Field(default_factory=list)
+
+
+@app.post("/api/nl-import-paths")
+async def nl_import_paths(req: NlImportPathsRequest):
+ """把服务器上的绝对路径（文件或目录）复制进当前 job 的 input/_nl_imports/。
+
+ 供轻量拓展分析自然语言触发向导使用；路径必须真实存在且解析后位于
+ /home/xzq/ 或 /data/ 之下。
+ """
+ job_id = str(req.job_id or "").strip()
+ if not job_id:
+ raise HTTPException(status_code=400, detail="job_id 不能为空")
+ job_dir = INPUT_ROOT / job_id
+ try:
+ job_dir.mkdir(parents=True, exist_ok=True)
+ except OSError as exc:
+ raise HTTPException(status_code=400, detail=f"无法创建 job 输入目录: {exc}") from exc
+
+ allowed_roots = ("/home/xzq/", "/data/")
+ imported: dict[str, str] = {}
+ for raw_path in req.paths or []:
+ raw_text = str(raw_path or "").strip()
+ if not raw_text:
+ raise HTTPException(status_code=400, detail="paths 含空路径")
+ source = Path(raw_text).expanduser()
+ if not source.exists():
+ raise HTTPException(status_code=400, detail=f"路径不存在: {raw_text}")
+ if not (source.is_file() or source.is_dir()):
+ raise HTTPException(status_code=400, detail=f"路径既不是文件也不是目录: {raw_text}")
+ try:
+ resolved = source.resolve()
+ except OSError as exc:
+ raise HTTPException(status_code=400, detail=f"路径无法解析: {raw_text}: {exc}") from exc
+ resolved_text = str(resolved)
+ if not any(resolved_text.startswith(prefix) for prefix in allowed_roots):
+ raise HTTPException(
+ status_code=400,
+ detail=f"路径必须位于 {', '.join(allowed_roots)} 之下: {raw_text}",
+ )
+ target_dir = job_dir / "_nl_imports"
+ target_dir.mkdir(parents=True, exist_ok=True)
+ if source.is_file():
+ destination = target_dir / source.name
+ candidate = destination
+ counter = 1
+ while candidate.exists():
+ candidate = target_dir / f"{source.stem}_{counter}{source.suffix}"
+ counter += 1
+ try:
+ shutil.copy2(source, candidate)
+ except OSError as exc:
+ raise HTTPException(status_code=400, detail=f"复制文件失败: {raw_text}: {exc}") from exc
+ name = candidate.name
+ else:
+ destination = target_dir / source.name
+ try:
+ shutil.copytree(source, destination, dirs_exist_ok=True)
+ except OSError as exc:
+ raise HTTPException(status_code=400, detail=f"复制目录失败: {raw_text}: {exc}") from exc
+ name = destination.name
+ imported[raw_text] = f"input/{job_id}/_nl_imports/{name}"
+ return {"success": True, "imported": imported}
+
+
@app.on_event("startup")
def initialize_agent_conversation_store() -> None:
"""Initialize the persistent conversation state used only by MainAgent."""
@@ -16354,7 +16421,7 @@ def _match_extension_stage_intent(message: str, job_id: str = "") -> dict[str, A
print(f"[ExtensionStage NL] list skipped: {exc}")
return {}
if not candidates:
- return {}
+ return _match_lightweight_extension_intent(text, job_id)
best: dict[str, Any] | None = candidates[0] if explicit_target_locked and len(candidates) == 1 else None
best_score = 1000 if best is not None else 0
@@ -16406,9 +16473,9 @@ def _match_extension_stage_intent(message: str, job_id: str = "") -> dict[str, A
best = ext
if not trigger:
- return {}
+ return _match_lightweight_extension_intent(text, job_id)
if not best or best_score <= 0:
- return {}
+ return _match_lightweight_extension_intent(text, job_id)
slots = resolve_extension_slots(text, best)
input_paths = slots.get("input_paths") or _extract_extension_input_paths(text)
runtime_parameters = _extract_extension_parameters(text, best)
@@ -16446,9 +16513,98 @@ def _match_extension_stage_intent(message: str, job_id: str = "") -> dict[str, A
}
+def _lightweight_settings():
+ """轻量拓展分析的 LightweightApiSettings（每次调用构造，避免启动期依赖）。"""
+ from gfa_core.extension_stage_lightweight.api import LightweightApiSettings
+
+ return LightweightApiSettings.from_environment(project_root=PROJECT_ROOT)
+
+
+def _lightweight_job_file_hints(job_id: str) -> list[str]:
+ """当前 job 的 input/workspace 下文件相对路径提示（供 LLM 匹配用户提到的文件）。"""
+ from gfa_core.extension_stage_lightweight import nl_intent
+
+ job_id = str(job_id or "").strip()
+ if not job_id:
+ return []
+ hints: list[str] = []
+ for root in (INPUT_ROOT / job_id, WORKSPACE_ROOT / job_id):
+ try:
+ if not root.is_dir():
+ continue
+ for path in root.rglob("*"):
+ if not path.is_file():
+ continue
+ try:
+ relative = path.relative_to(root)
+ except ValueError:
+ continue
+ prefix = "input" if str(root).startswith(str(INPUT_ROOT)) else "workspace"
+ hints.append(f"{prefix}/{job_id}/{relative.as_posix()}")
+ if len(hints) >= nl_intent.MAX_JOB_FILE_HINTS:
+ return hints
+ except OSError:
+ continue
+ return hints
+
+
+def _match_lightweight_extension_intent(message: str, job_id: str = "") -> dict[str, Any]:
+ """轻量拓展分析自然语言触发：粗筛命中后调用 DeepSeek 解析运行意图。
+
+ 返回 dict 含 lightweight=True；任何失败返回 {}，绝不抛异常。
+ """
+ try:
+ from gfa_core.extension_stage_lightweight import nl_intent
+
+ settings = _lightweight_settings()
+ records = nl_intent.load_published_lightweight_records(settings.store_root)
+ if not records:
+ return {}
+ resolution = nl_intent.resolve_lightweight_extension_intent(
+ message,
+ records,
+ environ=os.environ,
+ job_file_hints=_lightweight_job_file_hints(job_id),
+ )
+ except Exception as exc:
+ print(f"[LightweightNLIntent] matcher unavailable: {exc}")
+ return {}
+ if resolution is None or not resolution.extension_id:
+ return {}
+ params = {
+ "extension_id": resolution.extension_id,
+ "extensionName": resolution.extension_name or resolution.extension_id,
+ "extension_store": "lightweight",
+ "llm_slots": {
+ "input_overrides": resolution.input_overrides,
+ "parameter_overrides": resolution.parameter_overrides,
+ "explicit_prefix": resolution.explicit_prefix,
+ },
+ "prefill": resolution.prefill,
+ "confidence": resolution.confidence,
+ "parse_reason": resolution.reason,
+ }
+ return {
+ "stage": "extension_stage",
+ "stages": ["extension_stage"],
+ "primary_stage": "extension_stage",
+ "params": params,
+ "prefix": "",
+ "gene_family": "",
+ "species": "",
+ "task_kind": "extension_stage",
+ "taskType": "extension_stage",
+ "lightweight": True,
+ "confidence": resolution.confidence,
+ "message": f"已识别为轻量拓展分析：{params['extensionName']}。请按引导上传或选择输入文件并核对参数后执行。",
+ "job_id": job_id,
+ "workspace": str(PROJECT_ROOT / "workspace" / job_id) if job_id else "",
+ }
+
+
def parse_natural_language(message: str) -> dict:
"""解析自然语言，提取分析意图和参数
-
+
支持识别多个阶段，例如："帮我执行基因家族鉴定和绘制进化树"
会识别出 ['identify', 'phylogeny'] 两个阶段
"""
@@ -17757,6 +17913,42 @@ async def _run_agent_chat(
"handoff_request": request.message,
}
+ # 轻量拓展分析自然语言触发：预闸命中才查轻量 store（Phase A 命中在此不短路）
+ try:
+ from gfa_core.extension_stage_lightweight import nl_intent as _lightweight_nl
+
+ _lightweight_published = _lightweight_nl.load_published_lightweight_records(
+ _lightweight_settings().store_root
+ )
+ if _lightweight_published and _lightweight_nl._rough_trigger(request.message or "", _lightweight_published):
+ extension_intent = await asyncio.to_thread(
+ _match_extension_stage_intent, request.message, request.job_id or ""
+ )
+ if extension_intent and extension_intent.get("lightweight"):
+ assistant_message = extension_intent.get("message") or "已识别为轻量拓展分析。"
+ if conversation_turn is not None:
+ active_store.finish_turn(
+ conversation_turn,
+ assistant_message=assistant_message,
+ interaction_kind="independent",
+ action="propose_extension_stage",
+ intent=extension_intent,
+ )
+ return {
+ "success": True,
+ "status": "ok",
+ "action": "propose_extension_stage",
+ "intent": extension_intent,
+ "sources": [],
+ "tool_trace": [],
+ "mode": "analysis",
+ "completed_stages": completed_stages,
+ "model_used": "rule",
+ "session_id": session_id,
+ }
+ except Exception as exc:
+ print(f"[LightweightNLIntent] chat pre-gate skipped: {exc}")
+
global llm_manager
active_llm_manager = llm_manager
if active_llm_manager is None:
@@ -20960,7 +21152,7 @@ async def agent_plan(request: AgentPlanRequest):
if methodology_plan:
return await _methodology_saved_plan_response(request, session_id, methodology_plan, methodology_params)
extension_intent = _match_extension_stage_intent(request.message, request.job_id or "")
- if extension_intent:
+ if extension_intent and not extension_intent.get("lightweight"):
return _extension_stage_plan_response(extension_intent, session_id, request.job_id or "")
report_intent = dict(request.intent or {})
if report_intent.get("stages") == ["gene_family_report"]:
diff --git a/gfa_core/extension_stage_lightweight/api.py b/gfa_core/extension_stage_lightweight/api.py
index 0bb29a5da..d57108ac8 100644
--- a/gfa_core/extension_stage_lightweight/api.py
+++ b/gfa_core/extension_stage_lightweight/api.py
@@ -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(
diff --git a/gfa_core/extension_stage_lightweight/nl_intent.py b/gfa_core/extension_stage_lightweight/nl_intent.py
new file mode 100644
index 000000000..0d61eb879
--- /dev/null
+++ b/gfa_core/extension_stage_lightweight/nl_intent.py
@@ -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//<文件名>\" 或 "
+ "\"workspace//<文件名>\" 的 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,
+ )