467 lines
24 KiB
Python
467 lines
24 KiB
Python
"""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")
|