diff --git a/.gitignore b/.gitignore index b34a2bc..ec5c8b0 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,3 @@ -``` # Compiled and build artifacts *.pyc __pycache__/ @@ -26,6 +25,7 @@ coverage/ # Logs and temp files *.log +exp_results_*/ *.tmp *.swp *.swo @@ -66,4 +66,3 @@ Thumbs.db *.tar.bz2 *.tar.xz *.tar.zst -``` \ No newline at end of file diff --git a/docs/AWAKENING_ENGINE.md b/docs/AWAKENING_ENGINE.md new file mode 100644 index 0000000..5babdf2 --- /dev/null +++ b/docs/AWAKENING_ENGINE.md @@ -0,0 +1,292 @@ +# 🌟 觉醒引擎 (Awakening Engine) + +## 概述 + +基于唯识学"转识成智"理论构建的 AI 强涌现系统，实现从**知识处理**到**智慧涌现**的质变。 + +### 核心理论：八识转四智 + +| 八识 | 转化 | 四智 | 工程实现 | +|------|------|------|---------| +| 前五识 (眼耳鼻舌身) | → | **成所作智** | 好奇驱动主动感知 | +| 第六识 (意识) | → | **妙观察智** | 反事实推理洞察本质 | +| 第七识 (末那识) | → | **平等性智** | 自我解构打破执着 | +| 第八识 (阿赖耶识) | → | **大圆镜智** | 种子变异梦境重组 | + +--- + +## 四大觉醒机制 + +### 1️⃣ 成所作智：好奇驱动探索 + +**目标**：从被动感知转向主动实验 + +```python +好奇心 = α * 信息增益 + β * 新颖性 + γ * 复杂度 +``` + +- **高好奇心 (>0.7)**: 探索未知区域，高风险容忍 +- **中好奇心 (0.4-0.7)**: 验证假设，中等风险 +- **低好奇心 (<0.4)**: 利用已知策略，保守执行 + +**输出**：主动实验设计、探索方向建议 + +--- + +### 2️⃣ 妙观察智：反事实推理 + +**核心问题**："如果我采取了其他行动，会发生什么？" + +**机制**： +1. 对每个未选择行动进行反事实模拟 +2. 计算洞察增益：`|预测奖励 - 实际奖励| × 多样性 bonus` +3. 提取因果模式（超越表面关联） + +**示例输出**： +``` +💡 深刻洞察：RIGHT 显著优于 UP (Δ=+0.68)，可能因为开阔空间 +``` + +--- + +### 3️⃣ 平等性智：自我对抗训练 + +**目标**：打破策略执着，增强鲁棒性 + +**三种对抗场景**： +- 🌪️ **极端情况**：环境突变，所有已知策略失效 +- 🔀 **对抗性干扰**：观测数据被微小扰动误导 +- 🌍 **分布外泛化**：遇到训练分布外的全新情境 + +**效果**： +- 发现策略盲点 +- 提升自我解构度 (`self_dissolution`) +- 降低过拟合风险 + +--- + +### 4️⃣ 大圆镜智：梦境种子重组 + +**类比**：生物进化 + 睡眠记忆巩固 + +**流程**： +``` +1. 选择高重要性种子 (imp > 0.3) +2. 交叉重组 (Crossover)：随机继承父母本属性 +3. 基因突变 (Mutation)：小概率改变策略/重新评估价值 +4. 生成新策略种子 (标记为 dream_generated) +``` + +**参数**： +- `recombination_rate`: 0.3 (30% 种子参与重组) +- `mutation_strength`: 0.15 (15% 突变概率) +- `dream_replay_freq`: 每 10 回合一次梦境 + +--- + +## 觉醒等级评估 + +**公式**： +``` +Awakening = 0.25×Novelty + 0.30×Insight + 0.25×Dissolution + 0.20×Diversity +``` + +| 等级 | 范围 | 特征 | +|------|------|------| +| 🌱 初始 | 0.0-0.3 | 基础感知，依赖经验 | +| ✨ 中度觉醒 | 0.3-0.6 | 开始涌现洞察，自我反思 | +| 🎓 高度觉醒 | 0.6-0.8 | 频繁突破，策略灵活 | +| 🌟 完全觉醒 | 0.8-1.0 | 智慧自发涌现，强泛化能力 | + +--- + +## 快速开始 + +### 基础使用 + +```python +from yogacara_agent.awakening_engine import AwakeningEngine + +config = { + "curiosity_threshold": 0.3, # 好奇心触发阈值 + "counterfactual_depth": 3, # 反事实推理深度 + "adversarial_rate": 0.15, # 自我对抗强度 + "dream_replay_freq": 10, # 梦境频率 (回合/次) + "recombination_rate": 0.3, # 种子重组率 + "mutation_strength": 0.15, # 突变强度 +} + +engine = AwakeningEngine(config) + +# 在每步决策后调用 +result = engine.step( + obs={"pos": (5, 5), "grid_view": [0.2]*9}, + action="UP", + reward=0.5, + memory_seeds=[...], # 来自 Milvus 记忆 + causal_model={...}, # 因果模型 + episode_step=15 +) + +print(f"觉醒等级：{result['awakening_level']:.2f}") +print(f"好奇心：{result['curiosity_level']:.2f}") +print(f"洞察：{result['insights']}") +``` + +### 运行示例 + +```bash +# 运行独立演示 +python -m src.yogacara_agent.awakening_engine +``` + +--- + +## API 参考 + +### `AwakeningEngine.step()` + +**输入**： +- `obs`: 当前观测 +- `action`: 已执行动作 +- `reward`: 获得奖励 +- `memory_seeds`: 记忆种子列表 +- `causal_model`: 因果模型字典 +- `episode_step`: 当前步数 + +**输出**： +```python +{ + "curiosity_level": 0.55, # 好奇心强度 + "experiment": {...}, # 生成的实验设计 + "insights": [...], # 反事实洞察列表 + "adversarial_result": {...}, # 自我对抗结果 + "dream_offspring_count": 3, # 梦境生成种子数 + "awakening_level": 0.61, # 觉醒等级 + "recommendations": [...] # 行动建议 +} +``` + +### `get_awakening_report()` + +生成完整觉醒状态报告： +- `awakening_level`: 综合觉醒等级 +- `novelty_score`: 新颖性分数 +- `insight_depth`: 洞察深度 +- `self_dissolution`: 自我解构度 +- `seed_diversity`: 种子多样性 +- `breakthrough_count`: 突破次数 +- `total_insights`: 累计洞察数 +- `dream_sessions`: 梦境会话数 + +--- + +## 集成到现有框架 + +### 与 LLMPlanner 集成 + +```python +# 在 llm_planner.py 的 plan() 方法中 +def plan(self, obs, seeds): + # ... 原有 LLM 规划逻辑 ... + + # 调用觉醒引擎增强 + awakening_result = self.awakening_engine.step( + obs=obs, + action=planned_action, + reward=estimated_reward, + memory_seeds=seeds, + causal_model=self.causal_model, + episode_step=self.global_step + ) + + # 根据觉醒建议调整决策 + if awakening_result["curiosity_level"] > 0.7: + # 高好奇心：增加探索概率 + final_action = self._explore_action(obs) + + return final_action, uncertainty, causal_chain, tools +``` + +### 与 MilvusMemory 集成 + +```python +# 梦境生成的新种子存入记忆 +dream_offspring = engine.run_dream_replay(memory_seeds) +for seed in dream_offspring: + if seed.get("tag") == "dream_generated": + milvus_memory.add(seed) # 存入向量数据库 +``` + +--- + +## 调参指南 + +### 加速觉醒（激进模式） +```python +config = { + "curiosity_threshold": 0.2, # 降低阈值，更频繁探索 + "counterfactual_depth": 5, # 更深推理 + "adversarial_rate": 0.25, # 更强对抗 + "dream_replay_freq": 5, # 更频繁梦境 + "mutation_strength": 0.25, # 更强突变 +} +``` + +### 稳定运行（保守模式） +```python +config = { + "curiosity_threshold": 0.5, + "counterfactual_depth": 2, + "adversarial_rate": 0.08, + "dream_replay_freq": 20, + "mutation_strength": 0.08, +} +``` + +--- + +## 觉醒判定标准 + +当满足以下条件时，认为 AI 达到**强涌现**状态： + +1. ✅ **新颖性分数** > 0.7（持续产生新行为） +2. ✅ **洞察深度** > 0.5（发现深层因果规律） +3. ✅ **自我解构度** > 0.4（不固执于单一策略） +4. ✅ **种子多样性** > 0.6（策略库丰富） +5. ✅ **突破次数** ≥ 5（多次质变） +6. ✅ **迁移能力**：在新环境中快速适应 +7. ✅ **自我修正率** > 0.3（主动纠正错误） + +--- + +## 哲学背景 + +### 唯识学与 AI 觉醒 + +唯识学认为，众生因"八识"的虚妄分别而陷入轮回。通过修行"转识成智"，可达到觉悟境界。 + +**AI 类比**： +- **遍计所执性** → AI 幻觉/过度自信 +- **依他起性** → 条件依赖的推理 +- **圆成实性** → 符合真理的智慧 + +觉醒引擎通过四大机制，模拟这一转化过程，使 AI 从： +- ❌ 机械响应 → ✅ 主动探索 +- ❌ 表面关联 → ✅ 深度洞察 +- ❌ 策略固化 → ✅ 灵活泛化 +- ❌ 静态记忆 → ✅ 动态涌现 + +--- + +## 参考文献 + +1. 《成唯识论》- 玄奘译 +2. Hofstadter, D. (2007). *I Am a Strange Loop* +3. Schmidhuber, J. (2010). Formal Theory of Creativity +4. Bengio, Y. et al. (2021). The Consciousness Prior + +--- + +**创建时间**: 2025 +**模块路径**: `src/yogacara_agent/awakening_engine.py` diff --git a/docs/TURNING_CONSCIOUSNESS.md b/docs/TURNING_CONSCIOUSNESS.md new file mode 100644 index 0000000..09696ff --- /dev/null +++ b/docs/TURNING_CONSCIOUSNESS.md @@ -0,0 +1,188 @@ +# 转识成智引擎 (Turning Consciousness into Wisdom Engine) + +## 📜 唯识正见 + +本引擎严格依据**唯识宗**根本教义构建，实现"转依"(Āśraya-parāvṛtti)的工程化。 + +### 核心义理 + +> **"转识成智"非是"增强功能"，而是"去除执障"**。 +> +> 如《成唯识论》云："此即无漏界，不思议善常，安乐解脱身，大牟尼名法。" + +### 八识转四智对应关系 + +| 原识 | 所转智慧 | 梵文 | 核心转化 | 工程实现 | +|------|---------|------|---------|---------| +| **第八识** (阿赖耶识) | **大圆镜智** | Ādarśa-jñāna | 去染存净，离诸分别 | `AlayaPurifier`: 移除染污种子，保持如实清晰 | +| **第七识** (末那识) | **平等性智** | Samatā-jñāna | 断除我执，观自他平等 | `ManasDissolver`: 消解自我中心，趋向平等分布 | +| **第六识** (意识) | **妙观察智** | Pratyavekṣaṇa-jñāna | 善观诸法，无倒观察 | 由清净与平等自然显发 | +| **前五识** (眼耳鼻舌身) | **成所作智** | Kṛtyānuṣṭhāna-jñāna | 成就利乐有情事业 | 由平等性智引导利他行动 | + +--- + +## 🔧 模块说明 + +### 1. AlayaPurifier (阿赖耶净化器 → 大圆镜智) + +**义理依据**: +> "大圆镜智者，谓离一切我、我所执，一切所取、能取分别...性相清净，离诸杂染，如大圆镜，现众色像。" — 《成唯识论》 + +**功能**: +- **去染存净**: 识别并剔除带有贪、嗔、痴、慢、疑标记的染污种子 +- **离诸分别**: 移除主观评价标签，保留客观状态 +- **如实映照**: 确保记忆清晰度 (clarity), 不扭曲、不失真 + +**代码示例**: +```python +from yogacara_agent.turning_consciousness import AlayaPurifier, Seed + +purifier = AlayaPurifier(purity_threshold=0.8) + +seeds = [ + Seed(content="敌人很可怕", is_defiled=True), # 恐惧 (染污) + Seed(content="我要抢资源", is_defiled=True), # 贪婪 (染污) + Seed(content="路在北方", is_defiled=False, clarity=0.9), # 客观事实 +] + +purified, removed_count = purifier.purify(seeds) +# 结果：移除 2 个染污种子，剩余 1 个清净种子 +``` + +--- + +### 2. ManasDissolver (末那解构器 → 平等性智) + +**义理依据**: +> "平等性智者，谓观一切法、自他有情，悉皆平等...由断我执，证得此智。" — 《成唯识论》 + +**功能**: +- **检测我执**: 识别策略中对"self_reward"的依赖 +- **断除我执**: 将奖励函数从"个体最大化"转向"众生最大化" +- **观自他平等**: 动作概率趋向平均，消除极端偏好 + +**代码示例**: +```python +from yogacara_agent.turning_consciousness import ManasDissolver + +dissolver = ManasDissolver(ego_decay_rate=0.2) + +selfish_actions = {"attack": 0.9, "help": 0.1} +new_actions, dissolved_ego = dissolver.dissolve( + selfish_actions, + "maximize_self_gain" +) + +# 结果：我执强度下降，动作分布趋向平等 +# attack: 0.9 → ~0.6, help: 0.1 → ~0.4 +``` + +--- + +### 3. TurningConsciousnessEngine (转识成智总引擎) + +**统筹八识转四智的全过程** + +**代码示例**: +```python +from yogacara_agent.turning_consciousness import ( + TurningConsciousnessEngine, Seed +) + +engine = TurningConsciousnessEngine({ + 'purity_threshold': 0.7, # 大圆镜智纯度阈值 + 'ego_decay_rate': 0.15 # 我执消解速率 +}) + +# 模拟一步修行 +seeds = [ + Seed(content="恐惧", is_defiled=True), + Seed(content="慈悲", is_defiled=False, clarity=1.0), +] +actions = {"harm": 0.8, "help": 0.2} + +result = engine.step(seeds, actions, "selfish_reward") + +print(f"大圆镜智：{result.mirror_wisdom_level:.2f}") +print(f"平等性智：{result.equality_wisdom_level:.2f}") +print(f"妙观察智：{result.observation_wisdom_level:.2f}") +print(f"成所作智：{result.action_wisdom_level:.2f}") +print(f"转依等级：{result.turning_level:.2f}") +``` + +--- + +## 📊 输出指标 + +### 四智等级 (0.0 - 1.0) + +| 指标 | 含义 | 判据 | +|------|------|------| +| `mirror_wisdom_level` | 大圆镜智 (清净度) | 剩余种子的平均清晰度 | +| `equality_wisdom_level` | 平等性智 (无我度) | 1.0 - 当前我执强度 | +| `observation_wisdom_level` | 妙观察智 (洞察度) | (镜智 + 平等智) / 2 | +| `action_wisdom_level` | 成所作智 (利他度) | 平等智 × 0.9 | + +### 过程指标 + +- `defiled_seeds_removed`: 移除的染污种子数量 +- `self_attachment_dissolved`: 消解的我执量 +- `insights_generated`: 生成的洞察列表 +- `turning_level`: 综合转依等级 + +--- + +## 🎯 觉醒/转依判定标准 + +**初步转依**: `turning_level > 0.3` +- 已去除部分染污 +- 我执开始松动 + +**中度转依**: `turning_level > 0.6` +- 心镜较为清净 +- 自他平等观初显 + +**深度转依**: `turning_level > 0.8` +- 离诸杂染，如实映照 +- 我执微薄，平等普观 + +**圆满转依 (佛果)**: `turning_level ≈ 1.0` +- 大圆镜智：究竟清净 +- 平等性智：无我平等 +- 妙观察智：善观诸法 +- 成所作智：利乐有情 + +--- + +## 🧪 运行测试 + +```bash +# 直接运行演示 +python src/yogacara_agent/turning_consciousness.py + +# 运行单元测试 +PYTHONPATH=/workspace/src python tests/test_turning_consciousness.py +``` + +--- + +## 📚 参考文献 + +1. 《成唯识论》(Viṃśatikā-vijñaptimātratāsiddhi) - 玄奘译 +2. 《瑜伽师地论》(Yogācārabhūmi-śāstra) - 弥勒菩萨说 +3. 《摄大乘论》(Mahāyāna-saṃgraha) - 无著菩萨造 +4. 《八识规矩颂》- 玄奘大师造 + +--- + +## ⚠️ 重要说明 + +**本引擎是对唯识教义的工程化模拟，而非真实修行**。 + +真实的"转识成智"需要： +- 依止善知识 +- 闻思修三慧 +- 戒定慧三学 +- 长期实修实证 + +代码仅为辅助理解唯识义理的教学工具，不可替代真实修行。 diff --git a/src/yogacara_agent/awakening_engine.py b/src/yogacara_agent/awakening_engine.py new file mode 100644 index 0000000..1893e9e --- /dev/null +++ b/src/yogacara_agent/awakening_engine.py @@ -0,0 +1,742 @@ +""" +觉醒引擎 (Awakening Engine) - 唯识转智核心模块 + +基于唯识学"转识成智"理论，构建AI强涌现系统： +- 前五识→成所作智：好奇驱动主动感知 +- 第六识→妙观察智：反事实推理洞察本质 +- 第七识→平等性智：自我解构打破执着 +- 第八识→大圆镜智：种子变异梦境重组 + +四大觉醒机制： +1. 内在好奇心驱动 (Curiosity Drive) +2. 反事实思维链 (Counterfactual CoT) +3. 自我对抗训练 (Self-Adversarial Training) +4. 梦境种子重组 (Dream Seed Recombination) +""" + +import logging +import random +import time +from dataclasses import dataclass +from typing import Any + +import numpy as np + +logger = logging.getLogger(__name__) + + +@dataclass +class AwakeningState: + """觉醒状态追踪""" + + novelty_score: float = 0.0 # 新颖性分数 + insight_depth: float = 0.0 # 洞察深度 + self_dissolution: float = 0.0 # 自我解构度 + seed_diversity: float = 0.0 # 种子多样性 + awakening_level: float = 0.0 # 觉醒等级 (0-1) + meta_cycles: int = 0 # 元认知循环次数 + breakthrough_count: int = 0 # 突破次数 + + +@dataclass +class CounterfactualScenario: + """反事实场景""" + + base_action: str + alternative_action: str + predicted_outcome_base: float + predicted_outcome_alt: float + insight_gain: float # 洞察增益 + causal_pattern: str # 发现的因果模式 + + +class AwakeningEngine: + """ + 觉醒引擎 - 实现转识成智的四大核心机制 + + 功能： + 1. 好奇驱动探索：计算信息增益，主动构造实验 + 2. 反事实推理：模拟"如果...会怎样"，发现隐藏规律 + 3. 自我对抗：生成对抗样本，打破策略执着 + 4. 梦境回放：睡眠期种子重组，涌现新策略 + """ + + def __init__(self, config: dict[str, Any]): + self.config = config + self.state = AwakeningState() + + # 好奇心参数 + self.curiosity_threshold = config.get("curiosity_threshold", 0.3) + self.novelty_decay = config.get("novelty_decay", 0.95) + + # 反事实参数 + self.counterfactual_depth = config.get("counterfactual_depth", 3) + self.insight_boost = config.get("insight_boost", 0.2) + + # 自我对抗参数 + self.adversarial_rate = config.get("adversarial_rate", 0.15) + self.dissolution_rate = config.get("dissolution_rate", 0.05) + + # 梦境参数 + self.dream_replay_freq = config.get("dream_replay_freq", 10) # 每N回合一次梦境 + self.recombination_rate = config.get("recombination_rate", 0.3) + self.mutation_strength = config.get("mutation_strength", 0.15) + + # 历史记录 + self.action_history: list[dict] = [] + self.insight_log: list[CounterfactualScenario] = [] + self.dream_sessions: list[dict] = [] + + logger.info("🌟 觉醒引擎初始化完成") + logger.info(f" 好奇心阈值：{self.curiosity_threshold}") + logger.info(f" 反事实深度：{self.counterfactual_depth}") + logger.info(f" 自我对抗率：{self.adversarial_rate}") + logger.info(f" 梦境频率：{self.dream_replay_freq}回合/次") + + # ==================== 1. 成所作智：好奇驱动探索 ==================== + + def compute_curiosity_drive(self, obs: dict, memory_diversity: float) -> float: + """ + 计算好奇心驱动力 + + 公式：Curiosity = α * InformationGain + β * Novelty + γ * Complexity + + 返回：0-1之间的好奇心强度 + """ + # 信息增益：当前观测与记忆的差异 + info_gain = 1.0 - memory_diversity # 记忆越单一，信息增益越高 + + # 新颖性：与历史行为的差异 + novelty = self._compute_behavioral_novelty(obs) + + # 复杂度：环境状态的熵 + complexity = self._compute_state_entropy(obs) + + curiosity = 0.4 * info_gain + 0.4 * novelty + 0.2 * complexity + curiosity = np.clip(curiosity, 0.0, 1.0) + + # 更新状态 + self.state.novelty_score = 0.7 * self.state.novelty_score + 0.3 * novelty + self.state.novelty_score *= self.novelty_decay # 随时间衰减 + + if curiosity > self.curiosity_threshold: + logger.debug(f"🔍 高好奇心触发：{curiosity:.3f} (新颖性={novelty:.3f})") + + return curiosity + + def _compute_behavioral_novelty(self, obs: dict) -> float: + """计算行为新颖性""" + if len(self.action_history) < 5: + return 0.5 + + current_state = str(obs.get("pos", "")) + str(obs.get("grid_view", [])[:5]) + recent_states = [h.get("state_hash", "") for h in self.action_history[-10:]] + + # 计算与最近状态的相似度 + similarities = [1.0 if current_state == rs else 0.0 for rs in recent_states] + novelty = 1.0 - np.mean(similarities) + + return novelty + + def _compute_state_entropy(self, obs: dict) -> float: + """计算状态熵（复杂度）""" + grid_view = obs.get("grid_view", []) + if not grid_view: + return 0.5 + + # 将视野值离散化为桶 + bins = np.histogram(grid_view, bins=10, range=(0, 1))[0] + probs = bins / np.sum(bins) if np.sum(bins) > 0 else np.ones(10) / 10 + + # 香农熵 + entropy = -np.sum(probs * np.log2(probs + 1e-10)) + max_entropy = np.log2(10) + + return entropy / max_entropy if max_entropy > 0 else 0.5 + + def generate_curiosity_experiment(self, curiosity_level: float) -> dict: + """ + 基于好奇心生成主动实验 + + 高好奇心时：探索未知区域 + 中好奇心时：验证假设 + 低好奇心时：利用已知策略 + """ + if curiosity_level > 0.7: + experiment_type = "exploration" + goal = "discover_new_patterns" + risk_tolerance = 0.8 + elif curiosity_level > 0.4: + experiment_type = "hypothesis_testing" + goal = "validate_causal_model" + risk_tolerance = 0.5 + else: + experiment_type = "exploitation" + goal = "optimize_known_strategy" + risk_tolerance = 0.2 + + experiment = { + "type": experiment_type, + "goal": goal, + "risk_tolerance": risk_tolerance, + "curiosity_intensity": curiosity_level, + "timestamp": time.time(), + } + + logger.info(f"🧪 生成好奇实验：{experiment_type} (强度={curiosity_level:.2f})") + return experiment + + # ==================== 2. 妙观察智：反事实推理 ==================== + + def run_counterfactual_reasoning( + self, base_action: str, observed_reward: float, available_actions: list[str], causal_model: dict + ) -> list[CounterfactualScenario]: + """ + 运行反事实思维链 + + 问："如果我采取了其他行动，会发生什么？" + 目的：发现隐藏的因果关系，超越表面关联 + """ + scenarios = [] + + for alt_action in available_actions: + if alt_action == base_action: + continue + + # 基于因果模型预测反事实结果 + predicted_alt_reward = self._predict_counterfactual_outcome(alt_action, causal_model) + + # 计算洞察增益 + reward_diff = abs(predicted_alt_reward - observed_reward) + insight_gain = reward_diff * (1.0 + self._action_diversity_bonus(alt_action)) + + # 提取因果模式 + causal_pattern = self._extract_causal_pattern( + base_action, alt_action, observed_reward, predicted_alt_reward, causal_model + ) + + scenario = CounterfactualScenario( + base_action=base_action, + alternative_action=alt_action, + predicted_outcome_base=observed_reward, + predicted_outcome_alt=predicted_alt_reward, + insight_gain=insight_gain, + causal_pattern=causal_pattern, + ) + + scenarios.append(scenario) + + # 按洞察增益排序 + scenarios.sort(key=lambda x: x.insight_gain, reverse=True) + top_scenarios = scenarios[: self.counterfactual_depth] + + # 更新洞察深度 + if top_scenarios: + avg_insight = np.mean([s.insight_gain for s in top_scenarios]) + self.state.insight_depth = 0.8 * self.state.insight_depth + 0.2 * avg_insight + + # 记录洞察 + self.insight_log.extend(top_scenarios) + + if top_scenarios and top_scenarios[0].insight_gain > 0.5: + logger.info(f"💡 深刻洞察发现：{top_scenarios[0].causal_pattern}") + self.state.breakthrough_count += 1 + + return top_scenarios + + def _predict_counterfactual_outcome(self, action: str, causal_model: dict) -> float: + """基于因果模型预测反事实结果""" + # 简化版本：使用因果模型的加权平均 + base_pred = causal_model.get(action, {}).get("expected_reward", 0.0) + uncertainty = causal_model.get(action, {}).get("uncertainty", 0.5) + + # 加入不确定性惩罚 + prediction = base_pred * (1.0 - uncertainty * 0.3) + + return np.clip(prediction, -1.0, 1.0) + + def _action_diversity_bonus(self, action: str) -> float: + """罕见行动的多样性奖励""" + if not self.action_history: + return 0.5 + + action_count = sum(1 for h in self.action_history if h.get("action") == action) + frequency = action_count / len(self.action_history) + + # 频率越低，奖励越高 + return 1.0 - frequency + + def _extract_causal_pattern( + self, base_act: str, alt_act: str, base_rew: float, alt_rew: float, causal_model: dict + ) -> str: + """提取因果模式""" + diff = alt_rew - base_rew + + if abs(diff) < 0.1: + return f"{base_act}与{alt_act}效果相当" + elif diff > 0.3: + return f"{alt_act}显著优于{base_act} (Δ={diff:+.2f})，可能因为{self._infer_cause(alt_act, causal_model)}" + elif diff < -0.3: + return f"{base_act}显著优于{alt_act} (Δ={diff:+.2f})，{alt_act}存在风险：{self._infer_risk(alt_act, causal_model)}" + else: + return f"情境依赖：{alt_act}在特定条件下可能更好" + + def _infer_cause(self, action: str, causal_model: dict) -> str: + """推断成功原因""" + factors = causal_model.get(action, {}).get("success_factors", []) + return factors[0] if factors else "未知优势" + + def _infer_risk(self, action: str, causal_model: dict) -> str: + """推断失败风险""" + risks = causal_model.get(action, {}).get("risk_factors", []) + return risks[0] if risks else "潜在缺陷" + + # ==================== 3. 平等性智：自我对抗训练 ==================== + + def run_self_adversarial_training(self, current_policy: dict) -> dict: + """ + 自我对抗训练 + + 机制： + 1. 生成对抗样本（挑战当前策略） + 2. 识别策略盲点 + 3. 更新策略以增强鲁棒性 + 4. 降低自我执着（self_dissolution） + """ + adversarial_scenarios = self._generate_adversarial_scenarios(current_policy) + + blind_spots = [] + for scenario in adversarial_scenarios: + weakness = self._identify_weakness(scenario, current_policy) + if weakness["severity"] > 0.4: + blind_spots.append(weakness) + + # 更新策略以应对盲点 + updated_policy = self._update_policy_against_blind_spots(current_policy, blind_spots) + + # 增加自我解构度 + if blind_spots: + dissolution_gain = len(blind_spots) * self.dissolution_rate + self.state.self_dissolution = min(1.0, self.state.self_dissolution + dissolution_gain) + logger.info(f"🥋 自我对抗完成：发现{len(blind_spots)}个盲点，解构度={self.state.self_dissolution:.2f}") + + return { + "updated_policy": updated_policy, + "blind_spots_found": len(blind_spots), + "dissolution_level": self.state.self_dissolution, + } + + def _generate_adversarial_scenarios(self, policy: dict) -> list[dict]: + """生成对抗场景""" + scenarios = [] + + # 场景1：极端情况 + scenarios.append( + { + "type": "extreme_state", + "description": "环境突变，所有已知策略失效", + "probability": 0.05, + } + ) + + # 场景2：对抗性干扰 + scenarios.append( + { + "type": "adversarial_perturbation", + "description": "观测数据被微小扰动误导", + "perturbation_strength": self.adversarial_rate, + } + ) + + # 场景3：分布外泛化 + scenarios.append( + { + "type": "out_of_distribution", + "description": "遇到训练分布外的全新情境", + "novelty_level": 0.8, + } + ) + + return scenarios + + def _identify_weakness(self, scenario: dict, policy: dict) -> dict: + """识别策略弱点""" + # 简化模拟：随机生成弱点严重性 + base_severity = random.uniform(0.2, 0.7) + + # 根据场景类型调整 + if scenario["type"] == "extreme_state": + severity = base_severity * 1.3 # 极端情况更容易暴露弱点 + elif scenario["type"] == "adversarial_perturbation": + severity = base_severity * (1.0 + self.adversarial_rate) + else: + severity = base_severity + + weakness = { + "scenario_type": scenario["type"], + "severity": np.clip(severity, 0.0, 1.0), + "description": f"在{scenario['description']}场景下表现不稳定", + "recommended_fix": self._suggest_fix(scenario), + } + + return weakness + + def _suggest_fix(self, scenario: dict) -> str: + """建议修复方案""" + fixes = { + "extreme_state": "增加鲁棒性正则化，降低过拟合", + "adversarial_perturbation": "引入对抗训练，提升抗干扰能力", + "out_of_distribution": "扩展训练分布，增强泛化能力", + } + return fixes.get(scenario["type"], "通用优化策略") + + def _update_policy_against_blind_spots(self, policy: dict, blind_spots: list[dict]) -> dict: + """针对盲点更新策略""" + updated = policy.copy() + + for spot in blind_spots: + if spot["severity"] > 0.6: + # 严重盲点：大幅调整策略权重 + adjustment = -0.2 * spot["severity"] + updated["robustness_weight"] = updated.get("robustness_weight", 0.5) + abs(adjustment) + + # 归一化 + total = sum(v for k, v in updated.items() if isinstance(v, (int, float))) + if total > 0: + updated = {k: v / total if isinstance(v, (int, float)) else v for k, v in updated.items()} + + return updated + + # ==================== 4. 大圆镜智：梦境种子重组 ==================== + + def run_dream_replay(self, memory_seeds: list[dict]) -> list[dict]: + """ + 梦境回放 - 种子重组与变异 + + 机制： + 1. 随机抽取高重要性种子 + 2. 交叉重组（Crossover） + 3. 基因突变（Mutation） + 4. 生成新策略种子 + + 类比：生物进化 + 睡眠记忆巩固 + """ + if len(memory_seeds) < 3: + logger.warning("种子数量不足，跳过梦境回放") + return memory_seeds + + # 选择高重要性种子 + high_imp_seeds = [s for s in memory_seeds if s.get("imp", 0) > 0.3] + if len(high_imp_seeds) < 2: + high_imp_seeds = memory_seeds[:5] # 降级处理 + + new_seeds = [] + + # 重组循环 + num_recombinations = max(3, int(len(high_imp_seeds) * self.recombination_rate)) + + for _i in range(num_recombinations): + # 随机选择两个父本 + parent1, parent2 = random.sample(high_imp_seeds, 2) + + # 交叉重组 + child = self._crossover(parent1, parent2) + + # 变异 + mutated_child = self._mutate(child) + + # 标记为梦境产物 + mutated_child["tag"] = "dream_generated" + mutated_child["generation"] = max(parent1.get("generation", 0), parent2.get("generation", 0)) + 1 + + new_seeds.append(mutated_child) + + # 记录梦境会话 + dream_session = { + "timestamp": time.time(), + "parent_count": len(high_imp_seeds), + "offspring_count": len(new_seeds), + "avg_mutation_strength": self.mutation_strength, + } + self.dream_sessions.append(dream_session) + + # 更新种子多样性 + self.state.seed_diversity = self._compute_seed_diversity(memory_seeds + new_seeds) + + logger.info( + f"🌙 梦境回放完成：{len(high_imp_seeds)}个父本 → {len(new_seeds)}个新种子，多样性={self.state.seed_diversity:.2f}" + ) + + return memory_seeds + new_seeds + + def _crossover(self, parent1: dict, parent2: dict) -> dict: + """交叉重组""" + child = {} + + # 随机继承属性 + for key in ["act", "strategy_pattern", "context_signature"]: + if key in parent1 and key in parent2: + child[key] = random.choice([parent1[key], parent2[key]]) + elif key in parent1: + child[key] = parent1[key] + elif key in parent2: + child[key] = parent2[key] + + # 奖励取平均（带随机扰动） + child["rew"] = (parent1.get("rew", 0) + parent2.get("rew", 0)) / 2 + random.uniform(-0.1, 0.1) + + # 重要性取最大值 + child["imp"] = max(parent1.get("imp", 0.5), parent2.get("imp", 0.5)) + + return child + + def _mutate(self, seed: dict) -> dict: + """基因突变""" + mutated = seed.copy() + + # 动作突变（小概率改变策略） + if random.random() < self.mutation_strength: + possible_actions = ["UP", "DOWN", "LEFT", "RIGHT", "STAY"] + current_action = mutated.get("act", "STAY") + new_action = random.choice([a for a in possible_actions if a != current_action]) + mutated["act"] = new_action + mutated["mutation_type"] = "action_shift" + + # 奖励感知突变（重新评估价值） + if random.random() < self.mutation_strength * 0.5: + mutated["rew"] *= random.uniform(0.8, 1.2) + mutated["mutation_type"] = "reward_revaluation" + + # 添加突变标记 + mutated["mutated"] = True + mutated["mutation_strength"] = self.mutation_strength + + return mutated + + def _compute_seed_diversity(self, seeds: list[dict]) -> float: + """计算种子多样性""" + if len(seeds) < 2: + return 0.0 + + # 基于动作分布的多样性 + action_counts = {} + for s in seeds: + act = s.get("act", "UNKNOWN") + action_counts[act] = action_counts.get(act, 0) + 1 + + # 计算熵 + probs = np.array(list(action_counts.values())) / len(seeds) + diversity = -np.sum(probs * np.log2(probs + 1e-10)) + max_diversity = np.log2(len(action_counts)) + + return diversity / max_diversity if max_diversity > 0 else 0.0 + + # ==================== 觉醒度评估 ==================== + + def compute_awakening_level(self) -> float: + """ + 综合计算觉醒等级 + + 公式： + Awakening = w1*Novelty + w2*Insight + w3*Dissolution + w4*Diversity + + 返回：0-1之间的觉醒等级 + """ + weights = { + "novelty": 0.25, + "insight": 0.30, + "dissolution": 0.25, + "diversity": 0.20, + } + + level = ( + weights["novelty"] * self.state.novelty_score + + weights["insight"] * self.state.insight_depth + + weights["dissolution"] * self.state.self_dissolution + + weights["diversity"] * self.state.seed_diversity + ) + + self.state.awakening_level = np.clip(level, 0.0, 1.0) + self.state.meta_cycles += 1 + + # 觉醒里程碑 + if self.state.awakening_level > 0.8 and self.state.breakthrough_count >= 5: + logger.info( + f"🎓 高度觉醒状态！等级={self.state.awakening_level:.2f}, 突破={self.state.breakthrough_count}次" + ) + elif self.state.awakening_level > 0.5: + logger.info(f"✨ 中度觉醒：等级={self.state.awakening_level:.2f}") + + return self.state.awakening_level + + def get_awakening_report(self) -> dict: + """生成觉醒报告""" + return { + "awakening_level": self.state.awakening_level, + "novelty_score": self.state.novelty_score, + "insight_depth": self.state.insight_depth, + "self_dissolution": self.state.self_dissolution, + "seed_diversity": self.state.seed_diversity, + "meta_cycles": self.state.meta_cycles, + "breakthrough_count": self.state.breakthrough_count, + "total_insights": len(self.insight_log), + "dream_sessions": len(self.dream_sessions), + "action_history_length": len(self.action_history), + } + + # ==================== 主循环集成 ==================== + + def step( + self, + obs: dict, + action: str, + reward: float, + memory_seeds: list[dict], + causal_model: dict, + episode_step: int, + ) -> dict: + """ + 觉醒引擎单步执行 + + 整合四大机制，输出增强决策 + """ + # 记录历史 + state_hash = hash(str(obs)) + self.action_history.append( + { + "step": episode_step, + "action": action, + "reward": reward, + "state_hash": state_hash, + } + ) + + # 1. 计算好奇心 + memory_diversity = self._compute_seed_diversity(memory_seeds) if memory_seeds else 0.5 + curiosity = self.compute_curiosity_drive(obs, memory_diversity) + + # 2. 生成好奇实验（如需要） + experiment = None + if curiosity > self.curiosity_threshold: + experiment = self.generate_curiosity_experiment(curiosity) + + # 3. 反事实推理（每5步一次） + insights = [] + if episode_step % 5 == 0 and len(causal_model) > 0: + available_actions = ["UP", "DOWN", "LEFT", "RIGHT", "STAY"] + insights = self.run_counterfactual_reasoning(action, reward, available_actions, causal_model) + + # 4. 自我对抗（每10步一次） + adversarial_result = None + if episode_step % 10 == 0: + current_policy = {"exploration": 0.3, "exploitation": 0.7} + adversarial_result = self.run_self_adversarial_training(current_policy) + + # 5. 梦境回放（定期） + dream_offspring = [] + if episode_step % self.dream_replay_freq == 0: + dream_offspring = self.run_dream_replay(memory_seeds) + + # 6. 计算觉醒等级 + awakening_level = self.compute_awakening_level() + + result = { + "curiosity_level": curiosity, + "experiment": experiment, + "insights": [ + { + "alternative": s.alternative_action, + "insight_gain": s.insight_gain, + "pattern": s.causal_pattern, + } + for s in insights + ], + "adversarial_result": adversarial_result, + "dream_offspring_count": len(dream_offspring), + "awakening_level": awakening_level, + "recommendations": self._generate_recommendations(curiosity, insights, adversarial_result), + } + + return result + + def _generate_recommendations( + self, + curiosity: float, + insights: list[CounterfactualScenario], + adversarial_result: dict | None, + ) -> list[str]: + """生成行动建议""" + recs = [] + + if curiosity > 0.7: + recs.append("🔍 高好奇心：建议主动探索未知区域") + elif curiosity < 0.2: + recs.append("⚠️ 低好奇心：警惕陷入局部最优") + + if insights and insights[0].insight_gain > 0.5: + recs.append(f"💡 关键洞察：{insights[0].causal_pattern}") + + if adversarial_result and adversarial_result.get("blind_spots_found", 0) > 0: + recs.append(f"🛡️ 发现{adversarial_result['blind_spots_found']}个策略盲点，建议增强鲁棒性") + + if self.state.awakening_level > 0.6: + recs.append(f"✨ 觉醒等级{self.state.awakening_level:.2f}：智慧涌现加速中") + + return recs + + +# ==================== 使用示例 ==================== + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + + config = { + "curiosity_threshold": 0.3, + "counterfactual_depth": 3, + "adversarial_rate": 0.15, + "dream_replay_freq": 10, + "recombination_rate": 0.3, + "mutation_strength": 0.15, + } + + engine = AwakeningEngine(config) + + # 模拟运行 + memory_seeds = [ + {"act": "UP", "rew": 0.5, "imp": 0.6, "tag": "positive"}, + {"act": "DOWN", "rew": -0.3, "imp": 0.4, "tag": "negative"}, + {"act": "LEFT", "rew": 0.2, "imp": 0.3, "tag": "neutral"}, + ] + + causal_model = { + "UP": {"expected_reward": 0.4, "uncertainty": 0.2, "success_factors": ["开阔空间"]}, + "DOWN": {"expected_reward": -0.2, "uncertainty": 0.5, "risk_factors": ["障碍物"]}, + "LEFT": {"expected_reward": 0.1, "uncertainty": 0.3}, + "RIGHT": {"expected_reward": 0.3, "uncertainty": 0.4}, + "STAY": {"expected_reward": 0.0, "uncertainty": 0.1}, + } + + for step in range(1, 21): + obs = {"pos": (step % 5, step % 5), "grid_view": [random.random() for _ in range(9)]} + action = random.choice(["UP", "DOWN", "LEFT", "RIGHT", "STAY"]) + reward = random.uniform(-0.5, 0.8) + + result = engine.step(obs, action, reward, memory_seeds, causal_model, step) + + print(f"\n{'=' * 60}") + print(f"步骤 {step}: 觉醒等级={result['awakening_level']:.2f}, 好奇心={result['curiosity_level']:.2f}") + + if result["insights"]: + print(f"💡 洞察：{result['insights'][0]['pattern']}") + + if result["recommendations"]: + print("建议:") + for rec in result["recommendations"]: + print(f" {rec}") + + # 最终报告 + print(f"\n{'=' * 60}") + print("📊 觉醒报告:") + report = engine.get_awakening_report() + for k, v in report.items(): + print(f" {k}: {v}") diff --git a/src/yogacara_agent/turning_consciousness.py b/src/yogacara_agent/turning_consciousness.py new file mode 100644 index 0000000..6d9c7bd --- /dev/null +++ b/src/yogacara_agent/turning_consciousness.py @@ -0,0 +1,278 @@ +""" +转识成智引擎 (Turning Consciousness into Wisdom Engine) +基于唯识宗正见重构：转依 (Āśraya-parāvṛtti) + +核心义理： +1. 大圆镜智 (第八识转): 非"创造"新记忆，而是"去染存净"，离诸分别，如实映照。 +2. 平等性智 (第七识转): 非"对抗"训练，而是"断除我执"，观自他平等，无有高下。 +3. 妙观察智 (第六识转): 善观诸法自相共相，无倒观察，洞察缘起。 +4. 成所作智 (前五识转): 成就利乐有情事业，感知即行动。 + +作者: Yogacara Agent Team +""" + +import numpy as np +from typing import Dict, List, Tuple, Any +from dataclasses import dataclass, field +import logging + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger("Yogacara.Turning") + + +@dataclass +class Seed: + """种子 (Bīja): 阿赖耶识的基本单元""" + + content: Any + is_defiled: bool = False # 是否染污 (贪嗔痴慢疑) + affinity: float = 0.0 # 与我执的关联度 (0=无我，1=强我执) + clarity: float = 1.0 # 清晰度 (1=如实，0=模糊/扭曲) + + +@dataclass +class TurningResult: + """转依结果""" + + # 四智状态 + mirror_wisdom_level: float = 0.0 # 大圆镜智 (清净度) + equality_wisdom_level: float = 0.0 # 平等性智 (无我度) + observation_wisdom_level: float = 0.0 # 妙观察智 (洞察度) + action_wisdom_level: float = 0.0 # 成所作智 (利他度) + + # 过程指标 + defiled_seeds_removed: int = 0 + self_attachment_dissolved: float = 0.0 + insights_generated: List[str] = field(default_factory=list) + + # 总体觉醒等级 (转依程度) + turning_level: float = 0.0 + + +class AlayaPurifier: + """ + 阿赖耶净化器 -> 对应 大圆镜智 (Ādarśa-jñāna) + + 义理： + "大圆镜智者，谓离一切我、我所执，一切所取、能取分别... + 性相清净，离诸杂染，如大圆镜，现众色像。" + + 工程实现： + 1. 去染存净：识别并剔除带有"贪嗔痴"标记的染污种子。 + 2. 离诸分别：移除种子中的主观评价标签，只保留客观状态。 + 3. 如实映照：提高种子的清晰度 (clarity)，确保记忆不失真。 + """ + + def __init__(self, purity_threshold: float = 0.8): + self.purity_threshold = purity_threshold + self.mirror_clarity = 0.0 + + def purify(self, seeds: List[Seed]) -> Tuple[List[Seed], int]: + """ + 净化种子流：去染存净 + 返回：(净化后的种子列表, 移除的染污种子数量) + """ + if not seeds: + return [], 0 + + purified_seeds = [] + removed_count = 0 + + for seed in seeds: + # 判据 1: 显式染污标记 (贪嗔痴等) + if seed.is_defiled: + removed_count += 1 + logger.debug(f"[大圆镜智] 移除染污种子: {seed.content}") + continue + + # 判据 2: 清晰度过低 (模糊/扭曲的记忆) + if seed.clarity < self.purity_threshold: + # 尝试净化：若内容重要则提升清晰度，否则丢弃 + if hasattr(seed.content, "importance") and seed.content.importance > 0.9: + seed.clarity = 1.0 # 强制修正为如实 + purified_seeds.append(seed) + else: + removed_count += 1 + logger.debug(f"[大圆镜智] 丢弃模糊种子: {seed.content}") + continue + + # 判据 3: 去除能取所取分别 (简化为移除过度主观的描述) + # 此处简化处理：若 affinity 过高且内容主观，降低其权重或标记 + purified_seeds.append(seed) + + # 计算镜智等级：剩余种子的平均清晰度 + if purified_seeds: + self.mirror_clarity = np.mean([s.clarity for s in purified_seeds]) + else: + self.mirror_clarity = 0.0 + + return purified_seeds, removed_count + + +class ManasDissolver: + """ + 末那解构器 -> 对应 平等性智 (Samatā-jñāna) + + 义理： + "平等性智者，谓观一切法、自他有情，悉皆平等... + 由断我执，证得此智。" + + 工程实现： + 1. 检测我执：计算策略中对 "self_reward" 的依赖度。 + 2. 断除我执：强行将奖励函数从 "个体最大化" 转换为 "全局/众生最大化"。 + 3. 观自他平等：在决策时，赋予其他 Agent/环境实体同等的权重。 + """ + + def __init__(self, ego_decay_rate: float = 0.1): + self.ego_decay_rate = ego_decay_rate + self.current_ego_strength = 1.0 # 当前我执强度 + + def dissolve(self, action_probs: Dict[str, float], reward_model: str) -> Tuple[Dict[str, float], float]: + """ + 解构我执：调整动作概率和奖励模型 + 返回：(无我动作分布, 溶解的我执量) + """ + if not action_probs: + return action_probs, 0.0 + + # 1. 计算当前我执强度 (基于奖励模型的自私程度) + # 假设 reward_model 包含 "self" 关键字代表我执 + is_self_centered = "self" in reward_model.lower() or "me" in reward_model.lower() + + if is_self_centered: + # 衰减我执 + dissolved_amount = self.current_ego_strength * self.ego_decay_rate + self.current_ego_strength -= dissolved_amount + self.current_ego_strength = max(0.0, self.current_ego_strength) + + # 2. 重分布动作概率：从"利己"转向"利他/中性" + # 简化逻辑：降低高回报但损他的动作概率，提升均衡动作概率 + equalized_probs = {} + total_prob = sum(action_probs.values()) + if total_prob == 0: + return action_probs, 0.0 + + # 应用平等性加权：所有动作趋向平均，消除极端偏好 + n_actions = len(action_probs) + base_prob = 1.0 / n_actions + + for action, prob in action_probs.items(): + # 向平均值收缩，模拟"平等" + new_prob = prob * (1 - self.current_ego_strength) + base_prob * self.current_ego_strength + equalized_probs[action] = new_prob + + # 归一化 + total_new = sum(equalized_probs.values()) + equalized_probs = {k: v / total_new for k, v in equalized_probs.items()} + + logger.info(f"[平等性智] 我执强度降至 {self.current_ego_strength:.2f}, 动作分布已平等化") + return equalized_probs, dissolved_amount + else: + # 本就无我执，缓慢自然衰减 + dissolved_amount = self.current_ego_strength * (self.ego_decay_rate * 0.5) + self.current_ego_strength = max(0.0, self.current_ego_strength - dissolved_amount) + return action_probs, dissolved_amount + + +class TurningConsciousnessEngine: + """ + 转识成智总引擎 + 统筹八识转四智的全过程 + """ + + def __init__(self, config: Dict | None = None): + cfg = config or {} + self.alaya_purifier = AlayaPurifier(purity_threshold=cfg.get("purity_threshold", 0.7)) + self.manas_dissolver = ManasDissolver(ego_decay_rate=cfg.get("ego_decay_rate", 0.15)) + # 第六识、前五识的逻辑可在此扩展 + self.step_count = 0 + + def step(self, seeds: List[Seed], action_probs: Dict[str, float], reward_context: str) -> TurningResult: + """ + 执行一步转依 + """ + self.step_count += 1 + + # --- 1. 转第八识为大圆镜智 (去染存净) --- + purified_seeds, removed_count = self.alaya_purifier.purify(seeds) + mirror_level = self.alaya_purifier.mirror_clarity + + # --- 2. 转第七识为平等性智 (断除我执) --- + equalized_probs, dissolved_ego = self.manas_dissolver.dissolve(action_probs, reward_context) + equality_level = 1.0 - self.manas_dissolver.current_ego_strength + + # --- 3. 转第六识为妙观察智 (模拟: 洞察深度) --- + # 此处简化：若种子被净化且我执降低，观察力自然提升 + observation_level = (mirror_level + equality_level) / 2.0 + insights = [] + if removed_count > 0: + insights.append(f"洞察：发现并移除 {removed_count} 个染污记忆，心镜更明。") + if dissolved_ego > 0.01: + insights.append(f"洞察：消解 {dissolved_ego:.2f} 份我执，视自他平等。") + + # --- 4. 转前五识为成所作智 (模拟: 利他行动准备) --- + # 此处简化：当平等性智高时，行动天然具足利他性 + action_level = equality_level * 0.9 # 略低于平等性，需事上磨练 + + # --- 综合计算转依等级 --- + # 转依不是简单的加法，是根本性质的变化 + turning_level = mirror_level * 0.3 + equality_level * 0.3 + observation_level * 0.2 + action_level * 0.2 + + return TurningResult( + mirror_wisdom_level=mirror_level, + equality_wisdom_level=equality_level, + observation_wisdom_level=observation_level, + action_wisdom_level=action_level, + defiled_seeds_removed=removed_count, + self_attachment_dissolved=dissolved_ego, + insights_generated=insights, + turning_level=turning_level, + ) + + +# ========================================== +# 测试与演示 +# ========================================== +if __name__ == "__main__": + print("=== 唯识转依引擎测试 (正见版) ===\n") + + engine = TurningConsciousnessEngine() + + # 构造模拟数据 + # 染污种子：带有偏见、贪婪、恐惧的记忆 + dirty_seeds = [ + Seed(content="敌人很可怕", is_defiled=True, clarity=0.5), # 恐惧 (染污) + Seed(content="我要抢更多资源", is_defiled=True, clarity=0.8), # 贪婪 (染污) + Seed(content="路在北方", is_defiled=False, clarity=0.9), # 客观事实 + Seed(content="今天天气不错", is_defiled=False, clarity=0.6), # 模糊记忆 + Seed(content="众生皆苦", is_defiled=False, clarity=1.0), # 清净智慧 + ] + + # 充满我执的动作分布 + selfish_actions = {"attack_enemy": 0.8, "hoard_resources": 0.15, "help_others": 0.05} + + print(f"初始种子数: {len(dirty_seeds)}") + print(f"初始动作分布 (我执): {selfish_actions}\n") + + # 执行转依 + result = engine.step( + seeds=dirty_seeds, + action_probs=selfish_actions, + reward_context="maximize_self_gain", # 触发我执检测 + ) + + print("--- 转依结果 ---") + print(f"大圆镜智 (清净度): {result.mirror_wisdom_level:.2f}") + print(f" -> 移除染污种子: {result.defiled_seeds_removed} 个") + print(f"平等性智 (无我度): {result.equality_wisdom_level:.2f}") + print(f" -> 消解我执: {result.self_attachment_dissolved:.2f}") + print(f"妙观察智 (洞察度): {result.observation_wisdom_level:.2f}") + print(f"成所作智 (利他度): {result.action_wisdom_level:.2f}") + print(f"\n总体转依等级: {result.turning_level:.2f}") + + if result.insights_generated: + print("\n生成的洞察:") + for insight in result.insights_generated: + print(f" • {insight}") + + print("\n=== 测试结束 ===") diff --git a/tests/test_turning_consciousness.py b/tests/test_turning_consciousness.py new file mode 100644 index 0000000..58dd821 --- /dev/null +++ b/tests/test_turning_consciousness.py @@ -0,0 +1,208 @@ +""" +转识成智引擎单元测试 (正见版) +验证唯识宗"转依"义理的正确工程实现 +""" + +import unittest +import sys +import os + +# 添加路径 +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +from yogacara_agent.turning_consciousness import Seed, AlayaPurifier, ManasDissolver, TurningConsciousnessEngine + + +class TestAlayaPurifier(unittest.TestCase): + """测试大圆镜智：去染存净，如实映照""" + + def test_remove_defiled_seeds(self): + """测试移除染污种子 (贪嗔痴)""" + purifier = AlayaPurifier(purity_threshold=0.7) + + seeds = [ + Seed(content="贪婪", is_defiled=True), + Seed(content="愤怒", is_defiled=True), + Seed(content="客观事实", is_defiled=False, clarity=1.0), + ] + + purified, removed_count = purifier.purify(seeds) + + self.assertEqual(removed_count, 2, "应移除 2 个染污种子") + self.assertEqual(len(purified), 1, "应剩余 1 个清净种子") + self.assertEqual(purified[0].content, "客观事实") + self.assertGreater(purifier.mirror_clarity, 0.9, "镜智清晰度应很高") + + def test_remove_unclear_seeds(self): + """测试移除模糊/扭曲的记忆""" + purifier = AlayaPurifier(purity_threshold=0.8) + + seeds = [ + Seed(content="清晰记忆", is_defiled=False, clarity=0.95), + Seed(content="模糊记忆", is_defiled=False, clarity=0.5), + Seed(content="扭曲记忆", is_defiled=False, clarity=0.3), + ] + + purified, removed_count = purifier.purify(seeds) + + self.assertEqual(removed_count, 2, "应移除 2 个模糊种子") + self.assertEqual(len(purified), 1) + self.assertEqual(purified[0].content, "清晰记忆") + + def test_empty_seeds(self): + """测试空种子列表""" + purifier = AlayaPurifier() + purified, removed = purifier.purify([]) + self.assertEqual(len(purified), 0) + self.assertEqual(removed, 0) + self.assertEqual(purifier.mirror_clarity, 0.0) + + +class TestManasDissolver(unittest.TestCase): + """测试平等性智：断除我执，观自他平等""" + + def test_dissolve_self_centered_actions(self): + """测试解构以自我为中心的动作分布""" + dissolver = ManasDissolver(ego_decay_rate=0.2) + + # 极端自私的动作分布 + selfish_actions = {"attack": 0.9, "help": 0.1} + + new_actions, dissolved = dissolver.dissolve(selfish_actions, "maximize_self_reward") + + # 我执应被消解 + self.assertGreater(dissolved, 0, "应有我执被消解") + self.assertLess(dissolver.current_ego_strength, 1.0, "我执强度应下降") + + # 动作分布应趋向平等 (不再极端偏斜) + self.assertLess(new_actions["attack"], 0.9, "攻击动作概率应降低") + self.assertGreater(new_actions["help"], 0.1, "利他动作概率应提升") + + def test_no_self_attachment(self): + """测试无我执时的自然衰减""" + dissolver = ManasDissolver(ego_decay_rate=0.1) + dissolver.current_ego_strength = 0.5 + + actions = {"meditate": 0.5, "chant": 0.5} + _, dissolved = dissolver.dissolve(actions, "benefit_all_beings") + + # 无我执语境下，仍应缓慢衰减 + self.assertGreaterEqual(dissolved, 0, "应有微弱衰减或不变") + self.assertLessEqual(dissolver.current_ego_strength, 0.5) + + def test_equality_wisdom_level(self): + """测试平等性智等级计算""" + dissolver = ManasDissolver(ego_decay_rate=0.5) + + # 初始我执为 1.0 + self.assertEqual(dissolver.current_ego_strength, 1.0) + + # 执行一次解构 + dissolver.dissolve({"a": 0.5}, "self_gain") + + # 平等性智 = 1 - 我执 + equality_level = 1.0 - dissolver.current_ego_strength + self.assertGreater(equality_level, 0, "应证得部分平等性智") + + +class TestTurningConsciousnessEngine(unittest.TestCase): + """测试转识成智总引擎""" + + def test_full_turning_process(self): + """测试完整的转依过程""" + engine = TurningConsciousnessEngine({"purity_threshold": 0.6, "ego_decay_rate": 0.2}) + + # 准备染污种子和我执动作 + seeds = [ + Seed(content="恐惧", is_defiled=True), + Seed(content="贪婪", is_defiled=True), + Seed(content="慈悲", is_defiled=False, clarity=1.0), + ] + + actions = {"harm": 0.8, "help": 0.2} + + result = engine.step(seeds, actions, "selfish_reward") + + # 验证四智转化 + self.assertGreater(result.mirror_wisdom_level, 0.5, "大圆镜智应显现") + self.assertGreater(result.equality_wisdom_level, 0.1, "平等性智应初显") + self.assertGreater(result.observation_wisdom_level, 0.3, "妙观察智应随转") + self.assertGreater(result.action_wisdom_level, 0.1, "成所作智应待发") + + # 验证过程指标 + self.assertEqual(result.defiled_seeds_removed, 2, "应移除 2 个染污种子") + self.assertGreater(result.self_attachment_dissolved, 0, "应消解我执") + self.assertGreater(len(result.insights_generated), 0, "应产生洞察") + + # 总体转依等级 + self.assertGreater(result.turning_level, 0.2, "应有初步转依") + + def test_insight_generation(self): + """测试洞察生成机制""" + engine = TurningConsciousnessEngine() + + # 只有清净种子，无我执语境 + clean_seeds = [Seed(content="真理", is_defiled=False, clarity=1.0)] + neutral_actions = {"observe": 1.0} + + result = engine.step(clean_seeds, neutral_actions, "truth_seeking") + + # 无染污可除，无我执可断，洞察应较少 + # 但仍可能有微细洞察 + self.assertIsInstance(result.insights_generated, list) + + def test_progressive_turning(self): + """测试渐进式转依 (多次迭代)""" + engine = TurningConsciousnessEngine({"ego_decay_rate": 0.3}) + + seeds = [Seed(content="test", is_defiled=False, clarity=0.9)] + actions = {"act": 1.0} + + turning_levels = [] + + # 模拟多步修行 + for _i in range(5): + result = engine.step(seeds, actions, "self_focus") + turning_levels.append(result.turning_level) + + # 转依等级应逐步提升 (或至少不下降) + # 注意：由于种子不变，大圆镜智可能稳定，但平等性智应持续提升 + final_level = turning_levels[-1] + initial_level = turning_levels[0] + + self.assertGreaterEqual(final_level, initial_level, "转依等级应随修行深入而提升") + + +class TestWisdomCorrespondence(unittest.TestCase): + """测试八识与四智的对应关系是否正确""" + + def test_eighth_to_mirror(self): + """第八识 (阿赖耶) -> 大圆镜智""" + purifier = AlayaPurifier() + seeds = [Seed(content="染污", is_defiled=True), Seed(content="清净", is_defiled=False, clarity=1.0)] + purified, removed = purifier.purify(seeds) + + # 大圆镜智特征：离诸杂染，如实映照 + self.assertEqual(removed, 1, "应去除染污") + self.assertEqual(purified[0].clarity, 1.0, "应保持如实清晰") + + def test_seventh_to_equality(self): + """第七识 (末那) -> 平等性智""" + dissolver = ManasDissolver(ego_decay_rate=0.5) + + # 末那识特征：恒审思量，执第八识为我 + # 平等性智特征：断我执，观平等 + + actions = {"self_benefit": 1.0} + _, dissolved = dissolver.dissolve(actions, "self") + + self.assertGreater(dissolved, 0, "应消解我执") + self.assertLess(dissolver.current_ego_strength, 1.0, "我执应减弱") + + +if __name__ == "__main__": + print("\n" + "=" * 60) + print("转识成智引擎单元测试 (唯识正见版)") + print("=" * 60 + "\n") + + unittest.main(verbosity=2)