""" 功能区网络 - 6功能区 × 3柱 = 18功能柱类脑模型 """ import numpy as np from typing import List, Dict, Tuple, Optional from enum import Enum import os import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from functional_column import ( FunctionalColumn, create_sensory_column, create_memory_column, create_association_column, create_prefrontal_column, create_motor_column, create_thalamus_column ) from micro_columns.metacognition import MetacognitionModule class AreaType(Enum): """功能区类型""" SENSORY = 'sensory' # 感觉区 MEMORY = 'memory' # 记忆区 ASSOCIATION = 'association' # 联合区 PREFRONTAL = 'prefrontal' # 前额区 MOTOR = 'motor' # 运动区 THALAMUS = 'thalamus' # 丘脑区 class AggregationType(Enum): """聚合类型""" PARALLEL = 'parallel' # 并行 CASCADE = 'cascade' # 级联 RECURRENT = 'recurrent' # 循环 ATTENTION = 'attention' # 注意力 class FunctionalArea: """ 功能区 - 3个同类功能柱聚合 仿生结构： - 每个功能区 = 大脑皮层的一个区域 - 每个功能柱 = 皮层柱 (cortical column) - 3个功能柱 = 3个皮层柱组成功能区 """ def __init__( self, area_type: AreaType, num_columns: int = 3, connection_mode: str = 'default', neurons_per_micro: int = 100, neurons_per_column: int = 10, synaptic_tier: str = 'L1', ): self.area_type = area_type self.num_columns = num_columns self.neurons_per_micro = neurons_per_micro self.neurons_per_column = neurons_per_column self.synaptic_tier = synaptic_tier self.topology = self._infer_topology(area_type) # 创建3个功能柱 self.columns: List[FunctionalColumn] = [] self._create_columns(area_type, num_columns) # 循环聚合参数 - 增强复杂推理能力 self.recurrent_alpha = 0.6 # 反馈权重，略降低以保留更多原始信息 self.max_iterations = 10 # 扩展到10轮，支持深度推理 # 推理深度控制 - 根据任务复杂度自适应 self.reasoning_depth = 5 # 默认推理深度 self.convergence_threshold = 0.005 # 更严格的收敛检测 # 循环推理状态（前额区/联合区） self.recurrent_state: Dict = { 'iteration': 0, 'convergence': False, 'history': [], 'feedback_signal': None } # 丘脑调度状态 self.thalamus_state: Dict = { 'attention_weights': None, 'routing_targets': [], 'last_activation': None } # 统计 self._forward_count = 0 # 冻结模式: frozen=True时forward不修改任何self状态 self._frozen = False # 元认知模块 - 自我监控与策略调整 self._metacognition = MetacognitionModule( num_neurons=64, quality_threshold=0.5, confidence_threshold=0.3, max_history=20 ) # 元认知状态 self._meta_state: Dict = { 'last_quality': 0.0, 'last_confidence': 0.0, 'inference_count': 0, 'strategy': 'exploration' } def _infer_topology(self, area_type: AreaType) -> AggregationType: """根据功能区类型推断连接拓扑""" topologies = { AreaType.SENSORY: AggregationType.PARALLEL, # 并行提取多模态 AreaType.MEMORY: AggregationType.ATTENTION, # 注意力检索 AreaType.ASSOCIATION: AggregationType.RECURRENT, # 循环整合 AreaType.PREFRONTAL: AggregationType.RECURRENT, # 循环推理 AreaType.MOTOR: AggregationType.PARALLEL, # 并行输出(CASCADE级联9层衰减严重) AreaType.THALAMUS: AggregationType.PARALLEL, # 并行调度 } return topologies.get(area_type, AggregationType.PARALLEL) def _create_columns(self, area_type: AreaType, num_columns: int): """创建功能柱 - 按配置创建，支持自定义神经元数和突触层级""" creators = { AreaType.SENSORY: create_sensory_column, AreaType.MEMORY: create_memory_column, AreaType.ASSOCIATION: create_association_column, AreaType.PREFRONTAL: create_prefrontal_column, AreaType.MOTOR: create_motor_column, AreaType.THALAMUS: create_thalamus_column, } creator = creators[area_type] for i in range(num_columns): col = creator( col_index=i, neurons_per_micro=self.neurons_per_micro, synaptic_tier=self.synaptic_tier, num_mcs=self.neurons_per_column, ) self.columns.append(col) def forward(self, inputs: np.ndarray) -> Tuple[np.ndarray, Dict]: """前向传播""" if not self._frozen: self._forward_count += 1 if self.topology == AggregationType.PARALLEL: return self._forward_parallel(inputs) elif self.topology == AggregationType.CASCADE: return self._forward_cascade(inputs) elif self.topology == AggregationType.RECURRENT: return self._forward_recurrent(inputs) elif self.topology == AggregationType.ATTENTION: return self._forward_attention(inputs) else: return self._forward_parallel(inputs) def _forward_parallel(self, inputs: np.ndarray) -> Tuple[np.ndarray, Dict]: """并行聚合：所有柱独立处理，输出拼接""" outputs = [] for col in self.columns: out = col.forward(inputs) outputs.append(out) result = np.concatenate(outputs) metadata = { 'topology': 'parallel', 'n_columns': len(self.columns), 'intermediate_dims': [o.shape[0] for o in outputs] } return result, metadata def _forward_cascade(self, inputs: np.ndarray) -> Tuple[np.ndarray, Dict]: """级联聚合：前一柱输出作下一柱输入""" current = inputs.copy() intermediate_outputs = [] for col in self.columns: # 维度适配 expected = col.micro_columns[0].num_neurons if hasattr(col.micro_columns[0], 'num_neurons') else 100 if current.shape[0] != expected: current = self._project_dim(current, expected) current = col.forward(current) intermediate_outputs.append(current.copy()) result = current metadata = { 'topology': 'cascade', 'n_columns': len(self.columns), 'intermediate_dims': [o.shape[0] for o in intermediate_outputs] } return result, metadata def _forward_recurrent(self, inputs: np.ndarray) -> Tuple[np.ndarray, Dict]: """循环聚合：输出反馈迭代 + 元认知监控""" current = inputs.copy() history = [current.copy()] # 元认知初始化 iteration_outputs = [] max_iter = self.max_iterations for iteration in range(max_iter): # 收集所有列输出 column_outputs = [] for col in self.columns: out = col.forward(current) column_outputs.append(out) # 拼接输出 aggregated = np.concatenate(column_outputs) # 投影回输入维度 if aggregated.shape[0] != current.shape[0]: aggregated = self._project_dim(aggregated, current.shape[0]) # 遗忘更新 current = self.recurrent_alpha * aggregated + \ (1 - self.recurrent_alpha) * current iteration_outputs.append(current.copy()) history.append(current.copy()) # 元认知监控（每2次迭代评估一次） if iteration > 0 and iteration % 2 == 0 and len(iteration_outputs) >= 2: # 评估质量 quality = self._metacognition.assess_quality(iteration_outputs) # 计算置信度 confidence = self._metacognition.compute_confidence( current, iteration_outputs ) # 错误检测 errors = self._metacognition.detect_errors(iteration_outputs) # 更新元认知状态（冻结模式跳过） if not self._frozen: self._meta_state['inference_count'] += 1 self._meta_state['last_quality'] = quality self._meta_state['last_confidence'] = confidence # 策略调整（仅在前额区/联合区） if self.area_type in [AreaType.PREFRONTAL, AreaType.ASSOCIATION]: new_strategy = self._metacognition.adjust_strategy( quality, confidence, errors ) # 根据策略调整迭代次数（冻结模式跳过策略修改） if new_strategy.get('iterations') and new_strategy['iterations'] != max_iter: max_iter = min(new_strategy['iterations'], self.max_iterations) if not self._frozen: self._meta_state['strategy'] = new_strategy.get('mode', 'exploration') # 如果需要提前终止且质量足够好 if new_strategy.get('early_stop') and quality > self._metacognition.quality_threshold: break metadata = { 'topology': 'recurrent', 'n_columns': len(self.columns), 'iterations': len(iteration_outputs), 'history_len': len(history), 'metacognition': { 'quality': self._meta_state['last_quality'], 'confidence': self._meta_state['last_confidence'], 'strategy': self._meta_state['strategy'], 'inferences': self._meta_state['inference_count'] } } return current, metadata def _forward_attention(self, inputs: np.ndarray) -> Tuple[np.ndarray, Dict]: """注意力聚合：学习权重""" # 先获取各柱输出 column_outputs = [] for col in self.columns: out = col.forward(inputs) column_outputs.append(out) # 简单注意力：基于输出的范数计算权重 norms = np.array([np.linalg.norm(o) for o in column_outputs]) weights = norms / (norms.sum() + 1e-8) # 加权聚合 result = np.zeros_like(column_outputs[0]) for w, out in zip(weights, column_outputs): result += w * out metadata = { 'topology': 'attention', 'n_columns': len(self.columns), 'weights': weights.tolist() } return result, metadata def _project_dim(self, x: np.ndarray, target_dim: int) -> np.ndarray: """调整维度 — 压缩时用均值池化保留信号，扩展时用复制""" current_dim = x.shape[0] if current_dim == target_dim: return x elif current_dim < target_dim: # 扩展：重复+微小噪声防止梯度消失 repeats = target_dim // current_dim remainder = target_dim % current_dim result = np.tile(x, repeats) if remainder > 0: result = np.concatenate([result, x[:remainder]]) return result.astype(x.dtype) else: # 压缩：均值池化而非截断 chunk_size = current_dim / target_dim result = np.zeros(target_dim, dtype=x.dtype) for i in range(target_dim): start = int(i * chunk_size) end = int((i + 1) * chunk_size) result[i] = np.mean(x[start:end]) return result def get_config(self) -> Dict: """获取配置""" return { 'area_type': self.area_type.value, 'num_columns': self.num_columns, 'topology': self.topology.value, 'n_micro_columns': sum(len(c.micro_columns) for c in self.columns), 'recurrent_state': self.recurrent_state, 'thalamus_state': self.thalamus_state } # ============ 功能区级别循环推理控制 ============ def run_recurrent_reasoning( self, inputs: np.ndarray, memory_area: Optional['FunctionalArea'] = None, max_loops: int = 3 ) -> Tuple[np.ndarray, Dict]: """ 功能区级别循环推理（前额区/联合区使用） 与记忆区循环交互，直到收敛或达到最大循环次数 """ if self.area_type not in [AreaType.PREFRONTAL, AreaType.ASSOCIATION]: return self.forward(inputs) # 重置状态 self.recurrent_state = { 'iteration': 0, 'convergence': False, 'history': [], 'feedback_signal': None } current = inputs.copy() for loop in range(max_loops): self.recurrent_state['iteration'] = loop + 1 # 前向处理 out, meta = self.forward(current) self.recurrent_state['history'].append(out.copy()) # 与记忆区交互（如提供） if memory_area is not None: mem_out, _ = memory_area.forward(out) min_dim = min(out.shape[0], mem_out.shape[0]) # 反馈到输入 current = 0.6 * out[:min_dim] + 0.4 * mem_out[:min_dim] else: current = out # 收敛检测：与上一次输出相似 if len(self.recurrent_state['history']) >= 2: prev = self.recurrent_state['history'][-2] curr = self.recurrent_state['history'][-1] min_d = min(prev.shape[0], curr.shape[0]) diff = np.mean(np.abs(curr[:min_d] - prev[:min_d])) # 使用自适应收敛阈值 threshold = getattr(self, 'convergence_threshold', 0.01) if diff < threshold: self.recurrent_state['convergence'] = True break self.recurrent_state['feedback_signal'] = current return current, {'loops': loop + 1, 'converged': self.recurrent_state['convergence']} # ============ 丘脑调度功能 ============ def thalamus_dispatch( self, inputs: np.ndarray, target_areas: List['FunctionalArea'] ) -> Dict[AreaType, np.ndarray]: """ 丘脑调度功能：并行分发到多个目标区域 仿生：丘脑作为中继站，根据注意力权重路由信号 """ if self.area_type != AreaType.THALAMUS: raise ValueError("只有丘脑区可以执行调度") # 计算注意力权重 input_norm = np.linalg.norm(inputs) + 1e-8 # 对每个目标区域并行处理 outputs = {} for target in target_areas: # 注意力路由 target_out, _ = target.forward(inputs) outputs[target.area_type] = target_out # 更新丘脑状态 self.thalamus_state['routing__targets'] = [a.area_type.value for a in target_areas] self.thalamus_state['last_activation'] = input_norm return outputs def reset_recurrent_state(self): """重置循环状态""" self.recurrent_state = { 'iteration': 0, 'convergence': False, 'history': [], 'feedback_signal': None } class AreaNetwork: """ 6功能区网络 - 非对称类脑模型 支持配置化构建: - 默认: 8M模型(3柱×100N) - 200M: 非对称架构(前额叶300微柱×512N) 连接拓扑 (仿生神经环路): ``` 输入 → 感觉区 → 记忆区 → 联合区 → 前额区 ↔ 运动区 ↑ │ └──── 丘脑调度 ←─────┘ ``` """ # 默认8M配置 DEFAULT_CONFIG = { 'sensory': {'num_columns': 3, 'neurons_per_column': 10, 'neurons_per_micro': 100, 'synaptic_tier': 'L1'}, 'memory': {'num_columns': 3, 'neurons_per_column': 8, 'neurons_per_micro': 100, 'synaptic_tier': 'L1'}, 'association': {'num_columns': 3, 'neurons_per_column': 6, 'neurons_per_micro': 100, 'synaptic_tier': 'L1'}, 'prefrontal': {'num_columns': 3, 'neurons_per_column': 20, 'neurons_per_micro': 100, 'synaptic_tier': 'L3'}, 'motor': {'num_columns': 3, 'neurons_per_column': 3, 'neurons_per_micro': 100, 'synaptic_tier': 'L1'}, 'thalamus': {'num_columns': 3, 'neurons_per_column': 2, 'neurons_per_micro': 100, 'synaptic_tier': 'L1'}, } def __init__(self, config: dict = None): self.areas: Dict[AreaType, FunctionalArea] = {} self._config = config or self.DEFAULT_CONFIG self._build_network() self.input_mode = 'semantic' # 'semantic'(文字) 或 'perceptual'(语音/图像) self.semantic_vocab = None # jieba词表: {词: 索引} self.semantic_matrix = None # 词关联矩阵 (75, 75) 用于语义增强 self._assign_memory_roles() # 记忆微柱角色分配 def freeze(self): """冻结所有功能区: forward不修改任何内部状态""" for area in self.areas.values(): area._frozen = True # 递归冻结到微柱层 + 突触层 for col in area.columns: for mc in col.micro_columns: mc._frozen = True # 冻结突触底层 if hasattr(mc, '_synaptic'): mc._synaptic._frozen = True def unfreeze(self): """解冻: 恢复forward的状态更新""" for area in self.areas.values(): area._frozen = False for col in area.columns: for mc in col.micro_columns: mc._frozen = False if hasattr(mc, '_synaptic'): mc._synaptic._frozen = False def _build_network(self): """构建6功能区网络 — 支持非对称配置""" for area_type in AreaType: cfg = self._config.get(area_type.value, {}) num_columns = cfg.get('num_columns', 3) neurons_per_micro = cfg.get('neurons_per_micro', 100) neurons_per_column = cfg.get('neurons_per_column', 10) synaptic_tier = cfg.get('synaptic_tier', 'L1') area = FunctionalArea( area_type=area_type, num_columns=num_columns, connection_mode='default', neurons_per_micro=neurons_per_micro, neurons_per_column=neurons_per_column, synaptic_tier=synaptic_tier, ) self.areas[area_type] = area # 定义连接关系 self.connections: Dict[AreaType, List[AreaType]] = { AreaType.SENSORY: [AreaType.MEMORY], AreaType.MEMORY: [AreaType.ASSOCIATION], AreaType.ASSOCIATION: [AreaType.PREFRONTAL, AreaType.THALAMUS], AreaType.PREFRONTAL: [AreaType.MOTOR, AreaType.MEMORY], # 反馈到记忆 AreaType.MOTOR: [AreaType.THALAMUS], AreaType.THALAMUS: [AreaType.SENSORY, AreaType.PREFRONTAL], # 调度 } def forward(self, inputs: np.ndarray) -> Tuple[np.ndarray, Dict]: """ 前向传播: 感觉→记忆→联合→前额↔运动→丘脑 双模式感知: - semantic(文字): 语义增强，保留全部编码，补全关联词 - perceptual(语音/图像): WTA去噪，过滤噪音信号 """ current = inputs.copy() area_outputs = {} # 1. 感觉区 — 双模式处理 out, meta = self.areas[AreaType.SENSORY].forward(current) if self.input_mode == 'semantic': # 文字模式: 语义增强 — 保留原始编码 + 关联词扩散 out = self._semantic_enhance(out) else: # 感知模式: WTA稀疏化去噪(保留top30%最强激活) out = self._wta_sparsify(out, keep_ratio=0.3) area_outputs[AreaType.SENSORY] = out current = out sensory_signal = out.copy() # 残差：保留感觉区信号 # 2. 记忆区 out, meta = self.areas[AreaType.MEMORY].forward(current) area_outputs[AreaType.MEMORY] = out current = out # 3. 联合区 out, meta = self.areas[AreaType.ASSOCIATION].forward(current) area_outputs[AreaType.ASSOCIATION] = out current = out # 4. 前额区 (循环推理) out, meta = self.areas[AreaType.PREFRONTAL].forward(current) area_outputs[AreaType.PREFRONTAL] = out current = out # 5. 运动区 (级联输出) + 跨区残差连接 out, meta = self.areas[AreaType.MOTOR].forward(current) # 残差跳跃: MOTOR + 0.15*SENSORY(投影到同维) sensory_proj = self._area_project_dim(sensory_signal, out.shape[0]) out = 0.85 * out + 0.15 * sensory_proj area_outputs[AreaType.MOTOR] = out current = out # 5.5 反馈存储: 由learn()显式调用，不在forward中自动存储 # 原设计: self.memory_store(sensory_signal, motor_output, role='dialogue') # 改为: forward只做推理，避免隐式修改记忆导致非确定性 # 6. 丘脑区 (调度反馈，不参与最终输出) out, meta = self.areas[AreaType.THALAMUS].forward(current) area_outputs[AreaType.THALAMUS] = out # THALAMUS norm太大(17+)会淹没MOTOR信号，仅用于内部反馈 # 最终输出直接用MOTOR metadata = { 'area_outputs': {k: v for k, v in area_outputs.items()}, 'connections': {k.value: [v.value for v in lst] for k, lst in self.connections.items()} } return current, metadata def _area_project_dim(self, x: np.ndarray, target_dim: int) -> np.ndarray: """维度投影(均值池化压缩/复制扩展)""" current_dim = x.shape[0] if current_dim == target_dim: return x elif current_dim < target_dim: repeats = target_dim // current_dim remainder = target_dim % current_dim result = np.tile(x, repeats) if remainder > 0: result = np.concatenate([result, x[:remainder]]) return result.astype(x.dtype) else: chunk_size = current_dim / target_dim result = np.zeros(target_dim, dtype=x.dtype) for i in range(target_dim): start = int(i * chunk_size) end = int((i + 1) * chunk_size) result[i] = np.mean(x[start:end]) return result def _wta_sparsify(self, x: np.ndarray, keep_ratio: float = 0.3) -> np.ndarray: """赢者通吃稀疏化 — 只保留top-k最强激活，其余置零（感知模式用）""" k = max(1, int(len(x) * keep_ratio)) threshold = np.sort(np.abs(x))[-k] result = x.copy() result[np.abs(result) < threshold] = 0.0 return result def _semantic_enhance(self, x: np.ndarray) -> np.ndarray: """语义增强 — 保留原始编码 + 关联词扩散（文字模式用） 仿生: 类似大脑阅读时，看到"学习"会自动激活"记忆""知识"等关联概念 原理: 用词关联矩阵做一次扩散，微弱激活相关词 """ if self.semantic_matrix is not None and x.shape[0] == self.semantic_matrix.shape[0]: # 关联扩散: x' = x + 0.2 * (x @ M) # 0.2是扩散系数，防止单词激活太多关联词 enhanced = x + 0.2 * (x @ self.semantic_matrix) return enhanced # 没有关联矩阵时，直接透传（不过滤不丢弃） return x def set_semantic_vocab(self, vocab: dict, cooccurrence: np.ndarray = None): """设置语义编码表 Args: vocab: {词: 索引} 映射，如 jieba分词后的词典 cooccurrence: 词共现矩阵 (75,75)，可选，用于关联扩散 """ self.semantic_vocab = vocab self.semantic_matrix = cooccurrence self.input_mode = 'semantic' def _assign_memory_roles(self): """记忆微柱角色分配 12个Memory微柱分3组: - 词汇记忆(0-3): 字词→语义向量编码，类似词典 - 对话记忆(4-7): Q向量→A向量，类似经验 - 关联记忆(8-11): 上下文片段，类似情景记忆 """ memory_area = self.areas[AreaType.MEMORY] mc_idx = 0 for col in memory_area.columns: for mc in col.micro_columns: if mc.name == 'Memory': if mc_idx < 4: mc._role = 'lexical' # 词汇记忆 elif mc_idx < 8: mc._role = 'dialogue' # 对话记忆 else: mc._role = 'episodic' # 关联记忆 mc_idx += 1 def memory_store(self, key_vec: np.ndarray, value_vec: np.ndarray, role: str = 'auto'): """定向存储到记忆区 Args: key_vec: 键向量(输入编码) value_vec: 值向量(关联信息/回复编码) role: 'lexical'(词汇), 'dialogue'(对话), 'episodic'(关联), 'auto'(自动) """ memory_area = self.areas[AreaType.MEMORY] stored = False for col in memory_area.columns: for mc in col.micro_columns: if mc.name != 'Memory': continue mc_role = getattr(mc, '_role', 'lexical') # auto模式: 根据key特征自动分配 if role == 'auto': sparsity = float(np.count_nonzero(key_vec < 0.01)) / len(key_vec) if sparsity > 0.9: role = 'lexical' elif sparsity > 0.7: role = 'dialogue' else: role = 'episodic' if mc_role == role: # 维度适配: 投影到memory微柱的input_dim k = self._area_project_dim(key_vec, mc.input_dim) v = self._area_project_dim(value_vec, mc.input_dim) mc.store(k, v) stored = True break if stored: break def memory_retrieve(self, query_vec: np.ndarray, role: str = None, top_k: int = 3): """从记忆区检索，可按角色过滤 Args: query_vec: 查询向量 role: 可选过滤角色 top_k: 返回前k个结果 Returns: list of (value_vec, confidence, role) """ memory_area = self.areas[AreaType.MEMORY] results = [] for col in memory_area.columns: for mc in col.micro_columns: if mc.name != 'Memory': continue mc_role = getattr(mc, '_role', 'lexical') if role and mc_role != role: continue if mc.memory_count == 0: continue result, conf = mc.forward(query_vec, mode='read') results.append((result, conf, mc_role)) results.sort(key=lambda x: x[1], reverse=True) return results[:top_k] def learn(self, area_filter=None, **kwargs): """全网络赫布学习 — 不同区不同策略 改进：按区域分配不同学习策略 - SENSORY/MOTOR: 高学习概率（输入输出端需要快速适应） - MEMORY: 低学习概率（记忆需要稳定） - ASSOCIATION/PREFRONTAL: 中等学习概率 - THALAMUS: 不学习（调度不需要学习） Args: area_filter: 只训练这些区，如 [AreaType.SENSORY, AreaType.MOTOR] """ # 区域学习策略 area_learn_probs = { AreaType.SENSORY: 0.5, # 输入端，高学习率 AreaType.MOTOR: 0.5, # 输出端，高学习率 AreaType.MEMORY: 0.1, # 记忆需要稳定，低学习率 AreaType.ASSOCIATION: 0.3, # 联合区，中等 AreaType.PREFRONTAL: 0.3, # 前额区，中等 AreaType.THALAMUS: 0.0, # 丘脑不学习 } for area_type, area in self.areas.items(): if area_filter is not None and area_type not in area_filter: continue learn_prob = area_learn_probs.get(area_type, 0.3) if learn_prob <= 0: continue # 不学习 for col in area.columns: if hasattr(col, 'learn'): col.learn(learn_prob=learn_prob, **kwargs) def get_summary(self) -> Dict: """获取网络摘要""" total_columns = sum(len(area.columns) for area in self.areas.values()) total_micro = sum( sum(len(c.micro_columns) for c in area.columns) for area in self.areas.values() ) return { 'n_areas': len(self.areas), 'n_columns': total_columns, 'n_micro_columns': total_micro, 'area_types': [a.value for a in self.areas.keys()], 'connections': {k.value: [v.value for v in lst] for k, lst in self.connections.items()} } def create_six_area_network() -> AreaNetwork: """创建6功能区网络""" return AreaNetwork()