""" 功能柱 - 多种微柱混合组成的功能单元 每种功能柱按仿生比例混合不同类型微柱 """ import numpy as np from typing import List, Dict, Optional import sys import os # 确保能正确导入同目录下的micro_columns包 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from micro_columns import get_micro_column, MICRO_COLUMN_TYPES # ============ 功能柱微柱组成配方（参考docs/functional_columns/） ============ COLUMN_RECIPES = { # ============ 六大功能区定义 ============ # 数据流: SENSORY → ASSOCIATION → (PREFRONTAL/ MEMORY/MOTOR) → MOTOR → 输出 # 监控流: THALAMUS 监控所有区域，检测异常并中断 # 控制流: PREFRONTAL 作为总控，协调所有区域 'sensory': { 'name': '感知区', 'role': '输入信号预处理 - 接收原始输入，特征提取、归一化、降噪', 'composition': { 'sensory': 0.60, # 60% SensoryMicroColumn (特征提取) 'detector': 0.20, # 20% DetectorMicroColumn (模式检测) 'modulator': 0.20, # 20% ModulatorMicroColumn (增益调节) }, 'default_count': 10, 'count_range': (5, 20), 'connection_mode': 'hybrid', 'synaptic_tier': 'L1', 'inputs': ['raw_input'], # 外部输入 'outputs': ['association'], # 输出到联通区 }, 'association': { 'name': '联通区', 'role': '通讯中转 - 四大区的信息汇聚与分发，区域间路由', 'composition': { 'integrator': 0.50, # 50% IntegratorMicroColumn (信息融合) 'selector': 0.30, # 30% SelectorMicroColumn (路由选择) 'detector': 0.20, # 20% DetectorMicroColumn (一致性校验) }, 'default_count': 6, 'count_range': (3, 10), 'connection_mode': 'hybrid', 'synaptic_tier': 'L1', 'inputs': ['sensory', 'memory', 'prefrontal', 'motor'], # 汇聚四大区 'outputs': ['sensory', 'memory', 'prefrontal', 'motor'], # 分发到四大区 }, 'memory': { 'name': '记忆区', 'role': '记忆数据库 - 短期/长期记忆存储，检索，遗忘机制', 'composition': { 'memory': 0.50, # 50% MemoryMicroColumn (记忆存储) 'integrator': 0.30, # 30% IntegratorMicroColumn (关联整合) 'selector': 0.20, # 20% SelectorMicroColumn (检索选择) }, 'default_count': 8, 'count_range': (3, 50), # 支持大模型扩展 'connection_mode': 'hybrid', 'synaptic_tier': 'L1', 'inputs': ['association'], 'outputs': ['association', 'prefrontal'], }, 'prefrontal': { 'name': '额叶区', 'role': '分析推理总控 - 协调所有区域，推理决策，微柱/功能柱裂变', 'composition': { 'manager': 0.10, # 10% ManagerMicroColumn (调度+监控+拆分) 'memory': 0.20, # 20% MemoryMicroColumn (工作记忆) 'integrator': 0.25, # 25% IntegratorMicroColumn (推理融合) - 增强 'detector': 0.15, # 15% DetectorMicroColumn (一致性检测) 'motor': 0.15, # 15% MotorMicroColumn (决策输出) 'modulator': 0.10, # 10% ModulatorMicroColumn (增益调节) 'selector': 0.05, # 5% SelectorMicroColumn (模式选择) - 新增 }, 'default_count': 20, 'count_range': (10, 200), # 支持大模型扩展 'connection_mode': 'recurrent', # 改为循环模式，支持深度推理 'synaptic_tier': 'L3', # L3全突触，最强学习潜力(推理总控) 'inputs': ['association', 'memory', 'thalamus', 'sensory'], # 增加感官输入 'outputs': ['association', 'motor', 'thalamus'], }, 'motor': { 'name': '运动区', 'role': '结果输出与工具调用 - 决策执行，外部工具调用，响应输出', 'composition': { 'motor': 0.50, # 50% MotorMicroColumn (动作执行) 'selector': 0.30, # 30% SelectorMicroColumn (动作选择) 'modulator': 0.20, # 20% ModulatorMicroColumn (力度调节) }, 'default_count': 3, 'count_range': (1, 30), # 支持大模型扩展 'connection_mode': 'cascade', 'synaptic_tier': 'L1', 'inputs': ['association', 'prefrontal'], 'outputs': ['external', 'association'], # 外部输出 + 反馈 }, 'thalamus': { 'name': '丘脑区', 'role': '监控与异常中断 - 监控所有区域状态，检测异常，触发中断', 'composition': { 'selector': 0.50, # 50% SelectorMicroColumn (注意力监控) 'integrator': 0.30, # 30% IntegratorMicroColumn (状态评估) 'modulator': 0.20, # 20% ModulatorMicroColumn (门控中断) }, 'default_count': 2, 'count_range': (1, 3), 'connection_mode': 'parallel', 'synaptic_tier': 'L1', 'inputs': ['sensory', 'association', 'memory', 'prefrontal', 'motor'], # 监听所有区 'outputs': ['prefrontal', 'motor'], # 异常反馈到额叶/运动区 }, } class FunctionalColumn: """功能柱：多种微柱按仿生比例混合组成""" def __init__( self, column_type: str = 'sensory', num_micro_columns: int = None, col_index: int = 0, neurons_per_micro: int = 100, synaptic_tier: str = None, ): self.column_type = column_type recipe = COLUMN_RECIPES.get(column_type, COLUMN_RECIPES['sensory']) self.recipe = recipe self.neurons_per_micro = neurons_per_micro self._override_tier = synaptic_tier # 微柱数量：默认使用配方中的数量 if num_micro_columns is None: num_micro_columns = recipe['default_count'] lo, hi = recipe['count_range'] self.num_micro_columns = max(lo, min(hi, num_micro_columns)) self.connection_mode = recipe['connection_mode'] # 按配方比例创建混合微柱 self.micro_columns: List = [] self.micro_column_types: List[str] = [] self._create_mixed_micro_columns(recipe['composition'], self.num_micro_columns, col_index=col_index) # ============ Manager微柱集成 ============ self.manager = None self._setup_manager_scheduling() def _setup_manager_scheduling(self): """设置Manager微柱调度系统""" # 找出Manager微柱 for mc, mc_type in zip(self.micro_columns, self.micro_column_types): if mc_type == 'manager': self.manager = mc break if self.manager is None: return # 给Manager注册所有其他微柱（按类型标签） type_to_tags = { 'memory': ['memory', 'working', 'semantic'], 'integrator': ['reasoning', 'fusion', 'semantic'], 'detector': ['detection', 'pattern', 'semantic'], 'motor': ['action', 'output', 'spatial'], 'modulator': ['modulation', 'gain', 'temporal'], 'sensory': ['perception', 'input', 'spatial'], } for i, mc in enumerate(self.micro_columns): mc_type = self.micro_column_types[i] if mc_type == 'manager': continue tags = type_to_tags.get(mc_type, [mc_type]) unit_id = f"mc_{i}_{mc_type}" capacity = getattr(mc, 'num_neurons', 64) self.manager.register(unit_id, tags, capacity=capacity) print(f" [Manager] 已注册 {len(self.manager._registry)} 个微柱") def _create_mixed_micro_columns(self, composition: Dict[str, float], total: int, col_index: int = 0): """按比例分配微柱类型""" # 计算每种类型的数量 type_counts = {} remaining = total types_sorted = sorted(composition.keys(), key=lambda k: -composition[k]) for i, mc_type in enumerate(types_sorted): if i == len(types_sorted) - 1: # 最后一个类型取剩余 count = remaining else: count = max(1, round(total * composition[mc_type])) remaining -= count type_counts[mc_type] = count # 获取突触档位（优先使用外部覆盖） synaptic_tier = self._override_tier or self.recipe.get('synaptic_tier', 'L1') # 按顺序创建微柱（主类型在前）— 支持感受野 mc_idx = 0 for mc_type in types_sorted: for _ in range(type_counts[mc_type]): # 感觉柱的sensory类型mc可选感受野(默认关闭) rf_kwargs = {} # 实验结论: RF=10对composite无显著改善(p=0.91) # 保留参数供未来更大数据集实验 if False and self.column_type == 'sensory' and mc_type == 'sensory': input_dim = 75 rf_size = 10 rf_offset = (col_index * 31 + mc_idx * 17) % input_dim rf_kwargs = { 'receptive_field_size': rf_size, 'receptive_field_offset': rf_offset, } mc = get_micro_column(mc_type, num_neurons=self.neurons_per_micro, synaptic_tier=synaptic_tier, **rf_kwargs) self.micro_columns.append(mc) self.micro_column_types.append(mc_type) mc_idx += 1 def add_micro_column(self, mc_type: str, count: int = 1, approver: str = None, reason: str = None): """按需增加内部微柱（需审批） Args: mc_type: 微柱类型 count: 增加数量 approver: 审批人标识（必须提供） reason: 扩容原因（必须提供） Raises: ValueError: 未审批/类型无效/超出上限 """ # 1. 审批检查 if not approver or not reason: raise ValueError( f"扩容需审批！请提供 approver 和 reason。" f"当前{self.recipe['name']}有{self.num_micro_columns}微柱，" f"上限{self.recipe['count_range'][1]}" ) # 2. 类型检查 if mc_type not in MICRO_COLUMN_TYPES: raise ValueError(f"未知微柱类型: {mc_type}, 可选: {list(MICRO_COLUMN_TYPES.keys())}") # 3. 上限检查 max_count = self.recipe['count_range'][1] new_total = self.num_micro_columns + count if new_total > max_count: raise ValueError( f"{self.recipe['name']}微柱上限{max_count}，" f"当前{self.num_micro_columns}+{count}={new_total}，超出上限！" ) # 4. 执行扩容 for _ in range(count): mc = get_micro_column(mc_type, num_neurons=100) self.micro_columns.append(mc) self.micro_column_types.append(mc_type) self.num_micro_columns = len(self.micro_columns) # 5. 记录审批日志 if not hasattr(self, '_expansion_log'): self._expansion_log = [] self._expansion_log.append({ 'approver': approver, 'reason': reason, 'type': mc_type, 'count': count, 'total_after': self.num_micro_columns, }) def forward(self, inputs: np.ndarray) -> np.ndarray: """前向传播 - 支持Manager调度""" if self.manager is not None: # Manager调度：决定激活哪些微柱 active_unit_ids = self.manager.dispatch(inputs) if active_unit_ids: # 构建活跃微柱索引映射 unit_id_to_idx = {f"mc_{i}_{t}": i for i, t in enumerate(self.micro_column_types)} active_indices = [unit_id_to_idx[uid] for uid in active_unit_ids if uid in unit_id_to_idx] else: active_indices = list(range(len(self.micro_columns))) else: active_indices = list(range(len(self.micro_columns))) if self.connection_mode == 'parallel': return self._forward_parallel(inputs, active_indices) elif self.connection_mode == 'sequential': return self._forward_sequential(inputs, active_indices) elif self.connection_mode == 'hybrid': return self._forward_hybrid(inputs, active_indices) elif self.connection_mode == 'cascade': return self._forward_cascade(inputs, active_indices) else: return self._forward_parallel(inputs, active_indices) def _forward_parallel(self, inputs: np.ndarray, active_indices: list = None) -> np.ndarray: """并行模式：所有微柱独立处理，输出拼接""" if active_indices is None: active_indices = list(range(len(self.micro_columns))) outputs = [] for idx in active_indices: mc = self.micro_columns[idx] mc_type = self.micro_column_types[idx] out = self._call_micro_column(mc, mc_type, inputs) outputs.append(out) return np.concatenate(outputs) if outputs else inputs def _forward_sequential(self, inputs: np.ndarray, active_indices: list = None) -> np.ndarray: """顺序级联：上一个微柱输出作为下一个输入""" if active_indices is None: active_indices = list(range(len(self.micro_columns))) current = inputs for idx in active_indices: mc = self.micro_columns[idx] if hasattr(mc, 'num_neurons'): if current.shape[0] != mc.num_neurons: current = self._project_dim(current, mc.num_neurons) current = mc.forward(current) return current def _forward_hybrid(self, inputs: np.ndarray, active_indices: list = None) -> np.ndarray: """ 混合模式：按微柱类型分组 - 主类型组：并行处理同一输入 - 辅助类型组：级联细化 注意：Integrator需要双输入(sensory, memory)， Memory返回元组(result, confidence)， 这里做统一适配 """ if active_indices is None: active_indices = list(range(len(self.micro_columns))) composition = self.recipe['composition'] types_sorted = sorted(composition.keys(), key=lambda k: -composition[k]) primary_type = types_sorted[0] # 分组 primary_outputs = [] secondary_outputs = [] for idx in active_indices: mc = self.micro_columns[idx] mc_type = self.micro_column_types[idx] out = self._call_micro_column(mc, mc_type, inputs) primary_outputs.append(out) if mc_type == primary_type else secondary_outputs.append(out) # 主组并行拼接 if primary_outputs: result = np.concatenate(primary_outputs) else: result = inputs # 辅助组级联细化 for out in secondary_outputs: min_dim = min(result.shape[0], out.shape[0]) result = 0.7 * result[:min_dim] + 0.3 * out[:min_dim] if result.shape[0] < out.shape[0]: result = np.concatenate([result, out[min_dim:]]) return result def _call_micro_column(self, mc, mc_type: str, inputs: np.ndarray) -> np.ndarray: """统一微柱调用接口，处理不同微柱的参数差异 微柱返回类型分三类： 1. 单输入→单输出: sensory, detector, modulator, selector 2. 单输入→元组(result, confidence): memory, motor 3. 双输入→元组(result, coherence): integrator """ x = np.asarray(inputs, dtype=np.float32).flatten() # 维度适配：VariantB/C/D内部用inputs[index]索引，需要输入长度>=num_neurons target_dim = getattr(mc, 'num_neurons', getattr(mc, 'input_dim', 100)) if len(x) < target_dim: x = np.pad(x, (0, target_dim - len(x))) elif len(x) > target_dim: x = x[:target_dim] if mc_type == 'integrator': result, _coherence = mc.forward(x, x) elif mc_type == 'memory': result, _confidence = mc.forward(x) elif mc_type == 'motor': result, _confidence = mc.forward(x) else: result = mc.forward(x) return np.asarray(result, dtype=np.float32).flatten() def _forward_cascade(self, inputs: np.ndarray, active_indices: list = None) -> np.ndarray: """级联模式：逐级处理""" if active_indices is None: active_indices = list(range(len(self.micro_columns))) current = inputs for idx in active_indices: mc = self.micro_columns[idx] mc_type = self.micro_column_types[idx] current = self._call_micro_column(mc, mc_type, current) return current 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_state(self) -> Dict: """获取功能柱状态，供额叶区分析 Returns: 包含区域信息的字典 """ state = { 'column_type': self.column_type, 'num_micro_columns': len(self.micro_columns), 'active': True, } # 收集各微柱的状态信息 confidences = [] memory_states = [] activity_levels = [] for mc in self.micro_columns: # 置信度（SENSORY特有） if hasattr(mc, 'get_confidence'): conf = mc.get_confidence() confidences.append(conf) # 记忆状态（MEMORY特有） if hasattr(mc, 'get_memory_state'): mem_state = mc.get_memory_state() memory_states.append(mem_state) # 活动水平（通用） if hasattr(mc, 'get_strength'): activity_levels.append(mc.get_strength()) # 汇总信息 if confidences: state['avg_confidence'] = np.mean(confidences) state['min_confidence'] = np.min(confidences) if memory_states: # 聚合记忆信息 total_lt = sum(s.get('long_term_count', 0) for s in memory_states) total_st = sum(s.get('short_term_count', 0) for s in memory_states) state['long_term_memories'] = total_lt state['short_term_memories'] = total_st if activity_levels: state['avg_activity'] = np.mean(activity_levels) state['max_activity'] = np.max(activity_levels) return state def learn(self, force: bool = False, learn_prob: float = 0.3, area_states: Dict = None, **kwargs): """调用所有微柱的赫布学习 改进： - learn_prob: 学习概率（避免每次都学习，减少过度训练） - force: 强制学习（忽略概率） - 只在信号足够强时学习（避免噪声学习） PREFRONTAL专有： - area_states: 其他功能区的状态（来自额叶区调用） """ # 采样学习：不是每次都学习，减少过度训练 if not force and np.random.rand() > learn_prob: return # PREFRONTAL区：接收并分析其他区域状态 if self.column_type == 'prefrontal' and area_states: self._analyze_area_states(area_states) for mc in self.micro_columns: if hasattr(mc, 'learn'): # 检查微柱是否有有效的学习信号 if hasattr(mc, '_synaptic') and mc._synaptic is not None: pre = getattr(mc._synaptic, '_last_pre', None) post = getattr(mc._synaptic, '_last_post', None) if pre is not None and post is not None: # 只在信号强度足够时学习 signal_strength = np.mean(np.abs(post)) if signal_strength < 0.01: # 信号太弱，跳过 continue # PREFRONTAL微柱：传入区域状态供学习 if self.column_type == 'prefrontal' and area_states: mc.learn(area_states=area_states, **kwargs) else: mc.learn(**kwargs) def _analyze_area_states(self, area_states: Dict): """PREFRONTAL：分析其他区域状态，进行推理决策""" if not hasattr(self, '_area_analysis'): self._area_analysis = { 'history': [], # 历史分析结果 'low_confidence_count': 0, # 低置信度计数 'fission_candidates': [], # 待决裂变候选 } # 分析各区域状态 issues = [] for area_name, state in area_states.items(): if area_name == 'prefrontal': continue # 检查置信度 conf = state.get('avg_confidence', 1.0) if conf < 0.6: issues.append(f"{area_name}:低置信度({conf:.2f})") self._area_analysis['low_confidence_count'] += 1 # 检查记忆压力 lt = state.get('long_term_memories', 0) st = state.get('short_term_memories', 0) if lt > 1000: issues.append(f"{area_name}:长期记忆过载({lt})") # 检查活动异常 activity = state.get('avg_activity', 0) if activity < 0.05: issues.append(f"{area_name}:低活动({activity:.3f})") # 记录分析结果 if issues: self._area_analysis['history'].append({ 'issues': issues, 'timestamp': __import__('time').time() }) # 保持历史记录不过长 if len(self._area_analysis['history']) > 50: self._area_analysis['history'] = self._area_analysis['history'][-50:] def decide_fission(self, area_states: Dict = None) -> Dict: """裂变决策：决定是否需要分裂微柱/功能柱 PREFRONTAL专用：基于对各区域状态的分析，决定是否进行裂变 Args: area_states: 其他功能区的状态（可选） Returns: 裂变决策结果 { 'should_fission': bool, 'target_area': str, # 目标区域 'fission_type': str, # 'micro_column' / 'functional_column' 'reason': str, # 决策原因 'priority': int, # 优先级 1-5 } """ # 只有PREFRONTAL区可以做裂变决策 if self.column_type != 'prefrontal': return {'should_fission': False, 'reason': '非额叶区无权限'} decision = { 'should_fission': False, 'target_area': None, 'fission_type': None, 'reason': '状态正常', 'priority': 0, } # 分析区域状态 if area_states: # 1. 检查是否有区域需要更多微柱 for area_name, state in area_states.items(): if area_name == 'prefrontal': continue # 置信度低 → 建议增加微柱 conf = state.get('avg_confidence', 1.0) if conf < 0.4: decision['should_fission'] = True decision['target_area'] = area_name decision['fission_type'] = 'micro_column' decision['reason'] = f'置信度过低({conf:.2f})，需更多微柱增强' decision['priority'] = 5 break # 活动过低 → 可能是微柱不足 activity = state.get('avg_activity', 0.5) num_mc = state.get('num_micro_columns', 1) if activity < 0.1 and num_mc < 10: decision['should_fission'] = True decision['target_area'] = area_name decision['fission_type'] = 'micro_column' decision['reason'] = f'活动过低({activity:.3f})，需更多微柱激活' decision['priority'] = 4 # 2. 检查自身的分析历史 if hasattr(self, '_area_analysis'): low_conf_count = self._area_analysis.get('low_confidence_count', 0) # 连续多次低置信度 → 建议功能柱分裂 if low_conf_count > 5: decision['should_fission'] = True decision['target_area'] = 'association' # 默认关联区 decision['fission_type'] = 'functional_column' decision['reason'] = f'连续{low_conf_count}次低置信度，需扩展功能' decision['priority'] = 3 return decision def get_info(self) -> Dict: """获取功能柱完整信息""" # 统计各类型微柱数量 type_counts = {} for t in self.micro_column_types: type_counts[t] = type_counts.get(t, 0) + 1 recipe = self.recipe return { 'column_type': self.column_type, 'column_name': recipe['name'], 'role': recipe.get('role', ''), 'num_micro_columns': self.num_micro_columns, 'micro_column_types': type_counts, 'composition': recipe['composition'], 'connection_mode': self.connection_mode, 'inputs': recipe.get('inputs', []), 'outputs': recipe.get('outputs', []), } # ============ THALAMUS 专用：监控与异常中断 ============ def monitor_all(self, area_states: Dict) -> Dict: """THALAMUS专用：监控所有区域状态，检测异常""" if self.column_type != 'thalamus': return {'error': '只有丘脑区可执行监控'} if not hasattr(self, '_monitor_stats'): self._monitor_stats = { 'check_count': 0, 'anomaly_count': 0, 'interrupt_count': 0, 'anomaly_history': [], } self._monitor_stats['check_count'] += 1 # 检测各区域异常 anomalies = [] interrupt_triggered = False for area_name, state in area_states.items(): if area_name == 'thalamus': continue # 1. 检查活动水平异常 activity = state.get('avg_activity', 0.5) if activity < 0.02: anomalies.append({ 'area': area_name, 'type': 'low_activity', 'value': float(activity), 'severity': 4, }) interrupt_triggered = True # 2. 检查置信度异常 conf = state.get('avg_confidence', 1.0) if conf < 0.3 and area_name in ['sensory', 'memory', 'association']: anomalies.append({ 'area': area_name, 'type': 'low_confidence', 'value': float(conf), 'severity': 5, }) interrupt_triggered = True # 3. 检查记忆过载 lt = state.get('long_term_memories', 0) if lt > 20000: anomalies.append({ 'area': area_name, 'type': 'memory_overload', 'value': int(lt), 'severity': 3, }) # 记录异常 if anomalies: self._monitor_stats['anomaly_count'] += len(anomalies) import time self._monitor_stats['anomaly_history'].append({ 'timestamp': time.time(), 'anomalies': anomalies, }) if len(self._monitor_stats['anomaly_history']) > 20: self._monitor_stats['anomaly_history'] = self._monitor_stats['anomaly_history'][-20:] if interrupt_triggered: self._monitor_stats['interrupt_count'] += 1 return { 'check_count': self._monitor_stats['check_count'], 'anomaly_count': self._monitor_stats['anomaly_count'], 'interrupt_triggered': interrupt_triggered, 'anomalies': anomalies, } # ============ 便捷创建函数 ============ def create_sensory_column(num_mcs: int = 10, col_index: int = 0, neurons_per_micro: int = 100, synaptic_tier: str = None) -> FunctionalColumn: return FunctionalColumn(column_type='sensory', num_micro_columns=num_mcs, col_index=col_index, neurons_per_micro=neurons_per_micro, synaptic_tier=synaptic_tier) def create_memory_column(num_mcs: int = 8, col_index: int = 0, neurons_per_micro: int = 100, synaptic_tier: str = None) -> FunctionalColumn: return FunctionalColumn(column_type='memory', num_micro_columns=num_mcs, col_index=col_index, neurons_per_micro=neurons_per_micro, synaptic_tier=synaptic_tier) def create_association_column(num_mcs: int = 6, col_index: int = 0, neurons_per_micro: int = 100, synaptic_tier: str = None) -> FunctionalColumn: return FunctionalColumn(column_type='association', num_micro_columns=num_mcs, col_index=col_index, neurons_per_micro=neurons_per_micro, synaptic_tier=synaptic_tier) def create_prefrontal_column(num_mcs: int = 20, col_index: int = 0, neurons_per_micro: int = 100, synaptic_tier: str = None) -> FunctionalColumn: """前额区 - 默认20个微柱，支持复杂推理和学习""" return FunctionalColumn(column_type='prefrontal', num_micro_columns=num_mcs, col_index=col_index, neurons_per_micro=neurons_per_micro, synaptic_tier=synaptic_tier) def create_motor_column(num_mcs: int = 3, col_index: int = 0, neurons_per_micro: int = 100, synaptic_tier: str = None) -> FunctionalColumn: return FunctionalColumn(column_type='motor', num_micro_columns=num_mcs, col_index=col_index, neurons_per_micro=neurons_per_micro, synaptic_tier=synaptic_tier) def create_thalamus_column(num_mcs: int = 2, col_index: int = 0, neurons_per_micro: int = 100, synaptic_tier: str = None) -> FunctionalColumn: return FunctionalColumn(column_type='thalamus', num_micro_columns=num_mcs, col_index=col_index, neurons_per_micro=neurons_per_micro, synaptic_tier=synaptic_tier)