- commit
- fabbb69
- parent
- d00a2ea
- author
- im_wower
- date
- 2026-04-01 10:59:35 +0800 CST
integration phase2.2: Kernel Sanity Gate——修3个内核bug+6项验证 (53/53)
4 files changed,
+207,
-2
Raw patch view.
1diff --git a/cie/dynamics.py b/cie/dynamics.py
2index ea706aefd841cb5daf645b9cba3b9d431e76c988..c3dfcbb2a99034ffcd7a5b7dff01bd7b80aaf449 100644
3--- a/cie/dynamics.py
4+++ b/cie/dynamics.py
5@@ -106,6 +106,8 @@ class Dynamics:
6 self.state.J[(node_id, nb)] = (
7 self.state.J.get((node_id, nb), 0.0) * 0.9 + flow
8 )
9+ # attention accounting: 传播到邻居的激活也要记账
10+ self.state.attention.allocate(nb, flow)
11 # 记录经验命中
12 self.state.experience_hits[nb] = (
13 self.state.experience_hits.get(nb, 0) + 1
14@@ -113,6 +115,8 @@ class Dynamics:
15 self.state.active_region.add(nb)
16
17 new_mu[node_id] = mu_v - spread_amount
18+ # attention accounting: 释放从源节点流出的激活
19+ self.state.attention.release(node_id, spread_amount)
20
21 self.state.mu.update(new_mu)
22 # 清理极低激活
23diff --git a/cie/graph.py b/cie/graph.py
24index 35e41402d73508d0d3bd48dce5f101b9791d1d20..d24bd4dec8cccd9a05086df1c90ab401c9bdd648 100644
25--- a/cie/graph.py
26+++ b/cie/graph.py
27@@ -183,7 +183,7 @@ class Graph:
28 result = 0.0
29 # 对于每条边 (node_id, dst)
30 for dst, fwd_edge in self.fwd_edges.get(node_id, {}).items():
31- bwd_w = self.get_bwd_weight(node_id, dst)
32+ bwd_w = self.get_bwd_weight(dst, node_id) # dst←node_id 的反向权重
33 result += (fwd_edge.weight - bwd_w) * phi.get(dst, 0.0)
34
35 return result
36@@ -222,7 +222,7 @@ class Graph:
37 for nid, n in self.nodes.items()},
38 'edges': [
39 {'src': e.src, 'dst': e.dst, 'weight': e.weight,
40- 'bwd_weight': self.get_bwd_weight(e.src, e.dst),
41+ 'bwd_weight': self.get_bwd_weight(e.dst, e.src), # dst←src 的反向权重
42 'type': e.edge_type}
43 for src_edges in self.fwd_edges.values()
44 for e in src_edges.values()
45diff --git a/cie/runtime.py b/cie/runtime.py
46index cb4b3b5b09a25ea799d685f0490df9b265ca8f8d..a5ddd668d78e51e42247e0501180cb1ccee4df41 100644
47--- a/cie/runtime.py
48+++ b/cie/runtime.py
49@@ -55,9 +55,17 @@ class CIERuntime:
50 elif isinstance(anchors, (list, tuple)):
51 anchor_tokens = list(anchors)
52
53+ context_tokens = []
54+ if context:
55+ if isinstance(context, str):
56+ context_tokens = list(context)[:10]
57+ elif isinstance(context, (list, tuple)):
58+ context_tokens = list(context)[:10]
59+
60 signal = PendingSignal(
61 source="external",
62 tokens=tokens,
63+ context_tokens=context_tokens,
64 anchor_tokens=anchor_tokens,
65 strength=1.0,
66 polarity=1,
67@@ -118,6 +126,24 @@ class CIERuntime:
68 self.state.update_confidence(src, 0, amount=0.5 * signal.strength)
69 self.state.update_confidence(dst, 1, amount=0.5 * signal.strength)
70
71+ # ── context 消费：context tokens 与主 tokens 建立弱关联 ──
72+ for ctx in signal.context_tokens:
73+ if not ctx:
74+ continue
75+ if not self.graph.has_node(ctx):
76+ self.graph.add_node(ctx, label=ctx)
77+ self.state.init_node(ctx, phi_val=self.rng.gauss(0.0, 0.05))
78+ # context 与每个主 token 建立弱边
79+ for token in tokens[:5]: # 只取前 5 个避免爆炸
80+ if token and token != ctx:
81+ existing = self.graph.get_edge_weight(ctx, token)
82+ self.graph.add_edge(
83+ ctx, token,
84+ weight=existing + 0.3 * signal.strength,
85+ bwd_weight=existing + 0.2 * signal.strength,
86+ edge_type='context'
87+ )
88+
89 # ── 锚点注入 ──
90 for anchor in signal.anchor_tokens:
91 if not self.graph.has_node(anchor):
92diff --git a/tests/test_kernel_sanity.py b/tests/test_kernel_sanity.py
93new file mode 100644
94index 0000000000000000000000000000000000000000..aa7f237efb1d9cddcc1f81612679a2e44b2b4911
95--- /dev/null
96+++ b/tests/test_kernel_sanity.py
97@@ -0,0 +1,175 @@
98+"""
99+Phase 2.2: Kernel Sanity Gate Tests
100+=====================================
101+验证 Branch B 内核的 4 个关键接线：
102+1. asymmetry / backward-weight 读取正确性
103+2. Dirichlet per-node wiring（cat0/cat1/cat2 真正被更新）
104+3. context 参数真实消费
105+4. attention ledger 与 mu 对齐
106+"""
107+
108+import sys
109+import os
110+import math
111+
112+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
113+from cie import CIERuntime
114+from cie.graph import Graph
115+
116+
117+def test_asymmetry_correct():
118+ """asymmetry_at 使用正确的反向权重"""
119+ g = Graph()
120+ g.add_edge('a', 'b', weight=2.0, bwd_weight=0.5)
121+
122+ phi = {'a': 1.0, 'b': 1.0}
123+
124+ # asymmetry_at('a') 应该用 fwd('a','b')=2.0 和 bwd('a','b')
125+ # bwd_weight 是 b→a 方向的权重 = 0.5
126+ # asymmetry = (fwd - bwd) * phi(b) = (2.0 - 0.5) * 1.0 = 1.5
127+ asym = g.asymmetry_at('a', phi)
128+ expected = (2.0 - 0.5) * 1.0 # 1.5
129+
130+ assert abs(asym - expected) < 0.01, \
131+ f"asymmetry_at wrong: got {asym}, expected {expected}"
132+ print(f" PASS: asymmetry_at('a')={asym:.2f}, expected={expected:.2f}")
133+
134+
135+def test_bwd_weight_in_to_dict():
136+ """to_dict 导出正确的反向权重"""
137+ g = Graph()
138+ g.add_edge('x', 'y', weight=3.0, bwd_weight=1.0)
139+
140+ d = g.to_dict()
141+ edges = d['edges']
142+ xy_edge = next(e for e in edges if e['src'] == 'x' and e['dst'] == 'y')
143+
144+ assert abs(xy_edge['weight'] - 3.0) < 0.01, \
145+ f"fwd weight wrong: {xy_edge['weight']}"
146+ assert abs(xy_edge['bwd_weight'] - 1.0) < 0.01, \
147+ f"bwd weight wrong: {xy_edge['bwd_weight']}, expected 1.0"
148+ print(f" PASS: to_dict bwd_weight={xy_edge['bwd_weight']:.2f}")
149+
150+
151+def test_dirichlet_per_node_wiring():
152+ """普通 ingest 真正更新 per-node Dirichlet cat0/cat1"""
153+ rt = CIERuntime(seed=42)
154+ rt.ingest("你好世界")
155+ rt.step(n=1) # 消费信号
156+
157+ # bigram 你→好：你应该有 cat0 更新，好应该有 cat1 更新
158+ conf_ni = rt.state.confidence.get('你', None)
159+ conf_hao = rt.state.confidence.get('好', None)
160+
161+ assert conf_ni is not None, "Node '你' has no confidence"
162+ assert conf_hao is not None, "Node '好' has no confidence"
163+
164+ # cat0（左上下文）应该 > 1.0（先验）对于 '你'
165+ assert conf_ni[0] > 1.0, \
166+ f"'你' cat0 not updated: {conf_ni}"
167+
168+ # cat1（右上下文）应该 > 1.0 对于 '好'
169+ assert conf_hao[1] > 1.0, \
170+ f"'好' cat1 not updated: {conf_hao}"
171+
172+ # 不是全都一样（打破了均匀先验）
173+ assert not (abs(conf_ni[0] - conf_ni[1]) < 0.01 and abs(conf_ni[1] - conf_ni[2]) < 0.01), \
174+ f"'你' Dirichlet still uniform: {conf_ni}"
175+
176+ print(f" PASS: Dirichlet per-node — '你'={[round(x,2) for x in conf_ni]}, "
177+ f"'好'={[round(x,2) for x in conf_hao]}")
178+
179+
180+def test_context_consumed():
181+ """context 参数被真实消费——建立了 context→token 的边"""
182+ rt = CIERuntime(seed=42)
183+ rt.ingest("你好", context="背景")
184+ rt.step(n=1)
185+
186+ # context 字符应该存在于图中
187+ assert rt.graph.has_node('背'), f"Context char '背' not in graph"
188+ assert rt.graph.has_node('景'), f"Context char '景' not in graph"
189+
190+ # context 字符与主 token 之间应有边
191+ has_edge = False
192+ for ctx_char in ['背', '景']:
193+ for tok_char in ['你', '好']:
194+ if rt.graph.get_edge_weight(ctx_char, tok_char) > 0:
195+ has_edge = True
196+ break
197+ if has_edge:
198+ break
199+
200+ assert has_edge, "No edges between context and tokens"
201+ print(f" PASS: context consumed — '背','景' in graph with edges to tokens")
202+
203+
204+def test_attention_mu_alignment():
205+ """attention.used 与 sum(mu) 在受控场景下对齐"""
206+ rt = CIERuntime(seed=42)
207+ rt.ingest("测试")
208+ rt.step(n=5)
209+
210+ mu_sum = sum(v for v in rt.state.mu.values() if v > 0)
211+ attn_used = rt.state.attention.used
212+
213+ # 允许 20% 偏差（传播中有衰减和清理）
214+ if mu_sum > 0.1:
215+ ratio = attn_used / mu_sum if mu_sum > 0 else float('inf')
216+ # attention 应该 >= mu（因为还包含已衰减但未释放的部分）
217+ # 但不应该差太多
218+ assert 0.5 < ratio < 3.0, \
219+ f"attention/mu drift: attn={attn_used:.2f}, mu={mu_sum:.2f}, ratio={ratio:.2f}"
220+
221+ print(f" PASS: attention/mu aligned — attn={attn_used:.2f}, mu={mu_sum:.2f}")
222+
223+
224+def test_attention_no_unbounded_drift():
225+ """多轮运行后 attention 不会无界漂移"""
226+ rt = CIERuntime(seed=42)
227+
228+ for i in range(20):
229+ rt.ingest(f"轮次{i}")
230+ rt.step(n=3)
231+
232+ attn = rt.state.attention
233+ assert attn.used <= attn.total + 0.01, \
234+ f"Attention overflow: used={attn.used:.2f} > total={attn.total:.2f}"
235+ assert attn.free >= -0.01, \
236+ f"Attention went negative: free={attn.free:.2f}"
237+
238+ mu_sum = sum(v for v in rt.state.mu.values() if v > 0)
239+ print(f" PASS: no unbounded drift after 20 rounds — "
240+ f"attn_used={attn.used:.2f}, mu_sum={mu_sum:.2f}, free={attn.free:.2f}")
241+
242+
243+def run_all():
244+ tests = [
245+ ("asymmetry_correct", test_asymmetry_correct),
246+ ("bwd_weight_to_dict", test_bwd_weight_in_to_dict),
247+ ("dirichlet_per_node", test_dirichlet_per_node_wiring),
248+ ("context_consumed", test_context_consumed),
249+ ("attention_mu_align", test_attention_mu_alignment),
250+ ("attention_no_drift", test_attention_no_unbounded_drift),
251+ ]
252+
253+ passed = 0
254+ failed = 0
255+ for name, fn in tests:
256+ try:
257+ print(f"[KERNEL] {name}...")
258+ fn()
259+ passed += 1
260+ except Exception as e:
261+ print(f" FAIL: {e}")
262+ failed += 1
263+
264+ print(f"\n{'='*50}")
265+ print(f"Kernel Sanity Gate: {passed} passed, {failed} failed, {len(tests)} total")
266+ print(f"{'='*50}")
267+ return failed == 0
268+
269+
270+if __name__ == '__main__':
271+ success = run_all()
272+ sys.exit(0 if success else 1)