#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。
372 lines
13 KiB
Python
372 lines
13 KiB
Python
"""统一能力发现层(双 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
|