im_wower
·
2026-03-31
runtime.py
1"""
2CIE Runtime — 统一接口（SPEC §5）
3
4六个方法：ingest, step, emit, commit_feedback, snapshot_state, reset_session
5这是 Branch B 的工程稳健 runtime 实现。
6"""
7
8import random
9import math
10from typing import Optional
11from .graph import Graph
12from .state import CIEState
13from .dynamics import Dynamics
14
15
16class CIERuntime:
17 """
18 CIE 运行时——图原生认知推理引擎。
19
20 Branch B 定位：工程稳健增强 runtime。
21 先能跑、先能验证、先能出结果。
22 """
23
24 def __init__(self, seed: int = None):
25 self.graph = Graph()
26 self.state = CIEState()
27 self.dynamics = Dynamics(self.graph, self.state)
28 self.rng = random.Random(seed)
29
30 # ── 输出缓冲 ──
31 self._output_buffer: list[dict] = []
32 self._last_output: Optional[dict] = None
33
34 # ── 回灌标记 ──
35 self._feedback_pending = False
36
37 # ──────────────────────────────────────
38 # §5.1 ingest — 接收输入，注入图中
39 # ──────────────────────────────────────
40
41 def ingest(self, input_data, context=None, anchors=None):
42 """
43 接收新的输入、上下文、可选锚点提示，注入图中。
44
45 input_data: str 或 list[str]
46 - str: 文本，按字符拆分为 bigram 注入
47 - list[str]: 已分好的 token 列表
48 context: dict, optional
49 - 额外上下文信息
50 anchors: list[str], optional
51 - 锚点提示——谁、在哪、做什么
52
53 输出即输入：如果有上一轮输出，自动回灌。
54 """
55 # ── 输出即输入回灌 ──
56 if self._last_output is not None and self._feedback_pending:
57 self._feedback_loop(self._last_output)
58 self._feedback_pending = False
59
60 # ── 解析输入为 token 序列 ──
61 if isinstance(input_data, str):
62 tokens = list(input_data)
63 elif isinstance(input_data, (list, tuple)):
64 tokens = list(input_data)
65 else:
66 tokens = [str(input_data)]
67
68 if not tokens:
69 return
70
71 # ── 并行归位：所有 token 同时注入图 ──
72 # "一把种子同时撒在图的不同层级上"
73 for token in tokens:
74 if not self.graph.has_node(token):
75 self.graph.add_node(token, label=token)
76 self.state.init_node(token, phi_val=self.rng.gauss(0.0, 0.1))
77
78 # 注入激活
79 inject_amount = 100.0 / max(len(tokens), 1) * 0.5 # 半杯水
80 self.state.activate(token, inject_amount)
81
82 # ── 建立 bigram 边（非对称） ──
83 for i in range(len(tokens) - 1):
84 src, dst = tokens[i], tokens[i + 1]
85 existing_w = self.graph.get_edge_weight(src, dst)
86 # 正向强化，反向弱化——产生非对称
87 asym = self.rng.gauss(0.0, 0.1)
88 self.graph.add_edge(
89 src, dst,
90 weight=existing_w + 1.0 / (1.0 + existing_w * 0.1) + abs(asym),
91 bwd_weight=existing_w + 1.0 / (1.0 + existing_w * 0.1) - abs(asym) * 0.5,
92 edge_type='bigram'
93 )
94
95 # ── 锚点注入 ──
96 if anchors:
97 for anchor in anchors:
98 if not self.graph.has_node(anchor):
99 self.graph.add_node(anchor, label=anchor)
100 self.state.init_node(anchor, phi_val=1.0)
101 # 锚点高置信度
102 self.state.update_confidence(anchor, 2, amount=10.0) # 锚点/独立引用
103 # 锚点高势场
104 self.state.phi[anchor] = self.state.phi.get(anchor, 0.0) + 1.0
105
106 # ── 标记有输出需要回灌 ──
107 self._feedback_pending = True
108
109 # ──────────────────────────────────────
110 # §5.2 step — 推进动力学演化
111 # ──────────────────────────────────────
112
113 def step(self, n: int = 1):
114 """
115 推进 n 步内部动力学演化。
116 必须真实改变内部状态。
117 """
118 for _ in range(n):
119 self.dynamics.step()
120
121 # ──────────────────────────────────────
122 # §5.3 emit — 生成输出
123 # ──────────────────────────────────────
124
125 def emit(self) -> dict:
126 """
127 生成当前输出。允许完整输出或降级输出。
128
129 输出基于当前激活区域的行动释放值排序。
130 半杯水也要能流，不等满了再倒。
131 """
132 self.state.update_output_mode()
133
134 # 计算每个活跃节点的行动释放
135 releases = {}
136 for node_id in self.state.active_region:
137 u = self.dynamics.action_release(node_id)
138 if u > 1e-10:
139 releases[node_id] = u
140
141 # 按释放值排序
142 sorted_nodes = sorted(releases.items(), key=lambda x: -x[1])
143
144 # 根据输出模式决定输出多少
145 mode = self.state.output_mode
146 if mode == 'full':
147 top_nodes = sorted_nodes[:10]
148 elif mode == 'degraded':
149 top_nodes = sorted_nodes[:3]
150 else: # minimal
151 top_nodes = sorted_nodes[:1]
152
153 output = {
154 'mode': mode,
155 'activated': [
156 {
157 'node': nid,
158 'label': self.graph.get_node(nid).label if self.graph.get_node(nid) else nid,
159 'release': u,
160 'phi': self.state.phi.get(nid, 0.0),
161 'mu': self.state.mu.get(nid, 0.0),
162 'confidence': self.state.get_confidence(nid),
163 }
164 for nid, u in top_nodes
165 ],
166 'step': self.state.step_count,
167 'attention_free': self.state.attention.free,
168 'active_count': len(self.state.active_region),
169 }
170
171 self._last_output = output
172 self._output_buffer.append(output)
173 return output
174
175 # ──────────────────────────────────────
176 # §5.4 commit_feedback — 反馈写回
177 # ──────────────────────────────────────
178
179 def commit_feedback(self, feedback: dict):
180 """
181 把反馈写回系统——经验沉积、技能带修正、能力核慢更新。
182
183 feedback: dict
184 - 'correct': list[str] — 正确的节点，增强置信度
185 - 'wrong': list[str] — 错误的节点，降低置信度
186 - 'reward': float — 全局奖励信号
187 """
188 effect = {'reinforced': [], 'weakened': [], 'reward': 0.0}
189
190 # 正确的节点：增强置信度(cat2=独立引用) + 势场
191 for node_id in feedback.get('correct', []):
192 if self.graph.has_node(node_id):
193 self.state.update_confidence(node_id, 2, amount=2.0)
194 self.state.phi[node_id] = (
195 self.state.phi.get(node_id, 0.0) + 0.5
196 )
197 effect['reinforced'].append(node_id)
198
199 # 错误的节点：衰减势场 + 削弱置信度
200 for node_id in feedback.get('wrong', []):
201 if self.graph.has_node(node_id):
202 self.state.phi[node_id] = (
203 self.state.phi.get(node_id, 0.0) * 0.5
204 )
205 self.state.weaken_confidence(node_id, amount=3.0)
206 effect['weakened'].append(node_id)
207
208 # 全局奖励
209 reward = feedback.get('reward', 0.0)
210 effect['reward'] = reward
211 if reward > 0:
212 # 正奖励强化当前激活区域——按已有最强分量方向增强
213 for node_id in self.state.active_region:
214 if node_id in self.state.confidence:
215 alphas = self.state.confidence[node_id]
216 best_cat = alphas.index(max(alphas))
217 self.state.update_confidence(node_id, best_cat, amount=reward)
218 else:
219 self.state.update_confidence(node_id, 0, amount=reward)
220 elif reward < 0:
221 # 负奖励衰减当前激活区域（势场+置信度）
222 for node_id in self.state.active_region:
223 self.state.phi[node_id] *= max(0.1, 1.0 + reward)
224 self.state.weaken_confidence(node_id, amount=abs(reward))
225
226 self.state.last_feedback_effect = effect
227
228 # ──────────────────────────────────────
229 # §5.5 snapshot_state — 导出摘要
230 # ──────────────────────────────────────
231
232 def snapshot_state(self) -> dict:
233 """导出可比较的运行时摘要（SPEC §6）。"""
234 state_snap = self.state.snapshot()
235 state_snap['graph'] = {
236 'node_count': self.graph.node_count,
237 'edge_count': self.graph.edge_count,
238 }
239 return state_snap
240
241 # ──────────────────────────────────────
242 # §5.6 reset_session — 清理会话态
243 # ──────────────────────────────────────
244
245 def reset_session(self):
246 """
247 清理当前会话态，但不破坏长期结构层数据。
248 保留：φ（地形）、ability_cores、anchor_nodes
249 清理：μ（激活）、active_region、attention、output_buffer
250 """
251 # 清理激活
252 for node_id in list(self.state.active_region):
253 self.state.deactivate(node_id)
254 self.state.mu = {k: 0.0 for k in self.state.mu}
255 self.state.active_region.clear()
256 self.state.attention = type(self.state.attention)(total=100.0)
257
258 # 清理边流
259 self.state.J.clear()
260
261 # 清理输出缓冲
262 self._output_buffer.clear()
263 self._last_output = None
264 self._feedback_pending = False
265
266 # 保留：phi, confidence, anchor_nodes, ability_cores,
267 # experience_hits, experience_regions, skill_belt_candidates
268
269 self.state.output_mode = 'minimal'
270
271 # ──────────────────────────────────────
272 # 内部：输出即输入回灌
273 # ──────────────────────────────────────
274
275 def _feedback_loop(self, last_output: dict):
276 """
277 输出即输入——上一轮的激活结果直接成为下一轮输入的一部分。
278 不经过外部中转，闭合自指环路。
279 """
280 if not last_output or not last_output.get('activated'):
281 return
282
283 # 把上一轮输出的节点作为弱输入回灌
284 # 用 mu（激活量）而非 release（行动释放）——mu 是水量，release 是水压
285 for item in last_output['activated']:
286 node_id = item['node']
287 if self.graph.has_node(node_id):
288 mu_val = item.get('mu', 0.0)
289 # 回灌量 = 上轮激活的 5%（衰减版）
290 feedback_amount = mu_val * 0.05
291 if feedback_amount > 0.001:
292 self.state.activate(node_id, feedback_amount)
293 # 极微增强 φ（水流过的地方地形被改变）
294 self.state.phi[node_id] = (
295 self.state.phi.get(node_id, 0.0) + feedback_amount * 0.01
296 )
297
298
299 # ──────────────────────────────────────
300 # 便利方法
301 # ──────────────────────────────────────
302
303 def run(self, input_data, steps: int = 5, context=None, anchors=None) -> dict:
304 """便利方法：ingest + step + emit 一条龙"""
305 self.ingest(input_data, context=context, anchors=anchors)
306 self.step(n=steps)
307 return self.emit()