CANN算子白盒用例合并扩展

Step 5c合并 空 tensor 补全 data_range 展开【免费下载链接】cannbot-skillsCANNBot 是面向 CANN 开发的用于提升开发效率的系列智能体本仓库为其提供可复用的 Skills 模块。项目地址: https://gitcode.com/cann/cannbot-skills前置条件Step 5b 完成S5_mapped_cases_network.json 已写入概述输入文件输出文件S5_mapped_cases_path.json、S5_mapped_cases_network.jsonS5_merge_expand.py、S5_mapped_cases_low.json、S5_mapped_cases_high.json通过S5_merge_expand.py依次完成合并 path network 空 tensor → low.jsonall normal→ data_range 交叉展开 → high.json。S5_merge_expand.py脚本分为通用部分照搬和算子专属部分supplement_empty_cases见 §supplement_empty_cases 策略。子 agent 先照搬通用部分再根据算子 param_type 生成supplement_empty_cases并插入脚本最后按 §执行 中的步骤依次运行。S5_merge_expand.py — 合并 空 tensor 补全 元素数过滤 data_range 展开 import json, copy, math, os, re, sys out_dir os.path.dirname(os.path.abspath(__file__)) # 共享工具函数 NON_NORMAL [zero, extreme, negative, tiny_pos, all_ones, near_zero, with_inf, with_nan] _DOMAIN_EXCLUDE { positive: {zero, negative, near_zero, with_nan}, non_negative: {negative, near_zero, with_nan}, non_zero: {zero, near_zero, with_nan}, } def _allowed_ranges(value_domain): 根据 value_domain 返回允许的 NON_NORMAL 标签列表。 if not value_domain: return list(NON_NORMAL) vd_type value_domain[type] if vd_type range: lo, hi value_domain.get(min), value_domain.get(max) excluded {extreme, with_nan, with_inf} if lo is not None and lo 0: excluded.add(zero) excluded.add(negative) elif lo is not None and lo 0: excluded.add(negative) elif hi is not None and hi 0: excluded.add(zero) if hi is not None and hi 1e-7: excluded.add(tiny_pos) if (lo is not None and lo 1) or (hi is not None and hi 1): excluded.add(all_ones) return [dr for dr in NON_NORMAL if dr not in excluded] excluded _DOMAIN_EXCLUDE.get(vd_type, set()) return [dr for dr in NON_NORMAL if dr not in excluded] def set_all_data_range(case, dr): for spec in case[tensors][inputs].values(): if spec is None: continue if isinstance(spec, list): for sub in spec: sub[_data_range] dr else: spec[_data_range] dr case.pop(_data_range, None) return case def expand_high(cases): one-hot 全统一展开。按 _value_domain 过滤不兼容 data_range。 if not cases: return [] expanded [] input_names [n for n, s in cases[0][tensors][inputs].items() if s is not None] for c in cases: nc copy.deepcopy(c) nc[id] f{c[id]}_all_normal set_all_data_range(nc, normal) expanded.append(nc) all_allowed set(NON_NORMAL) for spec in c[tensors][inputs].values(): if spec is None: continue s spec[0] if isinstance(spec, list) else spec all_allowed set(_allowed_ranges(s.get(_value_domain))) for dr in sorted(all_allowed): nc copy.deepcopy(c) nc[id] f{c[id]}_all_{dr} set_all_data_range(nc, dr) expanded.append(nc) for inp in input_names: s c[tensors][inputs][inp] s0 s[0] if isinstance(s, list) else s allowed _allowed_ranges(s0.get(_value_domain)) for dr in allowed: nc copy.deepcopy(c) nc[id] f{c[id]}_{inp}_{dr} for name, spec in nc[tensors][inputs].items(): if spec is None: continue target_dr dr if name inp else normal if isinstance(spec, list): for sub in spec: sub[_data_range] target_dr else: spec[_data_range] target_dr expanded.append(nc) return expanded _COMPACT_KEYS (shape, dtype, id, _group, _data_range, _value_domain) def _compact_json(text): for key in _COMPACT_KEYS: text re.sub( rf({key}): \[\s*\n((?:\s[^\n],\n)*\s[^\n]\n\s*)\], lambda m: f{m.group(1)}: [{, .join(l.strip().rstrip(,) for l in m.group(2).strip().splitlines() if l.strip())}], text, flagsre.MULTILINE ) text re.sub( rf({key}): ([^]*|\d(?:\.\d)?(?:e[-]?\d)?), rf\1: \2, text, flagsre.MULTILINE ) return text # 算子专属supplement_empty_cases根据 param_type 生成见下方 §策略 def supplement_empty_cases(mapped_path_cases, seed42): 从 1 个随机 path case 模板生成空 tensor 变体。 REQUIRED 和 DYNAMIC 分支实现不同子 agent 根据算子 param_type 选择对应模板生成。 pass # 子 agent 根据 §supplement_empty_cases 策略 替换此实现 # 5c 入口合并 空 tensor 过滤 → low def _max_numel(spec): 计算 tensor spec 的最大元素数DYNAMIC 取各子 tensor 最大值 if spec is None: return 0 if isinstance(spec, list): return max((math.prod(sub[shape]) for sub in spec), default0) return math.prod(spec[shape]) def merge_low(): with open(os.path.join(out_dir, S5_mapped_cases_path.json)) as f: path_cases json.load(f)[cases] with open(os.path.join(out_dir, S5_mapped_cases_network.json)) as f: net_cases json.load(f)[cases] path_cases [c for c in path_cases if all(_max_numel(v) 100_000_000 for v in c[tensors][inputs].values())] net_cases [c for c in net_cases if all(_max_numel(v) 100_000_000 for v in c[tensors][inputs].values())] empty_cases supplement_empty_cases(path_cases) vd_map {} model_path os.path.join(out_dir, S2P1_operator_model.json) if os.path.exists(model_path): with open(model_path) as f: op_model json.load(f) for inp in op_model.get(inputs, []): vd inp.get(value_domain) if vd: vd_map[inp[name]] vd def _attach_vd(case): for name, spec in case[tensors][inputs].items(): if spec is None: continue vd vd_map.get(name) if not vd: continue if isinstance(spec, list): for sub in spec: sub[_value_domain] vd else: spec[_value_domain] vd return case combined path_cases net_cases empty_cases low [] for c in combined: c set_all_data_range(copy.deepcopy(c), normal) _attach_vd(c) low.append(c) with open(os.path.join(out_dir, S5_mapped_cases_low.json), w) as f: raw json.dumps({cases: low}, indent2, ensure_asciiFalse) f.write(_compact_json(raw)) # 5d 入口data_range 展开 → high def expand_high_main(): with open(os.path.join(out_dir, S5_mapped_cases_low.json)) as f: low_cases json.load(f)[cases] def _is_empty(c): cid c.get(id, ) return cid.endswith(_empty) empty_cases [c for c in low_cases if _is_empty(c)] path_cases [c for c in low_cases if not _is_empty(c) and c[params].get(_group) ! network] net_cases [c for c in low_cases if c[params].get(_group) network] high expand_high(path_cases) empty_cases expand_high(net_cases) with open(os.path.join(out_dir, S5_mapped_cases_high.json), w) as f: json.dump({cases: high}, f, ensure_asciiFalse) if __name__ __main__: if len(sys.argv) 1 and sys.argv[1] 5d: expand_high_main() else: merge_low()逻辑说明过滤规则剔除任意输入 tensor 元素数 1 亿的 path/network case规则相同。DYNAMIClist[dict]取各子 tensor 元素数最大值。supplement_empty_cases 策略函数supplement_empty_cases定义在 S5_merge_expand.py 中算子专属部分。 从 1 个随机 path case 模板生成空 tensor 变体。通用约束签名supplement_empty_cases(mapped_path_cases, seed42)返回list[dict]格式同 mapped_path_cases 元素含 id/params/tensors每个空 case 的_data_range normalID 后缀REQUIRED 统一为_empty含全零兜底DYNAMIC 为_empty位置变体或_all_empty全零兜底。_is_empty通过endswith(_empty)检测兼容两种REQUIRED 分支单 tensorREQUIRED 空 tensor 实现高度算子专属无通用代码模板。子 agent 根据 S5_mapping_spec.md 和 operator_model 生成具体实现。策略Phase 1 — 可空化维度识别从 S5_mapping_spec.md §shape 构造参数 中读取每个 shape 参数的控制关系参数名 → 控制的 tensor → 控制的 dim 索引确定哪些 input tensor 的哪些维度可以被置零。const input值和 shape 均固定的 tensor不随 case 参数变化不参与空化。Phase 2 — 变体生成对每个可空化的tensor, dim组合从 1 个随机 path case 模板生成 1 个变体仅将该维度置 0其余维度保持原值跨变体去重按所有 input shape 签名去重追加 1 个全零兜底变体仅将 Phase 1 识别的可空化 tensor 全维度置 0const input 保持原值按 sync_with 传播零化到关联 tensor更新 outputs将 params 中受 shape 变更影响的字段同步为与修改后 shape 一致的值不变约束签名supplement_empty_cases(mapped_path_cases, seed42)返回list[dict]格式同 mapped_path_cases 元素每个空 case 的_data_range normal设置params[_group]如{affected_key}_emptyID 后缀统一为_empty含全零兜底供_is_empty检测DYNAMIC 分支TensorListDYNAMIC 空 tensor 不逐子 tensor 逐维度枚举子 tensor 数量可达 50逐维度枚举导致场景爆炸改为按子 tensor 在列表中的位置采样。策略从 1 个随机 path case 模板生成变体筛选 ndim0 的子 tensor 索引跳过 scalarscalar 无法置零单个维度first/middle/last取筛选后索引列表的首/中/末位置各 1 个变体将该子 tensor 全维度置 0partial取 ndim 最高的子 tensor将其最大维度值置 0保留其余维度结构仅 ndim≥2 时生成all_empty所有子 tensor 全维度置 0兜底ID 后缀_all_empty位置去重若首/中/末指向同一索引仅 1 个非 scalar 子 tensor只生成 1 个变体sync_with 传播全维度置 0 → 目标对应位置全维度置 0partial → 目标同维度置 0_groupDYNAMIC 空 case 不设置 _group继承模板值空 tensor 检测依赖 ID 后缀见通用约束不依赖 _groupDYNAMIC 代码模板子 agent 替换name/out变量值L226-227 引号内的 {name}/{out}{target}/{target_out}文本替换L272 等引号内的占位符输出推导假设以下模板假设输出 shape 规则为same_as_input02 §4。若算子输出规则为derived或fixed子 agent 需修改模板中 outputs 更新逻辑。def supplement_empty_cases(mapped_path_cases, seed42): if not mapped_path_cases: return [] import random, copy rng random.Random(seed) template rng.choice(mapped_path_cases) name {name} out {out} max_id 0 for c in mapped_path_cases: cid c.get(id, case00000) num int(cid.replace(case, )) if num max_id: max_id num # Phase 1筛选 ndim0 的子 tensor 索引 valid_indices [i for i, sub in enumerate(template[tensors][inputs][name]) if len(sub[shape]) 0] positions [] seen set() for label, pos in [(first_empty, valid_indices[0] if valid_indices else -1), (middle_empty, valid_indices[len(valid_indices) // 2] if valid_indices else -1), (last_empty, valid_indices[-1] if valid_indices else -1)]: if pos 0 and pos not in seen: positions.append((label, pos, full)) seen.add(pos) if valid_indices: partial_idx max(valid_indices, keylambda i: len(template[tensors][inputs][name][i][shape])) partial_ndim len(template[tensors][inputs][name][partial_idx][shape]) if partial_ndim 2: positions.append((partial_empty, partial_idx, partial)) # Phase 2变体生成 empty_cases [] idx 0 for label, tensor_idx, mode in positions: variant copy.deepcopy(template) sub variant[tensors][inputs][name][tensor_idx] if mode full: sub[shape] tuple([0] * len(sub[shape])) else: # partial: 最大维度值置零 shape list(sub[shape]) max_dim shape.index(max(shape)) shape[max_dim] 0 sub[shape] tuple(shape) # 更新 outputssame_as_input 同步 variant[tensors][outputs][out][tensor_idx][shape] sub[shape] # sync_with 传播无 sync_with 时删除此段 if {target} in variant[tensors][inputs]: target_sub variant[tensors][inputs][{target}][tensor_idx] if mode full: target_sub[shape] tuple([0] * len(target_sub[shape])) else: target_shape list(target_sub[shape]) if len(target_shape) max_dim: target_shape[max_dim] 0 target_sub[shape] tuple(target_shape) if {target_out} in variant[tensors][outputs]: variant[tensors][outputs][{target_out}][tensor_idx][shape] target_sub[shape] variant[id] fcase{max_id1idx:05d}_{name}_{label} variant[_data_range] normal idx 1 empty_cases.append(variant) # all_empty 兜底 all_empty copy.deepcopy(template) for sub in all_empty[tensors][inputs][name]: sub[shape] tuple([0] * len(sub[shape])) for sub in all_empty[tensors][outputs][out]: sub[shape] tuple([0] * len(sub[shape])) if {target} in all_empty[tensors][inputs]: for sub in all_empty[tensors][inputs][{target}]: sub[shape] tuple([0] * len(sub[shape])) if {target_out} in all_empty[tensors][outputs]: for sub in all_empty[tensors][outputs][{target_out}]: sub[shape] tuple([0] * len(sub[shape])) all_empty[id] fcase{max_id1idx:05d}_{name}_all_empty all_empty[_data_range] normal empty_cases.append(all_empty) return empty_cases空 tensor 在 low / high 中的传递low由supplement_empty_cases生成后合并同其他 case 一样设_data_range: normalhigh按id后缀_empty从 low 中分离直接保留不经过expand_highlow 档位全 normal所有输入 tensor_data_range统一设为normal。输出格式文件格式S5_mapped_cases_low.json缩进 选择性压缩_compact_json压缩 shape/dtype/id/_group/_data_rangeS5_mapped_cases_high.json紧凑格式无缩进输出示例以下示例展示_compact_json()的实际压缩效果供子 agent 生成脚本时参考。示例S5_mapped_cases_low.json 压缩前 vs 压缩后_compact_json()将 5 个 keyshape、dtype、id、_group、_data_range从多行缩进展平为单行。压缩前JSON indent2 展开形式{ cases: [ { id: case00000, params: { _group: {group_id}, {param_1}: value_1, {param_2}: value_2 }, tensors: { inputs: { {tensor_a}: { shape: [ 4, 8, 16 ], dtype: float16 } }, outputs: { {tensor_out}: { shape: [ 4, 8, 16 ], dtype: float16 } } } } ] }压缩后low.json 实际写入格式5 个 key 展平为单行其他结构不变{ cases: [ { id: case00000, params: { _group: {group_id}, {param_1}: value_1, {param_2}: value_2 }, tensors: { inputs: { {tensor_a}: {shape: [4, 8, 16], dtype: float16} }, outputs: { {tensor_out}: {shape: [4, 8, 16], dtype: float16} } } } ] }压缩规则被压缩的 keyshape、dtype、id、_group、_data_range不被压缩的 key如params、tensors、各 tensor 名保持原有缩进S5_mapped_cases_low.json中所有 case 的_data_range均为normal约束_data_range存储在tensors.inputs.{name}._data_rangeper-input 独立set_all_data_range统一写入同时清除 5a 遗留的顶层_data_rangeDYNAMIC 输入list[dict]的每个子 tensor 均写入_data_range同一个 TensorList 内所有子 tensor 的值相同语义上共享同一个 data_range空 tensor 不参与 data_range 交叉固定_data_range: normal空 tensor 通过id后缀_empty检测不依赖_group字段每个输入 tensor 元素数 ≤ 100,000,0001 亿。超出的 path/network case 直接丢弃执行子 agent 须按以下顺序完成将上方通用部分写入{whitebox}/S5_merge_expand.py根据算子 param_type 生成supplement_empty_casesREQUIRED 或 DYNAMIC 分支替换脚本中的pass占位运行python S5_merge_expand.py→S5_mapped_cases_low.json运行python S5_merge_expand.py 5d→S5_mapped_cases_high.json检查清单S5_merge_expand.py已写入{whitebox}/supplement_empty_cases已根据算子 param_type 正确生成REQUIRED 或 DYNAMIC 分支python S5_merge_expand.py退出码 0S5_mapped_cases_low.json已写入缩进 压缩格式python S5_merge_expand.py 5d退出码 0S5_mapped_cases_high.json已写入紧凑格式空 tensor 变体数量 0【免费下载链接】cannbot-skillsCANNBot 是面向 CANN 开发的用于提升开发效率的系列智能体本仓库为其提供可复用的 Skills 模块。项目地址: https://gitcode.com/cann/cannbot-skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考