Files
qihuaiflow/backend/python/app/core/capability_discovery.py
T
QihuAI Design 33d0f9c32e feat(workflow #874): NodeRegistry 统一工作流节点抽象 + 双 Claw 事件对齐
#874 P3 工作流引擎 NodeRegistry 统一(加速带「工具即节点」思想落地,零源码拷贝重写):

- app/core/node_registry.py(新增):把 skill / MCP / Agent / DAG 四套割裂表达收敛为
  单一 Node 抽象。BaseNode(Protocol) 定义 node_type/name/input_schema/output_schema/
  risk_level + async run(ctx);NodeContext 携带 db/tenant_id/claw_type/trace_id/黑板/emit;
  NodeResult.to_blackboard() 与既有 orchestrator 黑板结构无缝互转。内建八型节点
  llm/reduce/local/external/agent/skill/mcp/loop,均继承 NodeBase 公共生命周期
  (发 start 步骤 -> 执行 -> ok/fail 覆盖同 index),异常统一归一为 NodeError(带结构化
  diagnosis)。SkillNode 用 asyncio.to_thread 包裹同步执行器避免阻塞事件循环;
  LoopNode 硬限流 20 次迭代 / 并发上限 5,子节点经 ctx.child() 复用同一 StepCounter 与
  emit,保证步骤序号全图唯一。CapabilitySnapshot 统一承载 skills/mcp_servers/agents/
  function_definitions/mcp_tool_map,供 _build_system_prompt、run_graph、yaml 校验三处
  共用同一份能力真相。normalize_claw() 将 im_wechat/im_platform 归一为 "im",避免
  Grafana 维度碎片化。

- app/core/capability_discovery.py:新增 discover_agents(),与 discover_skills 同构的
  三级降级策略(显式安装 -> published 兜底 -> rank_by_query 裁剪),补齐 Agent 维度的
  统一能力发现,杜绝各处内联第二份发现逻辑(#866 回退事故教训)。

- app/core/orchestrator.py:VALID_NODE_TYPES 与 validate_graph 改为运行期读
  registry.node_types()/requires_agent(),注册表成为节点类型的单一真相源;_run_node 改为
  拼装 NodeContext 后委托 registry.get(ntype).run(ctx),本体只保留调度与拓扑;run_graph
  新增 on_event/claw_type/tenant_id/snapshot 参数,编排全过程以统一 EngineEvent 下发。

- app/core/yaml_workflow.py:valid_node_types() 实时读注册表;validate_yaml 补 skill/mcp/
  loop 必填校验;yaml_to_orchestration / orchestration_to_yaml 双向透传节点扩展字段
  (label/skill_id/func_name/tool/server_id/args/body/items/split/max_iterations/
  concurrency/join),修复配置往返丢字段。

- app/core/engine_event.py:make_step 新增 node 参数,以 step["node"]={id,type,label}
  承载节点身份。设计书原文的 EngineEvent(kind="node") 与前端仅识别 thinking/skill/mcp/
  agent/kb/media 的现状冲突(未知 kind 会被静默丢帧),故映射为已知 kind + 附加字段,
  线格式对旧前端字节兼容。

- app/api/chat_service.py:Web Claw 编排调用点透传 on_event/claw_type/tenant_id,
  租户与通道标签复用 #875 的 ContextVar(get_route_ctx),不再新增一层参数透传。
  IM Claw 经 EngineCore 自动继承同一事件流(单引擎原则,im_llm.py 零改动)。

- tests/test_node_registry.py(新增):R-NODE-1~11 确定性哨兵,无网络无外部依赖。守卫
  内建八型齐全、注册表为单一真相源(orchestrator 与 yaml 校验实时同源)、describe 契约、
  双 Claw 事件对齐(SseAdapter 与 BufferAdapter 逐键相等且均带 node 身份)、claw 归一与
  tenant 标签、新节点必填校验、YAML 往返不吞字段、Loop 限流聚合、失败结构化诊断两端可见、
  禁止第二份能力发现、register 拒绝非法节点。

验证:容器内 py_compile 全通过;pytest tests/ 退出码 0(85 项 = 83 passed + 2 skipped,
0 FAILED/ERROR),含 R-NODE 16 项与 R-DUAL 双 Claw 哨兵。

safe-align: 仅复制 7 个 git-tracked 文件,未触碰 secrets/.venv/node_modules/dist/caches。
本地提交不 push。
2026-08-04 15:16:00 +08:00

372 lines
13 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""统一能力发现层(双 Claw 共用 · 单一实现)。
背景
----
七虎平台对外有两个 Claw 端口:
* Web Claw —— SSE 流式,用户可在 UI 勾选 skills / mcp / agents
* IM Bot Claw —— 微信等 IM,server push 整条回复,**没有任何勾选 UI**
历史问题(2026-08-04 线上实证):
1. `im_llm.py` 写死 `agents=[]` / `mcp=[]`,IM 端永远拿不到 MCP 工具;
2. `_discover_installed_skills()` 强依赖 `app_installations` 表,而线上该表
**0 行** → 97 个 published skill 在 Web / IM **双端同时发现不到**;
3. IM 端 skills 还被 `[:15]` 截断(截断一个空列表,属噪声)。
设计原则
--------
能力发现只允许有 **一份实现**(本模块)。任何 Claw 想要"技能/工具",
都必须调用这里的函数;禁止在端侧另写一套发现逻辑——历史上 #866 部署
就是因为存在第二份实现而把 #764/#765 的注入改动整体覆盖回退。
发现策略(三级降级,双端一致)
--------------------------
1) 用户显式安装(app_installations)—— 最高优先,尊重用户意愿
2) 无安装记录 → 降级为平台已发布(status='published')全集
3) 池子超过预算 → 按当前问题做轻量相关性排序,取 top-K 控制 token
运维旋钮(环境变量,改完重启生效)
--------------------------------
CAP_SKILL_LIMIT 单轮注入技能数上限,默认 12
CAP_MCP_LIMIT 单轮注入 MCP 服务数上限,默认 8
CAP_AGENT_LIMIT 单轮可见 Agent 数上限,默认 6
CAP_SKILL_FALLBACK_PUBLISHED 安装表为空时是否降级为 published,默认 开
CAP_AGENT_FALLBACK_PUBLISHED Agent 同上,默认 开
CAP_MCP_AUTO 未显式指定时是否自动挂载已启用 MCP,默认 开
"""
from __future__ import annotations
import logging
import os
import re
from typing import Any, Iterable
logger = logging.getLogger(__name__)
__all__ = [
"discover_skills",
"discover_mcp_servers",
"discover_agents",
"build_mcp_tool_defs",
"rank_by_query",
]
# --------------------------------------------------------------------------
# 环境旋钮
# --------------------------------------------------------------------------
def _env_int(name: str, default: int) -> int:
try:
raw = (os.environ.get(name) or "").strip()
return int(raw) if raw else default
except Exception:
return default
def _env_flag(name: str, default: bool = True) -> bool:
raw = (os.environ.get(name) or "").strip().lower()
if not raw:
return default
return raw in ("1", "true", "yes", "on")
# --------------------------------------------------------------------------
# 轻量相关性排序(中文友好:整词 + 2-gram)
# --------------------------------------------------------------------------
def _tokens(text: str) -> set:
text = re.sub(r"[^\w\u4e00-\u9fff]+", " ", (text or "").lower())
out: set = set()
for w in text.split():
if len(w) >= 2:
out.add(w)
for i in range(len(w) - 1):
out.add(w[i : i + 2])
return out
def rank_by_query(items: list, query: str, top_k: int) -> list:
"""按与 query 的词元重叠度排序取前 top_k。
池子本来就不超预算时原样返回,避免无谓抖动(同一问题两次调用结果稳定)。
"""
if not items:
return []
if top_k <= 0 or len(items) <= top_k:
return items[:top_k] if top_k > 0 else []
q = _tokens(query)
if not q:
return items[:top_k]
scored = []
for idx, it in enumerate(items):
blob = " ".join(
str(it.get(k) or "")
for k in ("name", "description", "category", "func_name")
)
t = _tokens(blob)
if t:
# 交集数按文档长度做平方根归一,抑制"长描述天然占优"
score = len(q & t) / (len(t) ** 0.5)
else:
score = 0.0
# idx 作为稳定次序的 tie-breaker,保证同分时结果可复现
scored.append((-score, idx, it))
scored.sort(key=lambda x: (x[0], x[1]))
return [it for _, _, it in scored[:top_k]]
# --------------------------------------------------------------------------
# 技能发现
# --------------------------------------------------------------------------
def discover_skills(
db,
user_id: str | None,
query: str = "",
limit: int | None = None,
claw: str = "web",
) -> list:
"""发现当前请求可用的技能列表(双 Claw 共用)。
返回 `_resolve_market_skills` 的结构:
[{id, name, description, skill_content, category, func_name, parameters}]
"""
limit = limit if limit is not None else _env_int("CAP_SKILL_LIMIT", 12)
if limit <= 0:
return []
# 延迟导入:避免 app.core ←→ app.api 循环依赖
from app.api.chat_skill import _resolve_market_skills
from app.models import AppInstallation, MarketApp
pool: list = []
source = "empty"
# 1) 用户显式安装
if user_id:
try:
app_ids = [
i.app_id
for i in db.query(AppInstallation)
.filter(AppInstallation.user_id == user_id)
.all()
]
if app_ids:
ids = [
a.id
for a in db.query(MarketApp)
.filter(
MarketApp.id.in_(app_ids),
MarketApp.type == "skill",
MarketApp.status == "published",
)
.all()
]
if ids:
pool = _resolve_market_skills(db, ids, user_id)
source = "installed"
except Exception as e:
logger.warning("[cap] 安装技能发现失败(降级继续): %s", e)
# 2) 降级:平台已发布全集
if not pool and _env_flag("CAP_SKILL_FALLBACK_PUBLISHED", True):
try:
ids = [
a.id
for a in db.query(MarketApp)
.filter(MarketApp.type == "skill", MarketApp.status == "published")
.all()
]
if ids:
pool = _resolve_market_skills(db, ids, user_id or "")
source = "published"
except Exception as e:
logger.warning("[cap] published 技能发现失败: %s", e)
# 3) 预算裁剪
picked = rank_by_query(pool, query, limit)
logger.info(
"[cap] skills claw=%s source=%s pool=%d picked=%d",
claw,
source,
len(pool),
len(picked),
)
return picked
# --------------------------------------------------------------------------
# MCP 发现
# --------------------------------------------------------------------------
def discover_mcp_servers(
db,
explicit_ids: Iterable[str] | None = None,
limit: int | None = None,
claw: str = "web",
) -> list:
"""发现当前请求可用的 MCP 服务连接(双 Claw 共用)。
Web Claw 传 explicit_ids(用户在 UI 勾选的);
IM Bot Claw 无勾选通道 → 传 None,自动挂载平台已启用的 MCP。
返回 [{id, name, description, transport, endpoint, url, headers}]
"""
limit = limit if limit is not None else _env_int("CAP_MCP_LIMIT", 8)
if limit <= 0:
return []
from app.api.chat_mcp import _list_enabled_mcp_servers, _resolve_active_mcp_servers
ids = [i for i in (explicit_ids or []) if i]
if ids:
out = _resolve_active_mcp_servers(db, ids)[:limit]
logger.info("[cap] mcp claw=%s source=explicit picked=%d", claw, len(out))
return out
if not _env_flag("CAP_MCP_AUTO", True):
return []
out = _list_enabled_mcp_servers(db)[:limit]
logger.info("[cap] mcp claw=%s source=auto picked=%d", claw, len(out))
return out
# --------------------------------------------------------------------------
# Agent 发现(#874:画布节点候选 / IM `/flow` 可选清单)
# --------------------------------------------------------------------------
def discover_agents(
db,
user_id: str | None,
query: str = "",
explicit_ids: Iterable[str] | None = None,
limit: int | None = None,
claw: str = "web",
) -> list:
"""发现当前请求可用的智能体(双 Claw 共用),策略与 discover_skills 同构。
返回 `_resolve_market_agents` 的结构:
[{id, name, role_prompt, instructions, skill_ids, knowledge_base_ids,
model_id, tools, description, tool_policy, platform, agent_endpoint,
agent_api_key, orchestration}]
用途定位:给 NodeRegistry 生成 agent 型节点候选、给 IM `/flow` 列可选智能体。
若要拿它往 system prompt 里注入,请自行收窄 limit —— 多个 Agent 人格叠加会串味。
"""
limit = limit if limit is not None else _env_int("CAP_AGENT_LIMIT", 6)
if limit <= 0:
return []
from app.api.chat_skill import _resolve_market_agents
from app.models import AppInstallation, MarketApp
# 0) 显式指定(Web Claw 勾选 / 画布已引用)—— 不做相关性裁剪,用户意愿最高
ids = [i for i in (explicit_ids or []) if i]
if ids:
out = _resolve_market_agents(db, ids, user_id or "")
logger.info("[cap] agents claw=%s source=explicit picked=%d", claw, len(out))
return out
pool: list = []
source = "empty"
# 1) 用户显式安装
if user_id:
try:
app_ids = [
i.app_id
for i in db.query(AppInstallation)
.filter(AppInstallation.user_id == user_id)
.all()
]
if app_ids:
aids = [
a.id
for a in db.query(MarketApp)
.filter(
MarketApp.id.in_(app_ids),
MarketApp.type == "agent",
MarketApp.status == "published",
)
.all()
]
if aids:
pool = _resolve_market_agents(db, aids, user_id)
source = "installed"
except Exception as e:
logger.warning("[cap] 安装 Agent 发现失败(降级继续): %s", e)
# 2) 降级:平台已发布全集
if not pool and _env_flag("CAP_AGENT_FALLBACK_PUBLISHED", True):
try:
aids = [
a.id
for a in db.query(MarketApp)
.filter(MarketApp.type == "agent", MarketApp.status == "published")
.all()
]
if aids:
pool = _resolve_market_agents(db, aids, user_id or "")
source = "published"
except Exception as e:
logger.warning("[cap] published Agent 发现失败: %s", e)
# 3) 预算裁剪
picked = rank_by_query(pool, query, limit)
logger.info(
"[cap] agents claw=%s source=%s pool=%d picked=%d",
claw,
source,
len(pool),
len(picked),
)
return picked
# --------------------------------------------------------------------------
# MCP → function definitions
# --------------------------------------------------------------------------
def _safe_name(raw: str, maxlen: int) -> str:
return re.sub(r"[^A-Za-z0-9_-]", "_", (raw or "")[:maxlen])
async def build_mcp_tool_defs(servers: list, idx_start: int = 0) -> tuple:
"""把 MCP 服务列表展开为 (function_definitions, tool_map)。
命名规则与 Web Claw 现有实现严格一致:`mcp__{server}__{tool}__{idx}`,
保证两端产生的工具名可互相对照、可被同一套审批/统计逻辑识别。
返回:
function_definitions: [{name, description, parameters}]
tool_map: {func_name: {server_id, name, server_name}}
"""
from app.mcp_client import list_mcp_tools
defs: list = []
tool_map: dict = {}
for offset, srv in enumerate(servers or []):
idx = idx_start + offset
try:
tools = await list_mcp_tools(srv)
except Exception as e:
logger.warning("[cap] MCP 拉取工具失败 [%s]: %s", srv.get("name"), e)
continue
for t in tools or []:
sname = _safe_name(srv.get("name") or "unknown", 30)
tname = _safe_name(t.get("name") or "tool", 50)
fname = f"mcp__{sname}__{tname}__{idx}"
tool_map[fname] = {
"server_id": srv.get("id"),
"name": t.get("name"),
"server_name": srv.get("name"),
}
defs.append(
{
"name": fname,
"description": (
f"[MCP:{srv.get('name')}] {t.get('description') or t.get('name') or ''}"
)[:900],
"parameters": t.get("inputSchema")
or {"type": "object", "properties": {}},
}
)
return defs, tool_map