Agent 创造性问题解决：类比推理、工具组合创新、假设驱动探索与创造性与可靠性的张力

作者：yumking | 2026 年 10 月 8 日 | 技术标签：创造性问题解决 / 类比推理 / 工具组合 / 假设驱动 / 探索与利用 / 可靠性张力

摘要

第 40 篇让记忆驱动规划——经验预避已知坑（纠错不如避错），但预避只能避"见过的坑"，任务没有历史经验、没有现成工具、没有明确流程时怎么办？第 41 篇让 Agent 有能力自知——能力画像内的任务自信执行、边界附近的求助，但能力边界之外的创造性任务（“帮我设计一个没人做过的监控方案”）既不在画像内也不该直接拒绝。第 43 篇闭合了纠错环——犯了能改、改了能记、记了能避，但纠错和预避都是"reactive"（对已发生错误的响应），创造性是"proactive"——没有路就造一条路。第 39 篇的 ToT/MCTS 是"搜索已知空间"（候选生成器知道方向），创造性是"探索未知空间"（候选生成器不知道答案在哪）。四篇停在"用已知方法解决已知类型的问题"，没走到"用已有能力解决没见过的问题"。本文补上创造性这一层：类比推理（结构映射理论——从源域提取关系结构映射到目标域，不是表面相似而是关系相似；跨域类比源检索+映射验证+迁移适配）、工具组合创新（组合空间爆炸问题——N 个工具的 2 组合是 N(N-1)/2、3 组合是 N(N-1)(N-2)/6；功能-效果矩阵+组合模式库=管道/守护/反馈/仲裁四模式+约束剪枝搜索）、假设驱动探索（科学方法式试错——观察异常→生成假设→设计最小实验→检验→修正；假设优先级排序=先验概率×信息增益/成本；顺序消解 vs 平行消解）、创造性与可靠性的张力（创造性方案天然不可靠——没有历史成功率数据；探索-利用平衡=ε-greedy 变体按任务风险调整；创造性方案的三级安全网=沙箱隔离验证/影子模式对照/渐进放量+金丝雀）。实测无先例任务解决率从 12%（纯 ReAct 硬试）提至 71%、类比迁移首次方案可用率 64%、工具组合发现有效组合所需尝试次数 47→6、假设驱动探索定位根因实验次数 12.3→3.1、创造性方案上线事故率 8.4%→0.9%（三级安全网）、探索成本可控（创造性任务平均 8.2 次调用 vs 预算 12）。

一、第 40/41/43 篇之后：会避坑、会纠错、有自知，但不会"发明"

1.1 四篇的接力与共同天花板

第 39 篇（规划深潜）→ ToT/MCTS/Reflexion——搜索已知空间
  → 解决了"已知任务类型怎么选最优路径"
  → 没解决："候选生成器不知道方向"的未知问题——ToT 的 K 个候选
           还是依赖"这类任务通常这么做"的先验

第 40 篇（记忆×规划）→ 经验预避已知坑
  → 解决了"见过的坑不再踩"
  → 没解决："没见过的坑"——记忆库里没有先例，预避检索返回空

第 41 篇（元认知）→ 能力边界自建模
  → 解决了"能力内自信执行/能力外求助"
  → 没解决："边界之外的创造性任务"——求助也没用（没人做过），
           拒绝又太保守（人类专家会"试着造一条路"）

第 43 篇（自我纠错）→ 犯了能改、记了能避
  → 解决了"错误的闭环"
  → 但纠错/预避都是 reactive——对已发生错误的响应
  → 没解决："proactive 的造路"——没有路时怎么走

四篇形成一个递进但未闭合的链条：

已知问题已知方法（第39篇规划搜索）
  → 已知坑预避（第40篇经验记忆）
  → 能力边界自知（第41篇元认知）
  → 错了能改能记（第43篇纠错环）
  → ??? 没见过的问题怎么解决（本文：创造性）

1.2 不会创造的三个翻车

翻车一：没有先例就硬试或拒绝
  用户："帮我们设计一个跨 3 个云厂商的成本监控方案"（Agent 没做过）
  → Agent 查经验记忆：预避检索返回空（没做过）
  → 查能力画像：越界任务，按第 41 篇该求助
  → 求助人类："这个任务我没做过"——但人类专家也不会手把手教，
    他会说"你参考一下单云的做法，想办法拼一下"
  → Agent 硬试：直接用单云方案的模板套 → 三云的账单格式完全不同 → 失败
  根因：不会类比推理——"单云监控"和"三云监控"的关系结构相似
        （数据采集→归一化→聚合→告警），只是源端格式不同
        应该迁移"监控管线"的关系结构，替换源端适配层

翻车二：有工具但不会组合
  任务："检测这个仓库的依赖是否有已知漏洞，并生成升级 PR"
  → Agent 的工具箱里有：code_search（搜代码）、web_search（搜 CVE）、
    file_edit（改文件）、git_pr（提 PR）、python_exec（跑脚本）
  → 单个工具都做不了这个任务
  → Agent 逐个尝试单工具 → 全失败 → 报告"任务超出能力范围"
  → 实际上正确的组合是：
    python_exec(解析依赖清单) → web_search(逐个查 CVE) →
    file_edit(升级版本号) → python_exec(本地验证) → git_pr(提 PR)
  根因：不会工具组合创新——单工具能力 ≠ 任务需求，
        但工具的组合可以覆盖。N 个工具的组合空间没被系统性搜索

翻车三：创造性方案直接上线
  Agent 为"日志聚合延迟"问题发明了一个新方案：
    用采样+聚合替代全量聚合（创造性！没历史先例）
  → 信心满满直接上线 → 采样丢掉了 3% 的 ERROR 日志 → 审计事故
  根因：创造性方案天然不可靠（没历史数据证明可靠），
        直接上线等于拿生产环境当实验田
        创造性需要安全网，创新不等于莽撞

1.3 创造性问题解决要回答的四个问题

问题一：没见过的问题，从哪找解法？
  → 类比推理——从相似领域迁移关系结构

问题二：现有能力够不着，怎么拼出新能力？
  → 工具组合创新——系统性搜索组合空间

问题三：连假设都没有，怎么动手？
  → 假设驱动探索——科学方法式试错

问题四：创造性方案天然不可靠，怎么安全地创新？
  → 创造性与可靠性的张力——探索-利用平衡+三级安全网

1.4 什么时候需要创造性（触发条件）

创造性不是默认模式，是特定条件触发的：

触发条件 1：经验检索返回空
  → 第 40 篇预避检索相似度 < 0.6（没有像样的先例）
  → 记忆库帮不上忙 → 进入创造性模式

触发条件 2：能力画像判定越界但任务可分解
  → 第 41 篇能力画像：成功率 < 0.4（越界）
  → 但任务可分解为若干"部分在能力内"的子任务
  → 求助成本高（用户不在/用户也不会）→ 进入创造性模式

触发条件 3：常规方案连续失败
  → 第 43 篇纠错环：同一任务连续 3 次失败，根因都是"方法本身不对"
  → 已知方法已穷尽 → 进入创造性模式

触发条件 4：用户显式要求创新
  → "设计一个新方案"/"没人做过这个"——显式创造性信号
  → 直接进入创造性模式

四个条件任一满足即触发，多条件同时满足提高创造性深度

二、类比推理：从相似领域迁移方法

2.1 结构映射：关系相似而非表面相似

类比推理的认知科学基础（Gentner 结构映射理论）：

表面相似：对象相似（都是"数据库"/都是"Python"）
  → 误导性强：表面像但关系结构不像 → 迁移失败
  → 例：单云监控和三云监控表面都"监控"，但难点不同

关系相似：关系结构相似（实体间的因果关系/流程结构相同）
  → 迁移价值高：关系结构相同 → 方法可迁移
  → 例：单云监控的关系结构 = 采集→归一化→聚合→告警
        三云监控的关系结构 = 采集(×3)→归一化(格式不同!)→聚合→告警
  → 结构映射：管线骨架相同，只需替换"归一化"节点的实现
  → 迁移方案：保留管线骨架，源端加适配层（正是翻车一的正确解法）

类比推理 = 找到关系结构相似的源问题 → 提取关系结构 → 映射到目标域 → 适配差异

2.2 结构映射实现

from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from typing import Any

@dataclass
class problem_schema:
    """问题的关系结构表示"""
    problem_id: str
    domain: str                       # 领域（monitoring/deploy/data_pipeline...）
    entities: list[str]               # 实体（collector/normalizer/aggregator...）
    relations: list[dict]             # 关系（采集→归一化 依赖...）
    goals: list[str]                  # 目标（统一视图/告警及时...）
    constraints: list[str]            # 约束（实时性/成本/合规...）
    pain_points: list[str]            # 难点（源端格式异构/数据量大...）

    def structural_signature(self) -> str:
        """结构签名：实体角色+关系形状（剥离表面特征）"""
        role_map = self._abstract_roles()
        relation_shape = [f"{role_map.get(r['from'], '?')}"
                          f"-{r['type']}->{role_map.get(r['to'], '?')}"
                          for r in self.relations]
        return "|".join(sorted(relation_shape))

    def _abstract_roles(self) -> dict[str, str]:
        """实体抽象为功能角色：data_source→SOURCE，processor→TRANSFORM..."""
        roles = {}
        for e in self.entities:
            roles[e] = self._infer_role(e)
        return roles

    def _infer_role(self, entity: str) -> str:
        e = entity.lower()
        if any(k in e for k in ("collect", "fetch", "pull", "source", "采集")):
            return "SOURCE"
        if any(k in e for k in ("normal", "clean", "transform", "适配", "转换")):
            return "TRANSFORM"
        if any(k in e for k in ("aggreg", "merge", "join", "聚合")):
            return "AGGREGATE"
        if any(k in e for k in ("alert", "notify", "告警", "通知")):
            return "SINK"
        if any(k in e for k in ("store", "save", "db", "存储")):
            return "STORAGE"
        return "PROCESS"

@dataclass
class analogy_candidate:
    """一个类比候选：源问题 → 目标问题"""
    source: problem_schema
    target: problem_schema
    structural_similarity: float     # 结构相似度（关系形状匹配）
    surface_similarity: float        # 表面相似度（领域/词汇匹配）
    mapping: dict[str, str]          # 实体映射：源实体 → 目标实体
    unmapped_relations: list[str]    # 无法映射的关系（迁移时要适配）
    confidence: float                # 类比置信度

    @property
    def is_high_quality(self) -> bool:
        """高质量类比 = 关系相似度高 且 表面相似度不作主导"""
        return self.structural_similarity >= 0.7

class analogical_reasoner:
    """类比推理器：源检索 → 结构映射 → 迁移适配"""

    def __init__(self, memory_store: Any):
        self.memory_store = memory_store   # 接第 40 篇经验/情景记忆

    async def find_analogies(self, target: problem_schema,
                              top_k: int = 3) -> list[analogy_candidate]:
        """跨域类比源检索：优先关系相似，抑制纯表面相似"""
        candidates = []

        broad_matches = await self.memory_store.search(
            query=" ".join(target.entities + target.goals),
            memory_type="scenario",     # 情景记忆存历史问题解法
            top_k=20,
            min_similarity=0.3,         # 宽召回，跨域类比往往表面不像
        )

        for match in broad_matches:
            source = self._to_schema(match)
            mapping, unmapped = self._structural_map(source, target)
            struct_sim = self._structural_similarity(source, target, mapping)
            surface_sim = match.get("similarity", 0.0)

            candidates.append(analogy_candidate(
                source=source,
                target=target,
                structural_similarity=struct_sim,
                surface_similarity=surface_sim,
                mapping=mapping,
                unmapped_relations=unmapped,
                confidence=self._analogy_confidence(struct_sim, surface_sim, len(unmapped)),
            ))

        candidates.sort(key=lambda c: c.confidence, reverse=True)
        return candidates[:top_k]

    def _structural_map(self, source: problem_schema,
                         target: problem_schema) -> tuple[dict[str, str], list[str]]:
        """结构映射：按功能角色对齐实体"""
        src_roles = source._abstract_roles()
        tgt_roles = target._abstract_roles()

        mapping = {}
        for tgt_entity, tgt_role in tgt_roles.items():
            for src_entity, src_role in src_roles.items():
                if src_role == tgt_role and src_entity not in mapping.values():
                    mapping[src_entity] = tgt_entity
                    break

        mapped_sources = set(mapping.keys())
        unmapped = [
            f"{r['from']}-{r['type']}->{r['to']}"
            for r in source.relations
            if r["from"] not in mapped_sources or r["to"] not in mapped_sources
        ]
        return mapping, unmapped

    def _structural_similarity(self, source: problem_schema,
                                target: problem_schema,
                                mapping: dict[str, str]) -> float:
        """结构相似度：映射后的关系形状匹配度"""
        src_sig = source.structural_signature()
        mapped_relations = []
        for r in source.relations:
            f = mapping.get(r["from"], r["from"])
            t = mapping.get(r["to"], r["to"])
            mapped_relations.append(f"{f}-{r['type']}->{t}")
        tgt_sig = "|".join(sorted(target._relation_shapes()))

        if not src_sig or not tgt_sig:
            return 0.0
        src_set = set(src_sig.split("|"))
        tgt_set = set(tgt_sig.split("|"))
        if not src_set and not tgt_set:
            return 1.0
        return len(src_set & tgt_set) / max(len(src_set | tgt_set), 1)

    def _analogy_confidence(self, struct_sim: float, surface_sim: float,
                             unmapped_count: int) -> float:
        """类比置信度：结构主导，表面辅助，未映射关系惩罚"""
        base = 0.75 * struct_sim + 0.15 * min(surface_sim, 0.5) + 0.10 * (1 - min(unmapped_count / 5, 1))
        return round(base, 3)

    def _to_schema(self, match: dict) -> problem_schema:
        return problem_schema(
            problem_id=match.get("id", "unknown"),
            domain=match.get("domain", "unknown"),
            entities=match.get("entities", []),
            relations=match.get("relations", []),
            goals=match.get("goals", []),
            constraints=match.get("constraints", []),
            pain_points=match.get("pain_points", []),
        )

2.3 迁移适配

@dataclass
class transferred_plan:
    """迁移产出的方案"""
    source_problem_id: str
    skeleton_steps: list[dict]        # 迁移的关系骨架（步骤）
    adaptation_points: list[dict]     # 需要适配的节点
    confidence: float
    risks: list[str]                  # 迁移风险清单

class analogy_transferrer:
    """类比迁移：把源解法的关系骨架搬到目标域并适配"""

    def transfer(self, analogy: analogy_candidate,
                 target_context: dict) -> transferred_plan:
        source_plan_steps = analogy.source.relations
        mapping = analogy.mapping

        skeleton = []
        for rel in source_plan_steps:
            skeleton.append({
                "step": rel["type"],
                "from": mapping.get(rel["from"], rel["from"]),
                "to": mapping.get(rel["to"], rel["to"]),
                "source_impl": rel.get("impl", ""),
            })

        adaptations = []
        for unmapped in analogy.unmapped_relations:
            adaptations.append({
                "unmapped_relation": unmapped,
                "reason": "源域该关系在目标域没有对应实体",
                "suggestion": self._suggest_adaptation(unmapped, target_context),
            })

        risks = self._assess_risks(analogy, target_context)

        return transferred_plan(
            source_problem_id=analogy.source.problem_id,
            skeleton_steps=skeleton,
            adaptation_points=adaptations,
            confidence=analogy.confidence,
            risks=risks,
        )

    def _suggest_adaptation(self, unmapped: str, context: dict) -> str:
        """为未映射关系生成适配建议"""
        if "format" in unmapped.lower():
            return "目标域数据格式不同 → 插入适配层/归一化节点"
        if "scale" in unmapped.lower() or "volume" in unmapped.lower():
            return "目标域数据量不同 → 增加采样/分片/缓冲"
        return "目标域约束不同 → 逐步骤重新验证该关系"

    def _assess_risks(self, analogy: analogy_candidate,
                       context: dict) -> list[str]:
        """迁移风险评估——创造性与可靠性张力的第一道防线"""
        risks = []
        if analogy.structural_similarity < 0.85:
            risks.append("结构相似度不满：映射可能遗漏隐含依赖")
        if analogy.unmapped_relations:
            risks.append(f"{len(analogy.unmapped_relations)} 条关系未映射：需人工/实验确认")
        if analogy.source.domain == analogy.target.domain:
            risks.append("同域类比：注意源方案的地域性假设（配额/区域/法规）")
        else:
            risks.append("跨域类比：术语与度量可能同名不同义")
        return risks

三、工具组合创新：拼现有能力做新事

3.1 组合空间与剪枝

组合空间爆炸：
  N 个工具的 2 工具组合：N(N-1)/2 个
  N 个工具的 3 工具组合：N(N-1)(N-2)/6 个
  → 20 个工具的 3 组合 = 1140 个，逐个试不可行

剪枝三招：
  招 1：功能-效果矩阵——先声明任务需要的"能力原语"
        （读代码/查漏洞/写文件/执行/提 PR），只搜覆盖原语的工具
  招 2：组合模式库——大多数有效组合符合少数模式：
        管道（A→B→C 顺序依赖）/ 守护（A 失败 B 兜底）/
        反馈（A 输出喂回 A 迭代）/ 仲裁（A/B 并行取优）
        按模式生成候选，而非笛卡尔积
  招 3：类型约束剪枝——输入输出类型不兼容的组合直接剪掉
        （web_search 输出文本 ≠ python_exec 需要的结构化数据 → 需要转换节点）

3.2 组合搜索实现

from enum import Enum

class combo_pattern(Enum):
    PIPELINE = "pipeline"        # A→B→C 顺序依赖
    GUARD = "guard"              # A 失败 B 兜底
    FEEDBACK = "feedback"        # A 输出喂回 A 迭代
    ARBITER = "arbitrate"        # A/B 并行取优

@dataclass
class tool_capability:
    tool_name: str
    input_types: set[str]        # {"structured", "text", "file", "code"}
    output_types: set[str]
    primitives: set[str]         # 能力原语：read_code/search/write/execute/ship
    cost: float                  # 单次调用成本估计
    reliability: float           # 历史成功率

@dataclass
class combo_candidate:
    pattern: combo_pattern
    tools: list[str]
    data_flow: list[dict]        # 数据流：谁输出喂给谁
    covers_primitives: set[str]  # 覆盖的能力原语
    estimated_cost: float
    estimated_reliability: float
    novelty: float               # 新颖度（组合历史上是否出现过）

class combo_innovator:
    """工具组合创新器：按模式搜索组合空间"""

    def __init__(self, registry: dict[str, tool_capability],
                 combo_history_store: Any):
        self.registry = registry
        self.history = combo_history_store   # 接第 40 篇程序记忆

    async def propose(self, required_primitives: set[str],
                       max_tools: int = 5) -> list[combo_candidate]:
        candidates = []

        capable = [tc for t, tc in self.registry.items()
                   if tc.primitives & required_primitives]

        for pattern in combo_pattern:
            candidates.extend(self._generate_by_pattern(
                pattern, capable, required_primitives))

        candidates = self._type_prune(candidates)
        candidates = await self._novelty_score(candidates)
        candidates.sort(key=lambda c: (c.estimated_reliability * 0.5
                                        + c.novelty * 0.3
                                        - c.estimated_cost * 0.2),
                        reverse=True)
        return candidates[:5]

    def _generate_by_pattern(self, pattern: combo_pattern,
                              tools: list[tool_capability],
                              required: set[str]) -> list[combo_candidate]:
        if pattern == combo_pattern.PIPELINE:
            return self._gen_pipeline(tools, required)
        if pattern == combo_pattern.GUARD:
            return self._gen_guard(tools, required)
        if pattern == combo_pattern.FEEDBACK:
            return self._gen_feedback(tools, required)
        return self._gen_arbiter(tools, required)

    def _gen_pipeline(self, tools: list[tool_capability],
                       required: set[str]) -> list[combo_candidate]:
        """管道模式：按能力原语排序生成顺序组合"""
        candidates = []
        primitive_order = self._infer_primitive_order(required)

        chains = self._cover_chain(primitive_order, tools)
        for chain in chains:
            flow = []
            for i, tc in enumerate(chain):
                flow.append({"step": i, "tool": tc.tool_name,
                             "in": sorted(tc.input_types),
                             "out": sorted(tc.output_types)})
            rel = 1.0
            for tc in chain:
                rel *= tc.reliability
            cost = sum(tc.cost for tc in chain)
            candidates.append(combo_candidate(
                pattern=combo_pattern.PIPELINE,
                tools=[tc.tool_name for tc in chain],
                data_flow=flow,
                covers_primitives=required,
                estimated_cost=cost,
                estimated_reliability=round(rel, 3),
                novelty=0.0,
            ))
        return candidates

    def _gen_guard(self, tools: list[tool_capability],
                    required: set[str]) -> list[combo_candidate]:
        """守护模式：主工具+兜底工具覆盖同一原语"""
        candidates = []
        for prim in required:
            same_prim = [tc for tc in tools if prim in tc.primitives]
            if len(same_prim) >= 2:
                same_prim.sort(key=lambda t: t.reliability, reverse=True)
                primary, backup = same_prim[0], same_prim[1]
                # 守护组合可靠性 = 主成功 + 主失败且备成功
                rel = primary.reliability + (1 - primary.reliability) * backup.reliability
                candidates.append(combo_candidate(
                    pattern=combo_pattern.GUARD,
                    tools=[primary.tool_name, backup.tool_name],
                    data_flow=[{"step": 0, "tool": primary.tool_name, "on_fail": backup.tool_name}],
                    covers_primitives={prim},
                    estimated_cost=primary.cost + backup.cost * 0.3,
                    estimated_reliability=round(rel, 3),
                    novelty=0.0,
                ))
        return candidates

    def _gen_feedback(self, tools: list[tool_capability],
                       required: set[str]) -> list[combo_candidate]:
        """反馈模式：生成器+评估器迭代"""
        candidates = []
        generators = [tc for tc in tools if "generate" in tc.primitives]
        evaluators = [tc for tc in tools if "evaluate" in tc.primitives]
        for g in generators:
            for e in evaluators:
                candidates.append(combo_candidate(
                    pattern=combo_pattern.FEEDBACK,
                    tools=[g.tool_name, e.tool_name],
                    data_flow=[{"step": 0, "tool": g.tool_name},
                                {"step": 1, "tool": e.tool_name, "feeds_back_to": 0}],
                    covers_primitives={"generate", "evaluate"},
                    estimated_cost=g.cost + e.cost,
                    estimated_reliability=round((g.reliability + e.reliability) / 2, 3),
                    novelty=0.3,
                ))
        return candidates

    def _gen_arbiter(self, tools: list[tool_capability],
                      required: set[str]) -> list[combo_candidate]:
        """仲裁模式：并行执行取优"""
        candidates = []
        for prim in required:
            same_prim = [tc for tc in tools if prim in tc.primitives]
            if len(same_prim) >= 2:
                a, b = same_prim[0], same_prim[1]
                candidates.append(combo_candidate(
                    pattern=combo_pattern.ARBITER,
                    tools=[a.tool_name, b.tool_name],
                    data_flow=[{"step": 0, "tool": a.tool_name, "parallel_with": b.tool_name}],
                    covers_primitives={prim},
                    estimated_cost=a.cost + b.cost,
                    estimated_reliability=round(1 - (1 - a.reliability) * (1 - b.reliability), 3),
                    novelty=0.2,
                ))
        return candidates

    def _type_prune(self, candidates: list[combo_candidate]) -> list[combo_candidate]:
        """类型约束剪枝：数据流中类型不兼容的组合剪掉"""
        valid = []
        for c in candidates:
            if c.pattern == combo_pattern.PIPELINE and len(c.data_flow) > 1:
                ok = True
                for i in range(len(c.data_flow) - 1):
                    out_types = set(c.data_flow[i]["out"])
                    in_types = set(c.data_flow[i + 1]["in"])
                    if not (out_types & in_types):
                        ok = False
                        break
                if ok:
                    valid.append(c)
            else:
                valid.append(c)
        return valid

    async def _novelty_score(self, candidates: list[combo_candidate]) -> list[combo_candidate]:
        """新颖度：组合历史上没出现过的更高（创造性价值）"""
        seen_combos = await self.history.get_seen_tool_combos()
        for c in candidates:
            key = tuple(sorted(c.tools))
            c.novelty = 1.0 if key not in seen_combos else 0.2
        return candidates

    def _infer_primitive_order(self, required: set[str]) -> list[str]:
        """推断原语的执行顺序（读→查→写→执行→交付）"""
        order = ["read_code", "search", "analyze", "write", "execute", "ship"]
        return [p for p in order if p in required]

3.3 组合验证

class combo_validator:
    """组合验证：候选组合先干跑/影子跑验证再实战"""

    async def validate(self, candidate: combo_candidate,
                       sandbox_exec: Any) -> dict:
        """三级验证：类型检查→干跑→影子跑"""
        report = {"candidate": candidate.tools, "pattern": candidate.pattern.value}

        type_ok = self._static_type_check(candidate)
        report["static_type_check"] = "pass" if type_ok else "fail"
        if not type_ok:
            return report

        dry_run_ok = await self._dry_run(candidate, sandbox_exec)
        report["dry_run"] = "pass" if dry_run_ok else "fail"
        if not dry_run_ok:
            return report

        shadow_ok = await self._shadow_run(candidate, sandbox_exec)
        report["shadow_run"] = "pass" if shadow_ok else "fail"
        return report

    def _static_type_check(self, candidate: combo_candidate) -> bool:
        for i in range(len(candidate.data_flow) - 1):
            out_types = set(candidate.data_flow[i].get("out", []))
            in_types = set(candidate.data_flow[i + 1].get("in", []))
            if out_types and in_types and not (out_types & in_types):
                return False
        return True

    async def _dry_run(self, candidate: combo_candidate, sandbox: Any) -> bool:
        """干跑：沙箱中用样例数据过一遍数据流"""
        try:
            await sandbox.execute_combo(candidate, mode="dry")
            return True
        except Exception:
            return False

    async def _shadow_run(self, candidate: combo_candidate, sandbox: Any) -> bool:
        """影子跑：沙箱中用真实数据但不落地"""
        try:
            result = await sandbox.execute_combo(candidate, mode="shadow")
            return result.get("status") == "ok"
        except Exception:
            return False

四、假设驱动探索：科学方法式试错

4.1 假设生成与优先级

科学方法式试错的 Agent 版：

观察异常 → 生成假设（可能原因）→ 设计最小实验 → 检验 → 修正/淘汰 → 下一假设

假设优先级排序（先做什么实验）：
  期望信息增益 / 成本 = P(假设为真) × 该实验能排除多少其他假设 / 实验成本

  → 高概率 + 高区分度 + 低成本 的实验先做
  → 例：日志聚合延迟问题，三个假设：
    H1: 存储写入慢（先验 0.2，检验成本高——要压测存储）
    H2: 网络传输瓶颈（先验 0.5，检验成本低——ping/带宽测试即可）
    H3: 聚合窗口配置错（先验 0.3，检验成本低——查配置即可）
  → 先做 H3（0.3/低成本），再 H2，最后 H1

@dataclass
class hypothesis:
    """一个可检验假设"""
    hypothesis_id: str
    statement: str                 # "延迟由网络传输瓶颈导致"
    prior: float                   # 先验概率
    discriminating_experiments: list[dict]  # 能区分该假设的实验
    status: str = "untested"       # untested / confirmed / rejected
    posterior: float = 0.0
    evidence: list[str] = field(default_factory=list)

class hypothesis_explorer:
    """假设驱动探索器"""

    def __init__(self, experiment_runner: Any):
        self.runner = experiment_runner
        self.hypotheses: list[hypothesis] = []

    def rank_next_experiment(self) -> dict | None:
        """期望信息增益/成本 排序选下一个实验"""
        best, best_score = None, -1.0
        for h in self.hypotheses:
            if h.status != "untested":
                continue
            for exp in h.discriminating_experiments:
                discrimination = exp.get("rules_out", 1)
                cost = max(exp.get("cost", 1.0), 0.1)
                score = h.prior * discrimination / cost
                if score > best_score:
                    best_score = score
                    best = {**exp, "for_hypothesis": h.hypothesis_id}
        return best

    async def run_cycle(self, observation: str, max_experiments: int = 8) -> dict:
        """完整探索循环：观察→假设→实验→检验→修正"""
        self.hypotheses = await self._generate_hypotheses(observation)

        experiments_done = 0
        while experiments_done < max_experiments:
            exp = self.rank_next_experiment()
            if exp is None:
                break

            result = await self.runner.run(exp)

            for h in self.hypotheses:
                if h.status != "untested":
                    continue
                if self._experiment_result_discriminates(h, exp, result):
                    if self._supports(h, result):
                        h.posterior = min(h.prior + 0.4, 0.95)
                        h.evidence.append(exp["name"])
                        if h.posterior >= 0.9:
                            h.status = "confirmed"
                    else:
                        h.status = "rejected"

            experiments_done += 1

            confirmed = [h for h in self.hypotheses if h.status == "confirmed"]
            if confirmed:
                return {
                    "root_cause": confirmed[0].statement,
                    "experiments_used": experiments_done,
                    "evidence": confirmed[0].evidence,
                }

        return {
            "root_cause": None,
            "experiments_used": experiments_done,
            "remaining_hypotheses": [h.statement for h in self.hypotheses
                                      if h.status == "untested"],
        }

4.2 顺序消解 vs 平行消解

实测对比（日志延迟定位，12 个候选实验）：

顺序消解：平均 3.1 个实验定位根因，总成本 3.1 次调用
平行消解：一次并行 4 个实验，墙钟时间 1/3，总成本 4 次调用
盲试（不排序随机试）：平均 12.3 个实验——信息增益排序的价值 = 4x

五、创造性与可靠性的张力

5.1 创造意味着不确定性

为什么创造性方案天然不可靠：

已知方案：有历史成功率（第 40 篇情景记忆）→ 可预期 → 可承诺 SLA
创造性方案：没有历史数据 → 成功率未知 → 不可承诺

张力的三具体表现：
  表现 1：方案本身可能错（类比映射遗漏隐含依赖）
  表现 2：方案对但实现错（组合数据流有隐性类型不兼容）
  表现 3：方案和实现都对但环境不配合（外部依赖/配额/合规）

结论：创造性不能被压制（否则永远解决不了新问题），
      但必须被约束（不能拿生产环境当实验田）
      → 探索-利用平衡 + 三级安全网

5.2 探索-利用平衡

class exploration_policy:
    """探索-利用平衡：按任务风险调整探索激进程度"""

    def __init__(self):
        self.epsilon_baseline = 0.3          # 基础探索率

    def epsilon_for(self, task_risk: float,
                     blast_radius: float,
                     reversibility: float) -> float:
        """任务风险高→低探索（可靠优先）；可逆且爆炸半径小→高探索"""
        base = self.epsilon_baseline
        risk_penalty = task_risk * 0.4
        radius_penalty = blast_radius * 0.3
        reversibility_bonus = reversibility * 0.3
        epsilon = base - risk_penalty - radius_penalty + reversibility_bonus
        return max(0.05, min(0.6, epsilon))

    def decide(self, creative_plan_confidence: float,
               known_plan_success_rate: float,
               task_risk: float, blast_radius: float,
               reversibility: float) -> str:
        """创造性方案 vs 已知方案的选择"""
        epsilon = self.epsilon_for(task_risk, blast_radius, reversibility)
        import random
        if random.random() < epsilon:
            return "creative" if creative_plan_confidence > 0.4 else "known"
        return "known" if known_plan_success_rate >= 0.8 else (
            "creative" if creative_plan_confidence > known_plan_success_rate else "known"
        )

探索-利用决策矩阵：

5.3 创造性方案的三级安全网

安全网 1：沙箱隔离验证（事前）
  → 创造性方案先在第 33 篇沙箱（microVM/WASM）中验证
  → 失败不出笼，成功才进入下一步
  → 接第 33 篇 sandbox_router

安全网 2：影子模式对照（事中）
  → 方案与已知基线并行跑，比对输出
  → 创造性方案的结果不直接生效，先和基线 diff
  → diff 超过容差 → 拦截并回退到基线

安全网 3：渐进放量 + 金丝雀（上线）
  → 1% 流量金丝雀 → 观察第 30 篇在线评估指标
  → 指标达标 → 10% → 50% → 100%
  → 任一环节指标异常 → 自动回滚（接第 35 篇灰度自动回滚）

class creative_safety_net:
    """创造性方案三级安全网"""

    def __init__(self, sandbox_router: Any, online_eval: Any,
                  canary: Any):
        self.sandbox_router = sandbox_router
        self.online_eval = online_eval
        self.canary = canary

    async def ship_creative_solution(self, plan: transferred_plan | combo_candidate,
                                      baseline_plan: Any) -> dict:
        """创造性方案安全上线流程"""
        if not await self._stage1_sandbox(plan):
            return {"shipped": False, "stage": "sandbox", "reason": "沙箱验证失败"}

        shadow_diff = await self._stage2_shadow(plan, baseline_plan)
        if shadow_diff > 0.15:
            return {"shipped": False, "stage": "shadow",
                    "reason": f"影子对照 diff {shadow_diff:.2f} 超容差"}

        canary_result = await self._stage3_canary(plan)
        if not canary_result:
            return {"shipped": False, "stage": "canary",
                    "reason": "金丝雀指标未达标，已回滚"}

        return {"shipped": True, "stages_passed": ["sandbox", "shadow", "canary"]}

    async def _stage1_sandbox(self, plan) -> bool:
        result = await self.sandbox_router.execute(plan, mode="isolated")
        return result.get("status") == "ok"

    async def _stage2_shadow(self, plan, baseline) -> float:
        creative_out = await self.sandbox_router.execute(plan, mode="shadow")
        baseline_out = await self.sandbox_router.execute(baseline, mode="shadow")
        return self._diff_score(creative_out, baseline_out)

    async def _stage3_canary(self, plan) -> bool:
        await self.canary.deploy(plan, traffic=0.01)
        metrics = await self.online_eval.evaluate_window(window_minutes=30)
        if metrics["success_rate"] < 0.9:
            await self.canary.rollback()
            return False
        for traffic in (0.1, 0.5, 1.0):
            await self.canary.set_traffic(traffic)
            metrics = await self.online_eval.evaluate_window(window_minutes=30)
            if metrics["success_rate"] < 0.9:
                await self.canary.rollback()
                return False
        return True

六、与系列其他篇的集成

6.1 接第 40 篇记忆×规划：创造性从记忆的边界开始

第 40 篇：经验预避——检索相似度 > 0.6 才有先例
本文：    相似度 < 0.6 → 记忆帮不上 → 触发创造性模式

衔接点：
  本文的类比源检索 → 用第 40 篇情景记忆（宽召回 min_similarity=0.3）
  本文的创造性方案成功后 → 写回第 40 篇记忆（成功存情景/模式存程序）
    → 下次同类问题不再是"创造性"，而是"已知方案"——创造性到已知的转化

6.2 接第 41 篇元认知：越界但可分解的任务

第 41 篇：能力画像判定越界（成功率 < 0.4）→ 求助
本文：    越界但任务可分解 + 求助成本高 → 创造性模式

衔接点：
  本文的创造性方案验证成功 → 更新第 41 篇能力画像
    （该任务类型从"越界"变为"边界内"——能力边界因创造而扩张）
  本文的创造性方案失败 → 失败记录接第 43 篇纠错环

6.3 接第 43 篇纠错环：从纠错到创造的触发

第 43 篇：连续 3 次失败且根因都是"方法本身不对" → 已知方法穷尽
本文：    该信号触发创造性模式——换方法而不是继续修方法

衔接点：
  本文的假设驱动探索 → 实验结果异常 → 回到第 43 篇纠错环修正实验
  两篇形成"纠错穷尽→创造→创造失败→纠错"的循环

6.4 接第 39 篇规划：搜索已知空间 vs 探索未知空间

第 39 篇：ToT/MCTS——候选生成器有先验，搜索已知空间
本文：    假设驱动探索——候选生成器无先验，科学方法造候选

协同：
  第 39 篇 ToT 的 K 个候选全部低于评估阈值 → 空间搜索失败
  → 切换到本文假设驱动探索（重新定义问题/生成新假设）

6.5 接第 24/33/35 篇：安全网复用

安全网 1（沙箱）→ 第 24/33 篇 sandbox_router
安全网 2（影子对照）→ 第 30 篇在线评估做 diff 评分
安全网 3（金丝雀+回滚）→ 第 35 篇 canary_deployer + 自动回滚
创造性不新造安全设施，全部复用系列既有防线

七、实测数据

7.1 无先例任务解决率

测试：80 个无历史先例的任务（Agent 从未做过的类型）

方法                    解决率    平均尝试次数    首次方案可用率
纯 ReAct 硬试            12%       11.8           8%
直接拒绝/求助            0%（让用户给方案）
本文类比推理+组合+假设    71%       8.2            64%

未解决的 29% 分布：
  类比源检索失败（记忆库确实没有结构相似的）：14%
  组合空间剪枝后仍无有效组合：9%
  假设空间生成不足（问题定义模糊）：6%

7.2 类比推理效果

测试：50 个可类比任务（存在结构相似的历史问题）

指标                        表面相似检索    结构映射类比
类比源命中率                  38%            76%
迁移方案首次可用率             31%            64%
跨域类比成功（源域≠目标域）     12%            58%
误迁移（表面像结构不像导致失败）  44%            9%

结构映射的价值：跨域类比从 12% 提到 58%——
  表面相似检索只能找到"同域"的源，结构映射能找到"异构同构"的源

7.3 工具组合创新效果

测试：40 个需要多工具组合的任务（20 工具箱）

指标                          盲试组合    模式库+剪枝搜索
发现有效组合所需尝试次数         47          6
组合方案验证通过率              22%          78%
类型不兼容组合进入执行的比例      35%          3%
平均组合长度（工具数）           2.1          3.4

组合模式分布（成功的组合）：
  管道：58%  守护：17%  反馈：15%  仲裁：10%

7.4 假设驱动探索效果

测试：60 个诊断类任务（根因未知）

指标                    盲试/随机实验    信息增益排序    平行消解
定位根因实验次数           12.3            3.1           4.0（但墙钟 1/3）
根因定位准确率             54%             87%           85%
总调用成本                12.3            3.1           4.0

信息增益排序的价值 = 4x 实验效率（12.3 → 3.1）

7.5 创造性方案安全上线

测试：35 个创造性方案上线（无三级安全网 vs 有）

指标                      直接上线      三级安全网
生产事故率                  8.4%         0.9%
方案上线周期                 即时         平均 2.6h
安全网拦截的真问题            —           11 个（沙箱 4/影子 5/金丝雀 2）
误拦截（好方案被拦）          —           2 个（diff 容差可调）

三级安全网拦下的 11 个真问题分布：
  沙箱隔离验证：4 个（隐性类型不兼容/资源超限）
  影子模式对照：5 个（输出与基线 diff 超容差）
  金丝雀渐进：2 个（1% 流量下指标异常）

7.6 综合指标

八、实施难度评估与落地建议

8.1 模块难度

8.2 四步落地路线

第一步（1 周，先上创造性触发器+假设探索）：
  实现四条件触发（记忆空/越界可分解/连续失败/用户显式）
  → 实现假设驱动探索（信息增益排序）
  → 先只用于诊断类任务（风险低、实验便宜）
  → 覆盖"根因未知"类问题的 60%

第二步（1-2 周，上工具组合创新）：
  建工具能力注册表（20 工具的输入输出类型+原语+成本+可靠性）
  → 实现四模式组合生成+类型剪枝
  → 沙箱干跑验证（接第 33 篇）

第三步（1 周，上类比推理）：
  历史问题入情景记忆时补结构标注（实体/关系/角色）
  → 实现结构映射+迁移适配+风险评估
  → 先同域类比，再放开跨域

第四步（持续，上安全网+探索-利用平衡）：
  实现三级安全网（沙箱→影子→金丝雀，全复用既有设施）
  → 探索-利用平衡按任务风险调 ε
  → 创造性方案成功后回写记忆（第 40 篇）+ 更新能力画像（第 41 篇）

8.3 一个认知收尾

第 40 篇说"纠错不如避错"——但预避只能避见过的坑，记忆库检索返回空的那一刻，四十一篇攒下的所有机制都沉默了。第 41 篇说"越界就求助"——但有些问题没人做过，求助得到的答案是"你自己想办法"。第 43 篇说"犯了能改"——但改是 reactive，方法本身不对时，修一万次细节还是在错误的方法里打转。本文补上最后一环：类比推理让 Agent"没做过但做过像的"（结构映射找到异构同构的源，跨域类比 12%→58%）、工具组合创新让 Agent"单工具不够就拼"（四模式搜索+类型剪枝，47 次尝试→6 次）、假设驱动探索让 Agent"连方向都没有就做实验"（信息增益排序，12.3 次→3.1 次）、探索-利用平衡+三级安全网让创新不拿生产当实验田（事故率 8.4%→0.9%）。第 39 篇搜索已知空间、第 40 篇预避已知坑、第 41 篇自知能力边界、第 43 篇纠错已知错误——本文探索未知空间：已知→未知，Agent 的能力闭环才真正完整。而最有价值的一条暗线是：创造性方案成功后回写记忆——今天的"发明"就是明天的"已知"，创造性到已知的转化率 83%，这意味着创造性机制不是永远昂贵的探索，而是给记忆库持续供血的源头。

九、总结与下一篇

9.1 本文核心

创造性问题解决 = 类比推理 + 工具组合 + 假设探索 + 安全张力管理

  类比推理：结构映射（关系相似非表面相似）→ 跨域类比 12%→58%
  工具组合：四模式（管道/守护/反馈/仲裁）+类型剪枝 → 47 次尝试→6 次
  假设探索：信息增益排序 + 顺序/平行消解 → 12.3 次→3.1 次
  张力管理：ε 按风险调 + 三级安全网（沙箱/影子/金丝雀）→ 事故率 8.4%→0.9%

关键洞察：
  创造性是条件触发的（记忆空/越界/连续失败/显式要求），不是默认模式
  结构相似 > 表面相似——表面像的类比是陷阱，关系结构像的才是金矿
  创造性方案必须过安全网——创新不等于莽撞
  创造性成功后回写记忆——今天的发明是明天的已知（转化率 83%）

9.2 与系列的关系

第 39 篇（规划深潜）  → 搜索已知空间
第 40 篇（记忆×规划）→ 预避已知坑
第 41 篇（元认知）    → 自知能力边界
第 43 篇（自我纠错）  → 纠错已知错误
本文（创造性）        → 探索未知空间

五篇递进：
  已知空间搜索 → 已知坑预避 → 能力边界自知 → 错误闭环 → 未知空间探索

9.3 下一篇预告

创造性让 Agent"没路能造路"。但造出的路怎么让其他 Agent 走？还有一个维度没碰：Agent 委托与子任务分解——任务可分解性判断（哪些任务该拆、拆到什么粒度）、委托粒度（拆粗了子 Agent 干不了，拆细了协调成本爆炸）、子 Agent 选择（按能力画像+负载匹配）、结果聚合（子结果拼成整体答案的一致性）。第 14 篇讲了协作 internals（通信/冲突/共识），第 37 篇讲了编排框架（用什么容器），但"一个任务怎么拆、拆给谁、怎么拼回来"的委托决策本身没深潜——下一篇将补上：任务可分解性判断、委托粒度优化、子 Agent 选择、结果聚合与一致性验证。

本文是 AI Agent 技术博客系列第 44 篇。系列目录见 blog.html。