import json import logging import uuid import os import asyncio from typing import List from agent.tools import tool, ToolResult from agent.llm.openrouter import openrouter_llm_call # 导入底层 Postgres 资产表依赖 from tool_agent.tool.tool_store import PostgreSQLToolStore from tool_agent.tool.capability import PostgreSQLCapabilityStore logger = logging.getLogger(__name__) SYSTEM_PROMPT_CAPABILITY = """你是一个专业的能力分析师。 你的任务是从给定的【待分析新工具】的使用介绍和它挂载的【相关背景知识文章】中，提取出它对整网【原子能力表】的贡献，并选择新建或是融合。 ## 定义与格式 1. 原子能力是面向需求、跨工具的独立完整业务单元。 2. 端到端型工具（如Midjourney）直接抽取能力；编排平台工具（如ComfyUI节点群）从实际搭建的工作流中提取能力视角（不要原子化平台或独立节点本身）。 请输出严格的 JSON 数组结构： [ { "action": "create", "tool_id": "<当前待分析的工具ID>", "knowledge_ids": ["<哪些传入的相关知识促成了这个能力，填入真实的知识ID>"], "capability_id": "NEW_<任意数字字母临时ID>", "name": "<总结提炼的新能力统称名>", "criterion": "<客观统一的判定标准>", "description": "<抽象的能力需求场景说明>", "implement_description": "<在该特定工具里是如何实现该能力的（具体操作或调用链）>" }, { "action": "attach", "tool_id": "<当前待分析的工具ID>", "knowledge_ids": ["<关联的知识ID，可为空数组>"], "capability_id": "<来自下面【已有全量能力库】字典中完全等价功能的真实ID>", "implement_description": "<在该特定工具里实现此老能力的具体手法>" }, { "action": "update_and_attach", "tool_id": "<当前待分析的工具ID>", "knowledge_ids": ["<关联的知识ID>"], "capability_id": "<来自【已有全量能力库】的真实ID>", "name": "<更统整包容的全局更好命名（如果不改就留空不变）>", "criterion": "<更新优化后的判定标准>", "description": "<更新优化后的描述>", "implement_description": "<在该特定工具中如何实现>" } ] 请绝对不要输出 markdown 包装，仅输出原生的合法 JSON。如果一个工具覆盖了多个独立原子能力，请为每个能力出具一条动作操作。 """ def _fetch_knowledge_map(cursor, k_ids: list): if not k_ids: return {} placeholders = ','.join(['%s'] * len(k_ids)) cursor.execute(f"SELECT row_to_json(knowledge) as data FROM knowledge WHERE id IN ({placeholders})", list(k_ids)) mapping = {} for r in cursor.fetchall(): d = r['data'] text = str(d.get('content', d.get('markdown', d.get('description', '')))) mapping[d['id']] = {"title": d.get('title', ''), "content": text[:4000]} return mapping async def extract_capabilities_with_claude(existing_caps, tool_batch, knowledge_map): cap_str = json.dumps([{"id": c["id"], "name": c["name"], "criterion": c.get("criterion", "")} for c in existing_caps], ensure_ascii=False) tool_str = json.dumps([{"id": t["id"], "name": t["name"], "desc": t["introduction"], "docs": t["tutorial"], "associated_knowledge": t.get("knowledge_ids", [])} for t in tool_batch], ensure_ascii=False) knowledge_str = json.dumps(knowledge_map, ensure_ascii=False) prompt = f"【现有全量原子能力库字典】:\n{cap_str}\n\n【相关背景知识文章】:\n{knowledge_str}\n\n【本次待分析抽取合并的工具列表】:\n{tool_str}\n\n请严格输出JSON操作数组：" messages = [ {"role": "system", "content": SYSTEM_PROMPT_CAPABILITY}, {"role": "user", "content": prompt} ] result_text = "" try: result = await openrouter_llm_call( messages=messages, model="anthropic/claude-sonnet-4-5", temperature=0.2 ) result_text = result.get("content", "") except Exception as e: logger.error(f"OpenRouter API failed: {e}") return [] with open("raw_capability_responses.log", "a", encoding="utf-8") as f: f.write(f"\n--- Synchronous Capability Batch Output ---\n{result_text}\n") try: clean_json = result_text.strip() if clean_json.startswith("```json"): clean_json = clean_json[7:] elif clean_json.startswith("```"): clean_json = clean_json[3:] if clean_json.endswith("```"): clean_json = clean_json[:-3] data = json.loads(clean_json.strip()) if isinstance(data, dict): if "action" in data: return [data] return [] elif isinstance(data, list): return [item for item in data if isinstance(item, dict) and "action" in item] return [] except Exception as e: logger.error(f"Failed to parse capability JSON: {e}") return [] @tool() async def sync_atomic_capabilities(target_tool_ids: List[str]) -> ToolResult: """ 一键式强同步工具（为Librarian等智能体量身打造）。 针对新发现的或发生变动的特定工具/知识源，它能在数十秒内完成关联获取、大模型分析并直接完成底层 PostgreSQL 更新操作。 直接返回应用成功后的增减战报。 Args: target_tool_ids: 必须提供。指定要被大模型执行能力审查提取的工具 ID 列表。建议每次传入数量极少（如 1-3个）以保证 15 秒内同步快速返回。 """ if not target_tool_ids: return ToolResult(title="❌ 参数错误", output="必须提供 target_tool_ids，独立系统不再允许发起全量全局扫描避免阻塞。", error="Missing target_tool_ids") logger.info(f"开启单通道同步能力萃取 (目标: {target_tool_ids})...") cap_store = PostgreSQLCapabilityStore() tool_store = PostgreSQLToolStore() k_cursor = cap_store._get_cursor() stats = {"created": 0, "attached": 0, "updated": 0, "knowledge_inherited": 0} try: existing_caps = cap_store.list_all(limit=5000) all_tools = tool_store.list_all(limit=2000) target_tools = [t for t in all_tools if t.get("id") in target_tool_ids] if not target_tools: return ToolResult(title="❌ 未找到工具", output=f"找不到任何由 {target_tool_ids} 制定的接入工具") # 拉取目标工具的强绑定相关知识 batch_k_ids = set([k for t in target_tools for k in t.get("knowledge_ids", [])]) k_map = _fetch_knowledge_map(k_cursor, list(batch_k_ids)) # Claude 执行抽象推理构建矩阵 ops = await extract_capabilities_with_claude(existing_caps, target_tools, k_map) if not ops: return ToolResult(title="ℹ️ 分析完成", output="大模型判定当前工具没有提取出任何有效或创新的功能资产。") temp_id_mapping = {} # 落地阶段一：先创造出新的原子能力 for op in ops: if op.get("action") == "create" and op.get("capability_id") and op.get("tool_id"): real_id = f"cap-{uuid.uuid4().hex[:12]}" temp_id_mapping[op.get("capability_id")] = real_id t_id = op.get("tool_id") inherited_knowledge = op.get("knowledge_ids", []) stats["knowledge_inherited"] += len(inherited_knowledge) cap_store.insert_or_update({ "id": real_id, "name": op.get("name", ""), "criterion": op.get("criterion", ""), "description": op.get("description", ""), "tool_ids": [t_id], "implements": {t_id: op.get("implement_description", "")}, "knowledge_ids": inherited_knowledge }) stats["created"] += 1 # 落地阶段二：处理老能力的依附和扩展刷新 for op in ops: action = op.get("action") if action in ("attach", "update_and_attach") and op.get("capability_id") and op.get("tool_id"): c_id = temp_id_mapping.get(op.get("capability_id"), op.get("capability_id")) existing_cap = cap_store.get_by_id(c_id) if not existing_cap: continue if action == "update_and_attach": existing_cap["name"] = op.get("name") or existing_cap.get("name") existing_cap["criterion"] = op.get("criterion") or existing_cap.get("criterion") existing_cap["description"] = op.get("description") or existing_cap.get("description") stats["updated"] += 1 t_id = op.get("tool_id") imp_desc = op.get("implement_description", "") tool_ids = existing_cap.get("tool_ids", []) if t_id not in tool_ids: tool_ids.append(t_id) existing_cap["tool_ids"] = tool_ids implements = existing_cap.get("implements", {}) implements[t_id] = imp_desc existing_cap["implements"] = implements op_k_ids = op.get("knowledge_ids", []) if op_k_ids: existing_k_ids = set(existing_cap.get("knowledge_ids", [])) new_k_ids = [k for k in op_k_ids if k not in existing_k_ids] if new_k_ids: existing_k_ids.update(new_k_ids) existing_cap["knowledge_ids"] = list(existing_k_ids) stats["knowledge_inherited"] += len(new_k_ids) cap_store.insert_or_update(existing_cap) stats["attached"] += 1 return ToolResult( title="✅ 强同步萃取完成", output=f"强同步萃取完毕并入库: 新生能力 {stats['created']}, 修缮扩写 {stats['updated']}, 同化挂载 {stats['attached']} (沿袭知识网脉络 {stats['knowledge_inherited']} 条).\n\n详情记录:\n" + json.dumps(ops, ensure_ascii=False, indent=2) ) except Exception as e: logger.error(f"Sync capability extraction failed: {e}") cap_store.conn.rollback() return ToolResult(title="❌ 系统异常", output=f"执行时发生错误: {str(e)}", error=str(e)) finally: k_cursor.close() cap_store.close() tool_store.close()