diff --git a/cheap-worker/cheap_service_app.py b/cheap-worker/cheap_service_app.py index bbc9a996..522f7586 100644 --- a/cheap-worker/cheap_service_app.py +++ b/cheap-worker/cheap_service_app.py @@ -444,6 +444,12 @@ def main() -> None: from worker import genconfig_nacos # noqa: PLC0415 —— 惰性 import,默认关时不牵 nacos 依赖 genconfig_nacos.build_and_attach_hot_source(tier="cheap") + # 知识包激活生效链接线(W-CFG-KB K3,与 genconfig 热源对称):TIER2_KB_ROOT 未配 → no-op(默认关、read_file + # 字节不变);已配 → 启动期按激活版物化知识根 +(Nacos 启用时)订阅知识包 dataId,激活推送即重新物化。best-effort: + # 任何异常只告警、不阻断 Service 启动 / 生成。 + from worker import kb_nacos # noqa: PLC0415 —— 惰性 import,默认关时不牵 nacos 依赖 + kb_nacos.setup_knowledge_activation() + uvicorn.run(build_cheap_app(), host=host, port=port) diff --git a/cheap-worker/tests/test_kb_externalize.py b/cheap-worker/tests/test_kb_externalize.py index a3942a4e..5d3992a7 100644 --- a/cheap-worker/tests/test_kb_externalize.py +++ b/cheap-worker/tests/test_kb_externalize.py @@ -267,3 +267,173 @@ def test_kb_addressing_anchors(): from worker import store as _src_store assert kb_store._KB_MYSQL_TABLE != _src_store._MYSQL_TABLE assert kb_store._KB_BUCKET_DEFAULT != "tier2-src" + + +# ─────────────────────────── K3:治理激活版本源接线(resolve 读 Nacos 激活版 · env 回落 · 变更即物化)─────────────────────────── + +from worker import kb_nacos # noqa: E402 + + +class _FakeSource: + """满足 kb_store 激活版本源 get(package_id)->version|None 契约的假源(单测注入,不连 Nacos)。""" + + def __init__(self, version, *, package_id="cheap-kb", raise_exc=None): + self._version = version + self._package_id = package_id + self._raise = raise_exc + + def get(self, package_id): + if self._raise is not None: + raise self._raise + if self._package_id is not None and package_id != self._package_id: + return None + return self._version + + +def test_resolve_prefers_governance_source(tmp_path, monkeypatch): + """K3 取值链:治理激活版本源(Nacos 下发)> env 钉版 > 最新 committed;detach 后回落。""" + store = _store(tmp_path) + latest = store.save([{"path": "a.md", "content": "A1"}], now_ts=_NOW, package_id="cheap-kb") + monkeypatch.setenv("TIER2_KB_ACTIVE_VERSION", "v-env-pinned") # env 钉了别的版 + kb_store.attach_active_version_source(_FakeSource("v-gov-active")) + try: + # 源优先于 env 与最新 committed。 + assert resolve_active_version(store, package_id="cheap-kb") == "v-gov-active" + finally: + kb_store.detach_active_version_source() + # detach 后回落 env 钉版。 + assert resolve_active_version(store, package_id="cheap-kb") == "v-env-pinned" + monkeypatch.delenv("TIER2_KB_ACTIVE_VERSION", raising=False) + # 无源无 env → 最新 committed(切片一行为、字节不变)。 + assert resolve_active_version(store, package_id="cheap-kb") == latest["versionId"] + + +def test_resolve_source_none_falls_back(tmp_path, monkeypatch): + """源返回 None(dataId 尚无激活版)→ 回落 env,再回落最新 committed。""" + store = _store(tmp_path) + v = store.save([{"path": "a.md", "content": "A1"}], now_ts=_NOW, package_id="cheap-kb") + monkeypatch.delenv("TIER2_KB_ACTIVE_VERSION", raising=False) + kb_store.attach_active_version_source(_FakeSource(None)) + try: + assert resolve_active_version(store, package_id="cheap-kb") == v["versionId"] + finally: + kb_store.detach_active_version_source() + + +def test_resolve_source_error_is_best_effort(tmp_path, monkeypatch): + """源取值抛异常 → best-effort 静默回落 env/最新 committed,绝不因治理源拖垮解析(→拖垮生成)。""" + store = _store(tmp_path) + v = store.save([{"path": "a.md", "content": "A1"}], now_ts=_NOW, package_id="cheap-kb") + monkeypatch.delenv("TIER2_KB_ACTIVE_VERSION", raising=False) + kb_store.attach_active_version_source(_FakeSource(None, raise_exc=RuntimeError("nacos down"))) + try: + assert resolve_active_version(store, package_id="cheap-kb") == v["versionId"] + finally: + kb_store.detach_active_version_source() + + +def test_resolve_source_package_mismatch_falls_back(tmp_path, monkeypatch): + """源里 dataId 标的是别的包 → get 返 None → 回落(不冒充本包激活版)。""" + store = _store(tmp_path) + v = store.save([{"path": "a.md", "content": "A1"}], now_ts=_NOW, package_id="cheap-kb") + monkeypatch.delenv("TIER2_KB_ACTIVE_VERSION", raising=False) + kb_store.attach_active_version_source(_FakeSource("v-other", package_id="some-other-kb")) + try: + assert resolve_active_version(store, package_id="cheap-kb") == v["versionId"] + finally: + kb_store.detach_active_version_source() + + +class _FakeNacosClient: + """假 v1 NacosClient:get_config 返预置内容,add_config_watcher 记回调供手动触发(不连真 Nacos)。""" + + def __init__(self, initial=None): + self._content = initial + self._cb = None + + def get_config(self, data_id, group): + return self._content + + def add_config_watcher(self, data_id, group, cb): + self._cb = cb + + def push(self, content): + """模拟 Nacos 长轮询推送:更新内容 + 以 v1 回调 dict 形态触发回调。""" + self._content = content + if self._cb is not None: + self._cb({"data_id": "gen-hot-params-knowledge", "group": "DEFAULT_GROUP", "content": content}) + + +def _kb_dataid_content(package_id, version): + """组一份知识包 dataId 内容(与 game-cloud codec 产的复合键载荷逐字对齐)。""" + import json + return json.dumps({"knowledge.packageId": package_id, "knowledge.activeVersion": version}) + + +def test_kb_source_parse_and_get(tmp_path): + """KbActiveVersionSource 解析 dataId 复合键 → get 返激活版;包 id 不符返 None(初读不物化)。""" + store = _store(tmp_path) + kb_root = tmp_path / "kb-active" + client = _FakeNacosClient(initial=_kb_dataid_content("cheap-kb", "v0001-aaa")) + src = kb_nacos.KbActiveVersionSource( + client=client, data_id="gen-hot-params-knowledge", group="DEFAULT_GROUP", + store=store, kb_root=kb_root, package_id="cheap-kb") + src.start() # 初读 populate 版本,不物化 + assert src.get("cheap-kb") == "v0001-aaa" + assert src.get("some-other-kb") is None # 包 id 不符不冒充 + assert not kb_root.exists() # 初读不物化(启动物化由 setup 显式做) + + +def test_kb_source_change_triggers_materialize(tmp_path, monkeypatch): + """激活推送(dataId 变更)→ 源 _on_change 触发重新物化 → read_file 读到新一版知识件。""" + store = _store(tmp_path) + kb_root = tmp_path / "kb-active" + v1 = store.save([{"path": "skill.md", "content": "GOV-V1"}], now_ts=_NOW, package_id="cheap-kb") + v2 = store.save([{"path": "skill.md", "content": "GOV-V2"}], now_ts=_NOW + 10, package_id="cheap-kb") + + client = _FakeNacosClient(initial=_kb_dataid_content("cheap-kb", v1["versionId"])) + src = kb_nacos.KbActiveVersionSource( + client=client, data_id="gen-hot-params-knowledge", group="DEFAULT_GROUP", + store=store, kb_root=kb_root, package_id="cheap-kb") + src.start() # 初读 v1(不物化) + kb_store.attach_active_version_source(src) + monkeypatch.setenv("TIER2_KB_ROOT", str(kb_root)) + try: + # 启动物化(setup 会做;这里显式模拟)→ resolve 读源 v1 → 物化 v1。 + materialize_active(store, kb_root) + assert cheap_run.read_file("skill.md")["content"] == "GOV-V1" + # 激活推送 v2 → 源 _on_change 触发重新物化。 + client.push(_kb_dataid_content("cheap-kb", v2["versionId"])) + assert src.get("cheap-kb") == v2["versionId"] + assert cheap_run.read_file("skill.md")["content"] == "GOV-V2" + finally: + kb_store.detach_active_version_source() + + +def test_kb_source_materialize_best_effort_on_missing(tmp_path, monkeypatch): + """激活推送到一个缺文件的版本 → 物化【显式失败】被 best-effort 接住(不抛)、kb_root 保留上一份完整包。""" + store = _store(tmp_path) + kb_root = tmp_path / "kb-active" + v1 = store.save([{"path": "skill.md", "content": "GOOD-V1"}], now_ts=_NOW, package_id="cheap-kb") + v2 = store.save( + [{"path": "skill.md", "content": "V2"}, {"path": "extra.md", "content": "E2"}], + now_ts=_NOW + 10, package_id="cheap-kb") + + client = _FakeNacosClient(initial=_kb_dataid_content("cheap-kb", v1["versionId"])) + src = kb_nacos.KbActiveVersionSource( + client=client, data_id="gen-hot-params-knowledge", group="DEFAULT_GROUP", + store=store, kb_root=kb_root, package_id="cheap-kb") + src.start() + kb_store.attach_active_version_source(src) + monkeypatch.setenv("TIER2_KB_ROOT", str(kb_root)) + try: + materialize_active(store, kb_root) # 物化 v1 + assert cheap_run.read_file("skill.md")["content"] == "GOOD-V1" + # 破坏 v2 的一个文件(manifest 列了但取不到)。 + victim = tmp_path / "_kb-store" / "cheap-kb" / v2["versionId"] / "files" / "extra.md" + victim.unlink() + # 推送 v2 → _on_change 触发物化 → 显式失败被 best-effort 接住(不抛),kb_root 保留 v1。 + client.push(_kb_dataid_content("cheap-kb", v2["versionId"])) # 绝不抛 + assert cheap_run.read_file("skill.md")["content"] == "GOOD-V1" # 仍读 v1 完整包 + finally: + kb_store.detach_active_version_source() diff --git a/game-admin/src/views/wanxiang/config/sanity.ts b/game-admin/src/views/wanxiang/config/sanity.ts index b8c4abb1..f5e28970 100644 --- a/game-admin/src/views/wanxiang/config/sanity.ts +++ b/game-admin/src/views/wanxiang/config/sanity.ts @@ -5,25 +5,48 @@ * 少一趟「定版→激活失败」往返。这是前端便利校验、非信任边界——真兜底仍是后端激活时的 codec 校验（走查断言 4 用真后端）。 * * 校验项（逐条对 codec：build / buildPathA / validateChatModelConfig）： - * 1. 至少一路：routeA 或 routeB（routeB 须非空对象）； + * 1. 至少一块：routeA / routeB（非空对象）/ readiness / knowledge（四者任一即有落点）； * 2. 有 routeA 则 agentId/sessionId 非空； * 3. 有 routeA 则 prompt 非空（system_prompt 是激活前置）； * 4. routeA.session.chat_model_config 存在且为对象，四字段 type/credential_id/model 非空、parameters 为对象； - * 5. max_tokens > thinking_budget（两者都给时；兼容扁平 thinking_budget 与嵌套 thinking.budget_tokens）。 + * 5. max_tokens > thinking_budget（两者都给时；兼容扁平 thinking_budget 与嵌套 thinking.budget_tokens）； + * 6. knowledge 块（W-CFG-KB K3，跨档知识包激活指针）：packageId/activeVersion 非空、与其余块互斥。 + * + * 说明：readiness（W-CFG-EXT 质量评分）与 knowledge（知识包激活）是两个跨档特例块，进程内/跨进程消费、不下发生成路。 + * 本预检对 readiness 只认「块存在=有落点」（其十字段数值校验是 admin 后续片，非本前端便利校验职责）；对 knowledge + * 给最小字段预检。真兜底仍是后端激活时的 codec.build（值域 sanity + 互斥）。 * * @param draftPrompt 草稿 prompt 正文 - * @param contentObj 已解析的 content_json 对象（routeA/routeB 嵌套信封） + * @param contentObj 已解析的 content_json 对象（routeA/routeB/readiness/knowledge 信封） * @returns null=校验通过；否则返回首个错误的中文说明 */ export function checkRouteASanity(draftPrompt: string | undefined, contentObj: any): string | null { const obj = contentObj && typeof contentObj === 'object' ? contentObj : {} const routeA = obj.routeA const routeB = obj.routeB + const readiness = obj.readiness + const knowledge = obj.knowledge const hasA = routeA != null && typeof routeA === 'object' const hasB = routeB != null && typeof routeB === 'object' && Object.keys(routeB).length > 0 + const hasReadiness = readiness != null && typeof readiness === 'object' + const hasKnowledge = knowledge != null && typeof knowledge === 'object' - if (!hasA && !hasB) { - return 'content_json 无任何可下发路由（至少需要 routeA 或非空 routeB 一路）' + if (!hasA && !hasB && !hasReadiness && !hasKnowledge) { + return 'content_json 无任何可下发路由（至少需要 routeA / routeB / readiness / knowledge 一块）' + } + + // knowledge 块（跨档知识包激活指针）：与其余块互斥 + packageId/activeVersion 非空（镜像后端 codec.buildKnowledgePathB）。 + if (hasKnowledge) { + if (hasA || hasB || hasReadiness) { + return 'knowledge 块与 routeA/routeB/readiness 互斥：知识包配置集只下发激活知识包版本' + } + if (!hasText(knowledge.packageId)) { + return 'knowledge.packageId 缺失或为空（知识包 id 必填）' + } + if (!hasText(knowledge.activeVersion)) { + return 'knowledge.activeVersion 缺失或为空（激活知识包须显式钉版本）' + } + return null } if (hasA) { diff --git a/game-cloud/game-module-aigc/game-module-aigc-api/src/main/java/com/wanxiang/huijing/game/module/aigc/enums/AigcConfigTierEnum.java b/game-cloud/game-module-aigc/game-module-aigc-api/src/main/java/com/wanxiang/huijing/game/module/aigc/enums/AigcConfigTierEnum.java index 37e262d2..a0257ae5 100644 --- a/game-cloud/game-module-aigc/game-module-aigc-api/src/main/java/com/wanxiang/huijing/game/module/aigc/enums/AigcConfigTierEnum.java +++ b/game-cloud/game-module-aigc/game-module-aigc-api/src/main/java/com/wanxiang/huijing/game/module/aigc/enums/AigcConfigTierEnum.java @@ -32,7 +32,17 @@ public enum AigcConfigTierEnum { * 账本指针前移」的纯账本收口。故其配置集内容与 routeA/routeB 互斥(codec 激活前置校验),也不参与按 cheap/tier2 * 拼 Nacos dataId 的下发逻辑。 */ - READINESS("readiness", "质量评分"); + READINESS("readiness", "质量评分"), + /** + * 知识包档(W-CFG-KB K3 · 跨档特例):承载便宜档生成 agent 的知识件(设计范式 skill/黄金脚手架/few-shot) + * 外置后的「当前激活知识包版本」指针,内容 = {@code {packageId, activeVersion}}。它与质量评分档一样不属任何 + * 生成档,但与质量评分档相反——要真下发到 Nacos:激活时经 {@link GenConfigContentCodec} 编成一条路 B 载荷 + * (复合键 {@code knowledge.packageId}/{@code knowledge.activeVersion}),按本档 dataId 模板 publish 到 + * {@code gen-hot-params-knowledge};便宜档 worker 起本档专用热源订阅它,读到激活版本号后据 kb_store 物化知识根, + * 改一版知识件 = 定版 + 激活即生效、不重部署。故其配置集内容与 routeA/routeB/readiness 互斥(codec 激活前置校验), + * 走独立 dataId、不冲便宜档/富游戏档的 gen-hot-params-cheap/tier2 预算门阈值键。 + */ + KNOWLEDGE("knowledge", "知识包"); /** 档位码（落库 tier 列） */ private final String tier; diff --git a/game-cloud/game-module-aigc/game-module-aigc-api/src/main/java/com/wanxiang/huijing/game/module/aigc/enums/ErrorCodeConstants.java b/game-cloud/game-module-aigc/game-module-aigc-api/src/main/java/com/wanxiang/huijing/game/module/aigc/enums/ErrorCodeConstants.java index 545d07b9..e51787a0 100644 --- a/game-cloud/game-module-aigc/game-module-aigc-api/src/main/java/com/wanxiang/huijing/game/module/aigc/enums/ErrorCodeConstants.java +++ b/game-cloud/game-module-aigc/game-module-aigc-api/src/main/java/com/wanxiang/huijing/game/module/aigc/enums/ErrorCodeConstants.java @@ -91,7 +91,7 @@ public interface ErrorCodeConstants { /** 值域 sanity 校验不通过：负预算/空模型名/阈值出域/max_tokens≤thinking/chat_model_config 不完整等（设计 §3.6，激活前置，拒绝不发任何一路）。占位符带明细。 */ ErrorCode AIGC_CONFIG_ACTIVATE_SANITY_FAILED = new ErrorCode(1_101_004_009, "配置值域校验不通过：{}"); /** 激活内容缺路由：content_json 未声明可激活的 routeA/routeB（无落点可下发）。 */ - ErrorCode AIGC_CONFIG_ACTIVATE_NO_ROUTE = new ErrorCode(1_101_004_010, "配置内容未声明任何可激活的下发路由（routeA/routeB/readiness 均缺）"); + ErrorCode AIGC_CONFIG_ACTIVATE_NO_ROUTE = new ErrorCode(1_101_004_010, "配置内容未声明任何可激活的下发路由（routeA/routeB/readiness/knowledge 均缺）"); /** 双路下发失败：某路 PATCH/publish 失败 → 整批判未生效、已把成功路补偿重推回上一激活版、指针不前移、可重试（设计 §3.6）。占位符带明细。 */ ErrorCode AIGC_CONFIG_ACTIVATE_DISPATCH_FAILED = new ErrorCode(1_101_004_011, "配置激活下发失败（已判未生效并补偿回上一激活版，可重试）：{}"); diff --git a/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigContentCodec.java b/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigContentCodec.java index 8fc06126..ba7b9ee9 100644 --- a/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigContentCodec.java +++ b/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigContentCodec.java @@ -61,6 +61,18 @@ import static com.wanxiang.huijing.framework.common.exception.util.ServiceExcept * 激活前置 readiness 值域 sanity(不过即拒):四维权重各 ∈[0,1] 且和 ≈1.0(±0.001)、中性/稳定性两级分 ∈[0,1]、 * 效率预算 > 0、修复轮 good_max ≤ ok_max;并与 routeA/routeB 互斥(readiness 存在时禁二者,反之亦然)。 * + *

第四块 knowledge(W-CFG-KB K3 · 知识包激活指针,与其余三块互斥):承载便宜档生成 agent 的知识件外置后的 + * 「当前激活知识包版本」指针,契约: + * 

+ * { "knowledge": { "packageId": "cheap-kb", "activeVersion": "v0001-abcdef123456" } }
+ *
+ * 它与 readiness 都是跨档特例,但走向相反:readiness 进程内消费产空载荷,knowledge 是跨进程(game-cloud 治理、 + * 便宜档 worker 消费),故激活时编成一条路 B 载荷 {@code {"knowledge.packageId":..,"knowledge.activeVersion":..}} + * (复合键与 routeB 的 area.key 同扁平口径),经 KNOWLEDGE 档 {@code gen-hot-params-knowledge} dataId publish 给 worker。 + * worker 读到激活版本号后据 kb_store 物化知识根,agent 的 read_file 即读到新一版设计范式——改一版知识件 = 定版 + 激活 + * 即生效、不重部署。激活前置 sanity:packageId 与 activeVersion 非空(激活知识包须显式钉版);与 routeA/routeB/readiness + * 互斥(知识包配置集只下发激活版本,不混生成路/评分块)。 + * * @author 绘境AI */ @Slf4j @@ -88,19 +100,33 @@ public final class GenConfigContentCodec { JsonNode routeA = root.get("routeA"); JsonNode routeB = root.get("routeB"); JsonNode readiness = root.get(ReadinessConfigKeys.BLOCK); + JsonNode knowledge = root.get(KnowledgeConfigKeys.BLOCK); boolean hasA = routeA != null && routeA.isObject(); boolean hasB = routeB != null && routeB.isObject() && routeB.size() > 0; boolean hasReadiness = readiness != null && readiness.isObject(); - if (!hasA && !hasB && !hasReadiness) { + boolean hasKnowledge = knowledge != null && knowledge.isObject(); + if (!hasA && !hasB && !hasReadiness && !hasKnowledge) { // 无任何可下发路由：没有落点（设计 §4 步骤3 交付以「双路下发」为核心，至少一路） throw exception(AIGC_CONFIG_ACTIVATE_NO_ROUTE); } - // 互斥(W-CFG-EXT):质量评分块是跨档、进程内消费、无外投影,不与生成下发路(routeA/routeB)共存于同一配置集。 + // 互斥①(W-CFG-KB):knowledge 是跨进程、走独立 dataId 的知识包激活指针,不与生成下发路(routeA/routeB)或进程内 + // 评分块(readiness)共存于同一配置集——混配会让激活按档拼错 dataId 或让投影语义打架,故激活前置拒(与 readiness + // 同纪律,让 KNOWLEDGE 档的 enum-value 建模对既有按档下发逻辑安全)。 + if (hasKnowledge && (hasA || hasB || hasReadiness)) { + throw sanity("knowledge 块与 routeA/routeB/readiness 互斥:知识包配置集只下发激活知识包版本,不得混配生成路/评分块"); + } + // 互斥②(W-CFG-EXT):质量评分块是跨档、进程内消费、无外投影,不与生成下发路(routeA/routeB)共存于同一配置集。 // 若一个质量评分配置集误混 routeB,激活会按 tier=readiness 拼出伪 dataId(resolveDataId 不认档但仍拼)真 publish; // 误混 routeA 则 resolveServiceUrl 返 null 下发失败。故激活前置拒,让 enum-value 建模对既有按档下发逻辑安全。 if (hasReadiness && (hasA || hasB)) { throw sanity("readiness 块与 routeA/routeB 互斥:质量评分配置集不下发 prompt/模型/预算/门阈值,不得混配生成路"); } + if (hasKnowledge) { + // 知识包配置集:值域 sanity + 产一条路 B 载荷(复合键 knowledge.packageId/activeVersion),按 KNOWLEDGE 档 + // dataId 模板 publish 到 gen-hot-params-knowledge;便宜档 worker 订阅它、读激活版本号后据 kb_store 物化知识根 + // (激活即生效)。pathA=null → dispatch 天然只走路 B(reconcile 同理只对账路 B),激活编排一行不改。 + return new GenConfigActivationPayload(null, buildKnowledgePathB(knowledge)); + } if (hasReadiness) { // 质量评分配置集:只做值域 sanity,不产任何外投影载荷。空载荷(pathA=pathB=null)→ 现有 dispatch/reconcile // 天然短路成「sanity + 账本收口、无投影、恒一致」(激活编排一行不改,见 GenConfigActivationServiceImpl.dispatch)。 @@ -177,6 +203,29 @@ public final class GenConfigContentCodec { return v.asInt(); } + // ==================== knowledge:知识包激活指针 → 路 B 载荷(W-CFG-KB K3)==================== + + /** + * 校验 content_json.knowledge 块并编成一条路 B 载荷(激活前置,不过即拒):packageId 与 activeVersion 均须非空—— + * 激活知识包必须显式钉一个版本(不像预算旋钮可留默认),worker 据 activeVersion {@code fetch} + 物化,取不到即物化 + * 显式失败、回落上一份完整包。产出 = {@code {"knowledge.packageId":"...","knowledge.activeVersion":"..."}} 扁平 JSON + * (与 routeB 的 area.key 复合键同口径),经 KNOWLEDGE 档 {@code gen-hot-params-knowledge} dataId publish 给 worker。 + */ + private String buildKnowledgePathB(JsonNode knowledge) { + String packageId = textOrNull(knowledge, KnowledgeConfigKeys.PACKAGE_ID); + String activeVersion = textOrNull(knowledge, KnowledgeConfigKeys.ACTIVE_VERSION); + if (!StringUtils.hasText(packageId)) { + throw sanity("knowledge." + KnowledgeConfigKeys.PACKAGE_ID + " 缺失或为空(知识包 id 必填)"); + } + if (!StringUtils.hasText(activeVersion)) { + throw sanity("knowledge." + KnowledgeConfigKeys.ACTIVE_VERSION + " 缺失或为空(激活知识包须显式钉版本)"); + } + ObjectNode out = JSON.createObjectNode(); + out.put(KnowledgeConfigKeys.NACOS_KEY_PACKAGE_ID, packageId); + out.put(KnowledgeConfigKeys.NACOS_KEY_ACTIVE_VERSION, activeVersion); + return writeJson(out); + } + // ==================== 路 A：/agent + /session 请求体组装 + 值域校验 ==================== private GenConfigActivationPayload.PathA buildPathA(String prompt, JsonNode routeA) { diff --git a/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/service/config/KnowledgeConfigKeys.java b/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/service/config/KnowledgeConfigKeys.java new file mode 100644 index 00000000..a3bd22b7 --- /dev/null +++ b/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/service/config/KnowledgeConfigKeys.java @@ -0,0 +1,51 @@ +package com.wanxiang.huijing.game.module.aigc.service.config; + +/** + * 知识包配置集(W-CFG-KB K3)的内容契约常量:content_json 里 {@code knowledge} 块的块名与字段名、 + * 知识包配置集的约定业务键,以及下发到 Nacos 的复合键。 + * + *

为何单独成类:这批字段名是「配置集内容(doc)」「激活编解码(code)」「worker 读激活版(跨进程)」三方之间的 + * 唯一约定——{@link GenConfigContentCodec} 激活前置的 knowledge 值域 sanity 与产路 B 载荷、便宜档 worker 侧 + * {@code kb_nacos.KbActiveVersionSource} 解析 Nacos dataId 都引用它,把字段名收进一处,避免多处各写一份字符串字面量、 + * 漂移无人对账。字段值(具体 packageId / versionId)不在这里,来自运营在配置集里填的当前激活知识包版本; + * 本类只管「用哪个键名读写」。 + * + *

与质量评分(readiness)块的区别:readiness 是进程内消费、无外投影(空载荷);knowledge 是跨进程—— + * game-cloud 治理它、便宜档生成 worker 消费它,故必须经 Nacos 下发一条路 B 载荷把激活版本号送到 worker。承载 dataId = + * {@code gen-hot-params-knowledge}(由 KNOWLEDGE 档套 {@code gen-hot-params-{tier}} 模板得到),内容是激活版本指针、 + * 不是预算门阈值热参,故与便宜档/富游戏档的生效 dataId 天然隔离、不相互冲刷。 + * + * @author 绘境AI(W-CFG-KB · K3 知识包治理接线) + */ +public final class KnowledgeConfigKeys { + + private KnowledgeConfigKeys() { + } + + /** content_json 里承载知识包激活指针的块名(平级于 routeA/routeB/readiness,互斥)。 */ + public static final String BLOCK = "knowledge"; + + /** + * 知识包稳定标识字段(一套便宜档知识语料一个 id;其下挂多版本)。与 worker 侧 + * {@code kb_store.DEFAULT_PACKAGE_ID} 对齐,默认 {@code cheap-kb}。 + */ + public static final String PACKAGE_ID = "packageId"; + /** + * 当前激活的知识包版本号字段。取值 = worker 侧 {@code kb_store} 落库时派生的 versionId(形如 + * {@code vXXXX-hash12});worker 据它 {@code fetch} + 物化,取不到即物化显式失败、回落上一份。 + */ + public static final String ACTIVE_VERSION = "activeVersion"; + + /** + * 知识包配置集的约定业务键(set_key)。跨档单例:全租户内知识包配置集用这一个业务键, + * uk_set_key 保证至多一条,寻址无歧义。运营在配置中心新建知识包配置集时须用此业务键。 + */ + public static final String KNOWLEDGE_SET_KEY = "knowledge-default"; + + // ---- 下发到 Nacos 的复合键(worker 侧 kb_nacos 读同名键;与 genconfig 的 area.key 复合键同扁平口径)---- + /** publish 到知识包 dataId 的「包 id」键 = {@code knowledge.packageId}。 */ + public static final String NACOS_KEY_PACKAGE_ID = BLOCK + "." + PACKAGE_ID; + /** publish 到知识包 dataId 的「激活版本」键 = {@code knowledge.activeVersion}。 */ + public static final String NACOS_KEY_ACTIVE_VERSION = BLOCK + "." + ACTIVE_VERSION; + +} diff --git a/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigActivationServiceImplTest.java b/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigActivationServiceImplTest.java index 8347cb8a..8fe068c0 100644 --- a/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigActivationServiceImplTest.java +++ b/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigActivationServiceImplTest.java @@ -65,6 +65,9 @@ class GenConfigActivationServiceImplTest extends BaseMockitoUnitTest { + "\"neutral_score\":0.5,\"efficiency_budget_rmb\":0.6," + "\"stability_repairs_good_max\":0,\"stability_repairs_ok_max\":2," + "\"stability_ok_score\":0.7,\"stability_bad_score\":0.4}}"; + /** 知识包配置集内容（knowledge-only，W-CFG-KB K3）：codec 产一条路 B 载荷、激活只 publish 到 gen-hot-params-knowledge。 */ + private static final String KNOWLEDGE_CONTENT = + "{\"knowledge\":{\"packageId\":\"cheap-kb\",\"activeVersion\":\"v0001-abcdef123456\"}}"; @BeforeEach void setUp() { @@ -239,6 +242,31 @@ class GenConfigActivationServiceImplTest extends BaseMockitoUnitTest { verify(nacosClient, never()).publish(anyString(), any()); } + // ==================== 知识包配置集（knowledge-only，W-CFG-KB K3）：只走路 B(publish gen-hot-params-knowledge) ==================== + + /** + * K3：知识包配置集（knowledge-only）激活 → codec 产一条路 B 载荷（knowledge.packageId/activeVersion）→ dispatch 只走路 B： + * tier=knowledge 经 {@link NacosConfigPublishClient#publish}("knowledge", ...) 下发到 gen-hot-params-knowledge dataId + * （便宜档 worker 订阅它读激活版物化知识根）；路 A（Agent Service）全程零触碰；账本正常收口。 + */ + @Test + void testActivate_knowledgeOnly_publishOnlyPathB() { + stubSet(1L, AigcConfigStatusEnum.PENDING_REVIEW, null, "knowledge"); // 跨档知识包档 + stubVersion(1L, 4000L, null, KNOWLEDGE_CONTENT); // 无 prompt、只有 knowledge 块 + when(genConfigService.beginActivating(1L, 4000L)) + .thenReturn(new GenConfigActivationSnapshot(AigcConfigStatusEnum.PENDING_REVIEW.getStatus(), null)); + when(nacosClient.publish(anyString(), any())).thenReturn(GenConfigDispatchResult.succeeded()); + + service.activate(1L, 4000L); + + // 只走路 B：按 knowledge 档 publish（resolveDataId 拼 gen-hot-params-knowledge）、路 A 零触碰 + verify(nacosClient).publish(eq("knowledge"), any()); + verify(agentClient, never()).patch(anyString(), any()); + // 账本正常收口（转 ACTIVE + 前移指针），不失败复位 + verify(genConfigService).markVersionActivated(1L, 4000L); + verify(genConfigService, never()).abortActivating(any(), any()); + } + // ==================== 前置拒绝：不写账本、不下发 ==================== /** 值域非法（负预算）→ 激活前置拒：不落 ACTIVATING、不发任何一路。 */ diff --git a/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigContentCodecTest.java b/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigContentCodecTest.java index 516f64c5..9295cae9 100644 --- a/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigContentCodecTest.java +++ b/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/service/config/GenConfigContentCodecTest.java @@ -305,4 +305,55 @@ class GenConfigContentCodecTest { assertEquals(AIGC_CONFIG_ACTIVATE_SANITY_FAILED.getCode(), sanityOf("sys", c).getCode()); } + // ==================== knowledge：知识包激活指针配置集（W-CFG-KB K3） ==================== + + /** 合法 knowledge 块（packageId + activeVersion）→ 一条路 B 载荷（复合键下发，worker 读激活版物化知识根）。 */ + @Test + void testBuild_knowledgeOnly_pathBOnly() { + GenConfigActivationPayload p = codec().build(null, + "{\"knowledge\":{\"packageId\":\"cheap-kb\",\"activeVersion\":\"v0001-abcdef123456\"}}"); + assertFalse(p.hasPathA(), "知识包配置集不下发路 A"); + assertTrue(p.hasPathB(), "知识包配置集走路 B（publish gen-hot-params-knowledge）"); + assertNull(p.pathA()); + // 复合键与 area.key 同扁平口径，worker 侧 kb_nacos 读同名键 + assertTrue(p.pathBJson().contains("knowledge.packageId")); + assertTrue(p.pathBJson().contains("knowledge.activeVersion")); + assertTrue(p.pathBJson().contains("cheap-kb")); + assertTrue(p.pathBJson().contains("v0001-abcdef123456")); + } + + /** knowledge 缺 activeVersion → sanity 拒（激活知识包须显式钉版）。 */ + @Test + void testKnowledgeSanity_missingActiveVersion() { + assertEquals(AIGC_CONFIG_ACTIVATE_SANITY_FAILED.getCode(), + sanityOf(null, "{\"knowledge\":{\"packageId\":\"cheap-kb\"}}").getCode()); + } + + /** knowledge packageId 空白 → sanity 拒。 */ + @Test + void testKnowledgeSanity_blankPackageId() { + assertEquals(AIGC_CONFIG_ACTIVATE_SANITY_FAILED.getCode(), + sanityOf(null, "{\"knowledge\":{\"packageId\":\" \",\"activeVersion\":\"v1\"}}").getCode()); + } + + /** 互斥：knowledge 与 routeB 共存 → sanity 拒（知识包块不得混配生成下发路）。 */ + @Test + void testKnowledge_mutexWithRouteB() { + String c = "{\"knowledge\":{\"packageId\":\"cheap-kb\",\"activeVersion\":\"v1\"}," + + "\"routeB\":{\"budget.cheap_rmb_hard_limit\":10.0}}"; + assertEquals(AIGC_CONFIG_ACTIVATE_SANITY_FAILED.getCode(), sanityOf(null, c).getCode()); + } + + /** 互斥：knowledge 与 readiness 共存 → sanity 拒（两个跨档特例块不得混配）。 */ + @Test + void testKnowledge_mutexWithReadiness() { + String c = "{\"knowledge\":{\"packageId\":\"cheap-kb\",\"activeVersion\":\"v1\"}," + + "\"readiness\":{\"playability_weight\":0.50,\"first_play_weight\":0.25," + + "\"stability_weight\":0.15,\"efficiency_weight\":0.10," + + "\"neutral_score\":0.5,\"efficiency_budget_rmb\":0.6," + + "\"stability_repairs_good_max\":0,\"stability_repairs_ok_max\":2," + + "\"stability_ok_score\":0.7,\"stability_bad_score\":0.4}}"; + assertEquals(AIGC_CONFIG_ACTIVATE_SANITY_FAILED.getCode(), sanityOf(null, c).getCode()); + } + } diff --git a/tier2/gen-worker/worker/kb_nacos.py b/tier2/gen-worker/worker/kb_nacos.py new file mode 100644 index 00000000..042dbc76 --- /dev/null +++ b/tier2/gen-worker/worker/kb_nacos.py @@ -0,0 +1,241 @@ +"""kb_nacos.py —— 知识包激活版本源:把 kb_store.resolve_active_version 的读取后端接到 Nacos(W-CFG-KB K3)。 + +【这份解决什么问题】 +切片一(kb_store)已把便宜档生成 agent 的知识件外置成版本化知识包(存 MinIO + manifest 落 MySQL),并留了 +「当前激活版本」接缝 resolve_active_version(切片一只会读 env 钉版 / 最新 committed)。K3 把这条接缝接到 game-cloud +的配置控制面:运营在配置控制台改一版知识件、定版、激活,game-cloud 把「当前激活知识包版本」publish 到一个专用 +Nacos dataId(gen-hot-params-knowledge);本模块在 worker 启动期订阅它,读到激活版本号后据 kb_store 物化知识根—— +激活推送到达即重新物化,agent 下次 read_file 就读到新一版设计范式,不重部署。 + +【与 genconfig 热源的关系:同范式、独立 dataId、独立槽】 +genconfig 的 Nacos 热源(genconfig_nacos)承载预算/门阈值旋钮(area.key 复合键),订阅便宜档 gen-hot-params-cheap; +知识包激活版本是另一码事(激活版本指针,不是热参),故走【独立 dataId】gen-hot-params-knowledge(由 game-cloud +KNOWLEDGE 档套 gen-hot-params-{tier} 模板得到),publish 一个不冲另一个的键。两者各 attach 各自的进程级槽:预算旋钮 +进 genconfig._hot_source,知识包激活版本进 kb_store._active_version_source。本模块复用 genconfig_nacos 的启用旗 +(TIER2_GENCONFIG_NACOS,Nacos 接入总开关)与内网代理旁路,不另造一套连接逻辑。 + +【默认关 = 现状字节不变】 +- TIER2_KB_ROOT 未配 → 知识件外置整体关闭:setup 直接返回,不物化、不订阅,read_file 回落仓根(切片一保证)。 +- TIER2_KB_ROOT 已配但 Nacos 未启用 → 只在启动期物化一次(env 钉版 / 最新 committed),不接 Nacos 激活版本源。 +- 两者都配 → 启动期物化 + 订阅 Nacos:激活推送即重新物化。 +全程 best-effort:构造/订阅/解析/物化任一步异常都只告警、不抛、不 attach —— 治理与激活绝不拖垮 worker 启动或生成。 + +【缺文件显式失败沿用】 +物化仍走 kb_store.materialize_active:激活版取不到 / manifest 列了但缺文件 / 指纹不匹配都【显式失败】,此时 kb_root +未被触碰 = 保留上一份已物化包,agent 仍读完整旧包。本模块把该显式失败在物化处 catch 成告警(best-effort 不咬生成), +但不削弱 materialize_active 内部的显式失败语义(它照抛,只是被这里的 best-effort 边界接住)。 +""" + +from __future__ import annotations + +import json +import os +import threading +from pathlib import Path +from typing import Any, Optional + +# genconfig_nacos 顶层零重依赖(不牵 agentscope / nacos),安全直接 import:复用其启用旗 + 内网代理旁路。 +try: + from . import genconfig_nacos # 包内相对导入(worker.kb_nacos) + from . import kb_store +except ImportError: # pragma: no cover —— PYTHONPATH=gen-worker 直跑时的绝对导入兜底 + from worker import genconfig_nacos # type: ignore + from worker import kb_store # type: ignore + + +# 知识包激活版本承载 dataId(与 game-cloud KNOWLEDGE 档 gen-hot-params-{tier} 模板产物逐字对齐)。 +# env TIER2_KB_NACOS_DATAID 显式覆盖(演示脚本指演示专用 dataId,不碰生产)。 +_DEFAULT_KB_DATAID = "gen-hot-params-knowledge" +_KB_DATAID_ENV = "TIER2_KB_NACOS_DATAID" + +# 下发到 dataId 的复合键(与 game-cloud KnowledgeConfigKeys.NACOS_KEY_* 逐字对齐;doc↔code 契约点)。 +_KEY_PACKAGE_ID = "knowledge.packageId" +_KEY_ACTIVE_VERSION = "knowledge.activeVersion" + + +class KbActiveVersionSource: + """知识包激活版本源(K3):订阅 Nacos 知识包 dataId、热读当前激活版本、变更即触发重新物化。 + + 对 kb_store.resolve_active_version 暴露 get(package_id)->version|None 契约(仅当 dataId 里的 packageId 与请求 + 一致才返版本,否则返 None 交回落);对 Nacos 侧订阅长轮询推送,激活版本变更即调 kb_store.materialize_active 把 + 知识根物化到新版(best-effort:物化异常保留上一份已物化包,不抛)。 + + 刻意不复用 nacos_hotconfig.NacosHotConfig:后者是通用只读视图(get 键值),本源要在变更时额外触发物化、且按 + packageId 匹配,语义不同;但订阅/回调/JSON 解析的骨架照 NacosHotConfig 同款(v1 watcher 后台线程 + 线程安全 + + best-effort),行为一致、无口径漂移。 + """ + + def __init__(self, *, client: Any, data_id: str, group: str, + store: Any, kb_root: Path, package_id: str) -> None: + """注入已建好的 v1 nacos.NacosClient + 知识包 dataId + kb_store(物化用) + 知识根 + 目标包 id。 + + client 由工厂建并负责代理旁路(建 client 前把 Nacos host 塞进 NO_PROXY);本类只消费。 + """ + self._client = client + self._data_id = data_id + self._group = group + self._store = store + self._kb_root = kb_root + self._package_id = package_id + self._lock = threading.RLock() + self._package: Optional[str] = None # 当前 dataId 里的 packageId + self._version: Optional[str] = None # 当前激活版本 + + def start(self) -> None: + """拉初值(populate 版本,不在初读时物化——启动物化由工厂显式做一次,避免双物化)+ 订阅推送(变更即物化)。""" + self._load_once() + try: + # v1 add_config_watcher(data_id, group, cb):第 3 位是回调,内部自起后台守护轮询线程,变更即回调。 + self._client.add_config_watcher(self._data_id, self._group, self._on_change) + print(f"[kb-nacos] 已订阅知识包激活版本 dataId={self._data_id} group={self._group}", flush=True) + except Exception as e: # noqa: BLE001 —— 订阅失败:留初值、不热更,但绝不炸(best-effort) + print(f"[kb-nacos] 订阅失败(留初值、不热更,best-effort):{type(e).__name__}: {e}", flush=True) + + def _load_once(self) -> None: + """经 get_config 拉一次初值(内建三级兜底);失败则版本停在 None、交 resolve 回落。初读不物化。""" + try: + content = self._client.get_config(self._data_id, self._group) + self._apply(content, materialize=False) + print(f"[kb-nacos] 初读完成 dataId={self._data_id} package={self._package} version={self._version}", + flush=True) + except Exception as e: # noqa: BLE001 —— Nacos 连兜底都没有:版本留 None,resolve 回落 env/最新 + print(f"[kb-nacos] 初读失败(版本留空、resolve 回落 env/最新,best-effort):" + f"{type(e).__name__}: {e}", flush=True) + + def _on_change(self, event: Any) -> None: + """v1 watcher 回调(后台 ThreadPool 线程):event 是 dict,取 content 解析;版本变了就重新物化。""" + content = event.get("content") if isinstance(event, dict) else event + self._apply(content, materialize=True) + + def _apply(self, content: Any, *, materialize: bool) -> None: + """解析 dataId 内容(area.key 扁平 JSON),更新缓存版本;materialize=True 且版本变了则触发重新物化。 + + best-effort:空内容 / 脏 JSON / 非对象都保留旧值、不抛。 + """ + if not content: + return # 空内容(get_config 三级全空返 None):dataId 尚无激活版,保留旧值(resolve 回落) + try: + parsed = json.loads(content) + except Exception as e: # noqa: BLE001 —— 脏内容不覆盖当前值,不抛 + print(f"[kb-nacos] 解析激活版本内容失败(保留旧值,best-effort):{type(e).__name__}: {e}", flush=True) + return + if not isinstance(parsed, dict): + print(f"[kb-nacos] 忽略非对象激活版本内容(保留旧值):{type(parsed).__name__}", flush=True) + return + pkg = parsed.get(_KEY_PACKAGE_ID) + ver = parsed.get(_KEY_ACTIVE_VERSION) + pkg = str(pkg).strip() if pkg and str(pkg).strip() else None + ver = str(ver).strip() if ver and str(ver).strip() else None + changed = False + with self._lock: + if pkg != self._package or ver != self._version: + self._package, self._version = pkg, ver + changed = True + if changed: + print(f"[kb-nacos] 知识包激活版本更新:package={pkg} version={ver}", flush=True) + # 变更 + 有版本 + 属本包(或 dataId 未标包 id)才物化;非本包的推送不误触发本 worker 物化。 + if materialize and changed and ver and (pkg is None or pkg == self._package_id): + self._materialize(ver) + + def _materialize(self, version: str) -> None: + """据激活版本把知识根物化到新版(显式传 version,不回查自身,避免 resolve 再绕一圈)。 + + best-effort 边界:materialize_active 内部对缺文件/取不到/指纹不符仍【显式失败】抛 + KnowledgeMaterializeError(kb_root 保留上一份),这里 catch 成告警——不削弱其显式失败语义,只保证 + 「激活版物化失败绝不拖垮生成」(agent 继续读上一份完整包 / 仓根)。 + """ + try: + res = kb_store.materialize_active( + self._store, self._kb_root, package_id=self._package_id, active_version=version) + print(f"[kb-nacos] 激活推送触发物化: {res}", flush=True) + except Exception as e: # noqa: BLE001 —— 含 KnowledgeMaterializeError:保留上一份,不抛(不咬生成) + print(f"[kb-nacos] ⚠ 激活版本物化失败(保留上一份已物化包/仓根,best-effort):" + f"{type(e).__name__}: {e}", flush=True) + + def get(self, package_id: str) -> Optional[str]: + """供 kb_store.resolve_active_version 查:返回当前激活版本;dataId 里的包 id 与请求不一致则返 None(交回落)。""" + with self._lock: + if self._package is not None and package_id != self._package: + return None # dataId 标了别的包 → 不冒充本包激活版 + return self._version + + +def _resolve_data_id() -> str: + """知识包 dataId:env TIER2_KB_NACOS_DATAID 显式覆盖 > 默认 gen-hot-params-knowledge。""" + override = os.environ.get(_KB_DATAID_ENV, "").strip() + return override or _DEFAULT_KB_DATAID + + +def build_and_attach_active_version_source(store: Any, kb_root: Path, *, + package_id: Optional[str] = None) -> Optional[KbActiveVersionSource]: + """worker 启动期:若 Nacos 已启用(复用 genconfig_nacos 总开关),构造知识包激活版本源、start()、attach 进 kb_store。 + + Returns: + 已 attach 的 KbActiveVersionSource;Nacos 未启用(默认)或构造失败 → None(resolve 回落 env/最新 committed)。 + 全程 best-effort:任何异常只告警、不抛、不 attach。 + """ + package_id = package_id or kb_store.DEFAULT_PACKAGE_ID + if not genconfig_nacos.is_enabled(): + # 默认关:不接 Nacos,resolve_active_version 走切片一链(env / 最新 committed)。 + return None + + nacos_addr = os.environ.get("NACOS_SERVER_ADDR", "100.64.0.8:8848") + namespace = os.environ.get("NACOS_NAMESPACE", "") # 默认公共空间 = 空串(勿写 public;前置任务口径) + username = os.environ.get("NACOS_USERNAME", "nacos") + password = os.environ.get("NACOS_PASSWORD", "") + group = os.environ.get("TIER2_GENCONFIG_NACOS_GROUP", "DEFAULT_GROUP") + data_id = _resolve_data_id() + + # 内网铁律:必须在【建 client 前】把 Nacos host 塞进 NO_PROXY(复用 genconfig_nacos 的旁路,不另写一份)。 + genconfig_nacos._bypass_proxy_for(nacos_addr.split(":")[0]) + + try: + import nacos # noqa: PLC0415 —— 惰性 import:默认关时不牵 nacos 依赖 + client = nacos.NacosClient(nacos_addr, namespace=namespace, username=username, password=password) + src = KbActiveVersionSource(client=client, data_id=data_id, group=group, + store=store, kb_root=kb_root, package_id=package_id) + src.start() # 初读 populate 版本(不物化;启动物化由 setup 显式做一次)+ 订阅(变更即物化) + kb_store.attach_active_version_source(src) + print(f"[kb-nacos] 已接知识包激活版本源:dataId={data_id} group={group} " + f"namespace={namespace!r} package={package_id} nacos={nacos_addr}", flush=True) + return src + except Exception as e: # noqa: BLE001 —— best-effort:接线失败不 attach、不抛,resolve 回落 env/最新 + print(f"[kb-nacos] 接知识包激活版本源失败(resolve 回落 env/最新 committed,best-effort):" + f"{type(e).__name__}: {e}", flush=True) + return None + + +def setup_knowledge_activation() -> Optional[KbActiveVersionSource]: + """worker(便宜档 Service / tier2)启动期一键接线知识包激活生效链(K3),与 genconfig 热源接线对称。 + + 做三件事(全 best-effort,任何异常只告警、绝不阻断 worker 启动 / 生成): + 1. TIER2_KB_ROOT 未配 → 知识件外置关闭,直接返回 None(默认关、read_file 字节不变); + 2. 已配 → 先(Nacos 启用时)attach 激活版本源,让启动物化就能读到治理激活版; + 3. 启动期物化一次:接了源就物化 Nacos 激活版,否则物化 env 钉版 / 最新 committed。 + + cheap_service_app.main / tier2 service.app.main 在起 uvicorn 前调它即可。 + """ + try: + kb_root = kb_store.kb_root_from_env() + if kb_root is None: + # 默认关:知识件外置未开,不物化、不订阅(现状字节不变)。 + return None + store = kb_store.default_kb_store() + # 先 attach Nacos 激活版本源(若启用):下面这次启动物化就能读到治理激活版(经 resolve → 源)。 + src = build_and_attach_active_version_source(store, kb_root) + # 启动期物化一次(源已 attach → 读 Nacos 激活版;否则 env 钉版 / 最新 committed)。 + try: + res = kb_store.materialize_active(store, kb_root) + print(f"[kb-nacos] 启动期物化知识根: {res}", flush=True) + except Exception as e: # noqa: BLE001 —— 含 KnowledgeMaterializeError:保留上一份/仓根,不阻断启动 + print(f"[kb-nacos] ⚠ 启动期物化失败(保留上一份已物化包/回落仓根,best-effort):" + f"{type(e).__name__}: {e}", flush=True) + return src + except Exception as e: # noqa: BLE001 —— 整条接线异常也不阻断 worker 启动 + print(f"[kb-nacos] ⚠ 知识包激活接线异常(跳过,best-effort):{type(e).__name__}: {e}", flush=True) + return None + + +def detach() -> None: + """卸载知识包激活版本源(K3 回退开关的显式入口;转调 kb_store.detach_active_version_source)。""" + kb_store.detach_active_version_source() diff --git a/tier2/gen-worker/worker/kb_store.py b/tier2/gen-worker/worker/kb_store.py index c47f79d6..477ee9fa 100644 --- a/tier2/gen-worker/worker/kb_store.py +++ b/tier2/gen-worker/worker/kb_store.py @@ -43,7 +43,7 @@ import os import shutil from abc import ABC, abstractmethod from pathlib import Path -from typing import Optional +from typing import Any, Optional # 纯内容寻址工具复用 store.py 的单一实现(整包指纹 / 版本 id 派生 / 相对路径安全校验)——不另抄一份, # 保证知识包与源工程用同一套「同内容 → 同指纹」「改源 → 新 versionId」语义。store.py 顶层只 import 标准库, @@ -627,17 +627,56 @@ def _minio_reader(client, bucket: str, prefix: str): return _read -# ── 当前激活版本解析(切片一简版;K3 的接缝点)──────────────────────────────────────────── +# ── 当前激活版本解析 + 治理激活版本源(K3 接缝点落地)────────────────────────────────────── +# 进程级单例槽:worker 启动期由 kb_nacos.build_and_attach_active_version_source() 注入一个「激活版本源」 +# (读 Nacos 下发的当前激活知识包版本);resolve_active_version 取值链最前端先查它。默认 None(未 attach)= +# 切片一行为:env 钉版 > 最新 committed(字节不变)。attach/detach 对称于 genconfig.attach_hot_source, +# 放本模块(纯 dict 槽、不牵 nacos 依赖,cheap venv 可 import);真正建 client + 订阅 + 变更即物化在 kb_nacos。 +_active_version_source: Any = None + + +def attach_active_version_source(source: Any) -> None: + """worker 启动期注入知识包激活版本源(K3);此后 resolve_active_version 最先查它。 + + source 只需满足 get(package_id)->version|None 契约(kb_nacos.KbActiveVersionSource 即是),本模块不依赖其 + 内部实现——故单测可注入任意 fake 源。幂等可重入:重复 attach 覆盖为最新源。 + """ + global _active_version_source + _active_version_source = source + + +def detach_active_version_source() -> None: + """卸载激活版本源,resolve_active_version 回落 env/最新 committed(K3 回退开关的进程内落点)。""" + global _active_version_source + _active_version_source = None + + def resolve_active_version(store: KnowledgePackageStore, *, package_id: str = DEFAULT_PACKAGE_ID) -> Optional[str]: - """解析「当前激活的知识包版本」:env TIER2_KB_ACTIVE_VERSION 显式钉版 > 最新 committed 版本。 + """解析「当前激活的知识包版本」:治理激活版本源(Nacos 下发) > env TIER2_KB_ACTIVE_VERSION 钉版 > 最新 committed。 - 这是 W-CFG-KB 治理面(K3)的接缝点:切片一用 env / 最新版这条简单来源;后续 K3 把它换成「读 yudao 配置集 - 当前激活的知识包版本」,并由激活推送触发 materialize_active。返回 None = 无激活版(物化 no-op,读回落仓根)。 + W-CFG-KB 治理面(K3)已把接缝落地:worker 启动期若接了知识包配置集的 Nacos 激活版本源(kb_nacos),这里最先查它 + ——改一版知识件经配置控制台定版 + 激活,激活推送到 Nacos,worker 读到新版本号即据它物化。未接源(默认 / TIER2_KB_ROOT + 未配 / Nacos 未启用)则回落切片一行为:env 钉版 > 最新 committed(字节不变)。 + + 治理源取值 best-effort:源不可达 / 无此包激活版 / 取值异常都静默回落 env/最新 committed,绝不让治理源拖垮物化 + (→拖垮生成)。返回 None = 无激活版(物化 no-op,read_file 回落仓根)。 """ + # ⓪ 治理激活版本源(K3 最高优先级):Nacos 下发的当前激活知识包版本。 + src = _active_version_source + if src is not None: + try: + gov = src.get(package_id) + if gov and str(gov).strip(): + return str(gov).strip() + except Exception as e: # noqa: BLE001 —— 治理源异常绝不连累物化:静默回落 env/最新 committed + print(f"[kb-store] ⚠ 读知识包激活版本源失败(回落 env/最新 committed): " + f"{type(e).__name__}: {e}", flush=True) + # ① env 显式钉版。 pinned = os.environ.get(KB_ACTIVE_VERSION_ENV) if pinned and pinned.strip(): return pinned.strip() + # ② 最新 committed 版本。 return store.latest_committed_version(package_id=package_id)