jerryandClaude Fable 5
aab24d2f85
- QA pipeline/answer_composer/agentic_controller: env 检查从 =1 改 ==0 (默认放行;运维显式设 0 才禁用) - SDK qa.py: capabilities(web_search/superpowers) 注入上游,面板开关直达 QA - answer_composer: superpowers 改会话开关门控(capabilities.superpowers) - UI: 去掉 gated_by 提示(开关即生效,不再误导) - env: QA_V1_ENABLE_WEB_SEARCH/SUPERPOWERS 默认 1(熔断位) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
102 lines
3.6 KiB
Python
102 lines
3.6 KiB
Python
"""services/capabilities.py —— BFF 侧能力清单与开关管理（S4）。
|
||
|
||
- MANIFEST: BFF 管辖的会话级能力（memory/web_search/humanize）+ 可见性条目
|
||
- DEFAULTS: 新会话默认开关（memory=true, web_search=false, humanize=false）
|
||
- sanitize_capabilities: 白名单键 + bool 强转（注入 {"evil":1} 丢弃）
|
||
- 与 QA 注册表合并：GET /api/capabilities 返回 BFF 清单 + QA load_warnings + 已装插件
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Dict, List
|
||
|
||
# BFF 管辖能力清单：key -> 展示信息
|
||
MANIFEST: Dict[str, Dict[str, Any]] = {
|
||
"memory": {
|
||
"name": "memory",
|
||
"label": "多轮记忆",
|
||
"description": "同会话连续提问携带上文语境（检索仍按当前问题独立执行）",
|
||
"provider": "sdk",
|
||
"session_default": True,
|
||
},
|
||
"web_search": {
|
||
"name": "web_search",
|
||
"label": "联网搜索",
|
||
"description": "知识库证据不足时补充 PubMed/Crossref 最新文献（引用 [WEB: url]）",
|
||
"provider": "qa_gateway",
|
||
"session_default": False,
|
||
},
|
||
"humanize": {
|
||
"name": "humanize",
|
||
"label": "答案润色",
|
||
"description": "去除 AI 行文痕迹（引用/表格/图片/代码块原样保留）",
|
||
"provider": "sdk",
|
||
"session_default": False,
|
||
},
|
||
"superpowers": {
|
||
"name": "superpowers",
|
||
"label": "Superpowers 提示词包",
|
||
"description": "证据审查与答案自检强化（提示词注入，无 UI 开关）",
|
||
"provider": "qa_gateway",
|
||
"session_default": False,
|
||
},
|
||
"mcp_plugins": {
|
||
"name": "mcp_plugins",
|
||
"label": "MCP 插件",
|
||
"description": "已安装的外部 MCP server 工具（聊天内自助安装）",
|
||
"provider": "qa_gateway",
|
||
"session_default": False,
|
||
"catalog": True,
|
||
},
|
||
}
|
||
|
||
DEFAULTS: Dict[str, bool] = {
|
||
name: bool(entry.get("session_default", False))
|
||
for name, entry in MANIFEST.items()
|
||
}
|
||
|
||
# 会话 meta.capabilities 白名单（允许 PATCH 写入的键）
|
||
SESSION_KEYS = ("memory", "web_search", "humanize")
|
||
|
||
|
||
def sanitize_capabilities(raw: Any) -> Dict[str, bool]:
|
||
"""白名单 + bool 强转：非白名单键丢弃、值强转 bool（"1"/"true"/1 → True）。"""
|
||
if not isinstance(raw, dict):
|
||
return {}
|
||
out: Dict[str, bool] = {}
|
||
for key, value in raw.items():
|
||
if key not in SESSION_KEYS:
|
||
continue
|
||
if isinstance(value, bool):
|
||
out[key] = value
|
||
elif isinstance(value, str):
|
||
out[key] = value.strip().lower() in ("1", "true", "yes", "on")
|
||
else:
|
||
out[key] = bool(value)
|
||
return out
|
||
|
||
|
||
def merged_capabilities(
|
||
session_caps: Dict[str, bool] | None,
|
||
qa_registry: List[Dict[str, Any]] | None = None,
|
||
qa_warnings: List[str] | None = None,
|
||
) -> Dict[str, Any]:
|
||
"""GET /api/capabilities 的响应体：BFF 清单 + 当前会话开关 + QA 侧信息。"""
|
||
qa_registry = qa_registry or []
|
||
qa_warnings = qa_warnings or []
|
||
current = dict(DEFAULTS)
|
||
current.update(sanitize_capabilities(session_caps or {}))
|
||
capabilities = []
|
||
for name, entry in MANIFEST.items():
|
||
item = dict(entry)
|
||
item["enabled"] = current.get(name, False)
|
||
capabilities.append(item)
|
||
return {
|
||
"capabilities": capabilities,
|
||
"defaults": dict(DEFAULTS),
|
||
"installed": [
|
||
{"name": item.get("name"), "description": item.get("description")}
|
||
for item in qa_registry
|
||
],
|
||
"load_warnings": list(qa_warnings),
|
||
}
|