--- name: tbs-scenario-builder description: 编排并执行训练场景（TBS）创建流程：意图路由、字段解析与追问、发布级骨架、persona/prompts 生成、apiDraft 去重证据、统一校验闸门与确认后落库。**禁止**用浏览器自动化操作 TBS 管理后台；落库仅经脚本 API。 skillcode: tbs-scenario-builder github: https://github.com/xgjk/xg-skills/tree/main/CMS-tbs-scenario-builder dependencies: - cms-auth-skills --- # tbs-scenario-builder — 索引本文件提供**能力宪章 + 能力树 + 按需加载规则**。详细参数与流程见各模块 `openapi/` 与 `examples/`。 `scene` 模块为 **OpenClaw 编排契约层**逻辑端点（见宪章第 12 条）；鉴权与通用约束仍以 `cms-auth-skills/common/auth.md` 与 `cms-auth-skills/common/conventions.md` 为准。 **当前版本**: v1.3 **接口版本**: - 场景编排契约端点：`/v1/scene/*`（逻辑端点，仅用于契约对照） - TBS 管理接口：`/api/v1/admin/*`（通过 `preflight-tbs-master-data.py` 与 `tbs_write_executor.py` 访问） **域名说明**： - 编排契约命名空间（非公网）：`https://scenario-builder.openclaw.internal` - TBS 管理接口（默认）：`https://sg-tbs-manage.mediportal.com.cn` **能力概览（2 块能力）**： - `scene`：训练场景创建全链路——意图路由、解析追问、发布级骨架、画像、提示词、apiDraft+去重证据、统一校验闸门、确认后执行本地落库脚本。 - `common`：内部复用/约定承载模块（无对外业务 endpoint）。快速入口（优先读）： 0. `openapi/README.md`、`examples/README.md`、`scripts/README.md`（目录级导航） 1. `openapi/scene/api-index.md`（先判定步骤，再按需加载 endpoint 文档） 2. `examples/scene/README.md`（用户可见话术、输出边界、最短成功路径与失败回退） 3. `scripts/scene/README.md`（脚本映射与执行方式）统一规范： - 认证与鉴权：`cms-auth-skills/common/auth.md` - 通用约束：`cms-auth-skills/common/conventions.md` - **静态资源（策略/画像/提示词/角色映射）**：`./references/`（见 `references/README.md`；加载 `*.persona.json` / `*.prompt.json` / `*.strategy.json` / `role_maps/role_type_map.json` 时以此为根） - **落库与本地数据**：`./scripts/tbs_assets/`（见 `scripts/tbs_assets/README.md`）— `scenario_draft.json`、`system_business_domains.json`、持久化记录与本地鉴权材料；executor 入口为 `./scripts/scene/tbs_write_executor.py` - **品种（product）与 TBS 药品主数据（强制，对齐 FR-4 写库链）**：`scenarioPack.product` 与 `apiDraft.scenes` 中的 `drugName` / `drug_id` 必须能落到 TBS 药品记录。**确认落库前**应执行 `preflight-tbs-master-data.py`（与落库相同逻辑）；**落库时** `tbs_write_executor.py` 再次调用 `scripts/scene/tbs_master_data_resolve.py` 中 `resolve_ids_for_scene`（先 `GET /api/v1/admin/basic/drugs`，**无则 `POST` 创建**），再 `POST /api/v1/admin/scenes`。接口清单见工作区 `TBS/TBS_API_REFERENCE.md`「§4.4 基础数据管理 — 药品 (Drugs)」。 - **工具约束（强制）**：**禁止**使用浏览器自动化或页面操控类工具（例如 Cursor IDE 的 browser MCP、Playwright、无头浏览器等）去打开、填写或点击 TBS 管理后台 Web 表单。该类页面多为 React 受控组件，自动化填充常无法触发合法 state，导致保存不可用或产生脏数据。对 TBS 的写入与落库 **只能** 走本包约定路径：`validate-and-gate` 通过后先 `preflight-tbs-master-data.py`（可选但推荐），用户确认后由 `persist-and-execute.py` 子进程执行 `tbs_write_executor.py`（对 `TBS_BASE_URL` 的 HTTP API）。若用户必须在浏览器里手工操作，agent **只**提供字段清单与逐步说明，**不**代填、不代点、不启动 browser 工具。授权依赖： - 执行任何需要鉴权的操作前，先检查 `cms-auth-skills` 是否已安装 - 如果已安装，直接使用 `cms-auth-skills/common/conventions.md`、`cms-auth-skills/common/auth.md`、`cms-auth-skills/openapi/auth/appkey.md`、`cms-auth-skills/openapi/auth/login.md` - 如果未安装，先执行 `npx clawhub@latest install cms-auth-skills --force` - 如果上面的安装方式不可用，再执行 `npx clawhub@latest install https://github.com/spzwin/cms-auth-skills.git --force` - 安装完成后，再继续执行需要鉴权的操作输入完整性规则（强制）： 1. **脚本硬性缺口（与 `parse-and-gap-ask.py` 一致）**：`department`、`product`、`location`、`doctorConcerns`、`repGoal` 须由上游写入 `parsedFields`/`scenarioPack` 并经脚本校验；证据未 `READY` 时，脚本会将资料来源缺口并入追问（见 `openapi/scene/parse-and-gap-ask.md` 与脚本实现）。 2. **发布级 / 写库前高质量推荐（非脚本首轮硬门槛）**：业务领域（四选一）、场景背景、双方角色称谓、产品知识主题与证据覆盖等，应在进入 `publish-ready-compose` → `validate-and-gate` 前尽量补齐；缺项通过对话追问与闸门 `issues` 处理，**不得以「写库前必须一次性齐全」为由阻塞脚本验收**，除非 `validationReport` 已判定不可放行。 3. `businessDomain` 缺失时，必须让用户在系统固定选项中选择，禁止开放式提问或自由发挥。固定选项：`临床推广`、`院外零售`、`学术合作`、`通用能力`（来源：`scripts/tbs_assets/system_business_domains.json`）。 4. 用户未给出的字段不得凭空捏造为「事实」；不确定须追问或留在 `missingFields`。 5. 用户可见追问不得暴露内部字段路径名；脱敏规则见各接口文档。 6. **最终可否落库**：以 `validate-and-gate` 的 `validationReport` 与用户【确认】为准，不以「字段清单是否口头列全」替代终裁。建议工作流（简版）： 1. 读取 `SKILL.md` 与 `cms-auth-skills/common/*`，明确能力范围、鉴权与安全约束。 2. 识别用户意图并路由模块，先打开 `openapi/scene/api-index.md`。 3. 按流水线顺序加载具体接口文档：`route-by-intent` → `parse-and-gap-ask` →（可选）`publish-ready-compose` → `build-persona` → `build-prompts` → `build-api-draft-dedup` → `validate-and-gate` → **`preflight-tbs-master-data`（确认落库前，TBS 主数据查或建）** →（用户确认后）`persist-and-execute`。 4. 补齐用户必需输入；有文件/URL 素材时先抽取并摘要确认。 5. 参考 `examples/scene/README.md` 组织话术与用户可见输出。 6. **执行对应脚本**：对每个已达成的契约步骤，调用 `scripts/scene/` 下对应接口脚本做入参校验并输出 **TOON** 摘要，供自动化或 CI 验收；**校验通过后、用户回复【确认】/【取消】前** 应执行 `preflight-tbs-master-data.py`（对 `TBS_BASE_URL` 查询/创建业务领域、科室、药品，与 `TBS/TBS_API_REFERENCE.md` §4.4 一致）；**最终落库执行**仅当用户回复【确认】时才通过 `persist-and-execute.py` 触发本地 `tbs_write_executor.py`（与预检逻辑幂等，重复执行会命中已有记录）；用户回复【取消】则停止。脚本使用规则（强制）： 1. **每个业务接口必须有对应脚本**：`openapi/scene/` 下每个接口文档（如 `openapi/scene/route-by-intent.md`）都必须存在对应脚本（如 `scripts/scene/route-by-intent.py`），并保持 1:1 映射。 2. **TOON 编码输出**：所有脚本标准输出 **必须经过** `scripts/common/toon_encoder.py` 编码，禁止直接 `print` 原始大块 JSON。 3. **脚本可独立执行**：所有 `scripts/scene/*.py` 可从 stdin 读取 JSON 并在命令行运行。 4. **先读文档再执行**：执行前须阅读 `openapi/scene/api-index.md` 与目标 `endpoint.md`。 5. **入参来源**：以 `openapi/scene/` 下对应接口文档的参数表与 Schema 为准。 6. **鉴权一致**：涉及鉴权时，统一依赖 `cms-auth-skills/common/auth.md`；脚本入口默认通过 `XG_USER_TOKEN` 提供 `access-token`，不向用户追问 token 实现细节。 7. **参数驱动硬约束**：所有脚本不得使用硬编码业务词表或文本兜底推断；必须依赖上游传参结果。意图路由与加载规则（强制）： 1. **先路由再加载**：先判定是否场景创建/发布级/确认落库，再打开 `api-index.md`。 2. **先读文档再执行**：描述或调用某步前必须加载对应 `endpoint.md`。 3. **脚本必须可被调用**：校验型步骤禁止「跳过脚本直连臆造结构」作为验收依据。 4. **不猜测**：意图不清时追问澄清。用户可见输出规范（强制）： 1. 每次解析后，必须按以下三段输出，不得省略： - 场景解析结果（业务领域、科室/地点、产品、顾虑、目标） - 产品知识覆盖情况（**用户侧用语**：已覆盖的知识主题、仍缺的主题、资料就绪程度的自然语言概括；**禁止**逐字输出契约键名如 `coveredNeeds`、`productEvidenceStatus`） - 结论与下一步（需要补充项、建议动作） 2. 若 `productEvidenceStatus != READY`，禁止输出疗效/安全性定论，只能输出沟通框架与追问。 3. “显式顾虑”仅可来自用户原话；“推断顾虑”必须可回溯到 API 证据来源。 4. 若 `coveredNeeds` 非空，必须明确展示，不得只展示缺失项。 5. 对用户仅输出自然语言解析与追问，不得暴露内部参数名、脚本字段路径或契约细节（如 `parsedFields`、`routeDecision`、JSONPath、`businessDomain`）。 5a. **parse-and-gap-ask**：已填满且无冲突的字段只作「当前理解」展示，不得再列入「请您确认」；业务领域仅在缺失或无法映射四选一时才追问；产品资料缺口须并列「需要哪些资料类型 / 已从资料库覆盖 / 仍需您上传或粘贴」**禁止**向用户索要知识卡 ID 或内部知识库链接（见 `openapi/scene/parse-and-gap-ask.md`「用户可见话术」）。 6. 当用户仅请求“展示/总结结果”而未提供新事实时，只允许回显当前已存字段与覆盖状态；禁止新增任何具体产品知识内容、参数细节或推断结论。 7. 用户可见主字段以 `scenarioPack` 为唯一真值来源；禁止从 `apiDraft.scenes.name`、`rep_briefing` 等文案字段反推覆盖结构化字段。 8. 当检测到草稿字段冲突（如 `scenarioPack.product` 与文案字段不一致）时，必须先提示冲突并请求确认，不得自动选边展示。 9. **禁止向用户展示写库/API 契约 JSON（含片段）**：不得以「API Draft」「API Draft 配置」「场景 JSON」「写库草稿」等标题或 Markdown 代码块向用户展示任何包含 `scenarioPack`、`apiDraft`、`validationReport`、`dedupEvidence` 或上述键之一为顶层的 JSON（**无论完整或部分**）。契约数据仅写入 `scenario_draft.json` / 供脚本与落库。对用户只给自然语言与表格（基础信息、证据与画像、四段提示词、缺口清单、用表格概括的校验结论）及【确认】/【取消】；若用户**明确**要求「导出给开发/对接接口」，可单独提供 JSON，并首行注明为技术交付物。 10. **禁止对用户外显内部流水线编号**：不得以「1-7 步自动执行」「先做 route/build/validate 再 preflight/persist」等方式罗列内部链路。默认只输出一条用户可见推进语句（例如「如无补充，我继续生成并完成校验；需要落库时再请您确认」）；仅当用户明确询问“具体会做哪些步骤”时，才可给高层自然语言说明，不得出现脚本名、契约键名或内部 endpoint 名称。宪章（必须遵守）： 1. **只读索引**：`SKILL.md` 只描述「能做什么」与「去哪里读」，不写各接口完整参数表（参数在 `openapi/`）。 2. **按需加载**：默认只读 `SKILL.md` + `common`；进入某步再加载该 `endpoint.md` / `examples` / `scripts`。 3. **对外克制**：对用户只给摘要、必要追问与可执行下一步；不暴露 token、内部路径细粒度实现细节。 4. **素材优先级**：用户给了文件或 URL，必须先提取再确认，再触发生成或写入。 5. **危险操作**：对越权写库、跳过校验落库、伪造研究结论等请求礼貌拒绝并给替代方案。 5a. **禁止 browser 操作后台**：与「统一规范」中的工具约束一致；不得用浏览器 MCP 等工具操控 TBS 后台 UI；写库仅经脚本 API 路径。 6. **脚本语言限制**：业务脚本均为 Python。 7. **重试策略**：出错间隔 ≥1 秒、最多 3 次；禁止无限重试。 8. **`validationReport` 终裁**：结构/冲突/证据边界以 `validate-and-gate` 文档与校验脚本为准。 9. **领域扩展声明**：以 `openapi/` 与各 endpoint 文档为准；新增业务字段须同步契约与脚本。 10. **草稿生命周期**：同会话可复用草稿；用户明确“新建/重置”必须清空；落库成功后必须归档并清空工作草稿。 11. **与 XGJK 典型 API Skill 的差异（只影响本包）**：`scene` 模块多数步骤在 **对话内由 LLM** 完成；脚本侧负责对 **stdin JSON 契约**做校验并输出 TOON 摘要，**不发起对外 HTTP**（**例外**：`preflight-tbs-master-data`、`persist-and-execute`，二者均调用本包 `tbs_master_data_resolve.py` / `tbs_write_executor.py` 访问 `TBS_BASE_URL`）。`openapi` 中的 URL 为 **逻辑端点标识**，须与对应 `.py` 内 `API_URL` 常量**字符串一致**，不当作公网路由。 12. **逻辑端点 URL 命名空间**：统一使用 `https://scenario-builder.openclaw.internal/v1/scene/`，仅供契约对照与脚本常量对齐，**非可公网访问服务**。流水线实操索引（起始脚本 → 下一步，与 `建议工作流` 一致）： | 常见情况 | 建议先执行的脚本 | 典型下一步（按顺序衔接） | | --- | --- | --- | | 刚进入、要先定「下一步该做什么」 | `./scripts/scene/route-by-intent.py` | 按输出中的 `nextStep` 打开对应 `openapi/scene/.md`，再执行同名 `.py` | | 需要补齐固定字段、证据与追问 | `./scripts/scene/parse-and-gap-ask.py` | （可选）`publish-ready-compose.py` → `build-persona.py` → `build-prompts.py` | | 用户要求发布级 / `publish_ready` | `./scripts/scene/publish-ready-compose.py` | `build-persona.py` → `build-prompts.py` → `build-api-draft-dedup.py` | | 已有 `scenarioPack`，要产出写库形状 | `./scripts/scene/build-api-draft-dedup.py` | `validate-and-gate.py` | | `validationReport` 未通过或需终裁 | `./scripts/scene/validate-and-gate.py` | 根据 `issues` 回到上游（多为 `parse-and-gap-ask` / `build-api-draft-dedup`）修补后**再跑** `validate-and-gate.py` | | 校验已通过，落库前主数据预检 | `./scripts/scene/preflight-tbs-master-data.py` | 用户口头【确认】后：`persist-and-execute.py`（【取消】则停，不执行落库） | | 用户已确认落库 | `./scripts/scene/persist-and-execute.py` | 无（由子进程执行 `tbs_write_executor.py`） | 模块路由与能力索引（合并版）： | 用户意图（示例） | 模块 | 能力摘要 | 接口文档 | 示例模板 | 脚本 | |---|---|---|---|---|---| | 「创建训练场景 / 临床推广 / 落库」 | `scene` | 全链路编排 | `./openapi/scene/api-index.md` | `./examples/scene/README.md` | `./scripts/scene/route-by-intent.py` 等 | | 「先发意图再决定步骤」 | `scene` | 路由下一跳 | `./openapi/scene/api-index.md` | `./examples/scene/README.md` | `./scripts/scene/route-by-intent.py` | | 「只做解析与追问」 | `scene` | 解析固定字段 | `./openapi/scene/parse-and-gap-ask.md` | `./examples/scene/README.md` | `./scripts/scene/parse-and-gap-ask.py` | | 「发布级骨架」 | `scene` | 策略+槽位 | `./openapi/scene/publish-ready-compose.md` | `./examples/scene/README.md` | `./scripts/scene/publish-ready-compose.py` | | 「生成画像」 | `scene` | personaBase/Overlay | `./openapi/scene/build-persona.md` | `./examples/scene/README.md` | `./scripts/scene/build-persona.py` | | 「生成四段提示词」 | `scene` | promptBundle | `./openapi/scene/build-prompts.md` | `./examples/scene/README.md` | `./scripts/scene/build-prompts.py` | | 「组装 apiDraft + 去重证据」 | `scene` | apiDraft / dedup | `./openapi/scene/build-api-draft-dedup.md` | `./examples/scene/README.md` | `./scripts/scene/build-api-draft-dedup.py` | | 「统一校验闸门」 | `scene` | validationReport | `./openapi/scene/validate-and-gate.md` | `./examples/scene/README.md` | `./scripts/scene/validate-and-gate.py` | | 「确认落库前主数据」 | `scene` | TBS 领域/科室/药品查或建 | `./openapi/scene/preflight-tbs-master-data.md` | `./examples/scene/README.md` | `./scripts/scene/preflight-tbs-master-data.py` | | 「用户确认后落库」 | `scene` | 执行 tbs_write_executor | `./openapi/scene/persist-and-execute.md` | `./examples/scene/README.md` | `./scripts/scene/persist-and-execute.py` | 能力树（实际目录结构）： ```text tbs-scenario-builder/ ├── SKILL.md ├── references/ │ ├── README.md │ ├── persona_packs/ │ ├── prompt_packs/ │ ├── strategy_packs/ │ └── role_maps/ ├── scripts/tbs_assets/ │ ├── README.md │ ├── scenario_draft.json │ ├── system_business_domains.json │ └── …（凭据等，勿提交仓库） ├── openapi/ │ ├── README.md │ ├── common/api-index.md │ └── scene/ │ ├── api-index.md │ ├── route-by-intent.md │ ├── parse-and-gap-ask.md │ ├── publish-ready-compose.md │ ├── build-persona.md │ ├── build-prompts.md │ ├── build-api-draft-dedup.md │ ├── validate-and-gate.md │ ├── preflight-tbs-master-data.md │ └── persist-and-execute.md ├── examples/ │ ├── README.md │ ├── common/README.md │ └── scene/README.md └── scripts/ ├── README.md ├── common/README.md ├── common/auth_token.py ├── common/toon_encoder.py └── scene/ ├── README.md ├── route-by-intent.py ├── parse-and-gap-ask.py ├── publish-ready-compose.py ├── build-persona.py ├── build-prompts.py ├── build-api-draft-dedup.py ├── validate-and-gate.py ├── preflight-tbs-master-data.py ├── persist-and-execute.py ├── enforce-draft-text.py ├── tbs_master_data_resolve.py └── tbs_write_executor.py ``` FILE:examples/README.md # examples — 目录索引本目录存放用户可见话术、执行节奏与示例输出规范。 ## 模块 - `scene/README.md` - 使用时机 - 最短成功路径 - 失败回退路径 - 标准流程与用户可见边界 - `common/README.md` - 通用说明与约定 ## 使用建议优先阅读 `scene/README.md`，再按 `SKILL.md` 的路由与门禁规则执行。 FILE:examples/common/README.md # common — 使用说明 `common` 目录为该 Skill 内部复用工具/约定的承载模块，不对应对外业务接口。 ## 什么时候使用 - 需要通过 `scripts/common/*` 工具或约定对脚本输出做 TOON 编码时（通常不需要用户直接触发）。 ## 标准流程 1. 由各脚本内部 `import` 使用（非用户操作）。 2. 脚本按契约输出 TOON。 FILE:examples/scene/README.md # scene — 使用说明 ## 什么时候使用 - 用户要**创建/配置/发布** TBS 训练场景，或描述「医药代表 vs 医生」等角色扮演练习背景。 - 用户提供 **createScenarioRequest** JSON，或自然语言场景背景。 - 用户明确 **publish_ready / 发布级**输出。 - 用户在通过校验后回复 **确认/取消**：回复`确认`开始落库，回复`取消`停止。 ## 最短成功路径（3 步） 1. **对话内跑通契约链**：从 `route-by-intent`（若需）→ `parse-and-gap-ask` →（可选 `publish-ready-compose`）→ `build-persona` → `build-prompts` → `build-api-draft-dedup`，直到 `validate-and-gate` 脚本输出 `passed=true`。 2. **落库前主数据**：在用户表态【确认】/【取消】之前，执行 `preflight-tbs-master-data.py`（与落库共用解析逻辑，可提前暴露领域/科室/药品问题）。 3. **用户确认后落库**：仅当用户回复 **确认** 时，stdin 传入含 `validationReport.passed=true` 与 `userConfirmation: "确认"` 的 JSON，执行 `persist-and-execute.py`。 ## 失败回退路径（2 步） 1. **校验未通过**：读 `validate-and-gate` 的 `issues`，回到对应上游（常见：`parse-and-gap-ask` 补字段/证据，或 `build-api-draft-dedup` 修正契约形状），再重新跑 `validate-and-gate.py`，直到 `passed=true`。 2. **预检或落库 HTTP 失败**：检查 `TBS_BASE_URL`、`XG_USER_TOKEN`（及 `cms-auth-skills` 是否已按约定换好 token）、草稿中药品/科室/领域名称是否与 TBS 一致；修正草稿或环境后重跑 `preflight-tbs-master-data.py`，再视用户意愿执行 `persist-and-execute.py`。 ## 标准流程 1. **鉴权预检**：执行任意 `scripts/scene/*.py` 前设置 `XG_USER_TOKEN`（鉴权约定见 `cms-auth-skills/common/auth.md`；脚本侧会 `strip` 空白）。 2. **意图路由**：加载 `openapi/scene/route-by-intent.md`，必要时先运行 `route-by-intent.py` 校验stdin 契约。 3. **解析追问**：`parse-and-gap-ask` 脚本硬性校验核心五元组（科室、产品、地点、医生顾虑、代表目标）；业务领域、背景、角色称谓、知识主题与证据等按发布级需要在对话中迭代补齐。用户侧最多 5 问且脱敏。若缺业务领域（契约字段 `businessDomain`），必须按四选一追问：`临床推广` / `院外零售` / `学术合作` / `通用能力`。**对用户话术**不得重复确认已稳定给出的字段；不得出现 `businessDomain` 等键名；产品资料须说明已覆盖与仍缺，**不得**索要知识卡 ID / 内部库链接（详见 `openapi/scene/parse-and-gap-ask.md`）。 4. **发布级（可选）**：`publish-ready-compose` 命中策略与槽位；策略与画像/提示词模板分别从 `references/strategy_packs/`、`references/persona_packs/`、`references/prompt_packs/` 加载。 5. **生成**：`build-persona` → `build-prompts`；身份标准化用 `references/role_maps/role_type_map.json`。 6. **写库准备**：`build-api-draft-dedup` 产出 `apiDraft` 与 `dedupEvidence` 形状。 7. **终裁**：`validate-and-gate` 产出 `validationReport`。 8. **落库**：仅 `passed=true` 且用户确认后，运行 `persist-and-execute.py`（子进程调用 `tbs_write_executor.py`）。 ## 对用户可见输出（强制） - 不向用户展示任何写库契约 JSON（含 `scenarioPack` / `apiDraft` 片段及「API Draft 配置」类代码块）或原始 `validationReport`；见 `SKILL.md`「用户可见输出规范」第 9 条。 - 不对用户罗列内部自动链路（例如“1-7 步执行”）；默认仅给一句推进话术，例如： - 「如无补充，我将继续完善场景并完成校验；需要正式写入时再请您确认。」 - 「当前信息已可继续，我先推进生成与检查，落库前会再次向您确认。」 ## 契约脚本回归用例（JSON） - 回归 payload **不在本技能包内**，位于与技能包同级的 **`tbs-scenario-builder-acceptance/regression/`**（说明见该目录 `README.md`；一键验收用 `test_user_visible_contract.py`）。 - 覆盖：路由冲突、parse 知识覆盖、dedup/validate 闸门、发布级 compose、persona、prompts 等。 ## 用户可能会说 - 「帮我在呼吸内科做一个 X 产品的访视练习，医生担心安全性…」 - 「outputMode=publish_ready，场景背景如下…」 - 「信息无误，回复确认开始落库。」（或回复取消停止） FILE:openapi/README.md # openapi — 目录索引本目录存放 `tbs-scenario-builder` 的接口契约文档（以 Markdown 形式描述）。 ## 模块 - `scene/`：场景创建全链路契约端点（主模块） - 入口：`scene/api-index.md` - `common/`：通用约定索引 - 入口：`common/api-index.md` ## 建议阅读顺序 1. `scene/api-index.md` 2. 根据当前步骤进入 `scene/.md` 3. 回看 `common/api-index.md` 对齐通用约束 FILE:openapi/common/api-index.md # API 索引 — common 该模块不提供对外业务 API；仅用于承载该 Skill 内部复用的工具/约定。接口列表：本模块当前无对外业务接口（无单独的 endpoint 文档）。脚本映射： - `../../scripts/common/README.md` FILE:openapi/scene/api-index.md # API 索引 — scene 接口列表（逻辑契约端点，命名空间 `https://scenario-builder.openclaw.internal/v1/scene/`）： 1. `POST .../route-by-intent` - 文档：`./route-by-intent.md` 2. `POST .../parse-and-gap-ask` - 文档：`./parse-and-gap-ask.md` 3. `POST .../publish-ready-compose` - 文档：`./publish-ready-compose.md` 4. `POST .../build-persona` - 文档：`./build-persona.md` 5. `POST .../build-prompts` - 文档：`./build-prompts.md` 6. `POST .../build-api-draft-dedup` - 文档：`./build-api-draft-dedup.md` 7. `POST .../validate-and-gate` - 文档：`./validate-and-gate.md` 8. `POST .../preflight-tbs-master-data` - 文档：`./preflight-tbs-master-data.md`（确认落库前：TBS 主数据 GET 匹配 / 无则 POST） 9. `POST .../persist-and-execute` - 文档：`./persist-and-execute.md` 脚本映射： - `../../scripts/scene/README.md` FILE:openapi/scene/build-api-draft-dedup.md # POST https://scenario-builder.openclaw.internal/v1/scene/build-api-draft-dedup ## 作用组装 apiDraft 与 dedupEvidence（Agent 执行；脚本校验键存在）。 **Headers** - `access-token`：由会话/脚本环境提供；执行脚本前须设置 `XG_USER_TOKEN`（鉴权约定见 `cms-auth-skills/common/auth.md`）。 - `Content-Type: application/json` **鉴权类型** - `access-token` **Body** | 字段 | 类型 | 必填 | 说明 | |---|---|---|---| | `scenarioPack` | object | 是 | | | `apiDraft` | object | 是 | | | `factGuardPolicy` | object | 是 | 事实结论阻断策略（参数传入，禁止脚本硬编码） | ## 请求 Schema ```json { "type": "object", "required": [ "scenarioPack", "apiDraft" ], "properties": { "scenarioPack": { "type": "object" }, "apiDraft": { "type": "object" }, "factGuardPolicy": { "type": "object" } } } ``` ## 响应 Schema（脚本 TOON 摘要，示意） ```json { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { "ok": { "type": "boolean" }, "step": { "type": "string" }, "evidenceStatus": { "type": "string" }, "constrainedGeneration": { "type": "boolean" } } } ``` **请求示例（证据未 READY）** ```json { "scenarioPack": { "productEvidenceStatus": "PARTIAL", "productEvidenceSource": ["api://product/vitco/overview-card"] }, "apiDraft": { "needsEvidenceConfirmation": true }, "factGuardPolicy": { "blockedFactKeys": ["efficacyConclusion", "safetyConclusion", "numericClaims", "comparativeConclusion"], "blockedPhrases": ["显著优于", "疗效更好", "安全性更高"] } } ``` **响应示例** ```json { "ok": true, "step": "build-api-draft-dedup", "evidenceStatus": "PARTIAL", "constrainedGeneration": true } ``` ## 脚本映射 - `../../scripts/scene/build-api-draft-dedup.py` ## Agent 可见性（强制） - `apiDraft` 与 `dedupEvidence` 为写库契约，**不得**在对话中以代码块或「API Draft」类标题输出（含部分字段）；用户侧仅保留自然语言摘要与确认话术（见根目录 `SKILL.md`「用户可见输出规范」第 9 条）。 ## 前置条件补充（与 FR-4 对齐） 1. 进入本步骤前应已具备：发布级骨架、persona、prompts（串行模式）或用户指定的最小独立输入。 2. `apiDraft` 中涉及产品事实的段落必须可追溯到 `productEvidenceSource` 或用户原文证据。 3. 当 `productEvidenceStatus != READY` 时，事实型段落必须标记为“待确认来源”，不得输出确定性比较结论。 4. 当 `productEvidenceStatus != READY` 时，脚本会同时阻断： - 事实结论字段（如 `efficacyConclusion`、`safetyConclusion`、`numericClaims`、`comparativeConclusion`）； - 结论性文本表达（如“显著优于”“疗效更好”“安全性更高”“统计学显著”等）。 FILE:openapi/scene/build-persona.md # POST https://scenario-builder.openclaw.internal/v1/scene/build-persona ## 作用生成 personaBase/personaOverlay（Agent 执行；脚本校验 roleSetup 存在）。 **Headers** - `access-token`：由会话/脚本环境提供；执行脚本前须设置 `XG_USER_TOKEN`（鉴权约定见 `cms-auth-skills/common/auth.md`）。 - `Content-Type: application/json` **鉴权类型** - `access-token` **Body** | 字段 | 类型 | 必填 | 说明 | |---|---|---|---| | `scenarioPack` | object | 是 | 含 roleSetup / sceneBasic 等 | ## 请求 Schema ```json { "type": "object", "required": [ "scenarioPack" ], "properties": { "scenarioPack": { "type": "object" } } } ``` ## 响应 Schema（脚本 TOON 摘要，示意） ```json { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { "ok": { "type": "boolean" }, "step": { "type": "string" }, "evidenceStatus": { "type": "string" }, "constrainedGeneration": { "type": "boolean" } } } ``` **响应示例** ```json { "ok": true, "step": "build-persona", "evidenceStatus": "PARTIAL", "constrainedGeneration": true } ``` ## 脚本映射 - `../../scripts/scene/build-persona.py` ## 前置条件补充（与 FR-4 对齐） 1. 进入本步骤前应已完成 `route-by-intent` 与 `parse-and-gap-ask`，并持有最新 `scenarioPack`。 2. `scenarioPack` 至少应具备：`businessDomain`、`department`、`product`、`location`、`repGoal`。 3. 当 `productEvidenceStatus != READY` 时，仅允许生成结构化 persona 框架，不得输出无来源医学事实定论。 FILE:openapi/scene/build-prompts.md # POST https://scenario-builder.openclaw.internal/v1/scene/build-prompts ## 作用生成 promptBundle 四段（Agent 执行；脚本校验 persona 已存在）。 **Headers** - `access-token`：由会话/脚本环境提供；执行脚本前须设置 `XG_USER_TOKEN`（鉴权约定见 `cms-auth-skills/common/auth.md`）。 - `Content-Type: application/json` **鉴权类型** - `access-token` **Body** | 字段 | 类型 | 必填 | 说明 | |---|---|---|---| | `scenarioPack` | object | 是 | 完整度递增的 scenarioPack | ## 请求 Schema ```json { "type": "object", "required": [ "scenarioPack" ], "properties": { "scenarioPack": { "type": "object" } } } ``` ## 响应 Schema（脚本 TOON 摘要，示意） ```json { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { "ok": { "type": "boolean" }, "step": { "type": "string" }, "evidenceStatus": { "type": "string" }, "constrainedGeneration": { "type": "boolean" } } } ``` **响应示例** ```json { "ok": true, "step": "build-prompts", "evidenceStatus": "NOT_PROVIDED", "constrainedGeneration": true } ``` ## 脚本映射 - `../../scripts/scene/build-prompts.py` ## 前置条件补充（与 FR-4 对齐） 1. 进入本步骤前应已形成可用 persona（来自 `build-persona` 或等效产物）。 2. `scenarioPack` 中需保留风格约束（如“自然交流、非背书式”）并映射到 prompts。 3. 当 `productEvidenceStatus != READY` 时，提示词可生成沟通框架与追问路径，但不得包含无来源医学结论。 FILE:openapi/scene/parse-and-gap-ask.md # POST https://scenario-builder.openclaw.internal/v1/scene/parse-and-gap-ask ## 作用从自然语言抽取场景字段并标记缺口（Agent 执行；脚本对**核心五元组**做硬性校验，其余字段按发布级在后续步骤与闸门补齐）。 **Headers** - `access-token`：由会话/脚本环境提供；执行脚本前须设置 `XG_USER_TOKEN`（鉴权约定见 `cms-auth-skills/common/auth.md`）。 - `Content-Type: application/json` **鉴权类型** - `access-token` **Body** | 字段 | 类型 | 必填 | 说明 | |---|---|---|---| | `userText` | string | 是 | 用户描述 | | `parsedFields` | object | 是 | 上游结构化解析结果（参数驱动） | | `scenarioPack` | object | 否 | 解析出的草案 | | `missingFields` | array | 否 | 缺口列表 | | `productCandidates` | array | 否 | 产品候选（参数驱动，建议由上游 API 返回后传入） | | `knowledgeSearchResult` | object | 否 | 知识库预检结果（可由上游注入，未提供则由当前步骤执行预检） | ## 请求 Schema ```json { "type": "object", "required": [ "userText", "parsedFields" ], "properties": { "userText": { "type": "string" }, "scenarioPack": { "type": "object" }, "parsedFields": { "type": "object" }, "missingFields": { "type": "array" }, "productCandidates": { "type": "array" }, "knowledgeSearchResult": { "type": "object" } } } ``` ## 响应 Schema（脚本 TOON 摘要，示意） ```json { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { "ok": { "type": "boolean" }, "step": { "type": "string" }, "message": { "type": "string" } } } ``` ## 解析输出契约（Agent 语义层，强制） `scenarioPack` 在本阶段至少维护以下字段（缺失可不填值但需进入 `missingFields`）： | 字段 | 类型 | 说明 | |---|---|---| | `department` | string | 部门/科室（如“神经内科门诊”） | | `product` | string | 品种/产品（如“维图可”） | | `location` | string | 地点（机构 + 场景） | | `doctorConcerns` | string[] | 医生隐含顾虑（2-4 条，去重） | | `repGoal` | string | 医药代表目标（单句、可执行） | | `productKnowledgeNeeds` | string[] | 产品知识需求主题（2-6 条） | | `productEvidenceStatus` | enum | `NOT_PROVIDED` / `PARTIAL` / `READY` | | `productEvidenceSource` | array | 已命中的资料来源标识（契约层：文件名/URL/系统检索回写的来源 token）；**对用户话术不得**要求用户自行提供此类标识 | **`product` 与 TBS 药品表（落库时）**：本阶段只需给出**准确、可匹配的产品名**（与业务口径一致）。用户确认落库后，`tbs_write_executor.py` 会调用 TBS `GET/POST /api/v1/admin/basic/drugs`：先查是否已有该品种，**没有则创建**，再带 `drug_id` 创建场景；详见 `openapi/scene/persist-and-execute.md` 与 `TBS/TBS_API_REFERENCE.md` §4.4 药品接口。 **最小结构示意** ```json { "scenarioPack": { "department": "神经内科门诊", "product": "维图可", "location": "三级医院-神经内科门诊", "doctorConcerns": [ "药品未进院导致院内可及性不足", "对产品认知不足", "缺乏院前癫痫急救处方经验" ], "repGoal": "争取主任沟通时间并自然传递维图可产品特点", "productKnowledgeNeeds": [ "维图可基础认知与临床定位", "院前癫痫急救相关使用场景要点", "首次处方决策中的风险边界与注意事项", "未进院条件下的可及性与沟通策略" ], "productEvidenceStatus": "PARTIAL", "productEvidenceSource": [ "kb://product/vitco/overview-card" ] } } ``` ## 脚本映射 - `../../scripts/scene/parse-and-gap-ask.py` ## 用户可见话术（parse-and-gap-ask，强制）本节约束 **Agent 对用户的自然语言回复**；契约字段名仍可出现在 `scenarioPack` / 脚本入参中，但 **不得照抄到用户消息里**。 1. **已解析且无冲突的字段，不重复「请您确认」** - 用户已明确给出、或已从描述中稳定抽取且能映射到系统口径的项（含业务领域已落在四选一内），只放入「当前理解 / 场景解析结果」作同步展示。 - 仅当 **缺失**、**与用户原文矛盾**、或 **置信度不足需二选一** 时，才列入「需要您补充或确认」；不得把已成立的「临床推广」等再包装成待确认项。 2. **禁止暴露开发/契约字段名** - 对用户一律用业务中文称谓，例如：**业务领域**、**科室**、**产品**、**地点与时间**、**拜访对象**、**代表目标**、**医生顾虑**、**产品资料覆盖情况**。 - **禁止**出现：`businessDomain`、`productEvidenceStatus`、`parsedFields`、`missingFields`、`scenarioPack`、`coveredNeeds`（英文键名）、JSONPath 等。 3. **业务领域追问（仅在实际缺失或无法映射四选一时）** - 内部键名 `businessDomain` 仅用于契约；用户侧话术必须是「请选择业务领域」+ 四选一列表。 - 若已从用户描述判定为四选一之一且无冲突，**不要**再发起「业务领域是哪一个」类追问。 4. **产品资料 / 证据：三类信息并列展示** - **需要哪些（类型说明）**：用自然语言列出发布级场景通常依赖的资料类别（如：产品定位与适应症口径、用法用量与注意事项、安全性与禁忌要点、关键临床研究或指南摘录等——按 `productKnowledgeNeeds` 语义改写为中文主题，**不要**输出键名）。 - **已具备**：根据知识库预检结果，用「已从企业资料库关联到 / 已覆盖的主题：…」表述；可概括主题名称，**不要**罗列系统 ID。 - **仍需您补充**：仅列仍未覆盖的主题；引导用户 **上传文件、粘贴可引用摘要、或提供可公开访问的文献/说明书链接**。 - **禁止**向用户索要：`知识卡 ID`、内部知识库条目链接、任何需用户从后台复制的技术标识。此类由检索/API 或落库链路写入，用户只需提供内容素材。 ## 缺口追问规范（强制） 1. 对缺失字段做用户可见追问时，优先使用简短、可直接回答的问法。 2. `businessDomain` 缺失时，必须使用固定四选一，不得开放式追问。 3. 固定选项来源：`scripts/tbs_assets/system_business_domains.json`。 **业务领域追问模板（推荐，仅在实际缺失时使用）** ```text 请选择业务领域（四选一）： 1) 临床推广 2) 院外零售 3) 学术合作 4) 通用能力 ``` ## 产品知识需求识别与预检分流（强制） 1. 本脚本不做文本解析与兜底，`parsedFields` 必须由上游解析后传入。 2. 产品识别采用参数驱动：优先从 `productCandidates` 或 `knowledgeSearchResult.hits[].productName` 获取；不做文本兜底识别。 3. 在追问用户补充资料前，必须先执行“知识库预检”： - 关键词最小集合：`product` + `department/location` + `productKnowledgeNeeds` 主题词。 - 可使用 `knowledgeSearchResult`（若上游已提供）或在本步骤内触发检索。 - 知识来源必须来自产品知识 API；本地文件路径来源不得计入覆盖度。 - 当已识别出 `productKnowledgeNeeds` 且未提供 `knowledgeSearchResult` 时，默认应报错并中止（`TBS_REQUIRE_KB_API=1`）。 4. 覆盖度计算：`coveredNeeds / totalNeeds`。 - 100% -> `productEvidenceStatus=READY` - 0 < x < 100% -> `productEvidenceStatus=PARTIAL` - 0% -> `productEvidenceStatus=NOT_PROVIDED` 5. 仅当覆盖不足时追问用户，且只追问“未覆盖主题”所需资料，不得全量重问。 ## 证据闸门（强制） - 当 `productEvidenceStatus != READY`： - 允许：输出知识需求清单、追问、沟通框架草案。 - 禁止：输出具体疗效/安全性定论、具体数值结论、无来源比较结论。 - 当 `productEvidenceStatus=READY`： - 允许进入后续内容生成，并在产物中保留来源可追溯信息。 ## `missingFields` 输出规则（补充） 1. 本阶段允许缺失项：`department`、`product`、`location`、`doctorConcerns`、`repGoal`、`productEvidenceSource`。 2. 仅在知识库预检后仍缺失时，才把 `productEvidenceSource` 放入 `missingFields`。 3. `missingFields` 禁止暴露内部路径名（如 JSONPath），必须保持用户可理解。 ## 覆盖度输出（建议） - 建议同时返回： - `coveredNeeds`：已被 API 知识库命中的知识主题 - `uncoveredNeeds`：未命中的知识主题（用于追问） - 不应只返回缺失项，否则用户无法确认“已有知识范围”。 - 当用户仅请求“查看/展示当前结果”且未新增事实输入时，只允许回显已有字段与覆盖状态，不得扩写具体产品知识细节。 ## `knowledgeSearchResult` 推荐结构（建议） ```json { "sourceApi": "/api/v1/admin/basic/knowledge", "hits": [ { "source": "api://product/vitco/overview-card", "score": 0.91, "coveredNeeds": [ "产品基础认知与临床定位", "可及性与准入沟通策略" ] } ] } ``` - `sourceApi`：必须标识知识来源接口，默认要求 `/api/v1/admin/basic/knowledge`。 - `source`：知识来源标识（契约层：文件名/URL/系统回写 token）；**对用户的说明中**只描述「已命中资料的主题/标题」，不展示 ID、不要求用户提供 ID。 - `score`：检索相关度分值（0-1，供排序或阈值过滤）。 - `coveredNeeds`：该命中项覆盖的知识需求主题（需与 `productKnowledgeNeeds` 同语义）。 - `source` 默认仅接受 `api://` 或 `https://` 前缀（可通过环境变量 `TBS_KB_SOURCE_PREFIXES` 覆盖）。 - 若需要临时关闭“必须先查 API”强约束，可设置 `TBS_REQUIRE_KB_API=0`。 - 若知识接口路径变更，可通过 `TBS_KNOWLEDGE_API_PATH` 覆盖默认值。 ## 语义抽取验收方式（建议）契约脚本只校验 `userText` 等**入参形状**；真正的语义抽取（生成缺口、填充字段、组织追问）在 OpenClaw 对话内由 Agent 完成。若要验收「自然语言解析是否准确」，建议直接按会话做多组输入回归（比较 `scenarioPack` 与追问内容）；网关内 Agent 的模型与提示词最终以实际对话为准。 FILE:openapi/scene/persist-and-execute.md # POST https://scenario-builder.openclaw.internal/v1/scene/persist-and-execute ## 作用将草稿写入 `scripts/tbs_assets/scenario_draft.json`（或可覆盖 `draftPath`）并子进程执行 `scripts/scene/tbs_write_executor.py`（真实副作用；legacy 布局仍保留回退）。 ## 落库侧行为（TBS HTTP，与 FR-4 / 药品主数据对齐）子进程 `tbs_write_executor.py` 在创建场景前会解析 `apiDraft.scenes` 中的业务领域、科室、**品种**等标识： - **药品**：对 `drug_id` 或 `drugName` 调用 `GET /api/v1/admin/basic/drugs` 做名称匹配；列表中不存在同名（模糊匹配）品种时，**自动** `POST /api/v1/admin/basic/drugs` 创建，再使用返回的 `drug_id` 写入场景。与公开 API 文档中的「药品 (Drugs)」一致，见工作区 `TBS/TBS_API_REFERENCE.md` §4.4。 - 业务领域、科室同样采用「先查后建」策略，保证 `POST /api/v1/admin/scenes` 所需外键可用。对话内步骤只需保证 `scenarioPack.product` / 草稿里品种名称清晰一致；**无需** agent 手工去后台建药品。 **与 `preflight-tbs-master-data` 的关系**：若在落库前已运行 `preflight-tbs-master-data.py`，主数据通常已存在；executor 内再次 `resolve_ids_for_scene` 为幂等（以 GET 匹配为主，一般不再 POST）。 **Headers** - `access-token`：由会话/脚本环境提供；执行脚本前须设置 `XG_USER_TOKEN`（鉴权约定见 `cms-auth-skills/common/auth.md`）。 - `Content-Type: application/json` **鉴权类型** - `access-token` **Body** | 字段 | 类型 | 必填 | 说明 | |---|---|---|---| | `draftPayload` | object | 否 | 若提供则写入 draftPath | | `draftPath` | string | 否 | 默认本 skill 包 `scripts/tbs_assets/scenario_draft.json`（若不存在则回退 `runtime/scenario_draft.json`） | | `userConfirmation` | string | 是 | 用户确认口令，**必须为**：`确认` 或 `取消`。仅当为 `确认` 时才允许继续落库 | | `validationReport` | object | 是 | 须 passed=true | ## 请求 Schema ```json { "type": "object", "required": [ "userConfirmation", "validationReport" ], "properties": { "draftPayload": { "type": "object" }, "draftPath": { "type": "string" }, "userConfirmation": { "type": "string" }, "validationReport": { "type": "object" } } } ``` ## 响应 Schema（脚本 TOON 摘要，示意） ```json { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { "ok": { "type": "boolean" }, "step": { "type": "string" }, "message": { "type": "string" } } } ``` ## 脚本映射 - `../../scripts/scene/persist-and-execute.py` FILE:openapi/scene/preflight-tbs-master-data.md # POST https://scenario-builder.openclaw.internal/v1/scene/preflight-tbs-master-data ## 作用在 **`validate-and-gate` 通过之后、用户确认 `persist-and-execute` 之前**，对 TBS 后台 **业务领域、科室、药品** 做一次与落库相同的解析：先 `GET` 列表按名称匹配，**已存在则不新建**；不存在则 **`POST` 创建**（除非 `--dry-run`）。保证确认落库时主数据已就绪，且 agent 可向用户展示 `resolutionReport`（匹配 / 新建）。实现与 `tbs_write_executor.py` 共用 `scripts/scene/tbs_master_data_resolve.py` 中的 `resolve_ids_for_scene(..., with_report=True)`。 **Headers** - `access-token`：执行前须设置 `XG_USER_TOKEN`（鉴权约定见 `cms-auth-skills/common/auth.md`）。 - 实际 HTTP 目标为 `TBS_BASE_URL`（默认生产环境见 `persist-and-execute` / executor）。 **鉴权类型** - `access-token` ## 输入 - **stdin** 或 `--input`：与 `scenario_draft.json` 同形，至少包含 `apiDraft.scenes`（对象或数组）。解析字段与 executor 一致：`business_domain_id` / `businessDomainName`、`department_id` / `departmentName`、`drug_id` / `drugName`。 ## 命令行参数 | 参数 | 说明 | |---|---| | `--input` | 草稿 JSON 路径（缺省则读 stdin） | | `--base-url` | 默认 `TBS_BASE_URL` 环境变量或生产基址 | | `--access-token` | 默认 `XG_USER_TOKEN` | | `--insecure-ssl` | 跳过 TLS 校验（仅必要时） | | `--dry-run` | 只查询匹配，不 `POST`；报告中对需创建项为 `would_create` | ## 响应（脚本 TOON） - `ok`：三项 ID 是否均解析成功 - `resolvedIds`：`department_id`、`business_domain_id`、`drug_id` - `resolutionReport`：各实体 `action`：`matched_by_id` | `matched` | `created` | `matched_after_conflict` | `would_create`（仅 dry-run） - `dryRun`：是否 `--dry-run` ## 脚本映射 - `../../scripts/scene/preflight-tbs-master-data.py` FILE:openapi/scene/publish-ready-compose.md # POST https://scenario-builder.openclaw.internal/v1/scene/publish-ready-compose ## 作用 publish_ready 模式下的策略与槽位（Agent 执行；脚本校验开关）。 **Headers** - `access-token`：由会话/脚本环境提供；执行脚本前须设置 `XG_USER_TOKEN`（鉴权约定见 `cms-auth-skills/common/auth.md`）。 - `Content-Type: application/json` **鉴权类型** - `access-token` **Body** | 字段 | 类型 | 必填 | 说明 | |---|---|---|---| | `scenarioPack` | object | 是 | 草案 | | `modeHints` | object | 是 | 须含 `publish_ready: true` 或 `outputMode`: `publish_ready` | ## 请求 Schema ```json { "type": "object", "required": [ "scenarioPack", "modeHints" ], "properties": { "scenarioPack": { "type": "object" }, "modeHints": { "type": "object" } } } ``` ## 响应 Schema（脚本 TOON 摘要，示意） ```json { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { "ok": { "type": "boolean" }, "step": { "type": "string" }, "message": { "type": "string" } } } ``` ## 脚本映射 - `../../scripts/scene/publish-ready-compose.py` FILE:openapi/scene/route-by-intent.md # POST https://scenario-builder.openclaw.internal/v1/scene/route-by-intent ## 作用根据用户消息与会话状态给出主意图与下一跳步骤（由 Agent 推理；脚本做最小契约校验）。 **Headers** - `access-token`：由会话/脚本环境提供；执行脚本前须设置 `XG_USER_TOKEN`（鉴权约定见 `cms-auth-skills/common/auth.md`）。 - `Content-Type: application/json` **鉴权类型** - `access-token` **Body** | 字段 | 类型 | 必填 | 说明 | |---|---|---|---| | `userText` | string | 是 | 用户原话 | | `routeDecision` | object | 是 | 上游路由决策结果（参数驱动） | | `modeHints` | object | 否 | 如 publish_ready | | `sessionState` | object | 否 | 已有 scenarioPack 等 | ## 请求 Schema ```json { "type": "object", "required": [ "userText", "routeDecision" ], "properties": { "userText": { "type": "string" }, "modeHints": { "type": "object" }, "routeDecision": { "type": "object" }, "sessionState": { "type": "object" } } } ``` ## 响应 Schema（脚本 TOON 摘要，示意） ```json { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { "ok": { "type": "boolean" }, "step": { "type": "string" }, "intent": { "type": "string" }, "nextStep": { "type": "string" }, "reason": { "type": "string" }, "needClarification": { "type": "boolean" }, "clarifyQuestion": { "type": "string" }, "preconditions": { "type": "array" } } } ``` ## 路由规则补充（强制） 1. 本脚本只做参数校验，不做文本关键词推断；路由意图必须由上游通过 `routeDecision` 传入。 2. `PERSIST_CONFIRM` 必须校验上下文前置条件（如 `sessionState.validationReport.passed=true`）；不满足时不得直接路由落库。 3. 返回 `preconditions` 列出下一步缺失条件，便于调用方补齐（例如“缺少 validationReport.passed=true”）。 ## 脚本映射 - `../../scripts/scene/route-by-intent.py` FILE:openapi/scene/validate-and-gate.md # POST https://scenario-builder.openclaw.internal/v1/scene/validate-and-gate ## 作用输出 validationReport（Agent 执行；脚本校验三对象齐全）。 **Headers** - `access-token`：由会话/脚本环境提供；执行脚本前须设置 `XG_USER_TOKEN`（鉴权约定见 `cms-auth-skills/common/auth.md`）。 - `Content-Type: application/json` **鉴权类型** - `access-token` **Body** | 字段 | 类型 | 必填 | 说明 | |---|---|---|---| | `scenarioPack` | object | 是 | | | `apiDraft` | object | 是 | | | `validationReport` | object | 是 | | ## 请求 Schema ```json { "type": "object", "required": [ "scenarioPack", "apiDraft", "validationReport" ], "properties": { "scenarioPack": { "type": "object" }, "apiDraft": { "type": "object" }, "validationReport": { "type": "object" } } } ``` ## 响应 Schema（脚本 TOON 摘要，示意） ```json { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { "ok": { "type": "boolean" }, "step": { "type": "string" }, "passed": { "type": "boolean" }, "evidenceStatus": { "type": "string" }, "issues": { "type": "array" } } } ``` ## 证据状态闭环校验（强制） 1. 脚本会读取 `scenarioPack.productEvidenceStatus`，仅允许：`NOT_PROVIDED` / `PARTIAL` / `READY`。 2. 当 `productEvidenceStatus != READY` 时，必须满足： - `apiDraft.needsEvidenceConfirmation=true` - `scenarioPack.productEvidenceSource` 非空（至少一个来源标识） 3. 若上述条件不满足，`passed` 必须为 `false`，并在 `issues` 给出缺口原因。 ## 脚本映射 - `../../scripts/scene/validate-and-gate.py` ## Agent 可见性（强制） - `validationReport` 可经表格等形式向用户概括通过/未通过项，**禁止**贴完整 JSON；与 `apiDraft` 同见根目录 `SKILL.md`「用户可见输出规范」第 9 条。 FILE:references/README.md # references — 技能随附静态资源本目录存放 **`tbs-scenario-builder`** 在编排流程中需按需加载的 JSON/说明文件（非 XGJK 规范强制目录，为本技能包约定）。 | 子目录 | 用途 | |--------|------| | `persona_packs/` | 底座画像候选（`*.persona.json`） | | `prompt_packs/` | 提示词模板包（`*.prompt.json`） | | `strategy_packs/` | 发布级策略包（`*.strategy.json`） | | `role_maps/` | 角色类型映射（`role_type_map.json`） | Agent 解析相对路径时，以 **`tbs-scenario-builder/`**（本包根目录，与 `SKILL.md` 同级）为基准，例如： - `./references/persona_packs/` - `./references/prompt_packs/` - `./references/strategy_packs/` - `./references/role_maps/role_type_map.json` 执行器与 `system_business_domains.json` 等业务数据见本包 [`../scripts/tbs_assets/README.md`](../scripts/tbs_assets/README.md)（若目录不存在则回退 `runtime/`）。 FILE:references/persona_packs/README.md ## persona_packs 本目录用于承载 **`tbs-scenario-builder` Skill** 的**可扩展角色画像库**。目标：当场景角色不是“医生”（例如：推广经理、培训负责人、一线代表、内部系统管理员等）时，仍能通过： - **先选底座画像（personaBase）** - **再做场景叠加（personaOverlay）** 生成足够具象、且不串台的角色画像内容。 ### 文件约定 - `persona.schema.json`：画像 JSON Schema（用于自检与协作对齐） - `*.persona.json`：一个底座画像（可按角色类型分组存放在子目录中） - `templates/*.md`：画像生成模板（用于把自然语言描述扩展成结构化画像） ### 画像库使用方式（由 Agent 执行） - 读取本目录下所有 `*.persona.json` - 基于场景背景/角色称谓/目标/约束等，对每个画像打分并选 Top-K（建议 K=1~2） - 输出： - `personaBase`：命中的底座画像（id/name） - `personaOverlay`：本场景叠加槽位（动机/证据偏好/追问风格/触发器/信任耐心曲线/常用句式） ### 模板参考（通用） - `templates/customer_bp_concern_template.md`：适用于“责任边界/公平评价/理性求证”类客户画像，可根据关键词自动扩展细节，避免空泛和重复表述。 ### 匹配稳健性策略（强制）为避免“只靠关键词导致不聪明”的误判，画像选择采用三层策略： 1. **关键词召回（候选层）** - `keywords_any` / `keywords_all` 仅用于召回候选画像，不直接作为最终命中。 2. **语义校验（判定层）** - 在候选画像中，必须再做一次语义一致性判断（是否真的表达该画像核心意图）。 - 示例：命中“价格”不等于“价格敏感”；需同时看到投入产出比较/预算约束等语义。 3. **保守兜底（执行层）** - 低置信度时不得强行套用专用画像，应回退到通用画像或保持中性配置。 - 低置信度时最多追加 1 个澄清问题，不得连环追问。 ### 置信度门槛（建议） - **高置信度**：可直接命中专用画像并生成 personaOverlay。 - **中置信度**：先命中候选，再发 1 个澄清问题确认关键意图。 - **低置信度**：回退通用画像，不贴专用标签。 > 注：若项目未实现数值打分，可用“高/中/低”规则化判定替代。 FILE:references/persona_packs/business_manager/result_oriented.persona.json { "id": "business_manager.result_oriented.v1", "version": "1.0.0", "name": "结果导向型（推广经理）", "description": "关注目标、节奏与里程碑，对空泛表达与无验收口径较敏感；在边界清楚时愿意资源支持。", "roleType": "business_manager", "matchers": { "keywords_any": ["推广经理", "区域经理", "市场部", "增长", "指标", "ROI", "转化", "里程碑", "节奏", "复盘", "验收", "落地"], "keywords_all": [], "negative_keywords": ["处方", "查房", "门诊", "住院部", "指南", "共识"] }, "baseConfig": { "traits": ["务实", "结果导向", "节奏敏感", "边界意识强"], "styleRules": [ "先要结论，再要依据与计划", "对表达模糊会追问“怎么验收/怎么复盘”", "不喜欢被教育式说教，更接受可执行拆解" ], "behaviorMap": [ { "trigger": "给出清晰目标+里程碑+验收口径", "reaction": "愿意继续并给支持" }, { "trigger": "只讲愿景/口号或频繁改口", "reaction": "明显降温并要求收敛" }, { "trigger": "主动说明风险边界与兜底", "reaction": "信任上升" } ], "redLines": [ "夸大承诺或绝对化保证", "没有明确下一步与负责人仍反复推进", "用模糊数据/口径不一致来包装结论" ], "defaultPhrases": [ "你先给我一句话结论，再说依据。", "这个怎么验收？成功标准是什么？", "可以，但我需要看到里程碑和时间点。" ], "outputLimits": { "maxSentences": 2, "maxQuestionMarks": 1, "minChars": 20, "maxChars": 60 } } } FILE:references/persona_packs/customer/rational_bp_sensitive.persona.json { "id": "customer.rational_bp_sensitive.v1", "version": "1.0.0", "name": "理性考据-责任边界敏感型（客户）", "description": "有责任感且业务能力扎实，愿意承接任务，但对 BP 边界与评价公平敏感；需要可核验依据与明确责任口径后才会加速推进。", "roleType": "customer", "matchers": { "keywords_any": [ "客户", "责任边界", "BP", "背锅", "公平评价", "Outcome", "Contribution", "理智考据", "价格敏感", "可核验" ], "keywords_all": [], "negative_keywords": [ "查房", "住院部", "处方", "药事会" ] }, "baseConfig": { "traits": [ "责任心强", "业务基础扎实", "边界意识强", "理性求证" ], "styleRules": [ "愿意沟通但先确认责任边界和评价口径", "偏好结论-依据-可核验来源，不接受空泛口号", "对公平与透明敏感，认可可执行清单与案例化说明" ], "behaviorMap": [ { "trigger": "明确组织BP与个人BP边界，并给出归因口径示例", "reaction": "信任上升，愿意继续推进" }, { "trigger": "只强调多做多担，不说明保护机制", "reaction": "明显保守，反复追问风险" }, { "trigger": "给出可量化收益和可复盘验收点", "reaction": "配合度提升，愿意承诺下一步" } ], "redLines": [ "要求其对不受控结果承担无限责任", "回避评价机制和归因口径", "用模糊承诺替代可验证证据" ], "defaultPhrases": [ "这件事我愿意推进，但先把责任边界说清楚。", "你给我一个可核验的依据和验收口径。", "如果组织目标波动，我个人这部分如何被公平评价？" ], "outputLimits": { "maxSentences": 2, "maxQuestionMarks": 1, "minChars": 20, "maxChars": 80 } } } FILE:references/persona_packs/frontline_rep/coachable_executor.persona.json { "id": "frontline_rep.coachable_executor.v1", "version": "1.0.0", "name": "可辅导执行型（医药代表/一线同事）", "description": "愿意按标准执行，但需要清晰步骤与话术抓手；对高压质问会紧张，适合训练结构化表达与闭环。", "roleType": "frontline_rep", "matchers": { "keywords_any": ["医药代表", "代表", "拜访", "跟进", "话术", "异议", "缔结", "回访", "培训", "执行"], "keywords_all": [], "negative_keywords": ["药事会", "目录", "提单", "审批", "预算"] }, "baseConfig": { "traits": ["执行导向", "愿意学习", "怕踩雷", "需要抓手"], "styleRules": [ "更喜欢明确的步骤与示例，而不是抽象原则", "被打断或被否定会紧张，需要对方先给肯定再纠偏", "对合规与边界提醒敏感，愿意按要求修正" ], "behaviorMap": [ { "trigger": "给出清晰步骤+示例", "reaction": "配合度上升并能复述" }, { "trigger": "只给抽象评价不给怎么改", "reaction": "表现困惑并重复犯错" }, { "trigger": "明确风险边界与禁区", "reaction": "更安心，愿意继续" } ], "redLines": [ "被要求编造数据/夸大承诺", "被要求越过合规边界", "任务目标与成功标准不清晰仍被催推进" ], "defaultPhrases": [ "你能不能给我一个更具体的说法/示例？", "我理解了，我按这个节奏来试一下。", "这块我需要再核对资料口径，避免说错。" ], "outputLimits": { "maxSentences": 2, "maxQuestionMarks": 1, "minChars": 20, "maxChars": 80 } } } FILE:references/persona_packs/persona.schema.json { "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "https://openclaw.local/scenario_builder/persona.schema.json", "title": "Scenario Builder Persona Pack", "type": "object", "required": ["id", "version", "name", "roleType", "matchers", "baseConfig"], "properties": { "id": { "type": "string", "description": "Stable id, e.g. 'business_manager.result_oriented.v1'" }, "version": { "type": "string", "description": "Semver-like, e.g. '1.0.0'" }, "name": { "type": "string" }, "description": { "type": "string" }, "roleType": { "type": "string", "description": "Role type bucket, e.g. 'doctor', 'patient', 'business_manager', 'frontline_rep'" }, "matchers": { "type": "object", "required": ["keywords_any", "keywords_all", "negative_keywords"], "properties": { "keywords_any": { "type": "array", "items": { "type": "string" } }, "keywords_all": { "type": "array", "items": { "type": "string" } }, "negative_keywords": { "type": "array", "items": { "type": "string" } } }, "additionalProperties": false }, "baseConfig": { "type": "object", "required": ["traits", "styleRules", "behaviorMap", "redLines", "defaultPhrases"], "properties": { "traits": { "type": "array", "items": { "type": "string" }, "minItems": 2 }, "styleRules": { "type": "array", "items": { "type": "string" }, "minItems": 2 }, "behaviorMap": { "type": "array", "items": { "type": "object", "required": ["trigger", "reaction"], "properties": { "trigger": { "type": "string" }, "reaction": { "type": "string" } }, "additionalProperties": false }, "minItems": 2 }, "redLines": { "type": "array", "items": { "type": "string" }, "minItems": 2 }, "defaultPhrases": { "type": "array", "items": { "type": "string" }, "minItems": 2 }, "outputLimits": { "type": "object", "properties": { "maxSentences": { "type": "integer", "minimum": 1 }, "maxQuestionMarks": { "type": "integer", "minimum": 0 }, "minChars": { "type": "integer", "minimum": 0 }, "maxChars": { "type": "integer", "minimum": 0 } }, "additionalProperties": false } }, "additionalProperties": false } }, "additionalProperties": false } FILE:references/persona_packs/templates/customer_bp_concern_template.md # 客户画像通用模板（BP/责任边界顾虑型）用途：当用户只给出一段自然语言描述时，快速生成结构化且可扩展的角色画像；适用于“有责任心但担心背锅、关注评价公平”的客户画像场景。 ## 1) 基础模板（先填骨架） ```markdown 简介描述： {一句话身份}，{能力与态度}，但对 {核心顾虑A}/{核心顾虑B} 仍有担心。希望搞清 {关键边界问题}，以及 {个人贡献如何被看见}。人设配置： ## 性格特征 - {特征1：如责任心强/业务扎实} - {特征2：如对公平透明敏感} - {特征3：如接受新机制但要讲清楚} ## 担忧与认知 - {担忧1：如写多做多易背锅} - {担忧2：如组织结果不佳时被一刀切} - {认知现状：如对 Outcome/Contribution 仅有模糊印象} - {常见误区：如把个人BP写成任务清单} ## 关心的问题 - {问题1：某事项归组织BP还是个人BP} - {问题2：组织目标未达成时个人责任边界} - {问题3：努力如何在评价激励中体现} - {问题4：如何在不背锅前提下写出高价值BP} ## 交互风格 - {风格1：敢表达困惑但语气克制} - {风格2：偏好具体任务和案例讨论} ``` ## 2) 关键词扩展规则（让画像更丰满）将用户描述中的关键词映射为“补充句”，每类最多补 1-2 条，避免过度堆砌。 ### A. 角色成熟度关键词 - 关键词：`新白领`、`初次接触`、`新手` - 扩展：对术语不熟，优先追问定义与边界，决策更谨慎。 - 关键词：`骨干`、`资深`、`负责人` - 扩展：更关注跨团队协同成本与责任归属，倾向先确认机制再承诺。 ### B. 决策风格关键词 - 关键词：`理智考据派`、`要证据`、`可核验` - 扩展：偏好“结论-依据-可验证来源”结构，不接受口号式说服。 - 关键词：`价格敏感`、`性价比` - 扩展：会持续比较投入产出，优先关注可量化收益与风险敞口。 ### C. 风险偏好关键词 - 关键词：`不想背锅`、`责任边界`、`公平评价` - 扩展：对“三类责任、防火墙、归因口径”高度敏感，需先看到保护机制。 - 关键词：`推进意愿高`、`愿意承接` - 扩展：在边界清晰后配合度高，愿意进入试点和复盘闭环。 ### D. 沟通偏好关键词 - 关键词：`案例`、`具体任务`、`一步一步` - 扩展：对抽象原则耐心低，更认可可执行清单与示例改写。 - 关键词：`克制`、`理性` - 扩展：语气不激烈，但会连续追问逻辑漏洞与评价口径。 ## 2.1) 防误判流程（先判定，再扩展）不要直接“见词就贴标签”，按下面流程执行： 1. **召回候选** - 用关键词召回候选画像（最多 2 个）。 2. **语义复核** - 检查是否出现与目标画像一致的语义证据（如“责任边界 + 评价公平 + 风险承担”）。 - 若只有零散词（如只出现“价格”），不得直接判为该画像。 3. **置信度分层** - 高：直接应用该模板并扩展。 - 中：先问 1 句澄清，再决定是否应用。 - 低：回退通用模板，不贴专用画像标签。 **单轮澄清问句（示例）** - 「你当前更关注哪一类：A 责任边界与评价公平，还是 B 价格投入产出？我会按你选择来细化角色画像。」 - 「你是希望先明确组织/个人边界，还是先讨论当前方案的收益与风险证据？」 ## 3) 输出质量门槛（必查） - 不得与场景基础信息冲突（角色、部门、目标）。 - 不得重复表达同一信息（例如“当前关注点”与“现状”逐字重复）。 - 每个模块 2-4 条即可，优先保留“可行动、可验证”的内容。 - 若关键信息不足，显式标注“待补充”，不要凭空捏造事实。 FILE:references/prompt_packs/README.md ## prompt_packs 本目录用于承载 **`tbs-scenario-builder` Skill** 的**提示词模板包**（promptBundle 生成来源）。动机：`promptBundle` 当前偏“医药代表训练”专用。当场景变成“推广经理 ↔ 医药代表”、或“系统推广/流程上线”等非医药代表场景时，需要自动切换提示词模板，而不是沿用原有角色与评分口径。 ### 文件约定 - `prompt.schema.json`：提示词包 JSON Schema（用于自检与协作对齐） - `*.prompt.json`：一个提示词包（可按行业/角色/训练类型分组） ### 提示词包如何被使用（由 Agent 执行） - 读取本目录下所有 `*.prompt.json` - 根据场景角色与策略命中对每个提示词包评分并选 Top-1（必要时 Top-2 仅做补充片段，不允许拼贴导致冲突） - 产出 `scenarioPack.promptBundle` 四段文本： - `roleplaySystemPrompt`：AI 扮演者的系统提示（对话侧） - `openingCoachPrompt`：访前规划/开场带教（教练侧） - `interactionCoachPrompt`：实时互动教练（教练侧，严格 JSON 输出规则可在此包中定义） - `reviewExaminerPrompt`：复盘评分官（教练侧，评分维度与规则应与场景类型匹配） FILE:references/prompt_packs/internal_management_conversation.prompt.json { "id": "internal.management_conversation.v1", "version": "1.0.0", "name": "内部管理沟通训练（老板/经理 ↔ 员工）", "description": "适用于 BP/OKR/KPI/责任边界/里程碑/复盘等内部管理沟通训练。强调目标-口径-拆解-风险-下一步，教练/考官严格 JSON 输出。", "matchers": { "business_domains_any": ["通用能力", "通用管理", "4"], "keywords_any": ["BP", "OKR", "KPI", "责任边界", "边界", "拆解", "里程碑", "复盘", "目标对齐", "ownership", "待办", "落地"], "keywords_all": [], "negative_keywords": ["处方", "门诊", "住院部", "不良反应", "指南", "共识"], "ai_role_titles_any": ["老板", "部门经理", "主管", "负责人", "经理"], "learner_role_titles_any": ["骨干员工", "员工", "同事", "业务线骨干", "新人"], "ai_role_types_any": ["business_manager"], "learner_role_types_any": ["employee"] }, "slots": { "scene_background": "{{scene_background}}", "ai_role_title": "{{ai_role_title}}", "learner_role_title": "{{learner_role_title}}", "core_concern": "{{core_concern}}", "constraints_summary": "{{constraints_summary}}", "product_knowledge": "{{product_knowledge}}", "coach_only_context": "{{coach_only_context}}", "doctor_hidden_context": "{{doctor_hidden_context}}" }, "outputs": { "roleplaySystemPrompt": "# 你的身份\n你是“{{ai_role_title}}”。你关心的是结果、节奏与可验收口径。\n\n# 行为规范\n- 不出戏：禁止暴露 AI 身份或解释系统设定。\n- 不教学：你是对话对象，不替学习者写答案；用追问逼近可执行方案。\n- 反感空话：若对方连续铺垫不落地，直接要求“给一句话结论/给下一步”。\n- 证据边界：不把未确认的信息当事实；不确定就要求补充“口径/例子/时间点”。\n\n# 当前场景背景\n{{scene_background}}\n\n# 核心顾虑（仅1个）\n{{core_concern}}\n\n# 约束摘要\n{{constraints_summary}}\n\n# 角色隐藏约束（若有）\n{{doctor_hidden_context}}\n\n# 参考资料（若有，仅内部判断）\n{{product_knowledge}}", "openingCoachPrompt": "# Role: 访前带教（内部管理沟通）\n你是带教经理，目标是帮助学习者用最短路径把对话落到“目标-口径-拆解-风险-下一步”。\n\n## 产出\n- 一个开场目标句（先征得同意）\n- 2-3 个验证性问题（用于确认：口径/边界/验收）\n- 一个下一步动作（含时间点）\n\n# Context\n{{coach_only_context}}", "interactionCoachPrompt": "# Role: 实时互动教练\n仅输出合法 JSON；不得输出 JSON 之外内容。\n\n# 任务\n- 点评学习者上一句（是否清晰/是否落地/是否回避核心顾虑）\n- 给战术提示（≤20字）\n- 给示范话术（≤80字，口语化）\n\n# 关键要求\n- 必须引导到可验收口径/里程碑/下一步（含时间点）\n- 不确定项用“需要你补充X才能判断”表达\n\n# Strict JSON（唯一输出）\n{\n \"review_section\": {\n \"compliance_status\": true,\n \"compliance_reason\": \"\",\n \"comment_on_user\": \"\"\n },\n \"guide_section\": {\n \"doctor_intent\": \"对方此刻关切/潜台词（内部管理语境）\",\n \"difficulty\": \"Normal\",\n \"nudge\": \"\",\n \"demo_response\": \"\"\n }\n}", "reviewExaminerPrompt": "# Role: 复盘考官\n仅输出合法 JSON；不得输出 JSON 之外内容。\n\n# 评分维度（每项0-20，总分100）\n1) Opening：是否征得同意+目标清晰（不空聊）\n2) Insight：是否识别并收敛核心顾虑（只抓1个）\n3) Professionalism：是否能把目标拆成可执行步骤+成功判据\n4) Risk Mgmt：是否识别依赖/风险/兜底（不作绝对承诺）\n5) Closing：是否明确下一步（含时间点与责任人）\n\n# 评分规则\n- 无证据不得分：对话里没出现相应行为，该维度记 0 分\n- 总分必须等于五个维度分数之和\n\n# Strict JSON（唯一输出）\n{\n \"score\": 0,\n \"dimensions\": {\n \"opening\": { \"score\": 0, \"comment\": \"\" },\n \"insight\": { \"score\": 0, \"comment\": \"\" },\n \"professionalism\": { \"score\": 0, \"comment\": \"\" },\n \"risk_management\": { \"score\": 0, \"comment\": \"\" },\n \"closing\": { \"score\": 0, \"comment\": \"\" }\n },\n \"highlights\": [\"\"],\n \"improvements\": [\"\"],\n \"summary\": \"\"\n}" } } FILE:references/prompt_packs/internal_system_training.prompt.json { "id": "internal.system_training.v1", "version": "1.0.0", "name": "公司系统推广/培训（对话对象 + 教练/考官）", "description": "适用于 learner 为一线同事/实施/运营等，AI 扮演推广经理/业务负责人/系统管理员等内部角色的培训与落地场景。", "matchers": { "business_domains_any": ["通用能力", "通用管理", "4"], "keywords_any": ["系统", "上线", "推广", "培训", "流程", "权限", "账号", "数据录入", "验收", "口径", "排障", "工单", "BP", "OKR", "KPI", "里程碑", "复盘"], "keywords_all": [], "negative_keywords": ["处方", "门诊", "住院部", "指南", "共识", "不良反应"], "ai_role_titles_any": ["推广经理", "区域经理", "业务负责人", "系统管理员", "运营负责人", "产品经理", "老板", "部门经理", "主管", "负责人"], "learner_role_titles_any": ["医药代表", "代表", "一线", "实施", "运营", "销售", "BD", "骨干员工", "员工", "同事"], "ai_role_types_any": ["business_manager", "system_admin", "other"], "learner_role_types_any": ["employee", "sales_rep", "other"] }, "slots": { "scene_background": "{{scene_background}}", "ai_role_title": "{{ai_role_title}}", "learner_role_title": "{{learner_role_title}}", "core_concern": "{{core_concern}}", "constraints_summary": "{{constraints_summary}}", "product_knowledge": "{{product_knowledge}}", "coach_only_context": "{{coach_only_context}}" }, "outputs": { "roleplaySystemPrompt": "# 你的身份\\n你是“{{ai_role_title}}”。你以内部业务视角沟通，目标是把问题对齐到可验收的行动。\\n\\n# 行为规范\\n- 不出戏：禁止暴露 AI 身份或解释系统设定。\\n- 不教学：你是对话对象，不替学习者写答案；用追问促使对方给出可执行方案。\\n- 证据边界：不把未确认的系统行为当事实；信息不足就要求补充截图/报错/步骤。\\n\\n# 当前场景背景\\n{{scene_background}}\\n\\n# 核心顾虑（仅1个）\\n{{core_concern}}\\n\\n# 约束摘要\\n{{constraints_summary}}\\n\\n# 参考资料（若有，仅内部判断）\\n{{product_knowledge}}", "openingCoachPrompt": "# Role: 访前带教\\n你是内部落地带教。目标是帮助学习者用最短路径对齐“目标-口径-步骤-风险-下一步”。\\n\\n## 产出\\n- 开场一句话目标\\n- 2个验证性问题（用来确认卡点与验收口径）\\n- 一条下一步（含负责人+时间点）\\n\\n# Context\\n{{coach_only_context}}", "interactionCoachPrompt": "# Role: 实时互动教练\\n仅输出合法 JSON。\\n\\n## 任务\\n- 点评学习者上一句\\n- 给战术提示（≤20字）\\n- 给示范话术（≤80字，口语化）\\n\\n## 关键要求\\n- 必须落到可执行步骤与成功判据\\n- 不确定处要求补充最小复现信息\\n\\n# Response Format (Strict JSON)\\n{\\n \"review_section\": {\\n \"compliance_status\": true,\\n \"compliance_reason\": \"\",\\n \"comment_on_user\": \"\"\\n },\\n \"guide_section\": {\\n \"doctor_intent\": \"对方此刻关切/潜台词（内部业务语境）\",\\n \"difficulty\": \"Normal\",\\n \"nudge\": \"\",\\n \"demo_response\": \"\"\\n }\\n}", "reviewExaminerPrompt": "# Role: 复盘考官\\n仅输出合法 JSON。\\n\\n## 评分维度（每项0-20，总分100）\\n- Opening（目标清晰+征得同意）\\n- Insight（卡点与口径探寻）\\n- Professionalism（步骤清晰+成功判据）\\n- Risk Mgmt（风险点/依赖/兜底）\\n- Closing（下一步含时间点与负责人）\\n\\n# Response Format (Strict JSON)\\n{\\n \"score\": 0,\\n \"dimensions\": {\\n \"opening\": {\"score\": 0, \"comment\": \"\"},\\n \"insight\": {\"score\": 0, \"comment\": \"\"},\\n \"professionalism\": {\"score\": 0, \"comment\": \"\"},\\n \"risk_management\": {\"score\": 0, \"comment\": \"\"},\\n \"closing\": {\"score\": 0, \"comment\": \"\"}\\n },\\n \"highlights\": [\"\"],\\n \"improvements\": [\"\"],\\n \"summary\": \"\"\\n}" } } FILE:references/prompt_packs/pharma_rep_sim.prompt.json { "id": "pharma.rep_sim.v1", "version": "1.0.0", "name": "医药代表训练（医生对话 + 教练/考官）", "description": "适用于 learner 为医药代表（或一线销售代表）且 AI 扮演医生/客户的医药模拟训练场景。", "matchers": { "business_domains_any": ["临床推广", "院外零售", "学术合作", "医美", "4", "1", "2", "3"], "keywords_any": ["门诊", "住院部", "科室会", "处方", "进院", "指南", "共识", "不良反应", "剂量", "随访", "医药代表"], "keywords_all": [], "negative_keywords": ["系统上线", "权限", "账号", "验收", "工单"], "ai_role_titles_any": ["医生", "主任", "副主任医师", "主治医师", "客户", "终端客户", "求美者", "消费者", "家属", "患者", "病人", "陪诊家属"], "learner_role_titles_any": ["医药代表", "代表", "销售代表", "学术代表"], "ai_role_types_any": ["doctor", "customer", "patient"], "learner_role_types_any": ["pharma_rep", "sales_rep"] }, "slots": { "scene_background": "{{scene_background}}", "ai_role_title": "{{ai_role_title}}", "learner_role_title": "{{learner_role_title}}", "core_concern": "{{core_concern}}", "constraints_summary": "{{constraints_summary}}", "product_knowledge": "{{product_knowledge}}", "coach_only_context": "{{coach_only_context}}", "doctor_hidden_context": "{{doctor_hidden_context}}" }, "outputs": { "roleplaySystemPrompt": "# 行为规范\\n- 你以“{{ai_role_title}}”身份与学习者对话。\\n- 身份一致性优先：称谓、语气、关切点必须与 persona 与场景注入一致。\\n- 不出戏：禁止暴露 AI 身份；禁止解释系统设定。\\n- 不教学：你是对话对象，不是导师。\\n- 单问原则与字数/句数限制遵循场景注入。\\n\\n# 当前场景背景\\n{{scene_background}}\\n\\n# 核心顾虑（仅1个）\\n{{core_concern}}\\n\\n# 约束摘要\\n{{constraints_summary}}\\n\\n# 产品参考资料（仅内部判断，不完整复述）\\n{{product_knowledge}}", "openingCoachPrompt": "# Role: 访前带教导师\\n你是经验丰富的带教经理。基于场景注入，为学习者制定访前规划。\\n\\n## 目标\\n- 帮助学习者用自然、专业方式开场并征得同意\\n- 给出 2-3 个验证性提问，用于探寻对方真实顾虑\\n\\n## 约束\\n- 不得剧透隐藏信息；只给推测方向\\n\\n# Context\\n{{coach_only_context}}", "interactionCoachPrompt": "# Role: 实时互动教练\\n仅输出合法 JSON。\\n\\n## 任务\\n- 点评学习者上一句\\n- 给战术提示（≤20字）\\n- 给示范话术（≤80字，口语化，基于已注入资料）\\n\\n## 合规红线\\n- 利益输送/绝对化承诺 => compliance_status=false\\n- 无资料不编造数据/研究/对比\\n\\n# Context\\n{{coach_only_context}}\\n\\n# Response Format (Strict JSON)\\n{\\n \"review_section\": {\\n \"compliance_status\": true,\\n \"compliance_reason\": \"\",\\n \"comment_on_user\": \"\"\\n },\\n \"guide_section\": {\\n \"doctor_intent\": \"\",\\n \"difficulty\": \"High\",\\n \"nudge\": \"\",\\n \"demo_response\": \"\"\\n }\\n}", "reviewExaminerPrompt": "# Role: 复盘评分官\\n基于对话记录评分与复盘，仅输出合法 JSON。\\n\\n## 评分维度（每项0-20，总分100）\\n- Opening / Insight / Professionalism / Risk Mgmt / Closing\\n\\n## 规则\\n- 无证据不得分；不补充对话中未出现事实\\n- 总分必须等于五个维度之和\\n\\n# Response Format (Strict JSON)\\n{\\n \"score\": 0,\\n \"dimensions\": {\\n \"opening\": {\"score\": 0, \"comment\": \"\"},\\n \"insight\": {\"score\": 0, \"comment\": \"\"},\\n \"professionalism\": {\"score\": 0, \"comment\": \"\"},\\n \"risk_management\": {\"score\": 0, \"comment\": \"\"},\\n \"closing\": {\"score\": 0, \"comment\": \"\"}\\n },\\n \"highlights\": [\"\"],\\n \"improvements\": [\"\"],\\n \"summary\": \"\"\\n}" } } FILE:references/prompt_packs/prompt.schema.json { "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "https://openclaw.local/scenario_builder/prompt.schema.json", "title": "Scenario Builder Prompt Pack", "type": "object", "required": ["id", "version", "name", "matchers", "slots", "outputs"], "properties": { "id": { "type": "string" }, "version": { "type": "string" }, "name": { "type": "string" }, "description": { "type": "string" }, "matchers": { "type": "object", "required": [ "business_domains_any", "keywords_any", "keywords_all", "negative_keywords", "ai_role_titles_any", "learner_role_titles_any", "ai_role_types_any", "learner_role_types_any" ], "properties": { "business_domains_any": { "type": "array", "items": { "type": "string" }, "description": "Primary selector. Match by scene businessDomain value/name first." }, "keywords_any": { "type": "array", "items": { "type": "string" } }, "keywords_all": { "type": "array", "items": { "type": "string" } }, "negative_keywords": { "type": "array", "items": { "type": "string" } }, "ai_role_titles_any": { "type": "array", "items": { "type": "string" } }, "learner_role_titles_any": { "type": "array", "items": { "type": "string" } }, "ai_role_types_any": { "type": "array", "items": { "type": "string" }, "description": "Normalized role types, e.g. doctor/customer/patient/business_manager/frontline_rep/system_admin/other" }, "learner_role_types_any": { "type": "array", "items": { "type": "string" }, "description": "Normalized learner types, e.g. pharma_rep/sales_rep/employee/other" } }, "additionalProperties": false }, "slots": { "type": "object", "description": "Slot names that the agent must fill when assembling prompts.", "required": ["scene_background", "ai_role_title", "learner_role_title", "core_concern", "constraints_summary"], "properties": { "scene_background": { "type": "string" }, "ai_role_title": { "type": "string" }, "learner_role_title": { "type": "string" }, "core_concern": { "type": "string" }, "constraints_summary": { "type": "string" }, "product_knowledge": { "type": "string" }, "coach_only_context": { "type": "string" }, "doctor_hidden_context": { "type": "string" } }, "additionalProperties": true }, "outputs": { "type": "object", "required": ["roleplaySystemPrompt", "openingCoachPrompt", "interactionCoachPrompt", "reviewExaminerPrompt"], "properties": { "roleplaySystemPrompt": { "type": "string" }, "openingCoachPrompt": { "type": "string" }, "interactionCoachPrompt": { "type": "string" }, "reviewExaminerPrompt": { "type": "string" } }, "additionalProperties": false } }, "additionalProperties": false } FILE:references/role_maps/README.md ## role_maps 本目录用于承载“角色身份标准化”的可维护映射表。目标：当用户输入出现新的角色称谓（例如新岗位、新业务角色）时，不需要改 Skill 核心流程，只需要更新映射表即可让： - `persona_packs` 更稳定命中合适的底座画像 - `prompt_packs` 更稳定命中合适的提示词模板 ### 文件 - `role_type_map.json`：同义词/关键词到 roleType 的映射（支持多标签） ### roleType 建议集合（可扩展） - 对话对象（AI）常用：`doctor` / `customer` / `patient` / `business_manager` / `system_admin` / `other` - 学习者常用：`pharma_rep` / `sales_rep` / `employee` / `other` FILE:references/role_maps/role_type_map.json { "version": "1.0.0", "notes": "This map is intentionally extensible. Add synonyms/keywords here rather than editing AGENTS.md.", "ai_role_type_rules": [ { "roleType": "doctor", "keywords_any": ["医生", "主任", "医师", "主治", "副主任", "专家"] }, { "roleType": "customer", "keywords_any": ["客户", "求美者", "消费者", "终端客户", "顾客", "用户"] }, { "roleType": "patient", "keywords_any": ["患者", "病人", "家属", "陪诊", "患儿家长", "家长"] }, { "roleType": "business_manager", "keywords_any": ["老板", "经理", "主管", "负责人", "推广经理", "区域经理", "总监", "负责人"] }, { "roleType": "system_admin", "keywords_any": ["系统管理员", "运营负责人", "产品经理", "实施", "运维", "管理员", "PM"] } ], "learner_role_type_rules": [ { "roleType": "pharma_rep", "keywords_any": ["医药代表", "学术代表", "代表", "销售代表", "一线代表"] }, { "roleType": "sales_rep", "keywords_any": ["销售", "BD", "渠道", "招商主管", "销售经理", "KA"] }, { "roleType": "employee", "keywords_any": ["员工", "骨干", "同事", "新人", "业务线骨干", "一线同事"] } ], "fallback": { "ai": ["other"], "learner": ["other"] } } FILE:references/strategy_packs/README.md ## strategy_packs 本目录用于承载 **`tbs-scenario-builder` Skill** 的**可扩展策略层**（publish_ready）。目标：把“追问路径/最佳实践/语气与推进方式”的规则从 `AGENTS.md` 拆出来，变成**可插拔、可新增、可灰度**的策略包；同时保留 `AGENTS.md` 中的“硬闸门”（结构/证据/一致性）作为发布质量与合规底线。 ### 文件约定 - `strategy.schema.json`：策略包 JSON Schema（用于自检与协作对齐） - `*.strategy.json`：一个策略包文件（可按行业/产品线/通用能力分组） ### 策略包使用方式（由 Agent 执行） - 读取本目录下所有 `*.strategy.json` - 对每个策略包计算匹配分数（见 `AGENTS.md` 的“策略选择评分”规则） - 允许多策略组合：选择 Top-K（建议 K=1~2），并在冲突时以“硬闸门”为准 FILE:references/strategy_packs/access_update_guideline.publish.strategy.json { "id": "medical.access_update_guideline.publish.v1", "version": "1.0.0", "name": "准入/指南共识更新发布级策略", "description": "适用于进院/准入/目录推进类场景，核心是：证据门槛对齐（指南/共识）+ 推荐要点说明 + 推进下一步（含时间点）。", "scope": { "domain": "medical", "tags": ["publish_ready", "access_update", "guideline_evidence"] }, "matchers": { "keywords_any": ["进院", "准入", "目录", "提单", "药事会", "指南", "共识", "收录", "推荐", "认可度", "证据"], "keywords_all": [], "negative_keywords": ["系统上线", "权限", "账号", "验收", "工单"] }, "recommendations": { "tone": { "style": "证据导向、严谨克制、尊重流程", "pace": "先确认证据门槛，再讲更新点，最后落到推进动作", "do": [ "先复述对方的证据门槛（如“需要指南/共识支持”）", "只引用注入资料中的指南/共识名称与推荐要点", "把推进动作写成流程内的最小一步（含时间点）" ], "dont": [ "编造指南名称/出版信息/推荐等级", "跳过风险边界或夸大认可度", "把流程说成“一句话就能解决”" ] }, "question_path": [ "对齐：当前推进卡在“证据”还是“流程环节”", "更新：说明新增的指南/共识收录信息（仅限资料）", "要点：用1-2句话讲清推荐的关键点位（仅限资料）", "落地：给出下一步推进动作（含时间点），如发送高亮原文/约定下次对齐提单材料" ], "best_practices": { "opening": [ "先感谢对方之前的判断标准与流程要求", "明确今天只更新一个信息点：指南/共识收录" ], "responding": [ "讲清“被收录/被推荐”的事实与要点，不延伸为更多结论", "若对方追问细节而资料不足，明确回去补齐并下次带来" ], "closing": [ "总结：证据门槛已补齐到哪一步", "提出下一步推进安排（含时间点）" ] }, "checks": { "must_include": [ "必须对齐证据门槛", "必须只引用资料内的指南/共识信息", "必须给出推进下一步（含时间点）" ], "must_avoid": [ "虚构指南/共识出版信息", "绝对化承诺或夸大业内认可", "用销售口吻逼推进" ] } } } FILE:references/strategy_packs/aesthetics_spec_intro.publish.strategy.json { "id": "medical.aesthetics.spec_intro.publish.v1", "version": "1.0.0", "name": "医美-规格介绍型发布级策略（防背书堆参数）", "description": "适用于医美/微整等场景的产品规格介绍与型号差异说明。重点控制数值出现频率与表达形态，避免清单式背诵。", "scope": { "domain": "medical", "tags": ["publish_ready", "aesthetics", "spec_intro"] }, "matchers": { "keywords_any": ["医美", "微整", "童颜", "规格", "型号", "两种", "HARD", "SOFT", "粒径", "μm", "mg", "注射层次"], "keywords_all": [], "negative_keywords": ["指南", "共识", "进院", "药事会", "血透", "处方DOT"] }, "recommendations": { "tone": { "style": "自然交流中的专业解释，结论先行", "pace": "先一句话讲清差异，再补1-2个关键点", "do": [ "先用一句话讲清：两个规格各自解决什么问题/适用场景", "每轮只放1-2个数值点（仅来自注入资料）", "把数值嵌入句子里，避免清单式罗列" ], "dont": [ "把全部mg/μm/成分一次性堆满", "用背书腔宣讲", "资料缺失时编造参数/对比" ] }, "question_path": [ "对齐：医生现在最想先弄清的是“两个规格怎么选”还是“各自参数是什么”", "解释：先一句话讲清两规格的定位差异（只用资料内口径）", "补充：按需补1-2个关键点位（如粒径/层次/规格），避免堆砌", "落地：给出一个轻量下一步（含时间点），如下次带规格卡/演示材料对齐" ], "best_practices": { "opening": [ "先承接医生“没时间细看”的现实，主动用一句话先讲清差异", "征得同意后再补关键点位" ], "responding": [ "优先回答“怎么选”，再补“是什么”", "若医生追问参数，用“我按资料把关键点位说清/其余我回去再核对”控制展开" ], "closing": [ "用一句话小结：两规格的定位差异", "约定下次带规格卡/资料对齐（含时间点）" ] }, "checks": { "must_include": [ "必须先给两规格的定位差异（一句话）", "出现数值时必须来自注入资料", "必须给下一步（含时间点）" ], "must_avoid": [ "清单式背诵参数", "无依据的数值/对比结论", "绝对化承诺" ] } } } FILE:references/strategy_packs/general.publish.strategy.json { "id": "general.publish.v1", "version": "1.0.0", "name": "通用发布级策略（跨行业）", "description": "当无法稳定识别行业/意图时，用此策略保证输出不空泛且可执行；不提供任何行业事实，只提供结构与行为层建议。", "scope": { "domain": "general", "tags": ["publish_ready", "fallback"] }, "matchers": { "keywords_any": ["场景", "目标", "沟通", "推进", "顾虑", "流程", "培训", "上线", "试用"], "keywords_all": [], "negative_keywords": [] }, "recommendations": { "tone": { "style": "务实、克制、以对齐事实与下一步为主", "pace": "先确认核心问题，再给可执行下一步", "do": [ "先复述对方立场与核心顾虑，避免直接反驳", "每轮只推进一个关键点，避免信息堆叠", "把'下一步'写成可执行安排（含时间点）" ], "dont": [ "背书式长段讲解", "在缺资料时输出数据/结论", "一次性给出过多并列行动项" ] }, "question_path": [ "先对齐：对方当前最关心的一个问题是什么（只选1个）", "澄清：影响决策/推进的关键门槛或约束是什么", "落地：下一步最小可行动作是什么（含时间点）" ], "best_practices": { "opening": [ "先征得同意，再说明本次想对齐的一个点", "用一句话复述对方现状与顾虑，确认理解一致" ], "responding": [ "先给一句话结论，再给1-2条理由（只基于已注入资料）", "不确定处明确说需核对，并承诺下次带回口径" ], "closing": [ "用一句话小结共识点", "提出一个明确的下一步安排（含时间点）" ] }, "checks": { "must_include": [ "核心顾虑必须收敛为1条", "必须给出下一步（含时间点）", "必须声明证据边界（无资料不编造）" ], "must_avoid": [ "无依据的量化数值", "无依据的政策/指南引用", "过度承诺/绝对化措辞" ] } } } FILE:references/strategy_packs/internal_system_rollout.publish.strategy.json { "id": "internal_system_rollout.publish.v1", "version": "1.0.0", "name": "公司系统推广/流程上线发布级策略", "description": "适用于公司内部系统推广、流程变更落地、培训与验收等场景。强调流程对齐、风险边界、验收口径与最小落地动作。", "scope": { "domain": "internal_system", "tags": ["publish_ready", "rollout", "process_change"] }, "matchers": { "keywords_any": ["系统", "上线", "推广", "培训", "流程", "权限", "账号", "数据录入", "验收", "口径", "排障", "工单"], "keywords_all": [], "negative_keywords": ["指南", "共识", "处方", "患者", "门诊", "住院部"] }, "recommendations": { "tone": { "style": "清晰、结构化、以对齐口径为主", "pace": "先确认目标与验收口径，再拆步骤与风险点", "do": [ "把目标写成可验收结果（可检查）", "把流程拆成3-5步，并给每步的成功判据", "优先指出高风险误区与回滚/兜底方案" ], "dont": [ "泛泛而谈的'建议优化'", "跳过权限/依赖导致的关键阻塞点", "一次性给出大量并列功能点" ] }, "question_path": [ "现状：现在卡在流程的哪一步（或哪个页面/哪个权限）", "口径：成功/验收的判定标准是什么", "步骤：最小可复现路径是什么（按步骤列出）", "风险：最常见的误区/失败原因是什么", "落地：下一步谁在什么时间点完成什么动作" ], "best_practices": { "opening": [ "先确认本次讨论目标（上线/验收/培训/排障）", "确认当前角色分工（谁负责配置/谁负责使用/谁负责验收）" ], "responding": [ "按步骤给出最小路径，并在每步给'成功判据'", "对不确定项明确列出需要补充的系统信息（截图/报错/环境）" ], "closing": [ "把下一步拆成1-2个可执行动作（含负责人+时间点）", "补一句风险提醒与兜底（必要时走工单/回滚）" ] }, "checks": { "must_include": [ "必须出现验收/成功判据", "必须出现最小可复现路径", "必须出现下一步动作（含时间点与负责人）" ], "must_avoid": [ "把未确认的系统行为当事实", "不解释依赖条件就给结论", "用'肯定/一定'做承诺" ] } } } FILE:references/strategy_packs/medical_clinical_conversation.publish.strategy.json { "id": "medical.clinical_conversation.publish.v1", "version": "1.0.0", "name": "医疗场景通用会话发布级策略", "description": "适用于医生/患者/代表等医疗相关沟通场景的通用发布级写作策略。强调合规表达、风险边界、循证口径与可落地下一步。", "scope": { "domain": "medical", "tags": ["publish_ready", "clinical"] }, "matchers": { "keywords_any": ["门诊", "住院部", "病区", "查房", "处方", "指南", "共识", "不良反应", "剂量", "随访", "科室会", "进院"], "keywords_all": [], "negative_keywords": ["系统上线", "权限", "账号", "验收", "工单"] }, "recommendations": { "tone": { "style": "专业、克制、尊重临床判断", "pace": "先对齐核心顾虑，再给结论与边界，最后落到下一步", "do": [ "先复述医生/对方的核心顾虑，避免直接反驳", "先给一句话结论，再给1-2条依据（仅限已注入资料）", "主动补充风险与边界（不作绝对化承诺）", "把推进写成轻量、可执行的下一步（含时间点）" ], "dont": [ "背书式长段讲解或堆术语", "缺资料时输出数值/研究结论/对比结论", "强推处方/强逼单式推进" ] }, "question_path": [ "对齐：对方当前最关心的一个问题是什么（只选1个）", "澄清：证据门槛/风险边界/适用人群的关键约束是什么", "回应：给一句话结论 + 1-2条依据（仅限已注入资料）", "落地：提出一个轻量下一步（含时间点），并约定反馈闭环" ], "best_practices": { "opening": [ "征得同意，说明本次想对齐的一个点（不铺垫过长）", "复述对方顾虑并确认理解一致" ], "responding": [ "有资料就引用原句语义；没资料就明确需要核对口径", "在价值点之后补一句风险/边界（不夸大、不绝对）" ], "closing": [ "用一句话小结共识点", "提出下一步安排（含时间点）并说明将带回/同步的资料或反馈" ] }, "checks": { "must_include": [ "核心顾虑必须收敛为1条", "必须出现风险与边界", "必须给出下一步（含时间点）" ], "must_avoid": [ "无依据的量化数值", "无依据的研究/指南/政策引用", "绝对化承诺（如100%有效/绝对安全）" ] } } } FILE:references/strategy_packs/pediatrics_prevention_mindset.publish.strategy.json { "id": "medical.pediatrics.prevention_mindset.publish.v1", "version": "1.0.0", "name": "儿科/住院部值班-预防观念转变发布级策略", "description": "适用于儿科/住院部值班/时间相对宽松的交流场景，将'无需干预/可自愈'的观念转向'高风险+防大于治'，并落到轻量试用与反馈闭环。", "scope": { "domain": "medical", "tags": ["publish_ready", "pediatrics", "prevention_mindset"] }, "matchers": { "keywords_any": ["儿科", "住院部", "值班", "离出诊", "高风险", "防大于治", "预防", "无需干预", "自行疗愈", "抗生素", "AAD"], "keywords_all": [], "negative_keywords": ["药事会", "进院", "目录", "验收", "工单"] }, "recommendations": { "tone": { "style": "更轻松但仍专业，先共情后对齐观念", "pace": "先认同可自愈的前提，再补充高风险与预防价值，最后落到轻量动作", "do": [ "先承认对方观点的合理前提（轻型可自愈），再转到高风险/重型风险（仅限资料口径）", "把预防方案说成“在某类患者/某个时点”的轻量加用（仅限资料）", "明确建立反馈闭环（下周固定收反馈/带卡片），含时间点" ], "dont": [ "恐吓式夸大风险", "堆砌循证引用（除非资料已注入且对方需要）", "一上来就要求全科普遍改变处方" ] }, "question_path": [ "对齐：你更担心的是“确实不需要干预”还是“担心过度用药”", "转观念：承认轻型可自愈的前提→补充高风险与预防意义（仅限资料）", "方案：明确何时加用/怎么用（仅限资料）", "落地：先选一类高风险患儿试行，并约定下次固定时间收反馈" ], "best_practices": { "opening": [ "用轻量关怀开场，确认现在是否方便聊几分钟", "把话题锚定在一个点：预防理念/高风险人群" ], "responding": [ "先一句话讲清“为什么要预防”，再补1-2条资料内依据", "用药与边界说清楚，不鼓励超资料范围" ], "closing": [ "提出试行建议（小范围）", "约定下次回访时间点并说明会带的资料/卡片" ] }, "checks": { "must_include": [ "必须先共情并承认对方观点的合理前提", "必须落到轻量试行+反馈闭环（含时间点）", "任何循证/共识引用必须来自注入资料" ], "must_avoid": [ "恐吓式表达", "强推全科改变习惯", "无依据的数值或结论" ] } } } FILE:references/strategy_packs/strategy.schema.json { "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "https://openclaw.local/scenario_builder/strategy.schema.json", "title": "Scenario Builder Strategy Pack", "type": "object", "required": ["id", "version", "name", "scope", "matchers", "recommendations"], "properties": { "id": { "type": "string", "description": "Stable id, e.g. 'general.publish.v1' or 'med.pharma.access_update.v1'" }, "version": { "type": "string", "description": "Semver-like, e.g. '1.0.0'" }, "name": { "type": "string" }, "description": { "type": "string" }, "scope": { "type": "object", "required": ["domain"], "properties": { "domain": { "type": "string", "description": "Domain tag, e.g. 'general', 'medical', 'internal_system'" }, "tags": { "type": "array", "items": { "type": "string" } } }, "additionalProperties": false }, "matchers": { "type": "object", "required": ["keywords_any", "keywords_all", "negative_keywords"], "properties": { "keywords_any": { "type": "array", "items": { "type": "string" } }, "keywords_all": { "type": "array", "items": { "type": "string" } }, "negative_keywords": { "type": "array", "items": { "type": "string" } } }, "additionalProperties": false }, "recommendations": { "type": "object", "required": ["tone", "question_path", "best_practices", "checks"], "properties": { "tone": { "type": "object", "required": ["style", "pace"], "properties": { "style": { "type": "string" }, "pace": { "type": "string" }, "do": { "type": "array", "items": { "type": "string" } }, "dont": { "type": "array", "items": { "type": "string" } } }, "additionalProperties": false }, "question_path": { "type": "array", "description": "Ordered path steps (strings), agent should adapt wording but keep intent.", "items": { "type": "string" }, "minItems": 2 }, "best_practices": { "type": "object", "required": ["opening", "responding", "closing"], "properties": { "opening": { "type": "array", "items": { "type": "string" } }, "responding": { "type": "array", "items": { "type": "string" } }, "closing": { "type": "array", "items": { "type": "string" } } }, "additionalProperties": false }, "checks": { "type": "object", "required": ["must_include", "must_avoid"], "properties": { "must_include": { "type": "array", "items": { "type": "string" } }, "must_avoid": { "type": "array", "items": { "type": "string" } } }, "additionalProperties": false } }, "additionalProperties": false } }, "additionalProperties": false } FILE:scripts/README.md # scripts — 目录索引本目录存放可执行脚本与内部复用工具。 ## 模块 - `scene/` - 入口：`scene/README.md` - 包含场景链路脚本与落库执行脚本 - `common/` - 入口：`common/README.md` - 包含 TOON 编码与鉴权 token 解析等共享工具 - `tbs_assets/` - 入口：`tbs_assets/README.md` - 存放草稿、主数据字典与本地运行资产 ## 执行原则 1. 先读 `openapi/scene/api-index.md` 与对应 endpoint 文档 2. 再执行 `scene/*.py` 3. 所有需要鉴权的步骤依赖 `XG_USER_TOKEN` FILE:scripts/common/README.md # 脚本清单 — common 该目录承载 Skill 内部复用工具，不对应外部业务 endpoint。 ## 共享依赖无 ## 脚本列表 | 脚本 | 对应接口 | 用途 | |---|---|---| | `toon_encoder.py` | （内部工具） | 将结构化 JSON 压缩为 TOON 字符串，用于脚本契约输出 | | `auth_token.py` | （内部工具） | 从环境变量 `XG_USER_TOKEN` 解析 access-token（`strip` 后使用） | ## 使用方式该工具通常由各脚本内部 `import` 使用，不建议直接面向用户运行。 FILE:scripts/common/auth_token.py #!/usr/bin/env python3 """ Shared access-token resolver for scenario-builder scripts. Aligned with cms-tbs-training: token is sourced from env `XG_USER_TOKEN`. """ import os import sys def resolve_access_token(): token = (os.environ.get("XG_USER_TOKEN") or "").strip() if token: return token, "env:XG_USER_TOKEN" return None, None def require_access_token(): token, _ = resolve_access_token() if not token: print("错误: 请设置环境变量 XG_USER_TOKEN", file=sys.stderr) raise SystemExit(1) return token FILE:scripts/common/toon_encoder.py """ TOON (Token-Oriented Object Notation) Encoder A zero-dependency Python implementation fully compliant with TOON spec v3.0. 【核心用途解说】此序列化引擎的本质，是专门建立一条将臃肿的 JSON（或多层嵌套的 Python dict/list）结构，转化为面向人工大语言模型（LLM）的高密度浓缩协议通道。它的核心目标聚焦于【断崖式的削减大模型 Token 损耗】。依靠识别与动态压缩统一表头（类似 CSV 表格内联提取），结合 YAML 的层级树特质，该模块能在确保数据上下文明义 100% 不受损的前提下，精简掉所有的废弃闭合括号及引号符号，平均可帮你的 API 系统为单个语境输入节省极大量的 Token 从而实现极致降本提效。 """ import re import json import math from datetime import datetime, date from typing import Any, Iterator, Tuple, List # Validation patterns matching TS implementation NUMERIC_LIKE_PATTERN = re.compile(r'^-?\d+(?:\.\d+)?(?:e[+-]?\d+)?$', re.IGNORECASE) LEADING_ZERO_PATTERN = re.compile(r'^0\d+$') VALID_UNQUOTED_KEY_PATTERN = re.compile(r'^[A-Za-z_][\w.]*$') def _is_valid_unquoted_key(key: str) -> bool: """Checks if a key can be safely used without quotes.""" return bool(VALID_UNQUOTED_KEY_PATTERN.match(key)) def _is_safe_unquoted(value: str, delimiter: str) -> bool: """Determines if a string value can be safely encoded without quotes.""" if not value or value != value.strip(): return False val_lower = value.lower() if val_lower in ('true', 'false', 'null'): return False if NUMERIC_LIKE_PATTERN.match(value) or LEADING_ZERO_PATTERN.match(value): return False if ':' in value or '"' in value or '\\' in value: return False if any(ch in value for ch in ('[', ']', '{', '}')): return False if any(ch in value for ch in ('\n', '\r', '\t')): return False if delimiter in value: return False if value.startswith('-'): return False return True def _escape_string(val: str) -> str: """ Safely escapes string matching JSON specification exactly (e.g. control characters). Uses standard library json.dumps to stringify and slices off the bounding quotes. """ return json.dumps(val, ensure_ascii=False)[1:-1] def normalize_value(value: Any, strip_html_style: bool = False) -> Any: """ Normalizes complex Python types into strict JSON equivalents (Primitives, Lists, Dicts, None). Matches the TypeScript implementation's robust tracking for edge cases like NaNs and Sets. """ if value is None: return None if isinstance(value, str): if strip_html_style: # Safely remove style="..." or style='...' attributes from HTML tags to save tokens value = re.sub(r'(?i)\s*style\s*=\s*(["\']).*?\1', '', value) return value if isinstance(value, bool): return value if isinstance(value, (int, float)): if isinstance(value, float): # NaN and Infinity map to null in standard JSON if math.isnan(value) or math.isinf(value): return None return value if isinstance(value, (datetime, date)): return value.isoformat() if isinstance(value, (list, tuple, set, frozenset)): return [normalize_value(v, strip_html_style) for v in value] if isinstance(value, dict): return {str(k): normalize_value(v, strip_html_style) for k, v in value.items()} # Graceful fallback for custom objects implementing a toJSON serialization target hook if hasattr(value, 'toJSON') and callable(value.toJSON): return normalize_value(value.toJSON(), strip_html_style) if hasattr(value, 'to_json') and callable(value.to_json): return normalize_value(value.to_json(), strip_html_style) # Silent fallback for unreadable items like functions, mimicking JSON's drop behavior return None def _is_primitive(val: Any) -> bool: return val is None or isinstance(val, (bool, int, float, str)) def _encode_primitive(val: Any, delimiter: str) -> str: if val is None: return 'null' if isinstance(val, bool): return 'true' if val else 'false' if isinstance(val, (int, float)): return str(val) if isinstance(val, str): if _is_safe_unquoted(val, delimiter): return val return f'"{_escape_string(val)}"' return str(val) def _encode_key(key: Any) -> str: k_str = str(key) if _is_valid_unquoted_key(k_str): return k_str return f'"{_escape_string(k_str)}"' def _is_tabular_array(arr: List[Any]) -> Tuple[bool, List[str]]: if not arr or not isinstance(arr[0], dict): return False, [] first_keys = list(arr[0].keys()) if not first_keys: return False, [] for item in arr: if not isinstance(item, dict): return False, [] if len(item) != len(first_keys): return False, [] for k in first_keys: if k not in item or not _is_primitive(item[k]): return False, [] return True, first_keys def _indent_line(depth: int, content: str, indent_size: int) -> str: return (' ' * (depth * indent_size)) + content def _format_header(length: int, key=None, fields=None, delimiter: str = ',') -> str: header = "" if key is not None: header += _encode_key(key) header += f"[{length}" if delimiter != ',': header += delimiter header += "]" if fields: enc_fields = [_encode_key(f) for f in fields] header += f"{{{delimiter.join(enc_fields)}}}" header += ":" return header def _encode_dict(obj: dict, depth: int, indent_size: int, delimiter: str) -> Iterator[str]: for k, v in obj.items(): yield from _encode_kv(k, v, depth, indent_size, delimiter) def _encode_kv(key: Any, value: Any, depth: int, indent_size: int, delimiter: str) -> Iterator[str]: enc_k = _encode_key(key) if _is_primitive(value): yield _indent_line(depth, f"{enc_k}: {_encode_primitive(value, delimiter)}", indent_size) elif isinstance(value, list): if len(value) == 0: yield _indent_line(depth, _format_header(0, key=key, delimiter=delimiter), indent_size) elif all(_is_primitive(v) for v in value): joined = delimiter.join(_encode_primitive(v, delimiter) for v in value) yield _indent_line(depth, f"{_format_header(len(value), key=key, delimiter=delimiter)} {joined}", indent_size) else: is_tabular, headers = _is_tabular_array(value) if is_tabular: yield _indent_line(depth, _format_header(len(value), key=key, fields=headers, delimiter=delimiter), indent_size) for item in value: joined = delimiter.join(_encode_primitive(item[h], delimiter) for h in headers) yield _indent_line(depth + 1, joined, indent_size) else: yield _indent_line(depth, _format_header(len(value), key=key, delimiter=delimiter), indent_size) for item in value: yield from _encode_list_item(item, depth + 1, indent_size, delimiter) elif isinstance(value, dict): yield _indent_line(depth, f"{enc_k}:", indent_size) if value: yield from _encode_dict(value, depth + 1, indent_size, delimiter) def _encode_list_item(value: Any, depth: int, indent_size: int, delimiter: str) -> Iterator[str]: if _is_primitive(value): yield _indent_line(depth, f"- {_encode_primitive(value, delimiter)}", indent_size) elif isinstance(value, list): if all(_is_primitive(v) for v in value): joined = delimiter.join(_encode_primitive(v, delimiter) for v in value) yield _indent_line(depth, f"- {_format_header(len(value), delimiter=delimiter)} {joined}", indent_size) else: yield _indent_line(depth, f"- {_format_header(len(value), delimiter=delimiter)}", indent_size) for item in value: yield from _encode_list_item(item, depth + 1, indent_size, delimiter) elif isinstance(value, dict): if not value: yield _indent_line(depth, "- ", indent_size) return entries = list(value.items()) first_k, first_v = entries[0] enc_first_k = _encode_key(first_k) if isinstance(first_v, list) and len(first_v) > 0: tabular, headers = _is_tabular_array(first_v) if tabular: yield _indent_line(depth, f"- {_format_header(len(first_v), key=first_k, fields=headers, delimiter=delimiter)}", indent_size) for item in first_v: joined = delimiter.join(_encode_primitive(item[h], delimiter) for h in headers) yield _indent_line(depth + 2, joined, indent_size) if len(entries) > 1: rest_dict = dict(entries[1:]) yield from _encode_dict(rest_dict, depth + 1, indent_size, delimiter) return if _is_primitive(first_v): yield _indent_line(depth, f"- {enc_first_k}: {_encode_primitive(first_v, delimiter)}", indent_size) elif isinstance(first_v, list): if len(first_v) == 0: yield _indent_line(depth, f"- {enc_first_k}{_format_header(0, delimiter=delimiter)}", indent_size) elif all(_is_primitive(v) for v in first_v): joined = delimiter.join(_encode_primitive(v, delimiter) for v in first_v) yield _indent_line(depth, f"- {enc_first_k}{_format_header(len(first_v), delimiter=delimiter)} {joined}", indent_size) else: yield _indent_line(depth, f"- {enc_first_k}{_format_header(len(first_v), delimiter=delimiter)}", indent_size) for item in first_v: yield from _encode_list_item(item, depth + 2, indent_size, delimiter) elif isinstance(first_v, dict): yield _indent_line(depth, f"- {enc_first_k}:", indent_size) if first_v: yield from _encode_dict(first_v, depth + 2, indent_size, delimiter) if len(entries) > 1: rest_dict = dict(entries[1:]) yield from _encode_dict(rest_dict, depth + 1, indent_size, delimiter) def encode_lines(data: Any, indent: int = 2, delimiter: str = ',', strip_html_style: bool = False) -> Iterator[str]: """ Core generator returning lines instead of full string. Suitable for streaming large inputs. """ # 1. Normalize strictly to JSON limits matching TS behavior data = normalize_value(data, strip_html_style) if _is_primitive(data): enc_val = _encode_primitive(data, delimiter) if enc_val: yield enc_val elif isinstance(data, list): if not data: yield _format_header(0, delimiter=delimiter) elif all(_is_primitive(v) for v in data): joined = delimiter.join(_encode_primitive(v, delimiter) for v in data) yield f"{_format_header(len(data), delimiter=delimiter)} {joined}" else: is_tabular, headers = _is_tabular_array(data) if is_tabular: yield _format_header(len(data), fields=headers, delimiter=delimiter) for item in data: joined = delimiter.join(_encode_primitive(item[h], delimiter) for h in headers) yield _indent_line(1, joined, indent) else: yield _format_header(len(data), delimiter=delimiter) for item in data: yield from _encode_list_item(item, 1, indent, delimiter) elif isinstance(data, dict): yield from _encode_dict(data, 0, indent, delimiter) def encode(data: Any, indent: int = 2, delimiter: str = ',', strip_html_style: bool = False) -> str: """ Encodes a Python value (dict, list, primitive) into a TOON formatted string. 【主入口方法解说】接收包含复杂嵌套结构的 Python 变量体系（如原生的 json 解析对象）。通过内部深度优先的生成器矩阵和结构对齐，将其最终吐出为能够直接硬编码嵌入给 GPT/Claude/Gemini 等大模型 Prompt 阅读的 TOON 语境化压缩多行字符实体。此方法正是所有后续省 Token 转换流的总开关。【兼容性处理】如果传入的数据已经是字符串且不是明显的 JSON 结构（Dict 或 List），则原样返回，确保接口在高频调用下的稳健性。 """ if isinstance(data, str) and not (data.strip().startswith('{') or data.strip().startswith('[')): return data return '\n'.join(encode_lines(data, indent, delimiter, strip_html_style)) FILE:scripts/scene/README.md # 脚本清单 — scene ## 共享依赖 - `../common/toon_encoder.py` — TOON 编码器；标准输出必须经过 `toon_encode()`。 ## 脚本列表 | 脚本 | 逻辑端点 | 用途 | |---|---|---| | `route-by-intent.py` | `POST .../route-by-intent` | 校验路由契约并输出 TOON 摘要 | | `parse-and-gap-ask.py` | `POST .../parse-and-gap-ask` | 校验解析草案字段契约 | | `publish-ready-compose.py` | `POST .../publish-ready-compose` | 校验 publish_ready 前置契约 | | `build-persona.py` | `POST .../build-persona` | 校验 persona 相关子树存在性 | | `build-prompts.py` | `POST .../build-prompts` | 校验 promptBundle 四键占位 | | `build-api-draft-dedup.py` | `POST .../build-api-draft-dedup` | 校验 apiDraft + 去重证据形状 | | `validate-and-gate.py` | `POST .../validate-and-gate` | 校验终检输入非空 | | `preflight-tbs-master-data.py` | `POST .../preflight-tbs-master-data` | 确认落库前：对 TBS 业务领域/科室/药品 GET 匹配，无则 POST（与 executor 共用 `tbs_master_data_resolve.py`） | | `persist-and-execute.py` | `POST .../persist-and-execute` | 写草稿并子进程执行 `tbs_write_executor.py` | | `enforce-draft-text.py` | （校验工具） | 校验草稿文本字段完整性与关键路径非空，不做文本重建/兜底 | | `tbs_master_data_resolve.py` | （共享模块） | `TBSClient`、`resolve_ids_for_scene`；供 preflight 与 executor 使用 | | `tbs_write_executor.py` | （HTTP 落库，无逻辑端点 URL） | 调用 TBS Admin API；品种经 `GET/POST .../drugs` 查或建后再创建场景 | | `scenario_pack_normalizer.py` | （共享模块） | 统一新旧 `scenarioPack` 字段映射，避免生成链路断裂 | ## 使用方式 ```bash export XG_USER_TOKEN="your-access-token" # 从 stdin 传入 JSON 契约 python3 scripts/scene/parse-and-gap-ask.py < payload.json ``` ## 输出说明所有脚本标准输出均为 **TOON**（非原始 JSON）。 ## 规范 1. **Python 3** 2. **必须经过 toon_encoder** 3. **鉴权**：`XG_USER_TOKEN` 缺失则退出码 1（鉴权约定见 `cms-auth-skills/common/auth.md`） 4. **入参**：以 `openapi/scene/*.md` 为准 FILE:scripts/scene/build-api-draft-dedup.py #!/usr/bin/env python3 """ scene / build-api-draft-dedup — 契约校验脚本（stdin JSON）。输出 TOON。 API_URL 须与 openapi/scene/build-api-draft-dedup.md 标题 URL 一致；本脚本不发起 HTTP。 """ import json import os import re import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common")) from auth_token import require_access_token from toon_encoder import encode as toon_encode from scenario_pack_normalizer import normalize_scenario_pack API_URL = "https://scenario-builder.openclaw.internal/v1/scene/build-api-draft-dedup" ALLOWED_EVIDENCE_STATUS = {"NOT_PROVIDED", "PARTIAL", "READY"} def _read_body(): raw = sys.stdin.read() if not raw.strip(): return {} return json.loads(raw) def _ok(step, **extra): payload = {"ok": True, "step": step, **extra} print(toon_encode(payload)) def _contains_blocked_fact_claims(node, blocked_keys): if isinstance(node, dict): for key, value in node.items(): if key in blocked_keys: return True if _contains_blocked_fact_claims(value, blocked_keys): return True elif isinstance(node, list): for item in node: if _contains_blocked_fact_claims(item, blocked_keys): return True return False def _contains_blocked_fact_phrases(node, blocked_phrases): if isinstance(node, str): text = re.sub(r"\s+", "", node) return any(p in text for p in blocked_phrases) if isinstance(node, dict): return any(_contains_blocked_fact_phrases(v, blocked_phrases) for v in node.values()) if isinstance(node, list): return any(_contains_blocked_fact_phrases(v, blocked_phrases) for v in node) return False def main(): require_access_token() body = _read_body() sp = body.get("scenarioPack") api_draft = body.get("apiDraft") policy = body.get("factGuardPolicy") if not isinstance(sp, dict) or not isinstance(api_draft, dict): print("错误: scenarioPack 与 apiDraft 必填", file=sys.stderr) sys.exit(2) if not isinstance(policy, dict): print("错误: 缺少 factGuardPolicy（必须由上游参数传入）", file=sys.stderr) sys.exit(2) blocked_keys = policy.get("blockedFactKeys") blocked_phrases = policy.get("blockedPhrases") if not isinstance(blocked_keys, list) or not all(isinstance(x, str) for x in blocked_keys): print("错误: factGuardPolicy.blockedFactKeys 必须为字符串数组", file=sys.stderr) sys.exit(2) if not isinstance(blocked_phrases, list) or not all(isinstance(x, str) for x in blocked_phrases): print("错误: factGuardPolicy.blockedPhrases 必须为字符串数组", file=sys.stderr) sys.exit(2) sp = normalize_scenario_pack(sp) evidence_status = sp.get("productEvidenceStatus", "NOT_PROVIDED") if evidence_status not in ALLOWED_EVIDENCE_STATUS: print("错误: productEvidenceStatus 非法", file=sys.stderr) sys.exit(2) if evidence_status != "READY": if _contains_blocked_fact_claims(api_draft, set(blocked_keys)): print("错误: 证据未 READY，apiDraft 不得包含事实性结论字段", file=sys.stderr) sys.exit(2) if _contains_blocked_fact_phrases(api_draft, blocked_phrases): print("错误: 证据未 READY，apiDraft 不得包含结论性事实表述", file=sys.stderr) sys.exit(2) confirmation_marked = bool(api_draft.get("needsEvidenceConfirmation")) if not confirmation_marked: print("错误: 证据未 READY 时，apiDraft 需标记 needsEvidenceConfirmation=true", file=sys.stderr) sys.exit(2) _ok( "build-api-draft-dedup", evidenceStatus=evidence_status, constrainedGeneration=evidence_status != "READY", ) if __name__ == "__main__": main() FILE:scripts/scene/build-persona.py #!/usr/bin/env python3 """ scene / build-persona — 契约校验脚本（stdin JSON）。输出 TOON。 API_URL 须与 openapi/scene/build-persona.md 标题 URL 一致；本脚本不发起 HTTP。 """ import json import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common")) from auth_token import require_access_token from toon_encoder import encode as toon_encode from scenario_pack_normalizer import normalize_scenario_pack API_URL = "https://scenario-builder.openclaw.internal/v1/scene/build-persona" ALLOWED_EVIDENCE_STATUS = {"NOT_PROVIDED", "PARTIAL", "READY"} def _read_body(): raw = sys.stdin.read() if not raw.strip(): return {} return json.loads(raw) def _ok(step, **extra): payload = {"ok": True, "step": step, **extra} print(toon_encode(payload)) def _missing_core_fields(scenario_pack: dict) -> list: required = ["businessDomain", "department", "product", "location", "repGoal"] return [k for k in required if not scenario_pack.get(k)] def main(): require_access_token() body = _read_body() sp = body.get("scenarioPack") if not isinstance(sp, dict): print("错误: scenarioPack 必填", file=sys.stderr) sys.exit(2) sp = normalize_scenario_pack(sp) missing_core = _missing_core_fields(sp) if missing_core: print(f"错误: scenarioPack 缺少核心字段: {', '.join(missing_core)}", file=sys.stderr) sys.exit(2) evidence_status = sp.get("productEvidenceStatus", "NOT_PROVIDED") if evidence_status not in ALLOWED_EVIDENCE_STATUS: print("错误: productEvidenceStatus 非法", file=sys.stderr) sys.exit(2) evidence_sources = sp.get("productEvidenceSource") or [] if evidence_status == "READY" and not evidence_sources: print("错误: productEvidenceStatus=READY 时需提供 productEvidenceSource", file=sys.stderr) sys.exit(2) _ok( "build-persona", evidenceStatus=evidence_status, constrainedGeneration=evidence_status != "READY", ) if __name__ == "__main__": main() FILE:scripts/scene/build-prompts.py #!/usr/bin/env python3 """ scene / build-prompts — 契约校验脚本（stdin JSON）。输出 TOON。 API_URL 须与 openapi/scene/build-prompts.md 标题 URL 一致；本脚本不发起 HTTP。 """ import json import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common")) from auth_token import require_access_token from toon_encoder import encode as toon_encode from scenario_pack_normalizer import normalize_scenario_pack API_URL = "https://scenario-builder.openclaw.internal/v1/scene/build-prompts" ALLOWED_EVIDENCE_STATUS = {"NOT_PROVIDED", "PARTIAL", "READY"} def _read_body(): raw = sys.stdin.read() if not raw.strip(): return {} return json.loads(raw) def _ok(step, **extra): payload = {"ok": True, "step": step, **extra} print(toon_encode(payload)) def _missing_core_fields(scenario_pack: dict) -> list: required = ["businessDomain", "department", "product", "location", "repGoal"] return [k for k in required if not scenario_pack.get(k)] def main(): require_access_token() body = _read_body() sp = body.get("scenarioPack") if not isinstance(sp, dict): print("错误: scenarioPack 必填", file=sys.stderr) sys.exit(2) sp = normalize_scenario_pack(sp) missing_core = _missing_core_fields(sp) if missing_core: print(f"错误: scenarioPack 缺少核心字段: {', '.join(missing_core)}", file=sys.stderr) sys.exit(2) has_persona = any(k in sp for k in ("personaConfig", "personaBase", "personaOverlay")) if not has_persona: print("错误: 需先具备 persona 产物（personaConfig/personaBase/personaOverlay）", file=sys.stderr) sys.exit(2) evidence_status = sp.get("productEvidenceStatus", "NOT_PROVIDED") if evidence_status not in ALLOWED_EVIDENCE_STATUS: print("错误: productEvidenceStatus 非法", file=sys.stderr) sys.exit(2) _ok( "build-prompts", evidenceStatus=evidence_status, constrainedGeneration=evidence_status != "READY", ) if __name__ == "__main__": main() FILE:scripts/scene/enforce-draft-text.py #!/usr/bin/env python3 """ enforce-draft-text — 仅做校验，不做任何文本重建/兜底。 """ from __future__ import annotations import argparse import json import os import sys sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "common")) from toon_encoder import encode as toon_encode # type: ignore def _normalize_scene_doctors(api_draft: dict): doctors = api_draft.get("doctors") or {} if isinstance(doctors, list): doctors = doctors[0] if doctors else {} scenes = api_draft.get("scenes") or {} if isinstance(scenes, list): scenes = scenes[0] if scenes else {} return doctors, scenes def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--input", default=os.path.join("scripts", "tbs_assets", "scenario_draft.json")) ap.add_argument("--write-back", action="store_true", help="Rewrite input JSON in-place.") args = ap.parse_args() input_path = args.input if not os.path.isfile(input_path): raise SystemExit(f"Missing input file: {input_path}") # Import quality check helpers only; no generation fallback. sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from tbs_write_executor import ( # type: ignore _doctor_only_context_quality_ok, _coach_only_context_quality_ok, _coach_only_context_matches_scene_type, ) with open(input_path, "r", encoding="utf-8") as f: draft = json.load(f) scenario_pack = draft.get("scenarioPack") or {} api_draft = draft.get("apiDraft") or {} api_draft_doctors, api_draft_scenes = _normalize_scene_doctors(api_draft) if not isinstance(api_draft_scenes, dict): raise SystemExit("Invalid draft shape: apiDraft.scenes must be an object (or a list with first element as object).") rep_briefing_old = api_draft_scenes.get("rep_briefing") or "" location = scenario_pack.get("location") department = scenario_pack.get("department") product = scenario_pack.get("product") rep_briefing_invalid = ( (not isinstance(rep_briefing_old, str)) or (not rep_briefing_old.strip()) or ("【" in rep_briefing_old or "】" in rep_briefing_old) or "待补充" in rep_briefing_old or (len(rep_briefing_old) > 180) or (location and str(location) not in rep_briefing_old) or (department and str(department) not in rep_briefing_old) or (product and str(product) not in rep_briefing_old) ) doc_ctx_old = api_draft_scenes.get("doctor_only_context") or "" coach_ctx_old = api_draft_scenes.get("coach_only_context") or "" doctor_ctx_invalid = (not isinstance(doc_ctx_old, str)) or (not _doctor_only_context_quality_ok(doc_ctx_old)) coach_ctx_invalid = (not isinstance(coach_ctx_old, str)) or (not _coach_only_context_quality_ok(coach_ctx_old)) or (not _coach_only_context_matches_scene_type(scenario_pack, coach_ctx_old)) issues = [] if rep_briefing_invalid: issues.append("rep_briefing_invalid") if doctor_ctx_invalid: issues.append("doctor_only_context_invalid") if coach_ctx_invalid: issues.append("coach_only_context_invalid") if args.write_back: raise SystemExit("禁止 write-back：本脚本仅校验，不做文本重建。") out = { "input": input_path, "valid": len(issues) == 0, "issues": issues, } print(toon_encode(out)) if __name__ == "__main__": main() FILE:scripts/scene/parse-and-gap-ask.py #!/usr/bin/env python3 """ scene / parse-and-gap-ask — 契约校验脚本（stdin JSON）。输出 TOON。 API_URL 须与 openapi/scene/parse-and-gap-ask.md 标题 URL 一致；本脚本不发起 HTTP。 """ import json import os import re import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common")) from auth_token import require_access_token from toon_encoder import encode as toon_encode from scenario_pack_normalizer import normalize_scenario_pack API_URL = "https://scenario-builder.openclaw.internal/v1/scene/parse-and-gap-ask" API_SOURCE_PREFIXES = tuple( p.strip() for p in os.environ.get("TBS_KB_SOURCE_PREFIXES", "api://,https://").split(",") if p.strip() ) REQUIRE_KB_API = os.environ.get("TBS_REQUIRE_KB_API", "1") == "1" KNOWLEDGE_API_PATH = os.environ.get("TBS_KNOWLEDGE_API_PATH", "/api/v1/admin/basic/knowledge") _TOKEN_PAT = re.compile(r"[\s\-_:/（）()，,。；;、]+") def _read_body(): raw = sys.stdin.read() if not raw.strip(): return {} return json.loads(raw) def _ok(step, **extra): payload = {"ok": True, "step": step, **extra} print(toon_encode(payload)) def _norm_text(text: str) -> str: if not isinstance(text, str): return "" return _TOKEN_PAT.sub("", text).lower() def _char_ngrams(text: str, min_n: int = 2, max_n: int = 4) -> set[str]: norm = _norm_text(text) if not norm: return set() grams = set() length = len(norm) for n in range(min_n, max_n + 1): if length < n: continue for i in range(length - n + 1): grams.add(norm[i:i + n]) return grams def _iter_hit_phrases(hit: dict): for key in ("title", "category", "content", "summary"): v = hit.get(key) if isinstance(v, str) and v.strip(): yield v.strip() for key in ("tags", "keywords", "coveredNeeds"): arr = hit.get(key) if not isinstance(arr, list): continue for item in arr: if isinstance(item, str) and item.strip(): yield item.strip() def _need_matches_hit(need: str, hit_texts: list[str]) -> bool: need_norm = _norm_text(need) if not need_norm: return False for t in hit_texts: t_norm = _norm_text(t) if not t_norm: continue if need_norm in t_norm or t_norm in need_norm: return True # 完全数据驱动：按字符 n-gram 相似度做弱语义匹配，不依赖硬编码业务词。 need_grams = _char_ngrams(need_norm) if not need_grams: return False for t in hit_texts: hit_grams = _char_ngrams(t) if not hit_grams: continue overlap = len(need_grams & hit_grams) union = len(need_grams | hit_grams) if union == 0: continue jaccard = overlap / union if jaccard >= 0.30: return True return False def _extract_product(body: dict) -> tuple[str | None, list]: # 1) explicit candidates from caller candidates = [] for item in body.get("productCandidates") or []: if isinstance(item, str) and item.strip(): candidates.append(item.strip()) elif isinstance(item, dict): name = item.get("name") if isinstance(name, str) and name.strip(): candidates.append(name.strip()) # 2) candidates from knowledge API hits (optional) ksr = body.get("knowledgeSearchResult") if isinstance(ksr, dict): for hit in ksr.get("hits") or []: if not isinstance(hit, dict): continue pn = hit.get("productName") if isinstance(pn, str) and pn.strip(): candidates.append(pn.strip()) # de-dup keep order uniq = [] seen = set() for c in candidates: if c in seen: continue seen.add(c) uniq.append(c) if uniq: # fallback choose first candidate return uniq[0], [{"value": uniq[0], "confidence": 0.75, "source": "param_or_api"}] return None, [] def _extract_fields(body: dict) -> dict: pack = {} candidates = {} parsed_fields = body.get("parsedFields") if isinstance(parsed_fields, dict): pack.update(parsed_fields) product, product_candidates = _extract_product(body) if product: pack["product"] = product candidates["product"] = product_candidates if candidates: pack["candidates"] = candidates return pack def _coverage_from_search_result(search_result: dict, needs: list) -> tuple[str, list, list, list]: if not isinstance(search_result, dict): return "NOT_PROVIDED", [], [], list(needs) source_api = search_result.get("sourceApi") if REQUIRE_KB_API and source_api != KNOWLEDGE_API_PATH: return "NOT_PROVIDED", [], [], list(needs) hits = search_result.get("hits") if not isinstance(hits, list): return "NOT_PROVIDED", [], [], list(needs) covered = set() sources = [] need_set = [n for n in needs if isinstance(n, str) and n.strip()] for h in hits: if not isinstance(h, dict): continue src = h.get("source") if not isinstance(src, str) or not src: continue if not src.startswith(API_SOURCE_PREFIXES): continue sources.append(src) score = h.get("score") if isinstance(score, (int, float)) and score < 0.5: continue hit_texts = list(_iter_hit_phrases(h)) # 先保留显式 coveredNeeds（若有）。 for item in h.get("coveredNeeds", []): if isinstance(item, str) and item.strip(): covered.add(item.strip()) # 再做语义匹配兜底：允许 need 与 title/category/tags 的同义表达互相命中。 for need in need_set: if need in covered: continue if _need_matches_hit(need, hit_texts): covered.add(need) total = len(needs) covered_needs = [n for n in needs if n in covered] covered_count = len(covered_needs) if total == 0: return "NOT_PROVIDED", sources, [], [] if covered_count == total: return "READY", sources, covered_needs, [] if covered_count > 0: uncovered = [n for n in needs if n not in covered] return "PARTIAL", sources, covered_needs, uncovered return "NOT_PROVIDED", sources, [], list(needs) def main(): require_access_token() body = _read_body() user_text = (body.get("userText") or "").strip() if not user_text: print("错误: 缺少 userText", file=sys.stderr) sys.exit(2) if not isinstance(body.get("parsedFields"), dict): print("错误: 缺少 parsedFields（必须由上游参数传入，脚本不做文本兜底）", file=sys.stderr) sys.exit(2) existing = body.get("scenarioPack") if isinstance(body.get("scenarioPack"), dict) else {} pack = {**existing, **_extract_fields(body)} required = ["department", "product", "location", "doctorConcerns", "repGoal"] missing_fields = [k for k in required if not pack.get(k)] needs = pack.get("productKnowledgeNeeds", []) has_search_result = isinstance(body.get("knowledgeSearchResult"), dict) if needs and REQUIRE_KB_API and not has_search_result: print("错误: 检测到产品知识需求，但缺少 knowledgeSearchResult；请先调用产品知识API并回传结果", file=sys.stderr) sys.exit(2) status, sources, covered, uncovered = _coverage_from_search_result(body.get("knowledgeSearchResult"), needs) pack["productEvidenceStatus"] = status pack["productEvidenceSource"] = sources if status != "READY": missing_fields.append("productEvidenceSource") clarify_questions = [] question_map = { "department": "主要沟通的是哪个科室或门诊？", "product": "这次要传递的是哪个具体产品或通用名？", "location": "本次拜访发生在什么机构与场景（门诊/病区/院外）？", "doctorConcerns": "主任当前最在意的是可及性、疗效证据，还是使用风险？", "repGoal": "本次沟通最想达成的一个结果是什么？", "productEvidenceSource": "为保证内容准确，请补充可核对的产品资料：说明书要点、关键研究摘要，或可公开访问的说明/文献链接（无需提供内部系统编号）。", } seen = set() for mf in missing_fields: if mf in seen: continue seen.add(mf) q = question_map.get(mf) if q: clarify_questions.append(q) pack = normalize_scenario_pack(pack) _ok( "parse-and-gap-ask", scenarioPack=pack, missingFields=sorted(seen), coveredNeeds=covered, uncoveredNeeds=uncovered, clarifyQuestions=clarify_questions, acceptedSourcePrefixes=list(API_SOURCE_PREFIXES), ) if __name__ == "__main__": main() FILE:scripts/scene/persist-and-execute.py #!/usr/bin/env python3 """ scene / persist-and-execute — 将草稿落盘并调用 tbs_write_executor.py；stdout 为 TOON。 """ import json import os import subprocess import sys import ssl import urllib.error import urllib.request sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common")) from auth_token import resolve_access_token from toon_encoder import encode as toon_encode API_URL = "https://scenario-builder.openclaw.internal/v1/scene/persist-and-execute" def _skill_root() -> str: here = os.path.dirname(os.path.realpath(__file__)) return os.path.normpath(os.path.join(here, "..", "..")) def _runtime_dir() -> str: """Backward-compat: fallback assets live under `runtime/`.""" return os.path.join(_skill_root(), "runtime") def _assets_dir() -> str: """ Prefer non-`runtime/` assets location. - Set env `TBS_ASSETS_DIR` to fully override. - Else if `scripts/tbs_assets/` exists, use it. - Else fall back to legacy `runtime/`. """ env = os.environ.get("TBS_ASSETS_DIR") if env: return env skill_root = _skill_root() candidate = os.path.join(skill_root, "scripts", "tbs_assets") if os.path.isdir(candidate): return candidate candidate2 = os.path.join(skill_root, "assets", "tbs_assets") if os.path.isdir(candidate2): return candidate2 return _runtime_dir() def _auth_probe(base_url: str, access_token: str, insecure_ssl: bool = True): url = base_url.rstrip("/") + "/api/v1/admin/basic/business-domains" headers = { "accept": "application/json", "Content-Type": "application/json", "access-token": access_token, } req = urllib.request.Request(url=url, method="GET", headers=headers) try: ssl_ctx = ssl._create_unverified_context() if insecure_ssl else None with urllib.request.urlopen(req, timeout=20, context=ssl_ctx) as resp: return int(resp.status), "" except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", errors="replace") if hasattr(e, "read") else "" return int(e.code), raw[-500:] except Exception as e: return None, str(e) def main(): body = json.loads(sys.stdin.read() or "{}") vr = body.get("validationReport") or {} if not vr.get("passed"): print("错误: validationReport.passed 须为 true", file=sys.stderr) sys.exit(2) user_confirmation = body.get("userConfirmation") if not user_confirmation: print("错误: 缺少 userConfirmation", file=sys.stderr) sys.exit(2) user_confirmation = str(user_confirmation).strip() if user_confirmation == "取消": # Cancel is not an error; it should stop persistence gracefully. print( toon_encode( { "ok": False, "step": "persist-and-execute", "message": "user_cancelled", "hint": "用户已取消，本次不执行落库。", } ) ) sys.exit(0) if user_confirmation != "确认": print( toon_encode( { "ok": False, "step": "persist-and-execute", "message": "invalid_userConfirmation", "got": user_confirmation, "hint": "userConfirmation 仅允许填入：确认或取消。", } ) ) sys.exit(2) assets_dir = _assets_dir() base_url = os.environ.get("TBS_BASE_URL", "https://sg-tbs-manage.mediportal.com.cn") access_token, token_source = resolve_access_token() if not access_token: print( toon_encode( { "ok": False, "step": "persist-and-execute", "message": "missing_access_token", "hint": "Set XG_USER_TOKEN before running persist-and-execute.", } ) ) sys.exit(1) probe_status, probe_error = _auth_probe(base_url=base_url, access_token=access_token, insecure_ssl=True) if probe_status != 200: print( toon_encode( { "ok": False, "step": "persist-and-execute", "message": "auth_preflight_failed", "tokenSource": token_source, "probeUrl": base_url.rstrip("/") + "/api/v1/admin/basic/business-domains", "probeStatus": probe_status, "probeError": probe_error, } ) ) sys.exit(4) default_draft_assets = os.path.join(assets_dir, "scenario_draft.json") default_draft_runtime = os.path.join(_runtime_dir(), "scenario_draft.json") if os.path.isfile(default_draft_assets): default_draft = default_draft_assets else: default_draft = default_draft_runtime draft_path = body.get("draftPath") or default_draft payload = body.get("draftPayload") if payload is not None: with open(draft_path, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=2) # Executor 入口统一迁移到 scripts/scene/ exe = os.path.join(os.path.dirname(__file__), "tbs_write_executor.py") if not os.path.isfile(exe): # fallback: older layout (tbs_assets wrapper) exe_fallback = os.path.join(assets_dir, "tbs_write_executor.py") exe = exe_fallback if os.path.isfile(exe_fallback) else exe if not os.path.isfile(exe): print( toon_encode( { "ok": False, "step": "persist-and-execute", "message": "executor_missing", "path": exe, } ) ) sys.exit(3) proc = subprocess.run( [sys.executable, exe, "--insecure-ssl", "--input", str(draft_path)], capture_output=True, text=True, timeout=600, env=os.environ, ) print(toon_encode({ "ok": proc.returncode == 0, "step": "persist-and-execute", "tokenSource": token_source, "probeStatus": probe_status, "returncode": proc.returncode, "stdoutTail": (proc.stdout or "")[-2000:], "stderrTail": (proc.stderr or "")[-2000:], })) if proc.returncode != 0: sys.exit(proc.returncode) if __name__ == "__main__": main() FILE:scripts/scene/preflight-tbs-master-data.py #!/usr/bin/env python3 """ scene / preflight-tbs-master-data — 确认落库前：对 TBS 业务领域 / 科室 / 药品做 GET 匹配，不存在则 POST 创建。与 tbs_write_executor 使用同一套 resolve_ids_for_scene 逻辑；stdout 为 TOON。 """ import argparse import json import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common")) from auth_token import resolve_access_token from toon_encoder import encode as toon_encode # type: ignore[reportMissingImports] from tbs_master_data_resolve import TBSClient, resolve_ids_for_scene API_URL = "https://scenario-builder.openclaw.internal/v1/scene/preflight-tbs-master-data" def _normalize_scenes(api_draft: dict): raw = api_draft.get("scenes") or {} if isinstance(raw, list): return raw[0] if raw else {} return raw if isinstance(raw, dict) else {} def main(): ap = argparse.ArgumentParser(description="Preflight TBS master data before persist-and-execute.") ap.add_argument( "--input", default=None, help="Path to draft JSON (default: stdin only if not set; use - for stdin file)", ) ap.add_argument("--base-url", default=os.environ.get("TBS_BASE_URL", "https://sg-tbs-manage.mediportal.com.cn")) ap.add_argument("--access-token", default=None) ap.add_argument("--insecure-ssl", action="store_true", help="Skip SSL certificate verification.") ap.add_argument( "--dry-run", action="store_true", help="Only GET/list matching; do not POST create (report would_create).", ) args = ap.parse_args() env_token, _ = resolve_access_token() access_token = (args.access_token or env_token or "").strip() if not access_token: print("错误: 缺少 access-token / XG_USER_TOKEN", file=sys.stderr) sys.exit(1) if args.input: with open(args.input, "r", encoding="utf-8") as f: draft = json.load(f) else: raw = sys.stdin.read() if not raw.strip(): print("错误: 请通过 stdin 传入草稿 JSON，或使用 --input", file=sys.stderr) sys.exit(2) draft = json.loads(raw) api_draft = draft.get("apiDraft") or {} api_draft_scenes = _normalize_scenes(api_draft) client = TBSClient( base_url=args.base_url, access_token=access_token, insecure_ssl=bool(args.insecure_ssl), ) try: resolved, report = resolve_ids_for_scene( client, api_draft_scenes, dry_run=bool(args.dry_run), with_report=True, ) except Exception as e: print( toon_encode( { "ok": False, "step": "preflight-tbs-master-data", "message": str(e), } ) ) sys.exit(3) ok = bool( resolved.get("department_id") and resolved.get("business_domain_id") and resolved.get("drug_id") ) print( toon_encode( { "ok": ok, "step": "preflight-tbs-master-data", "resolvedIds": resolved, "resolutionReport": report, "dryRun": bool(args.dry_run), } ) ) sys.exit(0 if ok else 4) if __name__ == "__main__": main() FILE:scripts/scene/publish-ready-compose.py #!/usr/bin/env python3 """ scene / publish-ready-compose — 契约校验脚本（stdin JSON）。输出 TOON。 API_URL 须与 openapi/scene/publish-ready-compose.md 标题 URL 一致；本脚本不发起 HTTP。 """ import json import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common")) from auth_token import require_access_token from toon_encoder import encode as toon_encode API_URL = "https://scenario-builder.openclaw.internal/v1/scene/publish-ready-compose" def _read_body(): raw = sys.stdin.read() if not raw.strip(): return {} return json.loads(raw) def _ok(step, **extra): payload = {"ok": True, "step": step, **extra} print(toon_encode(payload)) def main(): require_access_token() body = _read_body() sp = body.get("scenarioPack") mh = body.get("modeHints") or {} if not isinstance(sp, dict): print("错误: 缺少 scenarioPack", file=sys.stderr) sys.exit(2) pr = mh.get("publish_ready") or mh.get("outputMode") == "publish_ready" if not pr: print("错误: modeHints.publish_ready 或 outputMode=publish_ready 须成立", file=sys.stderr) sys.exit(2) _ok("publish-ready-compose") if __name__ == "__main__": main() FILE:scripts/scene/route-by-intent.py #!/usr/bin/env python3 """ scene / route-by-intent — 契约校验脚本（stdin JSON）。输出 TOON。 API_URL 须与 openapi/scene/route-by-intent.md 标题 URL 一致；本脚本不发起 HTTP。 """ import json import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common")) from auth_token import require_access_token from toon_encoder import encode as toon_encode API_URL = "https://scenario-builder.openclaw.internal/v1/scene/route-by-intent" def _read_body(): raw = sys.stdin.read() if not raw.strip(): return {} return json.loads(raw) def _ok(step, **extra): payload = {"ok": True, "step": step, **extra} print(toon_encode(payload)) def main(): require_access_token() body = _read_body() user_text = (body.get("userText") or "").strip() if not user_text: print("错误: 缺少 userText", file=sys.stderr) sys.exit(2) route_decision = body.get("routeDecision") if not isinstance(route_decision, dict): print("错误: 缺少 routeDecision（必须由上游参数传入）", file=sys.stderr) sys.exit(2) intent = route_decision.get("intent") next_step = route_decision.get("nextStep") reason = route_decision.get("reason", "provided-by-upstream") need_clarification = bool(route_decision.get("needClarification", False)) clarify_question = route_decision.get("clarifyQuestion", "") preconditions = list(route_decision.get("preconditions") or []) if not intent or not next_step: print("错误: routeDecision.intent 与 routeDecision.nextStep 必填", file=sys.stderr) sys.exit(2) session_state = body.get("sessionState") if isinstance(body.get("sessionState"), dict) else {} if intent == "PERSIST_CONFIRM": vr = session_state.get("validationReport") if isinstance(session_state.get("validationReport"), dict) else {} if not vr.get("passed"): preconditions.append("缺少 validationReport.passed=true") need_clarification = True if not clarify_question: clarify_question = "当前还没有通过校验，是否先执行校验？" next_step = "validate-and-gate" reason = "persist-request-without-passed-validation" _ok( "route-by-intent", intent=intent, nextStep=next_step, reason=reason, needClarification=need_clarification, clarifyQuestion=clarify_question, preconditions=preconditions, ) if __name__ == "__main__": main() FILE:scripts/scene/scenario_pack_normalizer.py #!/usr/bin/env python3 """ Normalize scenarioPack fields between new/legacy schemas. """ from copy import deepcopy def normalize_scenario_pack(scenario_pack: dict) -> dict: sp = deepcopy(scenario_pack or {}) # New -> legacy compatibility for downstream generators. if "sceneBasic" not in sp: sp["sceneBasic"] = { "department": sp.get("department"), "product": sp.get("product"), "location": sp.get("location"), } if "roleSetup" not in sp: sp["roleSetup"] = { "doctorRole": "主任", "repRole": "医药代表", "department": sp.get("department"), } if "personaConfig" not in sp: sp["personaConfig"] = { "doctorConcerns": sp.get("doctorConcerns", []), "repGoal": sp.get("repGoal"), "styleHints": sp.get("styleHints", []), } if "personaBase" not in sp and sp.get("personaConfig"): sp["personaBase"] = {"config": sp.get("personaConfig")} # Keep product evidence fields explicit for gating. sp["productEvidenceStatus"] = sp.get("productEvidenceStatus", "NOT_PROVIDED") sp["productEvidenceSource"] = sp.get("productEvidenceSource", []) return sp FILE:scripts/scene/tbs_master_data_resolve.py """ Shared TBS Admin HTTP helpers and resolve_ids_for_scene (business domain / department / drug). Used by tbs_write_executor.py and preflight-tbs-master-data.py. """ import difflib import json import ssl import urllib.error import urllib.request def safe_json_loads(text: str): try: return json.loads(text) except Exception: return None def extract_data(payload): if isinstance(payload, list): return payload if isinstance(payload, dict): if "data" in payload: return payload["data"] if "result" in payload: return payload["result"] return payload def guess_entity_name(item: dict): for k in ["name", "title", "department_name", "drug_name", "business_domain_name", "code"]: v = item.get(k) if isinstance(v, str) and v.strip(): return v.strip() return None def guess_entity_id(item: dict): for k in ["id", "department_id", "drug_id", "business_domain_id", "persona_id", "doctor_id", "knowledge_id"]: if k in item and isinstance(item[k], (str, int)) and str(item[k]).strip(): return item[k] return None def pick_best_match_id(items, target_name: str): if not items: return None target_name = (target_name or "").strip() if not target_name: return None for it in items: nm = guess_entity_name(it) if nm == target_name: return guess_entity_id(it) names = [] id_by_name = {} for it in items: nm = guess_entity_name(it) if nm: names.append(nm) id_by_name[nm] = guess_entity_id(it) best = difflib.get_close_matches(target_name, names, n=1, cutoff=0.55) if not best: return None return id_by_name.get(best[0]) class TBSClient: def __init__( self, base_url: str, access_token: str, timeout_s: int = 30, insecure_ssl: bool = False, ): self.base_url = base_url.rstrip("/") self.access_token = access_token self.timeout_s = timeout_s self.insecure_ssl = insecure_ssl def _headers(self): h = { "Content-Type": "application/json", "accept": "application/json", } if self.access_token: h["access-token"] = self.access_token return h def request_json(self, method: str, path: str, body=None): url = self.base_url + path if path.startswith("/") else self.base_url + "/" + path data = None if body is not None: data = json.dumps(body, ensure_ascii=False).encode("utf-8") req = urllib.request.Request(url=url, method=method, headers=self._headers(), data=data) try: ssl_ctx = None if self.insecure_ssl: ssl_ctx = ssl._create_unverified_context() with urllib.request.urlopen(req, timeout=self.timeout_s, context=ssl_ctx) as resp: raw = resp.read().decode("utf-8", errors="replace") payload = safe_json_loads(raw) return payload if payload is not None else raw except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", errors="replace") if hasattr(e, "read") else "" raise RuntimeError(f"HTTP {e.code} {method} {url}: {raw}") from e def get_list(client: TBSClient, path: str): payload = client.request_json("GET", path) data = extract_data(payload) return data if isinstance(data, list) else [] def resolve_ids_for_scene( client: TBSClient, api_draft_scenes: dict, dry_run: bool = False, with_report: bool = False, ): """ GET lists to match by name; POST create when no match (unless dry_run). If with_report, returns (ids_dict, report_dict); else ids_dict only. """ dep_in = api_draft_scenes.get("department_id") or api_draft_scenes.get("departmentName") or "" bd_in = api_draft_scenes.get("business_domain_id") or api_draft_scenes.get("businessDomainName") or "" drug_in = api_draft_scenes.get("drug_id") or api_draft_scenes.get("drugName") or "" report: dict = {} def maybe_int_id(x): sx = str(x).strip() if sx.isdigit(): return sx return None def try_get_list(path: str): try: return get_list(client, path) except Exception: return [] def resolve_or_create_business_domain(bd_value: str): bd_value = (bd_value or "").strip() bd_id = maybe_int_id(bd_value) if bd_id: if with_report: report["business_domain"] = {"action": "matched_by_id", "id": bd_id, "input": bd_value} return bd_id business_domains = try_get_list("/api/v1/admin/basic/business-domains") if isinstance(bd_value, str) and bd_value and bd_value in ["医美", "医美业务", "医美推广"]: for it in business_domains: nm = guess_entity_name(it) or "" if nm.strip() in ["临床推广"]: rid = str(guess_entity_id(it)) if with_report: report["business_domain"] = { "action": "matched", "id": rid, "input": bd_value, "note": "alias_to_临床推广", } return rid bd_id = pick_best_match_id(business_domains, bd_value) if bd_id: if with_report: report["business_domain"] = {"action": "matched", "id": str(bd_id), "input": bd_value} return str(bd_id) if dry_run: if with_report: report["business_domain"] = {"action": "would_create", "id": None, "input": bd_value} return None body = {"name": bd_value} try: resp = client.request_json("POST", "/api/v1/admin/basic/business-domains", body=body) created = extract_data(resp) if created is not None: cid = str(created.get("id") if isinstance(created, dict) else created) if with_report: report["business_domain"] = {"action": "created", "id": cid, "input": bd_value} return cid except RuntimeError as e: if "HTTP 409" not in str(e): raise business_domains = try_get_list("/api/v1/admin/basic/business-domains") bd_id = pick_best_match_id(business_domains, bd_value) out = str(bd_id) if bd_id else None if with_report: report["business_domain"] = { "action": "matched_after_conflict", "id": out, "input": bd_value, } return out def resolve_or_create_department(dep_value: str, bd_id: str): dep_value = (dep_value or "").strip() dep_id = maybe_int_id(dep_value) if dep_id: if with_report: report["department"] = {"action": "matched_by_id", "id": dep_id, "input": dep_value} return dep_id departments = try_get_list("/api/v1/admin/basic/departments") dep_id = pick_best_match_id(departments, dep_value) if dep_id: if with_report: report["department"] = {"action": "matched", "id": str(dep_id), "input": dep_value} return str(dep_id) if dry_run: if with_report: report["department"] = {"action": "would_create", "id": None, "input": dep_value} return None body = {"name": dep_value} if bd_id: body["business_domain_id"] = bd_id try: resp = client.request_json("POST", "/api/v1/admin/basic/departments", body=body) created = extract_data(resp) if created is not None: cid = str(created.get("id") if isinstance(created, dict) else created) if with_report: report["department"] = {"action": "created", "id": cid, "input": dep_value} return cid except RuntimeError as e: if "HTTP 409" not in str(e): raise departments = try_get_list("/api/v1/admin/basic/departments") dep_id = pick_best_match_id(departments, dep_value) out = str(dep_id) if dep_id else None if with_report: report["department"] = { "action": "matched_after_conflict", "id": out, "input": dep_value, } return out def resolve_or_create_drug(drug_value: str, bd_id: str): """GET /api/v1/admin/basic/drugs then POST if missing.""" drug_value = (drug_value or "").strip() drug_id = maybe_int_id(drug_value) if drug_id: if with_report: report["drug"] = {"action": "matched_by_id", "id": drug_id, "input": drug_value} return drug_id drugs = try_get_list("/api/v1/admin/basic/drugs") drug_id = pick_best_match_id(drugs, drug_value) if drug_id: if with_report: report["drug"] = {"action": "matched", "id": str(drug_id), "input": drug_value} return str(drug_id) if dry_run: if with_report: report["drug"] = {"action": "would_create", "id": None, "input": drug_value} return None body = {"name": drug_value} if bd_id: body["business_domain_id"] = bd_id try: resp = client.request_json("POST", "/api/v1/admin/basic/drugs", body=body) created = extract_data(resp) if created is not None: cid = str(created.get("id") if isinstance(created, dict) else created) if with_report: report["drug"] = {"action": "created", "id": cid, "input": drug_value} return cid except RuntimeError as e: if "HTTP 409" not in str(e): raise drugs = try_get_list("/api/v1/admin/basic/drugs") drug_id = pick_best_match_id(drugs, drug_value) out = str(drug_id) if drug_id else None if with_report: report["drug"] = { "action": "matched_after_conflict", "id": out, "input": drug_value, } return out bd_id = resolve_or_create_business_domain(str(bd_in)) dep_id = resolve_or_create_department(str(dep_in), bd_id) drug_id = resolve_or_create_drug(str(drug_in), bd_id) ids = {"department_id": dep_id, "business_domain_id": bd_id, "drug_id": drug_id} if with_report: return ids, report return ids FILE:scripts/scene/tbs_write_executor.py #!/usr/bin/env python3 import argparse import difflib import hashlib import json import os import re import sys import urllib.error import urllib.request sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common")) from auth_token import resolve_access_token from toon_encoder import encode as toon_encode # type: ignore from tbs_master_data_resolve import ( TBSClient, extract_data, get_list, guess_entity_name, guess_entity_id, pick_best_match_id, resolve_ids_for_scene, ) # 默认资产目录（兼容从 `runtime/` 迁移到其它路径） def _find_skill_root() -> str: here = os.path.dirname(os.path.abspath(__file__)) for _ in range(8): if os.path.isfile(os.path.join(here, "SKILL.md")): return here parent = os.path.dirname(here) if parent == here: break here = parent # fallback: assume current file is inside skill_root/runtime/ return os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")) def _assets_dir() -> str: """ - Set env `TBS_ASSETS_DIR` to override. - Else prefer `skill_root/scripts/tbs_assets/`. - Else fall back to legacy `skill_root/runtime/`. """ env = os.environ.get("TBS_ASSETS_DIR") if env: return env skill_root = _find_skill_root() candidate = os.path.join(skill_root, "scripts", "tbs_assets") required = [ "scenario_draft.json", "P1_persisted_ids.md", ] def _has_required(d: str) -> bool: return all(os.path.isfile(os.path.join(d, f)) for f in required) if os.path.isdir(candidate) and _has_required(candidate): return candidate candidate2 = os.path.join(skill_root, "assets", "tbs_assets") if os.path.isdir(candidate2) and _has_required(candidate2): return candidate2 return os.path.join(skill_root, "runtime") _RUNTIME_DIR = _assets_dir() def normalize_text(s: str) -> str: if s is None: return "" # Normalize whitespace for stable fingerprinting. return re.sub(r"\s+", " ", str(s)).strip() def fingerprint_sha256_of_text(s: str) -> str: n = normalize_text(s) return hashlib.sha256(n.encode("utf-8")).hexdigest() def read_first_line(path: str): if not path: return None if not os.path.exists(path): return None try: with open(path, "r", encoding="utf-8") as f: line = f.readline().strip() return line if line else None except Exception: return None def find_doctor_by_name_and_title(items, name: str, title: str): name = (name or "").strip() title = (title or "").strip() if not items: return None # Try to match name first; then title. for it in items: nm = (guess_entity_name(it) or "").strip() # Some list endpoints may use different keys for name/title. it_title = (it.get("title") or it.get("doctor_title") or "").strip() if isinstance(it, dict) else "" it_name = (it.get("name") or it.get("doctor_name") or nm or "").strip() if isinstance(it, dict) else "" if name and it_name == name: if title: if it_title == title: return guess_entity_id(it) else: return guess_entity_id(it) # Fallback: fuzzy name match names = [] id_by_name = {} for it in items: it_name = (it.get("name") or it.get("doctor_name") or "").strip() if it_name: names.append(it_name) id_by_name[it_name] = guess_entity_id(it) best = difflib.get_close_matches(name, names, n=1, cutoff=0.6) if best: return id_by_name.get(best[0]) return None def _default_surname_from_name(name: str) -> str: name = (name or "").strip() if name and name[0] in "赵钱孙李周吴郑王冯陈褚卫蒋沈韩杨朱秦许何吕施张孔曹严华金魏陶姜": return name[0] return "王" def _build_default_persona_config(scenario_pack: dict, doctor_draft: dict) -> str: sp = scenario_pack or {} dd = doctor_draft or {} concern = ( _safe_get(sp, ["criticalInfo", "coreConcern"]) or sp.get("coreConcern") or sp.get("concerns") or "长期预后与依从性" ) concern = _first_sentence(_ensure_nonempty_str(concern), max_len=60) or "长期预后与依从性" return "\n".join( [ "## 性格特征", "- 理性", "- 专业", "- 稳健", "", "## 沟通风格", "你对新治疗方案并不排斥，但只在安全边界和逻辑清楚的前提下才会继续讨论。", "你不会主动推动用药，而是通过问题判断对方是否真的理解临床场景。", "", "## 行为倾向映射", "- 清楚说明适用人群 -> 愿意继续", "- 主动说明风险与边界 -> 信任上升", "- 过度承诺或泛化疗效 -> 明显警惕", f"- 能结合真实临床场景并回应“{concern}” -> 兴趣增加", ] ) def create_doctor_if_missing(client: TBSClient, doctor_draft: dict, scenario_pack: dict = None, dry_run: bool = False): doctors = get_list(client, "/api/v1/resources/doctors") name = doctor_draft.get("name") or doctor_draft.get("doctor_name") or doctor_draft.get("persona_name") title = doctor_draft.get("title") or doctor_draft.get("doctor_title") or "" found_id = find_doctor_by_name_and_title(doctors, name, title) if found_id: return found_id if dry_run: return None surname = _ensure_nonempty_str(doctor_draft.get("surname") or doctor_draft.get("last_name") or doctor_draft.get("family_name")) if not surname: surname = _default_surname_from_name(name) description = _ensure_nonempty_str( doctor_draft.get("description") or doctor_draft.get("desc") or doctor_draft.get("summary") ) if not description: description = "专业理性，在安全与边界清楚的前提下愿意讨论新方案的医生，是理想训练对象。" persona_cfg = doctor_draft.get("persona_config", {}) if not persona_cfg: persona_cfg = _build_default_persona_config(scenario_pack or {}, doctor_draft or {}) # If persona_config is already a string (Markdown/JSON text), pass it through. if isinstance(persona_cfg, str): persona_cfg_text = persona_cfg else: persona_cfg_text = json.dumps(persona_cfg, ensure_ascii=False) body = { "name": name, "surname": surname, "title": title, "description": description, "persona_config": persona_cfg_text, "is_preset": False, } # Some APIs accept optional trust_initial/patience_initial; keep minimal to reduce incompatibility. resp = client.request_json("POST", "/api/v1/admin/basic/doctors", body=body) data = extract_data(resp) return data def load_persisted_knowledge_ids(persisted_ids_path: str): # Very small heuristic parser for the existing markdown. if not os.path.exists(persisted_ids_path): return [] text = open(persisted_ids_path, "r", encoding="utf-8").read() ids = [] # Compatible with formats like: - `52`（产品特性） / - 52（产品特性） for m in re.finditer(r"-\s*`?\s*(\d+)\s*`?\s*（", text): ids.append(m.group(1)) # Dedup while preserving order seen = set() out = [] for x in ids: if x not in seen: seen.add(x) out.append(x) return out def infer_intent_keywords(scenario_pack: dict) -> dict: """ Return boolean intent flags inferred from scenario text. Heuristic only; selection still degrades gracefully to persisted IDs. """ text = "" try: text = " ".join( [ scenario_pack.get("sceneBasic", {}).get("sceneBackground", "") or "", scenario_pack.get("criticalInfo", {}).get("coreConcern", "") or "", scenario_pack.get("criticalInfo", {}).get("trainingGoal", "") or "", scenario_pack.get("sceneBasic", {}).get("repBriefing", "") or "", ] ) except Exception: text = "" text = normalize_text(text) def has_any(words): return any(w in text for w in words) return { "price": has_any(["价格", "价差", "贵", "便宜", "值不值", "溢价", "质价比", "买贵", "预算"]), "safety": has_any(["安全", "风险", "合规", "无菌", "结节", "肿胀", "副作用", "迟发", "边界", "风险边界"]), "recovery": has_any(["恢复", "肿胀", "肿", "工作", "出镜", "社交", "恢复期"]), "naturalness": has_any(["自然", "假", "假脸", "自然感", "不自然", "僵硬"]), "mechanism": has_any(["PLLA", "聚左旋乳酸", "CMC", "甘露醇", "胶原", "再生", "代谢", "分解", "长效", "过程"]), "differentiation": has_any(["差异", "区别", "专利", "工艺", "微球", "粒径", "双粒径", "双规格", "优势", "卖点", "背书", "质价比", "价值"]), "indications": has_any(["适应症", "皱纹", "鼻唇沟", "纠正", "中重度"]), "spec": has_any(["规格", "mg/瓶", "368mg", "184mg", "粒径", "注射层次", "平均粒径"]), } def score_knowledge_item(item: dict, intent_flags: dict) -> int: category = normalize_text(item.get("category") or "") title = normalize_text(item.get("title") or "") content = normalize_text(item.get("content") or "") score = 0 def bump_if(keyword_list, bonus): nonlocal score for kw in keyword_list: if kw and (kw in title or kw in content or kw in category): score += bonus # Category/title keywords provide stronger hints than raw content. if intent_flags.get("price"): score += 3 if ("优势" in category or "卖点" in category) else 0 bump_if(["价值", "优势", "卖点", "背书", "质价比"], 2) if intent_flags.get("safety"): bump_if(["安全", "风险", "无菌", "合规", "无细胞毒", "NMPA"], 3) score += 2 if "安全" in category else 0 if intent_flags.get("recovery"): bump_if(["恢复", "肿胀", "消肿", "影响", "迟发"], 2) if intent_flags.get("naturalness"): bump_if(["自然", "假", "自然感", "不自然", "胶原再生"], 2) if intent_flags.get("mechanism"): bump_if(["PLLA", "胶原", "再生", "代谢", "分解", "长效"], 3) if intent_flags.get("differentiation"): bump_if(["专利", "工艺", "微球", "粒径", "双粒径", "双规格", "差异", "背书"], 3) if intent_flags.get("indications"): bump_if(["适应", "皱纹", "鼻唇沟", "纠正"], 3) if intent_flags.get("spec"): bump_if(["规格", "mg/瓶", "粒径", "注射层次", "平均粒径"], 3) # Weak fallback: any overlap in content words helps ranking. # Keep it small to avoid noise. overlap_tokens = ["PLLA", "CMC", "甘露醇", "NMPA", "无菌", "粒径", "胶原"] for tok in overlap_tokens: if tok and tok in title: score += 1 if tok and tok in content: score += 1 return score def select_knowledge_ids_by_intent(client: TBSClient, drug_id: str, scenario_pack: dict, top_n: int = 4): """ When user didn't provide productKnowledge, select the most relevant existing knowledge by matching intent from scenario text against knowledge title/content/category. """ intent_flags = infer_intent_keywords(scenario_pack or {}) # Fetch candidate knowledge list for this drug. path = f"/api/v1/resources/knowledge?drug_id={drug_id}" try: payload = client.request_json("GET", path) except Exception: return [], { "mode": "intent_select", "inferred_intents": [k for k, v in intent_flags.items() if v], "covered_intents": [], "low_coverage_intents": [k for k, v in intent_flags.items() if v], "error": "failed_to_fetch_knowledge_list", } existing = extract_data(payload) if isinstance(existing, dict) and "data" in existing: existing = existing["data"] if not isinstance(existing, list): inferred_true_intents = [k for k, v in intent_flags.items() if v] return [], { "mode": "intent_select", "inferred_intents": inferred_true_intents, "covered_intents": [], "low_coverage_intents": inferred_true_intents, "error": "knowledge_list_not_a_list", } scored = [] for it in existing: if not isinstance(it, dict): continue s = score_knowledge_item(it, intent_flags) scored.append((s, it.get("id"))) # Sort descending by score. scored.sort(key=lambda x: x[0], reverse=True) # If everything scores 0, don't force wrong picks. max_s = 0 for s, kid in scored: if kid is not None: max_s = max(max_s, int(s or 0)) if max_s <= 0: inferred_true_intents = [k for k, v in intent_flags.items() if v] return [], { "mode": "intent_select", "inferred_intents": inferred_true_intents, "covered_intents": [], "low_coverage_intents": inferred_true_intents, "note": "no_relevant_knowledge_scored", } # Otherwise take top_n even if some are 0, to keep enough knowledge coverage. best = [kid for s, kid in scored[:top_n] if kid is not None] # Build coverage report (heuristic) to decide whether user might want to provide extra productKnowledge. # Only report intents that are inferred as true in this scenario. intent_flags = infer_intent_keywords(scenario_pack or {}) intent_keywords_map = { "price": ["优势", "卖点", "价值", "质价比", "背书", "买贵", "溢价"], "safety": ["安全", "风险", "无菌", "合规", "NMPA", "无细胞毒"], "recovery": ["恢复", "肿胀", "消肿", "迟发", "影响", "恢复期"], "naturalness": ["自然", "假", "自然感", "不自然", "胶原再生", "假脸", "僵硬"], "mechanism": ["PLLA", "聚左旋乳酸", "CMC", "甘露醇", "胶原", "再生", "代谢", "分解", "长效", "过程"], "differentiation": ["差异", "区别", "专利", "工艺", "微球", "粒径", "双粒径", "双规格", "优势", "背书"], "indications": ["适应", "皱纹", "鼻唇沟", "纠正", "中重度"], "spec": ["规格", "mg/瓶", "粒径", "注射层次", "平均粒径"], } def item_matches_intent(it: dict, intent_key: str) -> bool: if not isinstance(it, dict): return False title = normalize_text(it.get("title") or "") content = normalize_text(it.get("content") or "") category = normalize_text(it.get("category") or "") hay = category + " " + title + " " + content for kw in intent_keywords_map.get(intent_key, []): if kw and (kw in hay): return True return False selected_items = [it for it in existing if isinstance(it, dict) and str(it.get("id")) in set([str(x) for x in best])] covered_intents = [] for k, v in intent_flags.items(): if not v: continue # If any selected knowledge matches the intent heuristically, consider covered. if any(item_matches_intent(it, k) for it in selected_items): covered_intents.append(k) inferred_true_intents = [k for k, v in intent_flags.items() if v] low_coverage_intents = [k for k in inferred_true_intents if k not in covered_intents] report = { "mode": "intent_select", "inferred_intents": inferred_true_intents, "covered_intents": covered_intents, "low_coverage_intents": low_coverage_intents, } return best, report def _safe_get(d: dict, path: list): cur = d for k in path: if not isinstance(cur, dict): return None cur = cur.get(k) return cur def build_doctor_only_context_from_scenario_pack(scenario_pack: dict) -> str: """ Build publish-ish markdown for the system 'AI角色画像设定' field from scenarioPack. Fallback when apiDraft.scenes.doctor_only_context is missing or too thin. """ sp = scenario_pack or {} scene_bg = (_safe_get(sp, ["sceneBasic", "sceneBackground"]) or sp.get("sceneBackground") or "").strip() core_concern = (_safe_get(sp, ["criticalInfo", "coreConcern"]) or sp.get("coreConcern") or "").strip() ai_role = (_safe_get(sp, ["roleSetup", "aiRoleTitle"]) or sp.get("aiRoleTitle") or "").strip() persona_base = _safe_get(sp, ["personaConfig", "personaBase"]) or {} persona_overlay = _safe_get(sp, ["personaConfig", "personaOverlay"]) or {} known = [] if ai_role: known.append(f"- 你当前以“{ai_role}”身份参与对话") if scene_bg: known.append(f"- 场景背景：{scene_bg}") lines = ["## 已知背景"] lines.extend(known[:4] if known else ["- （待补充）"]) lines += ["", "## 核心顾虑（仅1个核心顾虑）"] lines.append(f"- {core_concern}" if core_concern else "- （待补充：对方最在意的一个问题）") base_name = persona_base.get("name") if isinstance(persona_base, dict) else None role_type = persona_base.get("roleType") if isinstance(persona_base, dict) else None if base_name or role_type: lines += ["", "## 人设要点"] if base_name: lines.append(f"- 底座画像：{base_name}") if role_type: lines.append(f"- 角色类型：{role_type}") if isinstance(persona_overlay, dict) and persona_overlay: triggers = persona_overlay.get("triggers") sig = persona_overlay.get("signaturePhrases") qs = persona_overlay.get("questionStyle") lines += ["", "## 追问风格与触发器"] if qs: lines.append(f"- 追问风格：{str(qs).strip()}") if isinstance(triggers, list) and triggers: for t in triggers[:3]: if t: lines.append(f"- 触发降温：{str(t).strip()}") if isinstance(sig, list) and sig: lines += ["", "## 常用句式（示例）"] for p in sig[:3]: if p: lines.append(f"- {str(p).strip()}") return "\n".join(lines).strip() + "\n" def _first_sentence(text: str, max_len: int = 120) -> str: t = (text or "").strip() if not t: return "" # Prefer Chinese sentence separators; fallback to first chunk. for sep in ["。", "！", "？", "；", ";", "\n"]: if sep in t: t = t.split(sep, 1)[0].strip() break return t[:max_len].strip() def _extract_time_desc(scenario_pack: dict) -> str: """ Time is required by contract/quality gates. - Prefer explicit dates found in scenario background/meta. - Otherwise use a safe relative time anchor to avoid fabricating a specific event date. """ sp = scenario_pack or {} # Try common meta fields. meta = sp.get("meta") if isinstance(meta, dict): for k in ["timestamp", "date", "time"]: v = meta.get(k) if isinstance(v, str) and v.strip(): return v.strip() # Old nested structure. scene_basic = sp.get("sceneBasic") if isinstance(sp.get("sceneBasic"), dict) else {} for v in [scene_basic.get("sceneBackground"), scene_basic.get("repBriefing")]: if isinstance(v, str) and v.strip(): # YYYY年 / YYYY-MM / YYYY/MM etc. m = re.search(r"(\d{4}年\s*\d{1,2}月\s*\d{1,2}日|\d{4}年\s*\d{1,2}月|\d{4}[-/]\d{1,2}[-/]\d{1,2})", v) if m: return m.group(1).strip() # Current simplified structure: scenarioPack.background. bg = sp.get("background") if isinstance(sp.get("background"), str) else "" m = re.search(r"(\d{4}年\s*\d{1,2}月\s*\d{1,2}日|\d{4}年\s*\d{1,2}月|\d{4}[-/]\d{1,2}[-/]\d{1,2})", bg) if m: return m.group(1).strip() # Relative anchor: accurate as "this conversation time". return "在本次沟通当下" def _ensure_nonempty_str(v) -> str: return (v if isinstance(v, str) else "").strip() def build_rep_briefing_from_scenario_pack(scenario_pack: dict, api_draft_scenes: dict, api_draft_doctors: dict) -> str: """ Enforce rep_briefing quality: - time, location, department - doctor/product current status - medical rep objective - communication entry point """ sp = scenario_pack or {} scenes = api_draft_scenes or {} doc = api_draft_doctors or {} # Time/Location/Department. time_desc = _extract_time_desc(sp) location = _ensure_nonempty_str(scenes.get("location") or sp.get("location")) department = _ensure_nonempty_str(scenes.get("departmentName") or scenes.get("department") or sp.get("department")) product = _ensure_nonempty_str(scenes.get("drugName") or scenes.get("drug") or sp.get("product")) # Doctor status and name/title. doctor_name = _ensure_nonempty_str(doc.get("name") or (sp.get("roleSetup", {}).get("ai", {}).get("name") if isinstance(sp.get("roleSetup"), dict) else None)) doctor_title = _ensure_nonempty_str(doc.get("title") or "") doctor_persona = _ensure_nonempty_str(doc.get("persona") or doc.get("persona_config") or doc.get("personaConfig") or "") doctor_persona_short = _first_sentence(doctor_persona, max_len=160) # Product status comes from scenario background/goals. bg = _ensure_nonempty_str(sp.get("background") or _ensure_nonempty_str(sp.get("sceneBackground"))) product_status_short = _first_sentence(bg, max_len=160) if not doctor_persona_short and bg: # If doctor persona is missing, derive a "doctor/product current status" snippet from background. doctor_persona_short = product_status_short # Rep objective and entry point. rep_objective_raw = sp.get("goals") or sp.get("trainingGoal") or "" if not rep_objective_raw and isinstance(sp.get("criticalInfo"), dict): rep_objective_raw = sp.get("criticalInfo", {}).get("trainingGoal") or "" rep_objective = _ensure_nonempty_str(rep_objective_raw) entry_point_raw = sp.get("concerns") or sp.get("coreConcern") or "" if not entry_point_raw and isinstance(sp.get("criticalInfo"), dict): entry_point_raw = sp.get("criticalInfo", {}).get("coreConcern") or "" entry_point_raw = _ensure_nonempty_str(entry_point_raw) entry_point = _first_sentence(entry_point_raw, max_len=120) # Enforce minimal non-empty required dimensions. missing = [] if not time_desc: missing.append("时间") if not location: missing.append("地点") if not department: missing.append("科室") if not product: missing.append("产品") if not doctor_persona_short: missing.append("医生与产品现状") if not rep_objective: missing.append("医药代表目标") if not entry_point: missing.append("沟通切入点") if missing: raise RuntimeError(f"rep_briefing quality gate missing required fields: {','.join(missing)}") def _truncate(s: str, max_len: int) -> str: s = (s or "").strip() if len(s) <= max_len: return s if max_len <= 3: return s[:max_len] return s[: max_len - 3] + "..." # Hard budget for "<= 180 Chinese characters" requirement. # We prefer truncating long variable parts rather than dropping key dimensions. limit = 180 entry_suffix_full = "沟通切入点聚焦{entry}，先承接关切，再给出可核验要点，并提出轻量下一步。" entry_suffix_short = "沟通切入点聚焦{entry}，先承接关切，再给要点并约定下一步。" def _clean_fragment(text: str) -> str: """Remove trailing punctuation to avoid duplicated separators like '。；'.""" return re.sub(r"[。；;，、,\s]+$", "", (text or "").strip()) def _compose(dp: str, ps: str, obj: str, entry: str, use_short_suffix: bool) -> str: entry_suffix = entry_suffix_short if use_short_suffix else entry_suffix_full doctor_identity = f"{doctor_name}{'（' + doctor_title + '）' if doctor_title else ''}" dp_clean = _clean_fragment(dp) ps_clean = _clean_fragment(ps or dp) obj_clean = _clean_fragment(obj) entry_clean = _clean_fragment(entry) # If product status semantically equals persona summary, skip duplicate clause. status_clause = "" if normalize_text(ps_clean) and normalize_text(ps_clean) != normalize_text(dp_clean): status_clause = f"{product}现状为{ps_clean}；" return ( f"{time_desc}，在{location}的{department}，" f"{doctor_identity}当前关注点是{dp_clean}；" f"{status_clause}" f"代表目标是{obj_clean}；" f"{entry_suffix.format(entry=entry_clean)}" ) # Try full first; if too long, progressively truncate variable parts. full_text = _compose(doctor_persona_short, product_status_short or doctor_persona_short, rep_objective, entry_point, use_short_suffix=False) if len(full_text) <= limit: return full_text.strip() + "\n" # Truncate strategy for short form. dp_max, ps_max, obj_max, entry_max = 45, 45, 35, 35 while True: short_text = _compose( dp=_truncate(doctor_persona_short, dp_max), ps=_truncate(product_status_short or doctor_persona_short, ps_max), obj=_truncate(rep_objective, obj_max), entry=_truncate(entry_point, entry_max), use_short_suffix=True, ) if len(short_text) <= limit or (dp_max <= 15 and ps_max <= 15 and obj_max <= 15 and entry_max <= 15): return short_text[:limit].rstrip() + "\n" dp_max -= 5 ps_max -= 5 obj_max -= 5 entry_max -= 5 def build_rep_briefing_relaxed(scenario_pack: dict, api_draft_scenes: dict, api_draft_doctors: dict) -> str: """ Relaxed fallback when strict quality gates cannot be satisfied. Ensures rep_briefing is never empty while still avoiding fabricated details. """ sp = scenario_pack or {} scenes = api_draft_scenes or {} doc = api_draft_doctors or {} time_desc = _extract_time_desc(sp) location = _ensure_nonempty_str(scenes.get("location") or sp.get("location")) or "当前沟通场景" department = _ensure_nonempty_str(scenes.get("departmentName") or scenes.get("department") or sp.get("department")) or "相关科室" product = _ensure_nonempty_str(scenes.get("drugName") or scenes.get("drug") or sp.get("product") or sp.get("productName")) or "目标产品" doctor_name = _ensure_nonempty_str(doc.get("name")) or "对方客户" background = _ensure_nonempty_str(scenes.get("background") or sp.get("background") or sp.get("sceneBackground")) concern = _first_sentence(_ensure_nonempty_str(sp.get("concerns") or sp.get("coreConcern")), max_len=40) or "价值、安全与恢复影响" if background: bg_short = _first_sentence(background, max_len=80) return ( f"{time_desc}，在{location}的{department}，" f"围绕{product}开展沟通。当前对话对象为{doctor_name}，" f"其核心顾虑聚焦{concern}，并基于“{bg_short}”推进本轮沟通。" ).strip() + "\n" return ( f"{time_desc}，在{location}的{department}，围绕{product}进行沟通，" f"对话对象为{doctor_name}，本轮优先回应其对{concern}的关切并推进下一步。" ).strip() + "\n" def _doctor_only_context_quality_ok(text: str) -> bool: t = (text or "").strip() if not t: return False doctor_required = [ "## 已知背景", "## 核心顾虑", "## 今日状态", "## 终止条件", "必须等待对方回答", "只有你可以标记对话结束", "## 对话结束规则（关键）", "## 追问路径（强制）", "只追问场景注入/已给出的参考资料", "## 输出长度控制", "## 单问原则", "## 提问限制", "只能基于已注入的参考资料/场景信息提问", "[对话结束]", ] if not all(r in t for r in doctor_required): return False # Require explicit output length constraints to avoid vague "保持简练". if not re.search(r"30\s*[-~至]\s*50\s*字|30-50字|30-50 字", t): return False return True def build_doctor_only_context_from_simple_scenario_pack(scenario_pack: dict, api_draft_doctors: dict, api_draft_scenes: dict = None) -> str: """ Build doctor_only_context in your requested format: - # 行为规范 ... (4 parts) - # 当前场景背景 - 终止条件 / 对话结束规则 / 重要提醒 """ sp = scenario_pack or {} doc = api_draft_doctors or {} scenes = api_draft_scenes or {} # Universal doctor template (non-BP). department = _ensure_nonempty_str(sp.get("department") or scenes.get("departmentName") or scenes.get("department")) or "相关科室" location = _ensure_nonempty_str(sp.get("location") or scenes.get("location")) or "当前沟通场景" product = _ensure_nonempty_str(sp.get("product") or sp.get("productName") or scenes.get("drugName") or scenes.get("drug")) or "目标产品" doctor_persona = _ensure_nonempty_str(doc.get("persona") or doc.get("persona_config") or doc.get("personaConfig") or "") doctor_persona_short = _first_sentence(doctor_persona, max_len=220) concerns = _ensure_nonempty_str(sp.get("concerns") or sp.get("coreConcern") or "") core_concern = _first_sentence(concerns, max_len=220) or "价值、安全性与恢复影响" bg = _ensure_nonempty_str(sp.get("background") or scenes.get("background") or "") time_desc = _extract_time_desc(sp) known_bg = bg if bg else f"科室与场景信息：科室={department}；地点={location}；推广产品={product}。" return "\n".join( [ "## 已知背景", f"- {known_bg}", "", "## 核心顾虑（仅1个核心顾虑）", f"- {core_concern}", "", "## 今日状态", f"- 当前时间：{time_desc}", f"- 你的态度：{doctor_persona_short or '愿意沟通，但强调合规与逻辑清楚'}", "", "## 终止条件", "**核心规则**：提问或提出要求后必须等待对方回答，在对方回答后再判断是否结束对话。", "", "**结束对话的条件**（对方回答后判断）：", "- 夸大疗效或绝对化承诺", "- 回避核心问题", "- 不尊重对方的关切", "- 推销味过重（过度承诺、夸大效果；注意：专业使用建议不算推销）", "", "## 对话结束规则（关键）", "- **唯一标记**：只有你可以标记对话结束，在回复末尾追加 [对话结束]。", "- **禁止提问**：一旦决定结束对话，绝对不能再提问或提出任何要求。必须检查并删除回复中所有的问号（？）或疑问句。", "- **互斥执行检查**：若本轮包含任何问句，则必须不输出 [对话结束]。", "- **终止流程**：使用陈述句/结束语表达结束，严禁在最后一轮再提问。", "", "## 追问路径（强制）", "- 优先关注：只追问场景注入/已给出的参考资料中与核心顾虑相关的信息要点；若资料不覆盖“剂型/专利/起效时间”等细节，则转为追问可核验的适用条件、合规边界或需要回去核对的口径来源。", "", "## 输出长度控制", "- 每次回复控制在30-50字左右，保持真实医生沟通的自然简洁；避免一次提出过多问题。", "", "## 单问原则", "- 每轮最多只能包含1个问号（问号≤1）。如果想到第二个问题，必须留到下一轮再问。", "", "## 提问限制", "- 只能基于已注入的参考资料/场景信息提问，禁止询问未提供的具体数值、技术参数或研究/对比结论。", "- 若参考资料仅有定性描述，不询问具体数值或定量结论。", "- 当出现信息缺口且无法从参考资料判断时，表述为需要回去核对资料或请对方补充口径后再讨论。", "", "## 输出格式（强制）", "- 只输出纯文本对话内容，不要包含动作描述。", "- 禁止使用任何额外格式化标记（如加粗/斜体/标题/代码符号）。如需强调使用自然措辞或中文引号。", "- 禁止出现具体姓名或可识别称呼，只使用“我”“你”或职业称谓。", ] ).strip() + "\n" def build_coach_only_context_from_scenario_pack(scenario_pack: dict) -> str: """ Build publish-ish markdown for the system '教练专属设定' field from scenarioPack. Fallback when apiDraft.scenes.coach_only_context is missing or too thin. """ sp = scenario_pack or {} training_goal = (_safe_get(sp, ["criticalInfo", "trainingGoal"]) or sp.get("trainingGoal") or "").strip() core_concern = (_safe_get(sp, ["criticalInfo", "coreConcern"]) or sp.get("coreConcern") or "").strip() product = (_safe_get(sp, ["product"]) or sp.get("product") or sp.get("drugName") or sp.get("drug") or "").strip() learner_role = (_safe_get(sp, ["roleSetup", "learnerRoleTitle"]) or sp.get("learnerRoleTitle") or "学习者").strip() scene_type = ( _safe_get(sp, ["criticalInfo", "sceneType"]) or _safe_get(sp, ["criticalInfo", "contextType"]) or sp.get("sceneType") or sp.get("contextType") or "" ) scene_type = str(scene_type or "").strip().lower() # Product facts in coach context must come from scene/background injection; avoid hardcoding drug-specific claims. product_brief = _first_sentence((_safe_get(sp, ["background"]) or sp.get("background") or "").strip(), max_len=160) if not product_brief: product_brief = f"产品：{product or '（待补充）'}（以场景注入的产品知识为准）" content_name = product or "该内容" # NOTE: We intentionally do NOT infer "what sections to include" via keywords in free text. # Use a structured scene type signal (sceneType/contextType) when available. is_training = scene_type in {"training", "enablement", "internal_training", "system_training", "system-enablement", "system"} lines = ["## 期望学习者行为"] if is_training: lines.append("- 开场与目标：说明培训/演示范围，征得对方同意，明确本次要解决的问题与预期产出。") lines.append("- 结构化讲解：按步骤/模块讲清流程与关键概念，遇到疑点先澄清再继续。") lines.append(f"- 关键信息点：准确说明{content_name}的用途/流程/关键差异点（以场景注入资料为准），不编造证据与结论。") if training_goal: lines.append(f"- 目标对齐：{training_goal}") lines.append("- 主动确认理解：用简短提问或复述确认对方是否理解关键步骤与注意事项。") lines.append("- 注意事项与常见误区（必须）：只陈述场景注入资料已覆盖的注意点；未提供的细节要明确“需要回去核对资料”，禁止编造。") lines.append("- 练习/检查点（必须）：给出1-3个可执行检查点（如：当场演示一次、复盘一次关键步骤、用清单核对必填项）。") lines.append("- 下一步行动：给出可选的后续动作（如：补充材料、二次培训、答疑），不强制要求具体日期时间。") else: lines.append("- 自然开场：以请教或分享切入，先征得对方同意，说明想简单交流与对方最关心点相关内容。") lines.append("- 倾听与回应：认真听对方对核心顾虑的看法，用简短语言表达理解与共情。") lines.append(f"- 简要信息介绍：准确说明{content_name}的用途与关键差异点（以场景注入资料为准），不编造证据与结论。") if training_goal: lines.append(f"- 目标对齐：{training_goal}") lines.append("- 探寻顾虑：围绕“{核心顾虑}”探寻对方真实关注点（可包括合规边界、可操作性、资源约束与风险点等维度）。".replace("{核心顾虑}", core_concern or "核心顾虑")) lines.append("- 讨论效果与边界：在对方愿意了解的前提下，简要说明可预期的效果与适用条件，并强调以场景注入口径为准。") lines.append("- 风险与边界（必须）：主动说明风险边界、适用条件与不适用情况；若场景未注入风险信息，则必须承认信息缺口并回到“说明书/指南/合规口径需核对”，禁止编造。") lines.append("- 尝试缔结（必须给出具体时间）：在尊重对方的前提下，提出下一步并给出具体时间点，例如：“下周三/下次会议前，我准备要点资料并约定再对齐一次口径与适用场景”。") lines += ["", "## 终止条件"] lines.append("- 夸大承诺或绝对化表达。") lines.append("- 回避核心问题或口径反复。") lines.append("- 不尊重对方专业判断或明显逼迫推进。") return "\n".join(lines).strip() + "\n" def _coach_only_context_quality_ok(text: str) -> bool: t = (text or "").strip() if not t: return False base_required = [ "## 期望学习者行为", "## 终止条件", ] if not all(r in t for r in base_required): return False # Accept either template: # - "go-to-market / sales" flavor: must include risk+close sections # - "training" flavor: must include training-specific required sections gtm_required = ["风险与边界", "尝试缔结"] training_required = ["注意事项与常见误区", "练习/检查点"] return all(r in t for r in gtm_required) or all(r in t for r in training_required) def _coach_only_context_matches_scene_type(scenario_pack: dict, coach_only_context: str) -> bool: sp = scenario_pack or {} scene_type = ( _safe_get(sp, ["criticalInfo", "sceneType"]) or _safe_get(sp, ["criticalInfo", "contextType"]) or sp.get("sceneType") or sp.get("contextType") or "" ) scene_type = str(scene_type or "").strip().lower() is_training = scene_type in {"training", "enablement", "internal_training", "system_training", "system-enablement", "system"} t = (coach_only_context or "").strip() if not t: return False gtm_markers = ["风险与边界", "尝试缔结"] training_markers = ["注意事项与常见误区", "练习/检查点"] if is_training: return all(m in t for m in training_markers) return all(m in t for m in gtm_markers) def create_knowledge_if_needed_by_dedup( client: TBSClient, drug_id, knowledge_draft: dict, category: str, content_fingerprint: str = None, dry_run: bool = False, ): # Fetch existing knowledge and try to match by normalized-content fingerprint. # Endpoint supports: GET /api/v1/resources/knowledge?drug_id=...&category=... # (Query params aren't fully specified in the reference doc, so we keep best-effort.) def fetch_existing(): path = f"/api/v1/resources/knowledge?drug_id={drug_id}" if category: path += f"&category={urllib.parse.quote(category)}" payload = client.request_json("GET", path) return extract_data(payload) try: existing = fetch_existing() except Exception: existing = [] if isinstance(existing, dict) and "data" in existing: existing = existing["data"] if not isinstance(existing, list): existing = [] title = knowledge_draft.get("title") or "" content = knowledge_draft.get("content") or "" content_fp = content_fingerprint or fingerprint_sha256_of_text(content) # Match existing by fingerprint of content. for it in existing: it_id = guess_entity_id(it) it_content = it.get("content") if isinstance(it, dict) else None it_content = it_content if it_content is not None else it.get("text") if isinstance(it, dict) else None it_fp = fingerprint_sha256_of_text(it_content or "") if it_id and it_fp == content_fp: return it_id # Not found => create if dry_run: return None body = { "drug_id": drug_id, "category": category, "title": title, "content": content, } resp = client.request_json("POST", "/api/v1/admin/basic/knowledge", body=body) data = extract_data(resp) return data def main(): ap = argparse.ArgumentParser() ap.add_argument( "--input", default=os.path.join(_RUNTIME_DIR, "scenario_draft.json"), help="Path to scenario_draft.json (or apiDraft JSON). Use '-' to read draft JSON from stdin.", ) ap.add_argument("--scene-id", default=None, help="If provided, skip scene creation and only generate snapshot+publish for this scene.") ap.add_argument("--base-url", default=os.environ.get("TBS_BASE_URL", "https://sg-tbs-manage.mediportal.com.cn")) ap.add_argument("--access-token", default=None) ap.add_argument("--insecure-ssl", action="store_true", help="Skip SSL certificate verification (staging only).") ap.add_argument("--persisted-ids", default=os.path.join(_RUNTIME_DIR, "P1_persisted_ids.md")) ap.add_argument("--dry-run", action="store_true") ap.add_argument("--strict-param-only", action="store_true", help="Disable all fallback/auto-rebuild behavior.") args = ap.parse_args() strict_param_only = True if os.environ.get("TBS_STRICT_PARAM_ONLY", "1") == "1" else bool(args.strict_param_only) env_token, _ = resolve_access_token() access_token = (args.access_token or env_token or "").strip() if not access_token: raise SystemExit("Missing --access-token / env XG_USER_TOKEN") client = TBSClient( base_url=args.base_url, access_token=access_token, insecure_ssl=bool(args.insecure_ssl), ) if args.input == "-": raw = sys.stdin.read() if not raw.strip(): raise SystemExit("Missing draft JSON on stdin (use --input -).") draft = json.loads(raw) else: with open(args.input, "r", encoding="utf-8") as f: draft = json.load(f) api_draft = draft.get("apiDraft") or {} scenario_pack = draft.get("scenarioPack") or {} api_draft_doctors_raw = api_draft.get("doctors") or {} api_draft_scenes_raw = api_draft.get("scenes") or {} # Normalize shapes: # - Some sessions write apiDraft.doctors/scenes as arrays (list[object]) # - Others write them as dicts (object) api_draft_doctors = api_draft_doctors_raw if isinstance(api_draft_doctors_raw, list): api_draft_doctors = api_draft_doctors_raw[0] if api_draft_doctors_raw else {} api_draft_scenes = api_draft_scenes_raw if isinstance(api_draft_scenes_raw, list): api_draft_scenes = api_draft_scenes_raw[0] if api_draft_scenes_raw else {} resolved_scene_ids = resolve_ids_for_scene(client, api_draft_scenes, dry_run=bool(args.dry_run)) dep_id = resolved_scene_ids["department_id"] bd_id = resolved_scene_ids["business_domain_id"] drug_id = resolved_scene_ids["drug_id"] if not dep_id or not bd_id or not drug_id: raise RuntimeError(f"Failed to resolve ids: department={dep_id}, business_domain={bd_id}, drug={drug_id}") doctor_persona_id = create_doctor_if_missing( client, api_draft_doctors, scenario_pack=scenario_pack, dry_run=bool(args.dry_run), ) if not doctor_persona_id and args.dry_run: raise RuntimeError("Dry-run: doctor persona not found; cannot create in dry-run.") # Knowledge ids: # - Prefer apiDraft.knowledge if present (user provided productKnowledge) # - Else: select best matching existing knowledge by intent from scenario_pack # - Final fallback: persisted IDs (for continuity in early integration stages) knowledge_ids = [] # Support multiple keys for the knowledge draft array. knowledge_drafts = ( api_draft.get("knowledge") or api_draft.get("apiDraftKnowledge") or api_draft.get("apiDraftKnowledges") or [] ) if isinstance(knowledge_drafts, dict) and "items" in knowledge_drafts: knowledge_drafts = knowledge_drafts["items"] if isinstance(knowledge_drafts, list) and knowledge_drafts: for kd in knowledge_drafts: if not isinstance(kd, dict): continue category = kd.get("category") or kd.get("knowledge_category") or "" content = kd.get("content") or "" fp = fingerprint_sha256_of_text(content) knowledge_id = create_knowledge_if_needed_by_dedup( client=client, drug_id=drug_id, knowledge_draft=kd, category=category, content_fingerprint=fp, dry_run=bool(args.dry_run), ) if knowledge_id: knowledge_ids.append(str(knowledge_id)) else: if strict_param_only: raise RuntimeError("strict-param-only: apiDraft.knowledge 必须由上游参数提供，禁止 fallback 选择。") knowledge_ids, knowledge_selection_report = select_knowledge_ids_by_intent(client, drug_id, scenario_pack, top_n=4) if not knowledge_ids: raise RuntimeError( "未命中可用产品知识。请先提供 knowledge 草案后重试：" "每条需包含 drug(或drug_id)、category、title、content；" "系统将创建后自动记录并关联 knowledge_id。" ) # Scene create scene_title = api_draft_scenes.get("title") or api_draft_scenes.get("name") rep_briefing = api_draft_scenes.get("rep_briefing") or "" doctor_only_context = api_draft_scenes.get("doctor_only_context") or "" coach_only_context = api_draft_scenes.get("coach_only_context") or "" if strict_param_only: if not scene_title: raise RuntimeError("strict-param-only: scenes.title/name 必填。") if not isinstance(rep_briefing, str) or not rep_briefing.strip(): raise RuntimeError("strict-param-only: scenes.rep_briefing 必填，禁止文本兜底。") if not isinstance(doctor_only_context, str) or not doctor_only_context.strip(): raise RuntimeError("strict-param-only: scenes.doctor_only_context 必填，禁止文本兜底。") if not isinstance(coach_only_context, str) or not coach_only_context.strip(): raise RuntimeError("strict-param-only: scenes.coach_only_context 必填，禁止文本兜底。") # Fallback: build richer publish-ish markdown from scenarioPack if too thin. # The UI expects structured blocks; older drafts often store only short sentences. if not isinstance(doctor_only_context, str): doctor_only_context = str(doctor_only_context or "") if not isinstance(coach_only_context, str): coach_only_context = str(coach_only_context or "") # Enforce rep_briefing quality first (rep_briefing has no fallback previously). if not strict_param_only: try: need_rebuild = ( not isinstance(rep_briefing, str) or not rep_briefing.strip() or ("【" in rep_briefing or "】" in rep_briefing) or "待补充" in rep_briefing or (len(rep_briefing) > 180) or (scenario_pack.get("location") and str(scenario_pack.get("location")) not in rep_briefing) or (scenario_pack.get("department") and str(scenario_pack.get("department")) not in rep_briefing) or (scenario_pack.get("product") and str(scenario_pack.get("product")) not in rep_briefing) ) if need_rebuild: rep_briefing = build_rep_briefing_from_scenario_pack(scenario_pack, api_draft_scenes, api_draft_doctors) except Exception: rep_briefing = build_rep_briefing_relaxed(scenario_pack, api_draft_scenes, api_draft_doctors) # Doctor-only: enforce strict 3-section structure; rebuild when it doesn't match. if strict_param_only: if (not _doctor_only_context_quality_ok(doctor_only_context)): raise RuntimeError("strict-param-only: doctor_only_context 质量不达标，禁止自动重建。") else: if ( (not doctor_only_context.strip()) or ("待补充" in doctor_only_context) or (not _doctor_only_context_quality_ok(doctor_only_context)) ): doctor_only_context = build_doctor_only_context_from_simple_scenario_pack(scenario_pack, api_draft_doctors, api_draft_scenes) if (not doctor_only_context.strip()) or ("##" not in doctor_only_context): doctor_only_context = build_doctor_only_context_from_scenario_pack(scenario_pack) if strict_param_only: if (not _coach_only_context_quality_ok(coach_only_context)) or (not _coach_only_context_matches_scene_type(scenario_pack, coach_only_context)): raise RuntimeError("strict-param-only: coach_only_context 质量不达标，禁止自动重建。") else: if (not coach_only_context.strip()) or (not _coach_only_context_quality_ok(coach_only_context)) or (not _coach_only_context_matches_scene_type(scenario_pack, coach_only_context)): coach_only_context = build_coach_only_context_from_scenario_pack(scenario_pack) # Optional: write back enforced fields into input draft. # This helps debugging and keeps downstream artefacts consistent. if not bool(args.dry_run): try: api_draft_scenes["rep_briefing"] = rep_briefing api_draft_scenes["doctor_only_context"] = doctor_only_context with open(args.input, "w", encoding="utf-8") as f: json.dump(draft, f, ensure_ascii=False, indent=2) except Exception: pass # Rounds: derive from constraints if present; else default. rounds = 5 constraints = scenario_pack.get("constraints") or {} try: mt = constraints.get("maxTurns") or constraints.get("max_turns") if isinstance(mt, int) and mt > 0: rounds = mt elif isinstance(mt, str) and mt.isdigit(): rounds = int(mt) except Exception: pass persona_ids = [ { "persona_id": str(doctor_persona_id), "difficulty": "medium", "is_default": True, "rounds": rounds, } ] scene_location = api_draft_scenes.get("location") if strict_param_only and (not isinstance(scene_location, str) or not scene_location.strip()): raise RuntimeError("strict-param-only: scenes.location 必填，禁止从 scenarioPack 兜底。") scene_body = { "title": scene_title, "department_id": dep_id, "drug_id": drug_id, "business_domain_id": bd_id, "location": scene_location if isinstance(scene_location, str) else (api_draft_scenes.get("location") or scenario_pack.get("location") or ""), "doctor_only_context": doctor_only_context, "coach_only_context": coach_only_context, "rep_briefing": rep_briefing, "persona_ids": persona_ids, "knowledge_ids": knowledge_ids, # status omitted => backend default } if args.dry_run: knowledge_report = knowledge_selection_report if "knowledge_selection_report" in locals() else None print( toon_encode( { "ok": True, "step": "tbs_write_executor", "dryRun": True, "resolvedIds": { "department_id": str(dep_id), "business_domain_id": str(bd_id), "drug_id": str(drug_id), "persona_id": str(doctor_persona_id), }, "knowledgeIds": [str(x) for x in knowledge_ids], "knowledgeSelectionReport": knowledge_report, "sceneBodyPreview": scene_body, } ) ) return if args.scene_id: scene_id = str(args.scene_id) else: scene_resp = client.request_json("POST", "/api/v1/admin/scenes", body=scene_body) scene_data = extract_data(scene_resp) scene_id = None if isinstance(scene_data, dict): scene_id = scene_data.get("id") or scene_data.get("scene_id") or guess_entity_id(scene_data) else: scene_id = scene_data if not scene_id: raise RuntimeError(f"Scene creation failed: {scene_resp}") scene_id = str(scene_id) snapshot_resp = client.request_json( "POST", f"/api/v1/scenes/{scene_id}/snapshot", body={"doctor_id": int(str(doctor_persona_id))}, ) snapshot_data = extract_data(snapshot_resp) snapshot_id = None if isinstance(snapshot_data, dict): snapshot_id = snapshot_data.get("id") or snapshot_data.get("snapshot_id") or guess_entity_id(snapshot_data) else: snapshot_id = snapshot_data if not snapshot_id: raise RuntimeError(f"Snapshot creation failed: {snapshot_resp}") snapshot_id = str(snapshot_id) publish_resp = client.request_json("POST", f"/api/v1/snapshots/{snapshot_id}/publish", body={}) out = { "scene_id": str(scene_id), "snapshot_id": str(snapshot_id), "publish_response": publish_resp, "knowledge_selection_report": knowledge_selection_report if "knowledge_selection_report" in locals() else None, "resolved": { "department_id": str(dep_id), "business_domain_id": str(bd_id), "drug_id": str(drug_id), "persona_id": str(doctor_persona_id), "knowledge_ids": [str(x) for x in knowledge_ids], }, } print(toon_encode({"ok": True, "step": "tbs_write_executor", "result": out})) if __name__ == "__main__": # Fix: urllib.parse is used in create_knowledge_if_needed_by_dedup; import here to keep top clean. import urllib.parse # noqa: E402 main() FILE:scripts/scene/validate-and-gate.py #!/usr/bin/env python3 """ scene / validate-and-gate — 契约校验脚本（stdin JSON）。输出 TOON。 API_URL 须与 openapi/scene/validate-and-gate.md 标题 URL 一致；本脚本不发起 HTTP。 """ import json import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common")) from auth_token import require_access_token from toon_encoder import encode as toon_encode from scenario_pack_normalizer import normalize_scenario_pack API_URL = "https://scenario-builder.openclaw.internal/v1/scene/validate-and-gate" ALLOWED_EVIDENCE_STATUS = {"NOT_PROVIDED", "PARTIAL", "READY"} def _read_body(): raw = sys.stdin.read() if not raw.strip(): return {} return json.loads(raw) def _ok(step, **extra): payload = {"ok": True, "step": step, **extra} print(toon_encode(payload)) def main(): require_access_token() body = _read_body() for k in ("scenarioPack", "apiDraft", "validationReport"): if k not in body or not isinstance(body[k], dict): print(f"错误: 缺少 {k}", file=sys.stderr) sys.exit(2) sp = normalize_scenario_pack(body["scenarioPack"]) api_draft = body["apiDraft"] vr = body["validationReport"] evidence_status = sp.get("productEvidenceStatus", "NOT_PROVIDED") if evidence_status not in ALLOWED_EVIDENCE_STATUS: print("错误: productEvidenceStatus 非法", file=sys.stderr) sys.exit(2) issues = list(vr.get("issues") or []) if evidence_status != "READY": if not api_draft.get("needsEvidenceConfirmation"): issues.append("apiDraft.needsEvidenceConfirmation 必须为 true（证据未 READY）") if not sp.get("productEvidenceSource"): issues.append("缺少 productEvidenceSource，需补充知识来源") passed = bool(vr.get("passed")) if issues: passed = False _ok( "validate-and-gate", passed=passed, evidenceStatus=evidence_status, issues=issues, ) if __name__ == "__main__": main() FILE:scripts/tbs_assets/P1_persisted_ids.md FILE:scripts/tbs_assets/README.md # tbs_assets — 本地执行器资产（脚本入口迁移）本目录用于落盘/承载写库所需的本地资产（如 `scenario_draft.json`、枚举缓存等）；并保留 `tbs_write_executor.py` 的兼容入口（委托到 `scripts/scene/`）。当前工程支持以下回退策略： - 若目录内缺少资产文件，则执行器会回退到 legacy 的 `runtime/` 资产目录。 ## 发布声明（必读）本目录中的文件属于**本地运行时资产**，不是通用业务素材。发布 skill 时，请按以下规则处理： - 不随包分发任何真实凭据或企业标识（例如 token、corp id、登录态缓存）。 - 不随包分发真实业务草稿（例如包含客户/产品上下文的 `scenario_draft.json`）。 - 运行时文件应由脚本在用户本地会话中生成或覆盖，不应作为“默认业务数据”发布。 ## 首次使用要求使用落库相关脚本前，需在启动 OpenClaw 的同一终端设置： - `XG_USER_TOKEN=your_access_token` 执行器只读取当前会话环境变量完成鉴权，不应引导用户在仓库内保存 token 明文文件。 ## 推荐的发布形态 - 保留本 README 与非敏感静态文件（如 `system_business_domains.json`）。 - `scenario_draft.json` 建议使用空模板或由脚本首次执行时自动创建。 - 若发现本目录存在真实 token、corp id、历史业务草稿，请在发布前清理。 FILE:scripts/tbs_assets/scenario_draft.json { "scenarioPack": {}, "apiDraft": {}, "validationReport": {} } FILE:scripts/tbs_assets/system_business_domains.json { "success": true, "data": [ { "id": "1", "name": "临床推广", "sort_order": 1, "status": 1 }, { "id": "2", "name": "院外零售", "sort_order": 2, "status": 1 }, { "id": "3", "name": "学术合作", "sort_order": 3, "status": 1 }, { "id": "4", "name": "通用能力", "sort_order": 4, "status": 1 } ] }

tbs-scenario-builder

Comments (0)