"""Background executor for one skill x target matrix item.""" from __future__ import annotations import asyncio import json import logging from pathlib import Path from typing import Any import httpx from deerflow.assistant_knowledge import export_package_from_skill_ir from deerflow.integrations.weknora.client import WeKnoraError from deerflow.integrations.weknora.runtime import build_weknora_client, get_resolved_llmwiki_runtime from deerflow.skill_knowledge import build_skill_ir, enrich_converted_documents, scan_skill from deerflow.skill_knowledge.graph_writer import parameterized_statements, plan_from_ir from deerflow.skills.parser import parse_skill_file from deerflow.skills.storage import get_or_new_skill_storage from deerflow.skills.types import SkillCategory logger = logging.getLogger(__name__) def _skill_directory(app_config: Any, skill_name: str) -> Path: storage = get_or_new_skill_storage(app_config=app_config) root = storage.get_skills_root_path().resolve() for category in SkillCategory: direct = root / category.value / skill_name if direct.resolve().is_relative_to((root / category.value).resolve()) and (direct / "SKILL.md").is_file(): return direct for category in SkillCategory: category_root = root / category.value if not category_root.is_dir(): continue for skill_md in category_root.rglob("SKILL.md"): if skill_md.is_symlink(): continue parsed = parse_skill_file(skill_md, category=category, relative_path=skill_md.parent.relative_to(category_root)) if parsed and parsed.name == skill_name: return skill_md.parent raise FileNotFoundError(f"Skill '{skill_name}' not found under running skill roots") def _graph_projection(ir: dict[str, Any]) -> str: rows = [] entities = {str(item.get("local_id")): item for item in ir.get("entities", [])} for relation in ir.get("relations", []): if float(relation.get("confidence") or 0) < 0.75 or not relation.get("evidence"): continue source = entities.get(str(relation.get("source_local_id")), {}) target = entities.get(str(relation.get("target_local_id")), {}) rows.append( json.dumps( { "source": source.get("name"), "source_type": source.get("type"), "predicate": relation.get("predicate"), "target": target.get("name"), "target_type": target.get("type"), "description": relation.get("description"), "confidence": relation.get("confidence"), "evidence": relation.get("evidence"), }, ensure_ascii=False, ) ) return "\n".join(rows) class SkillKnowledgeJobExecutor: def __init__(self, app: Any) -> None: self.app = app self.store = app.state.skill_knowledge_store async def _cancel_if_requested(self, item: dict[str, Any]) -> bool: job = await self.store.get_job(str(item["job_id"])) if not (job or {}).get("cancel_requested"): return False await self.store.update_item( str(item["id"]), status="canceled", phase="completed", progress=100, ) return True async def _assert_source_unchanged(self, item: dict[str, Any], expected_digest: str) -> None: root = _skill_directory(self.app.state.config, str(item["skill_name"])) current = await asyncio.to_thread(scan_skill, str(item["skill_name"]), root) current = await enrich_converted_documents(current) if current.source_digest == expected_digest: return await self.store.mark_skill_stale(str(item["skill_name"])) raise RuntimeError("技能知识文件在归纳期间发生变化，旧版本保持激活，请重试") async def _write_direct_graph( self, item: dict[str, Any], snapshot_id: str, ir: dict[str, Any], ) -> dict[str, Any] | None: config = self.app.state.config.llmwiki.weknora.graph if not config.direct_write_enabled: return None if not config.neo4j_http_url or not config.username or not config.password: raise RuntimeError("Neo4j direct writer 已开启但后端连接配置不完整") plan = plan_from_ir( ir, target_type=str(item["target_type"]), target_id=str(item["target_id"]), binding_id=str(item["binding_id"]), snapshot_id=snapshot_id, job_id=str(item["job_id"]), ) statements = parameterized_statements(plan) if not statements: return {"nodes": 0, "edges": 0, "statement_count": 0} endpoint = f"{config.neo4j_http_url.rstrip('/')}/db/{config.database}/tx/commit" async with httpx.AsyncClient( auth=(config.username, config.password), timeout=config.timeout_seconds, ) as client: response = await client.post(endpoint, json={"statements": statements}) response.raise_for_status() payload = response.json() errors = payload.get("errors") or [] if errors: raise RuntimeError(f"Neo4j transaction failed: {str(errors[0].get('code') or 'unknown')}") return { "nodes": len(plan.nodes), "edges": len(plan.edges), "statement_count": len(statements), "scope": plan.scope, } async def run_item(self, item: dict[str, Any]) -> None: item_id = str(item["id"]) try: await self.store.update_item(item_id, status="running", phase="scanning", progress=5) root = _skill_directory(self.app.state.config, str(item["skill_name"])) scan = await asyncio.to_thread(scan_skill, str(item["skill_name"]), root) scan = await enrich_converted_documents(scan) if await self._cancel_if_requested(item): return await self.store.update_item(item_id, phase="extracting", progress=30) skill_record = await self.app.state.skill_store.get_any(str(item["skill_name"])) ir = await asyncio.to_thread( build_skill_ir, scan, display_name=(skill_record or {}).get("name_zh"), description=str((skill_record or {}).get("description") or ""), ) snapshot = await self.store.get_or_create_snapshot( skill_name=str(item["skill_name"]), source_digest=scan.source_digest, extractor_version=str(ir["extractor_version"]), prompt_hash=str(ir["prompt_hash"]), model_name=str(ir["model_name"]), manifest=scan.manifest, ir=ir, counts=scan.counts, ) snapshot_id = str(snapshot["id"]) await self.store.update_item(item_id, phase="awaiting_review", progress=45, snapshot_id=snapshot_id) job = await self.store.get_job(str(item["job_id"])) review_mode = str((job or {}).get("review_mode") or "auto_high_confidence") review_objects: list[tuple[str, dict]] = [] if review_mode == "required": review_objects = [("entity", obj) for obj in ir.get("entities", [])] + [("relation", obj) for obj in ir.get("relations", [])] elif review_mode == "auto_high_confidence": review_objects = [(kind, obj) for kind, rows in (("entity", ir.get("entities", [])), ("relation", ir.get("relations", []))) for obj in rows if float(obj.get("confidence") or 0) < 0.9] if review_objects: existing = await self.store.list_reviews(item_id=item_id, status=None) if not existing: await self.store.create_reviews( job_id=str(item["job_id"]), item_id=item_id, binding_id=str(item["binding_id"]), snapshot_id=snapshot_id, objects=review_objects, ) await self.store.update_item(item_id, status="awaiting_review", phase="awaiting_review", progress=50) return await self._project_and_activate(item, snapshot_id, ir, scan) except asyncio.CancelledError: raise except Exception as exc: # noqa: BLE001 logger.exception("Skill knowledge item failed: item=%s", item_id) await self.store.update_item( item_id, status="failed", phase="completed", progress=100, error_code=type(exc).__name__, error_message=str(exc)[:2000], ) async def resume_reviewed_item(self, item_id: str) -> None: if await self.store.pending_review_count(item_id): return jobs = await self.store.list_jobs(limit=100) item = None for job in jobs: detail = await self.store.get_job(str(job["id"])) item = next((row for row in (detail or {}).get("items", []) if row["id"] == item_id), None) if item: break if not item or not item.get("snapshot_id"): return snapshot = await self.store.get_snapshot(str(item["snapshot_id"])) ir = await self.store.effective_ir(str(item["snapshot_id"]), item_id) if not snapshot or not ir: return try: await self.store.update_item(item_id, status="running", phase="merging", progress=55) await self._project_and_activate(item, str(item["snapshot_id"]), ir, None) except Exception as exc: # noqa: BLE001 logger.exception("Reviewed skill knowledge item failed: item=%s", item_id) await self.store.update_item( item_id, status="failed", phase="completed", progress=100, error_code=type(exc).__name__, error_message=str(exc)[:2000], ) async def _project_and_activate(self, item: dict[str, Any], snapshot_id: str, ir: dict[str, Any], scan: Any | None) -> None: if await self._cancel_if_requested(item): return expected_digest = str(ir.get("skill", {}).get("source_digest") or "") if scan is None: root = _skill_directory(self.app.state.config, str(item["skill_name"])) scan = await asyncio.to_thread(scan_skill, str(item["skill_name"]), root) scan = await enrich_converted_documents(scan) await self._assert_source_unchanged(item, expected_digest) if item["target_type"] == "weknora" and item["target_mode"] == "wiki": await self._generate_vectorize_and_deposit(item, snapshot_id, ir, scan) return await self.store.update_item(str(item["id"]), phase="writing_wiki", progress=65) target_type = str(item["target_type"]) artifacts: list[dict[str, Any]] = [] binding_status = "synced" page = ir["wiki"]["skill_page"] if target_type == "assistant": package = export_package_from_skill_ir(ir) await self.app.state.assistant_knowledge_store.import_package( package=package, source_type="skill", source_key=str(item["skill_name"]), source_name=str(ir["skill"].get("display_name") or item["skill_name"]), trigger="skill_direct", created_by=(await self.store.get_job(str(item["job_id"]))).get("created_by"), skill_name=str(item["skill_name"]), ) artifacts.append({"kind": "assistant_wiki_revision", "slug": page["slug"], "content": page["content"], "business_key": f"assistant:wiki:{page['slug']}"}) elif str(item["target_mode"]) == "readonly_sidecar": artifacts.append({"kind": "sidecar_wiki", "slug": page["slug"], "content": page["content"], "business_key": f"sidecar:{item['target_id']}:{page['slug']}"}) artifacts.append({"kind": "sidecar_graph", "slug": f"skills/{item['skill_name']}/graph-projection.jsonl", "content": _graph_projection(ir)}) binding_status = "pending_remote_write" else: mapping = await self.app.state.llmwiki_store.get_authorized(str(item["target_id"]), "system", write=True, is_admin=True) if mapping is None: raise ValueError("Target WeKnora mapping does not exist") runtime = get_resolved_llmwiki_runtime(self.app.state.config) if not runtime.weknora_enabled: raise RuntimeError("普通知识库服务未配置") client = build_weknora_client(runtime) remote_kb_id = str(mapping["weknora_id"]) if scan is not None: for file in scan.files: source_path = Path(scan.source_root) / file.path if file.text: remote = await client.create_manual_document(remote_kb_id, title=f"skills/{item['skill_name']}/source/{file.path}", content=file.text) else: remote = await client.upload_document( remote_kb_id, filename=f"{item['skill_name']}-{source_path.name}", content=await asyncio.to_thread(source_path.read_bytes), ) remote_id = str(remote.get("id") or remote.get("knowledge_id") or "") or None artifacts.append( { "kind": "raw_source", "slug": file.path, "remote_id": remote_id, "business_key": f"weknora:document:{remote_id}" if remote_id else None, "remote_refs": [{"provider": "weknora", "remote_kind": "document", "remote_id": remote_id}] if remote_id else [], } ) target_mode = str(item["target_mode"]) if target_mode == "wiki": remote_page: dict[str, Any] try: await client.get_wiki_page(remote_kb_id, page["slug"]) except WeKnoraError as exc: if exc.status_code != 404: raise remote_page = await client.create_wiki_page( remote_kb_id, { "slug": page["slug"], "title": page["title"], "content": page["content"], "summary": str(ir["skill"].get("description") or ""), "page_type": "skill", }, ) else: remote_page = await client.update_wiki_page( remote_kb_id, page["slug"], { "title": page["title"], "content": page["content"], "summary": str(ir["skill"].get("description") or ""), }, ) remote_page_id = str(remote_page.get("id") or remote_page.get("page_id") or page["slug"]) artifacts.append( { "kind": "wiki_page", "slug": page["slug"], "content": page["content"], "remote_id": remote_page_id, "business_key": f"weknora:wiki_page:{remote_page_id}", "remote_refs": [ { "provider": "weknora", "remote_kind": "wiki_page", "remote_id": remote_page_id, "slug": page["slug"], } ], } ) else: document_content = page["content"] artifact_kind = "summary_document" if target_mode == "faq": artifact_kind = "faq_projection" document_content = f"问题：{page['title']} 能解决什么问题，如何使用？\n\n回答：\n{page['content']}" elif target_mode == "conversation_archive": artifact_kind = "conversation_archive" document_content = f"# 技能归纳会话归档：{page['title']}\n\n以下内容由技能知识归纳任务生成。\n\n{page['content']}" remote_document = await client.create_manual_document( remote_kb_id, title=f"skills/{item['skill_name']}/{target_mode}.md", content=document_content, ) remote_document_id = str(remote_document.get("id") or remote_document.get("knowledge_id") or "") or None artifacts.append( { "kind": artifact_kind, "slug": f"skills/{item['skill_name']}/{target_mode}.md", "content": document_content, "remote_id": remote_document_id, "business_key": (f"weknora:document:{remote_document_id}" if remote_document_id else None), "remote_refs": ( [ { "provider": "weknora", "remote_kind": "document", "remote_id": remote_document_id, } ] if remote_document_id else [] ), } ) projection = await client.create_manual_document(remote_kb_id, title=f"skills/{item['skill_name']}/graph-projection.jsonl", content=_graph_projection(ir)) projection_id = str(projection.get("id") or projection.get("knowledge_id") or "") or None artifacts.append({"kind": "graph_projection", "slug": f"skills/{item['skill_name']}/graph-projection.jsonl", "remote_id": projection_id}) try: graph_audit = await self._write_direct_graph(item, snapshot_id, ir) if graph_audit is not None: artifacts.append( { "kind": "neo4j_direct_write_audit", "slug": f"skills/{item['skill_name']}/neo4j-audit.json", "content": json.dumps(graph_audit, ensure_ascii=False), "metadata": graph_audit, } ) except Exception as exc: # noqa: BLE001 logger.exception("Optional Neo4j direct graph write failed: item=%s", item["id"]) binding_status = "partial_failed" artifacts.append( { "kind": "neo4j_direct_write_audit", "slug": f"skills/{item['skill_name']}/neo4j-audit.json", "content": json.dumps( {"status": "failed", "error": type(exc).__name__}, ensure_ascii=False, ), "metadata": { "status": "failed", "error_type": type(exc).__name__, }, } ) await self._assert_source_unchanged(item, expected_digest) await self.store.update_item(str(item["id"]), phase="activating", progress=90) await self.store.stage_and_activate(item_id=str(item["id"]), snapshot_id=snapshot_id, ir=ir, artifacts=artifacts, status=binding_status) async def _generate_vectorize_and_deposit(self, item, snapshot_id, ir, scan): mapping = await self.app.state.llmwiki_store.get_authorized(str(item["target_id"]), "system", write=True, is_admin=True) if mapping is None: raise ValueError("目标普通知识库不存在") sync = getattr(self.app.state, "llmwiki_sync_service", None) if sync is None: raise ValueError("请先配置 Wiki 编码模型并启用本地向量索引；归纳需要完成向量化后同步知识梳理") client = build_weknora_client(get_resolved_llmwiki_runtime(self.app.state.config)) remote_id = mapping["weknora_id"] remote = await client.get_knowledge_base(remote_id) if not (remote.get("indexing_strategy") or {}).get("wiki_enabled"): raise ValueError("目标普通知识库尚未开启 Wiki") artifacts = [] document_ids = [] await self.store.update_item(str(item["id"]), phase="writing_sources", progress=55) for file in scan.files: if await self._cancel_if_requested(item): return if not file.text: continue document = await client.create_manual_document(remote_id, title=f"{ir['skill'].get('display_name') or item['skill_name']} · {file.path}", content=file.text) doc_id = str(document.get("id") or document.get("knowledge_id") or "") if not doc_id: raise ValueError("资料上传未返回文档标识") document_ids.append(doc_id) artifacts.append({"kind": "raw_source", "slug": file.path, "remote_id": doc_id}) if not document_ids: raise ValueError("该技能没有可归纳的知识正文，代码、脚本和代码块不参与归纳") await self.store.update_item(str(item["id"]), phase="generating_wiki", progress=65) # WeKnora performs model extraction, categorization and linking. Wait # for both parsing and Wiki's debounced finalize before vectorization. stable = 0 for _ in range(720): if await self._cancel_if_requested(item): return docs = [await client.get_document(doc_id) for doc_id in document_ids] if any(doc.get("parse_status") in {"failed", "error"} for doc in docs): raise RuntimeError("WeKnora 资料解析失败，请查看普通知识库中的文档状态") stats = await client.get_wiki_stats(remote_id) done = all(doc.get("parse_status") == "completed" for doc in docs) and not stats.get("is_active") and not stats.get("pending_tasks") stable = stable + 1 if done else 0 if stable >= 3: if not int(stats.get("total_pages") or 0): raise RuntimeError("WeKnora 解析已结束，但未生成 Wiki；请检查 Wiki 生成模型的调用错误、余额及资料是否含实质知识，再重试") break await asyncio.sleep(5) else: raise TimeoutError("Wiki 分析尚未完成；请检查普通知识库的生成任务后重新归纳") await self.store.update_item(str(item["id"]), phase="vectorizing_wiki", progress=80) result = await sync.sync_mapping(mapping, force=False, notify=False) if result.get("status") != "completed": raise RuntimeError("Wiki 向量化未全部完成，请检查向量化进度") await self.store.update_item(str(item["id"]), phase="syncing_assistant", progress=92) from app.gateway.knowledge_transfer import sync_to_global deposited = await sync_to_global(self.app, mapping) artifacts.append({"kind": "assistant_deposit", "slug": "knowledge-summary", "metadata": deposited}) await self._assert_source_unchanged(item, str(ir["skill"]["source_digest"])) await self.store.stage_and_activate(item_id=str(item["id"]), snapshot_id=snapshot_id, ir=ir, artifacts=artifacts, status="synced")