机器智能开源代码

"""
BIM神经网络系统 - 基于《人工智能原理》张师定著
完整实现：六库 + 四耳神经元 + 关系矩阵 + 智能合约 + 学习/科研模式 + 安全加密 + 分布式模拟 + GUI
"""

import json
import hashlib
import uuid
import copy
import threading
import queue
import time
from typing import Dict, List, Any, Optional, Callable, Tuple, Set
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict
import sqlite3
import os
import pickle
import base64
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC

# ========================== 安全加密模块 ==========================
class SecurityManager:
"""安全加密管理，使用Fernet对称加密"""
def __init__(self, password: str = "default_password"):
self.password = password
self.key = self._derive_key(password)
self.cipher = Fernet(self.key)

def _derive_key(self, password: str) -> bytes:
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=b'static_salt_1234', # 实际应随机生成
iterations=100000,
)
return base64.urlsafe_b64encode(kdf.derive(password.encode()))

def encrypt(self, data: Any) -> str:
"""加密任意数据（序列化为JSON）"""
json_str = json.dumps(data, ensure_ascii=False)
encrypted = self.cipher.encrypt(json_str.encode())
return base64.b64encode(encrypted).decode()

def decrypt(self, encrypted_str: str) -> Any:
"""解密数据"""
encrypted = base64.b64decode(encrypted_str)
decrypted = self.cipher.decrypt(encrypted)
return json.loads(decrypted.decode())

# ========================== 存储层（内存+JSON持久化） ==========================
class Storage:
"""内存存储，支持JSON序列化与持久化"""
def __init__(self, db_path: str = "bim_agi_data.json"):
self.db_path = db_path
self.data = {}
self.load()

def create(self, key: str, value: Any):
self.data[key] = value
self.save()

def read(self, key: str) -> Any:
return self.data.get(key)

def update(self, key: str, value: Any):
if key in self.data:
self.data[key] = value
self.save()

def delete(self, key: str):
if key in self.data:
del self.data[key]
self.save()

def query(self, condition: Callable[[str, Any], bool]) -> List[Any]:
return [v for k, v in self.data.items() if condition(k, v)]

def save(self):
with open(self.db_path, 'w', encoding='utf-8') as f:
json.dump(self.data, f, ensure_ascii=False, indent=2)

def load(self):
if os.path.exists(self.db_path):
with open(self.db_path, 'r', encoding='utf-8') as f:
self.data = json.load(f)
else:
self.data = {}

# ========================== 六库系统 ==========================
class SemanticLibrary:
"""语义库：存储概念定义"""
def __init__(self, storage: Storage, security: SecurityManager):
self.storage = storage
self.security = security

def add_entity(self, entity_id: str, name: str, definition: str, data_type: str, unit: str = ""):
entity = {
'name': name,
'definition': definition,
'data_type': data_type,
'unit': unit,
'encrypted': self.security.encrypt({'id': entity_id, 'name': name})
}
self.storage.create(entity_id, entity)

def get_entity(self, entity_id: str) -> Dict:
return self.storage.read(entity_id)

def validate_term(self, term: str, expected_type: str = None) -> bool:
ent = self.get_entity(term)
if not ent:
return False
if expected_type and ent['data_type'] != expected_type:
return False
return True

class KnowledgeRuleLibrary:
"""知识规则库：IF-THEN规则"""
def __init__(self, storage: Storage, semantic_lib: SemanticLibrary, security: SecurityManager):
self.storage = storage
self.semantic_lib = semantic_lib
self.security = security

def add_rule(self, rule_id: str, conditions: List[Dict], actions: List[Dict], description: str = ""):
# 验证条件中的术语
for cond in conditions:
if not self.semantic_lib.validate_term(cond.get('term')):
raise ValueError(f"术语 '{cond.get('term')}' 未在语义库中定义")
rule = {
'conditions': conditions,
'actions': actions,
'description': description,
'encrypted': self.security.encrypt({'id': rule_id, 'conditions': conditions})
}
self.storage.create(rule_id, rule)

def evaluate_rules(self, context: Dict) -> List[str]:
triggered = []
for rule_id, rule in self.storage.data.items():
if not isinstance(rule, dict) or 'conditions' not in rule:
continue
conditions_met = True
for cond in rule['conditions']:
term = cond['term']
op = cond['operator']
val = cond['value']
if term not in context:
conditions_met = False
break
if op == 'eq' and context[term] != val:
conditions_met = False
break
elif op == 'gt' and context[term] <= val:
conditions_met = False
break
elif op == 'lt' and context[term] >= val:
conditions_met = False
break
if conditions_met:
triggered.append(rule_id)
return triggered

def get_rule(self, rule_id: str) -> Dict:
return self.storage.read(rule_id)

class AlgorithmLibrary:
"""算法库：存储可调用函数"""
def __init__(self, storage: Storage, semantic_lib: SemanticLibrary, security: SecurityManager):
self.storage = storage
self.semantic_lib = semantic_lib
self.security = security

def add_algorithm(self, algo_id: str, name: str, func: Callable, input_params: List[Dict], output_params: List[Dict], description: str = ""):
# 验证参数
for p in input_params + output_params:
if not self.semantic_lib.validate_term(p['term'], p.get('data_type')):
raise ValueError(f"参数 '{p['term']}' 未定义或类型不匹配")
# 序列化函数（使用pickle）
func_serialized = base64.b64encode(pickle.dumps(func)).decode()
algo = {
'name': name,
'func_serialized': func_serialized,
'input_params': input_params,
'output_params': output_params,
'description': description,
'encrypted': self.security.encrypt({'id': algo_id, 'name': name})
}
self.storage.create(algo_id, algo)

def execute(self, algo_id: str, inputs: Dict) -> Dict:
algo = self.storage.read(algo_id)
if not algo:
raise ValueError(f"算法 {algo_id} 不存在")
# 反序列化函数
func = pickle.loads(base64.b64decode(algo['func_serialized']))
# 检查输入
for p in algo['input_params']:
if p['term'] not in inputs:
raise ValueError(f"缺少输入参数: {p['term']}")
result = func(**inputs)
# 确保输出格式
return result

class ModelPropertyLibrary:
"""模型属性库：参数化信息模型模板与实例"""
def __init__(self, storage: Storage, semantic_lib: SemanticLibrary, rule_lib: KnowledgeRuleLibrary, algo_lib: AlgorithmLibrary, security: SecurityManager):
self.storage = storage
self.semantic_lib = semantic_lib
self.rule_lib = rule_lib
self.algo_lib = algo_lib
self.security = security

def add_template(self, template_id: str, name: str, parameters: List[Dict], rules: List[str] = None, algorithms: List[str] = None):
for p in parameters:
if not self.semantic_lib.validate_term(p['term'], p.get('data_type')):
raise ValueError(f"参数 '{p['term']}' 未定义")
template = {
'name': name,
'parameters': parameters,
'rules': rules or [],
'algorithms': algorithms or [],
'encrypted': self.security.encrypt({'id': template_id, 'name': name})
}
self.storage.create(template_id, template)

def create_instance(self, instance_id: str, template_id: str, parameter_values: Dict) -> Dict:
template = self.storage.read(template_id)
if not template:
raise ValueError(f"模板 {template_id} 不存在")
context = parameter_values.copy()
# 应用规则
triggered_rules = self.rule_lib.evaluate_rules(context)
for rule_id in triggered_rules:
rule = self.rule_lib.get_rule(rule_id)
if rule:
for action in rule['actions']:
if action['type'] == 'set_value':
context[action['parameter']] = action['value']
# 执行算法
for algo_id in template['algorithms']:
algo = self.algo_lib.storage.read(algo_id)
if algo:
algo_input = {p['term']: context.get(p['term']) for p in algo['input_params']}
result = self.algo_lib.execute(algo_id, algo_input)
for outp in algo['output_params']:
if outp['term'] in result:
context[outp['term']] = result[outp['term']]
# 保存实例
instance = {
'template_id': template_id,
'parameters': context,
'encrypted': self.security.encrypt({'id': instance_id, 'params': context})
}
self.storage.create(instance_id, instance)
return context

def get_instance(self, instance_id: str) -> Dict:
return self.storage.read(instance_id)

class CausalContractLibrary:
"""因果合约库：定义节点间逻辑关系与协作规则"""
def __init__(self, storage: Storage, semantic_lib: SemanticLibrary, model_lib: ModelPropertyLibrary, security: SecurityManager):
self.storage = storage
self.semantic_lib = semantic_lib
self.model_lib = model_lib
self.security = security

def add_contract(self, contract_id: str, name: str, precondition: List[Dict], action: List[Dict], postcondition: List[Dict]):
contract = {
'name': name,
'precondition': precondition,
'action': action,
'postcondition': postcondition,
'encrypted': self.security.encrypt({'id': contract_id, 'name': name})
}
self.storage.create(contract_id, contract)

def execute(self, contract_id: str, context: Dict) -> Tuple[bool, Dict]:
contract = self.storage.read(contract_id)
if not contract:
return False, f"合约 {contract_id} 不存在"
# 检查前置条件
for cond in contract['precondition']:
term = cond['term']
op = cond['operator']
val = cond['value']
if term not in context:
return False, f"前置条件缺失: {term}"
if op == 'eq' and context[term] != val:
return False, f"前置条件不满足: {term} != {val}"
elif op == 'gt' and context[term] <= val:
return False, f"前置条件不满足: {term} <= {val}"
elif op == 'lt' and context[term] >= val:
return False, f"前置条件不满足: {term} >= {val}"
# 执行动作
result_context = context.copy()
for act in contract['action']:
if act['type'] == 'create_model':
instance_id = act['instance_id']
template_id = act['template_id']
params = act['parameters']
resolved_params = {}
for k, v in params.items():
if isinstance(v, str) and v.startswith('$'):
resolved_params[k] = result_context.get(v[1:], v)
else:
resolved_params[k] = v
model_result = self.model_lib.create_instance(instance_id, template_id, resolved_params)
result_context.update(model_result)
# 检查后置条件
for cond in contract['postcondition']:
term = cond['term']
op = cond['operator']
val = cond['value']
if term not in result_context:
return False, f"后置条件缺失: {term}"
if op == 'eq' and result_context[term] != val:
return False, f"后置条件不满足: {term} != {val}"
elif op == 'gt' and result_context[term] <= val:
return False, f"后置条件不满足: {term} <= {val}"
elif op == 'lt' and result_context[term] >= val:
return False, f"后置条件不满足: {term} >= {val}"
return True, result_context

class TaskFlowLibrary:
"""任务流程库：存储流程分解"""
def __init__(self, storage: Storage, semantic_lib: SemanticLibrary, contract_lib: CausalContractLibrary, security: SecurityManager):
self.storage = storage
self.semantic_lib = semantic_lib
self.contract_lib = contract_lib
self.security = security

def add_workflow(self, workflow_id: str, name: str, steps: List[Dict], description: str = ""):
workflow = {
'name': name,
'steps': steps,
'description': description,
'encrypted': self.security.encrypt({'id': workflow_id, 'name': name})
}
self.storage.create(workflow_id, workflow)

def execute(self, workflow_id: str, initial_context: Dict = None) -> Tuple[bool, Dict]:
workflow = self.storage.read(workflow_id)
if not workflow:
return False, f"流程 {workflow_id} 不存在"
context = initial_context or {}
results = {}
for step in workflow['steps']:
step_type = step['type']
if step_type == 'contract':
contract_id = step['contract_id']
success, new_ctx = self.contract_lib.execute(contract_id, context)
if not success:
return False, f"步骤 {step.get('name')} 失败: {new_ctx}"
context.update(new_ctx)
results[step.get('name')] = new_ctx
elif step_type == 'manual':
# 模拟人工操作（在GUI中可弹出对话框）
print(f"【人工操作】{step.get('name')}: {step.get('description')}")
if 'output' in step:
context.update(step['output'])
return True, results

# ========================== 四耳神经元与神经网络 ==========================
class NeuralNode:
"""四耳神经元：每个节点代表一个智能体"""
def __init__(self, node_id: str, name: str):
self.node_id = node_id
self.name = name
self.shared_input = None # 共享输入
self.self_input = None # 自输入（包含感觉信息）
self.shared_output = None # 共享输出
self.self_output = None # 自输出
self.parent = None
self.children = []
self.predecessors = [] # 紧前节点
self.successors = [] # 紧后节点
self.contracts = [] # 关联合约ID
self.context = {} # 当前状态
self.sensor_data = {} # 感觉信息（作为自输入的一部分）

def add_child(self, child: 'NeuralNode'):
child.parent = self
self.children.append(child)

def add_predecessor(self, pred: 'NeuralNode'):
if pred not in self.predecessors:
self.predecessors.append(pred)

def add_successor(self, succ: 'NeuralNode'):
if succ not in self.successors:
self.successors.append(succ)

def set_sensor_data(self, data: Dict):
"""设置感觉信息，自动合并到自输入"""
self.sensor_data.update(data)
self.update_self_input()

def update_self_input(self):
"""更新自输入（包含感觉信息）"""
self.self_input = {**self.sensor_data, **self.self_input} if self.self_input else self.sensor_data.copy()

def set_shared_input(self, data: Dict):
self.shared_input = data
self.context.update(data)

def process(self, contract_lib: CausalContractLibrary) -> bool:
if self.contracts:
contract_id = self.contracts[0] # 简化
success, result = contract_lib.execute(contract_id, self.context)
if success:
self.shared_output = result
self.context.update(result)
return True
return False

def get_info(self) -> Dict:
return {
'node_id': self.node_id,
'name': self.name,
'parent': self.parent.node_id if self.parent else None,
'children': [c.node_id for c in self.children],
'predecessors': [p.node_id for p in self.predecessors],
'successors': [s.node_id for s in self.successors],
'contracts': self.contracts,
'shared_input': self.shared_input,
'self_input': self.self_input,
'shared_output': self.shared_output,
'self_output': self.self_output,
'context': self.context,
'sensor_data': self.sensor_data
}

class BIMNeuralNetwork:
"""BIM神经网络，包含节点和关系矩阵"""
def __init__(self):
self.nodes: Dict[str, NeuralNode] = {}
self.relation_matrix: List[List[int]] = [] # 方阵，索引对应节点列表
self.node_list: List[str] = [] # 节点ID列表，与矩阵行列对应

def add_node(self, node: NeuralNode):
if node.node_id in self.nodes:
raise ValueError(f"节点 {node.node_id} 已存在")
self.nodes[node.node_id] = node
self.node_list.append(node.node_id)
# 扩展矩阵
size = len(self.node_list)
for row in self.relation_matrix:
row.append(0)
self.relation_matrix.append([0]*size)

def add_relation(self, from_id: str, to_id: str, value: int = 1):
"""设置紧前-紧后关系，值为1表示有连接"""
if from_id not in self.node_list or to_id not in self.node_list:
raise ValueError("节点不存在")
i = self.node_list.index(from_id)
j = self.node_list.index(to_id)
self.relation_matrix[i][j] = value
# 同时更新节点的successors/predecessors
from_node = self.nodes[from_id]
to_node = self.nodes[to_id]
from_node.add_successor(to_node)
to_node.add_predecessor(from_node)

def get_relation_matrix(self) -> List[List[int]]:
return self.relation_matrix

def get_node_by_id(self, node_id: str) -> Optional[NeuralNode]:
return self.nodes.get(node_id)

def get_all_nodes(self) -> List[NeuralNode]:
return list(self.nodes.values())

def get_roots(self) -> List[NeuralNode]:
return [n for n in self.nodes.values() if n.parent is None]

def to_dict(self) -> Dict:
return {
'nodes': {nid: node.get_info() for nid, node in self.nodes.items()},
'node_list': self.node_list,
'relation_matrix': self.relation_matrix
}

@classmethod
def from_dict(cls, data: Dict):
net = cls()
for nid, info in data['nodes'].items():
node = NeuralNode(nid, info['name'])
node.context = info.get('context', {})
node.contracts = info.get('contracts', [])
net.nodes[nid] = node
net.node_list = data['node_list']
net.relation_matrix = data['relation_matrix']
# 重建父子、紧前紧后关系
for nid, info in data['nodes'].items():
node = net.nodes[nid]
if info['parent']:
parent = net.nodes.get(info['parent'])
if parent:
parent.add_child(node)
for pred_id in info['predecessors']:
pred = net.nodes.get(pred_id)
if pred:
node.add_predecessor(pred)
for succ_id in info['successors']:
succ = net.nodes.get(succ_id)
if succ:
node.add_successor(succ)
return net

# ========================== 学习与科研模式 ==========================
class LearningEngine:
"""学习模式：当知识缺乏时，尝试从已有规则中学习"""
def __init__(self, rule_lib: KnowledgeRuleLibrary, model_lib: ModelPropertyLibrary, algo_lib: AlgorithmLibrary):
self.rule_lib = rule_lib
self.model_lib = model_lib
self.algo_lib = algo_lib
self.learning_history = []

def learn(self, context: Dict, failed_rule: str = None) -> Tuple[bool, Dict]:
"""尝试通过已有规则学习，补充缺失的上下文"""
# 尝试匹配所有规则，看哪些规则的条件接近
best_match = None
best_score = 0
for rule_id, rule in self.rule_lib.storage.data.items():
if not isinstance(rule, dict) or 'conditions' not in rule:
continue
score = 0
for cond in rule['conditions']:
term = cond['term']
if term in context:
score += 1 # 已知条件
if score > best_score and score < len(rule['conditions']):
best_score = score
best_match = rule_id
if best_match:
# 尝试应用该规则来推导缺失值
rule = self.rule_lib.get_rule(best_match)
if rule:
# 假设规则只有一个动作且是set_value
for action in rule['actions']:
if action['type'] == 'set_value':
param = action['parameter']
if param not in context:
# 从规则中推导值（简化：直接使用规则中定义的值）
context[param] = action['value']
self.learning_history.append(f"从规则 {best_match} 学习到 {param}={action['value']}")
return True, context
return False, context

def research(self, context: Dict, problem: str) -> Tuple[bool, Dict]:
"""科研模式：当学习仍不足时，创造新规则或新模型"""
# 模拟科研：生成一个新规则，基于现有概念组合
new_rule_id = f"research_rule_{uuid.uuid4().hex[:8]}"
# 简单的科研：如果缺少某个参数，尝试用算法推导
# 这里我们假设可以调用算法库中的算法
for algo_id, algo in self.algo_lib.storage.data.items():
if not isinstance(algo, dict) or 'input_params' not in algo:
continue
# 检查是否可以用该算法推导缺失值
input_terms = [p['term'] for p in algo['input_params']]
output_terms = [p['term'] for p in algo['output_params']]
if all(t in context for t in input_terms) and any(t not in context for t in output_terms):
try:
inputs = {t: context[t] for t in input_terms}
result = self.algo_lib.execute(algo_id, inputs)
for out_term in output_terms:
if out_term in result and out_term not in context:
context[out_term] = result[out_term]
self.learning_history.append(f"通过科研算法 {algo_id} 推导出 {out_term}={result[out_term]}")
return True, context
except:
pass
# 如果还是不行，创建一个人工规则（科研产生新知识）
# 假设生成规则：如果问题包含"桥梁"，则设置bridge_type="预应力混凝土"
if '桥梁' in problem:
context['bridge_type'] = '预应力混凝土'
self.learning_history.append(f"科研模式生成新知识: bridge_type=预应力混凝土")
return True, context
return False, context

# ========================== 分布式模拟 ==========================
class DistributedNode(threading.Thread):
"""模拟分布式节点，运行在独立线程中"""
def __init__(self, node: NeuralNode, contract_lib: CausalContractLibrary, input_queue: queue.Queue, output_queue: queue.Queue):
super().__init__()
self.node = node
self.contract_lib = contract_lib
self.input_queue = input_queue
self.output_queue = output_queue
self.running = True

def run(self):
while self.running:
try:
data = self.input_queue.get(timeout=1)
if data is None:
break
# 处理接收到的共享输入
self.node.set_shared_input(data)
# 执行处理
success = self.node.process(self.contract_lib)
if success:
# 发送共享输出给后继节点（通过输出队列）
self.output_queue.put(self.node.shared_output)
except queue.Empty:
continue

def stop(self):
self.running = False
self.input_queue.put(None)

class DistributedSystem:
"""分布式系统管理器，模拟多节点部署"""
def __init__(self):
self.nodes = {}
self.queues = {}
self.threads = {}

def deploy_node(self, node: NeuralNode, contract_lib: CausalContractLibrary):
in_q = queue.Queue()
out_q = queue.Queue()
self.queues[node.node_id] = (in_q, out_q)
thread = DistributedNode(node, contract_lib, in_q, out_q)
self.threads[node.node_id] = thread
self.nodes[node.node_id] = node
thread.start()

def send_to_node(self, node_id: str, data: Dict):
if node_id in self.queues:
self.queues[node_id][0].put(data)

def get_output_from_node(self, node_id: str) -> Optional[Dict]:
if node_id in self.queues:
try:
return self.queues[node_id][1].get_nowait()
except queue.Empty:
return None
return None

def shutdown(self):
for thread in self.threads.values():
thread.stop()
for thread in self.threads.values():
thread.join()

# ========================== 图形界面 (Tkinter) ==========================
import tkinter as tk
from tkinter import ttk, messagebox, scrolledtext
import matplotlib.pyplot as plt
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
import networkx as nx

class BIMAGIGUI:
"""符合专著要求的通用智能体交互界面"""
def __init__(self, master, agi_system):
self.master = master
self.agi = agi_system
self.master.title("BIM-AGI 通用智能体交互界面")
self.master.geometry("1200x800")
self.current_node_id = None

# 主面板
main_pane = ttk.PanedWindow(self.master, orient=tk.HORIZONTAL)
main_pane.pack(fill=tk.BOTH, expand=True)

# 左侧：网络图
left_frame = ttk.Frame(main_pane, width=400)
main_pane.add(left_frame, weight=1)
ttk.Label(left_frame, text="BIM神经网络拓扑", font=('Arial', 12, 'bold')).pack(pady=5)
self.fig, self.ax = plt.subplots(figsize=(5, 5))
self.canvas = FigureCanvasTkAgg(self.fig, master=left_frame)
self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
self.draw_network()

# 右侧信息面板
right_frame = ttk.Frame(main_pane, width=600)
main_pane.add(right_frame, weight=2)

# 节点选择
select_frame = ttk.Frame(right_frame)
select_frame.pack(fill=tk.X, padx=5, pady=5)
ttk.Label(select_frame, text="选择节点:").pack(side=tk.LEFT, padx=5)
self.node_combo = ttk.Combobox(select_frame, state='readonly', width=30)
self.node_combo.pack(side=tk.LEFT, padx=5)
self.node_combo.bind('<<ComboboxSelected>>', self.on_node_selected)

# 节点信息
info_frame = ttk.LabelFrame(right_frame, text="节点详细信息", padding=5)
info_frame.pack(fill=tk.X, padx=5, pady=5)
self.info_text = scrolledtext.ScrolledText(info_frame, height=8, width=70)
self.info_text.pack(fill=tk.BOTH, expand=True)

# 四耳输入输出
io_frame = ttk.LabelFrame(right_frame, text="四耳信息", padding=5)
io_frame.pack(fill=tk.X, padx=5, pady=5)
# 共享输入
ttk.Label(io_frame, text="共享输入:").grid(row=0, column=0, sticky=tk.W, padx=5)
self.shared_input_text = scrolledtext.ScrolledText(io_frame, height=2, width=40)
self.shared_input_text.grid(row=0, column=1, padx=5, pady=2)
# 自输入（含感觉）
ttk.Label(io_frame, text="自输入 (感觉):").grid(row=1, column=0, sticky=tk.W, padx=5)
self.self_input_text = scrolledtext.ScrolledText(io_frame, height=2, width=40)
self.self_input_text.grid(row=1, column=1, padx=5, pady=2)
# 共享输出
ttk.Label(io_frame, text="共享输出:").grid(row=2, column=0, sticky=tk.W, padx=5)
self.shared_output_text = scrolledtext.ScrolledText(io_frame, height=2, width=40)
self.shared_output_text.grid(row=2, column=1, padx=5, pady=2)
# 自输出
ttk.Label(io_frame, text="自输出:").grid(row=3, column=0, sticky=tk.W, padx=5)
self.self_output_text = scrolledtext.ScrolledText(io_frame, height=2, width=40)
self.self_output_text.grid(row=3, column=1, padx=5, pady=2)

# 任务执行
task_frame = ttk.LabelFrame(right_frame, text="任务执行", padding=5)
task_frame.pack(fill=tk.X, padx=5, pady=5)
ttk.Label(task_frame, text="选择流程:").grid(row=0, column=0, sticky=tk.W, padx=5)
self.workflow_combo = ttk.Combobox(task_frame, state='readonly', width=30)
self.workflow_combo.grid(row=0, column=1, padx=5, pady=2)
ttk.Label(task_frame, text="参数(JSON):").grid(row=1, column=0, sticky=tk.W, padx=5)
self.params_entry = tk.Text(task_frame, height=3, width=40)
self.params_entry.grid(row=1, column=1, padx=5, pady=2)
self.execute_btn = ttk.Button(task_frame, text="执行流程", command=self.execute_workflow)
self.execute_btn.grid(row=2, column=1, pady=5, sticky=tk.E)

# 日志
log_frame = ttk.LabelFrame(right_frame, text="执行日志", padding=5)
log_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
self.log_text = scrolledtext.ScrolledText(log_frame, height=6)
self.log_text.pack(fill=tk.BOTH, expand=True)

# 刷新节点列表
self.refresh_node_list()
self.refresh_workflow_list()

def draw_network(self):
"""使用networkx绘制网络拓扑"""
self.ax.clear()
G = nx.DiGraph()
net = self.agi.network
for node_id, node in net.nodes.items():
G.add_node(node_id, label=node.name)
for i, from_id in enumerate(net.node_list):
for j, to_id in enumerate(net.node_list):
if net.relation_matrix[i][j] == 1:
G.add_edge(from_id, to_id)
if G.number_of_nodes() == 0:
self.ax.text(0.5, 0.5, "无节点", ha='center', va='center')
else:
pos = nx.spring_layout(G)
nx.draw(G, pos, ax=self.ax, with_labels=True, node_color='lightblue',
edge_color='gray', arrows=True, arrowsize=20)
self.canvas.draw()

def refresh_node_list(self):
"""更新下拉列表"""
node_ids = list(self.agi.network.nodes.keys())
self.node_combo['values'] = node_ids
if node_ids:
self.node_combo.current(0)
self.on_node_selected(None)

def refresh_workflow_list(self):
workflows = self.agi.task_lib.storage.query(lambda k, v: isinstance(v, dict) and 'steps' in v)
wf_names = [wf['name'] for wf in workflows]
self.workflow_combo['values'] = wf_names
if wf_names:
self.workflow_combo.current(0)

def on_node_selected(self, event):
node_id = self.node_combo.get()
if not node_id:
return
node = self.agi.network.get_node_by_id(node_id)
if not node:
return
info = node.get_info()
self.info_text.delete('1.0', tk.END)
self.info_text.insert('1.0', json.dumps(info, indent=2, ensure_ascii=False))
# 更新四耳文本框
self.shared_input_text.delete('1.0', tk.END)
self.shared_input_text.insert('1.0', json.dumps(info.get('shared_input', {}), indent=2, ensure_ascii=False))
self.self_input_text.delete('1.0', tk.END)
self.self_input_text.insert('1.0', json.dumps(info.get('self_input', {}), indent=2, ensure_ascii=False))
self.shared_output_text.delete('1.0', tk.END)
self.shared_output_text.insert('1.0', json.dumps(info.get('shared_output', {}), indent=2, ensure_ascii=False))
self.self_output_text.delete('1.0', tk.END)
self.self_output_text.insert('1.0', json.dumps(info.get('self_output', {}), indent=2, ensure_ascii=False))

def execute_workflow(self):
workflow_name = self.workflow_combo.get()
if not workflow_name:
messagebox.showwarning("警告", "请选择流程")
return
# 查找流程ID
wf_id = None
for kid, val in self.agi.task_lib.storage.data.items():
if isinstance(val, dict) and val.get('name') == workflow_name:
wf_id = kid
break
if not wf_id:
messagebox.showerror("错误", "流程未找到")
return
params_str = self.params_entry.get('1.0', tk.END).strip()
try:
params = json.loads(params_str) if params_str else {}
except:
messagebox.showerror("错误", "参数格式错误")
return
self.log(f"执行流程: {workflow_name}")
# 执行流程（可能需要先选择起始节点）
# 这里默认使用第一个节点作为上下文起点
nodes = self.agi.network.get_all_nodes()
if not nodes:
self.log("错误: 无节点")
return
# 将参数注入到第一个节点的上下文中
first_node = nodes[0]
first_node.context.update(params)
# 执行流程
success, result = self.agi.task_lib.execute(wf_id, first_node.context)
if success:
self.log(f"流程执行成功，结果: {json.dumps(result, indent=2, ensure_ascii=False)}")
# 更新节点共享输出
first_node.shared_output = result
# 刷新显示
self.on_node_selected(None)
self.draw_network()
else:
self.log(f"流程执行失败: {result}")

def log(self, msg):
self.log_text.insert(tk.END, msg + "\n")
self.log_text.see(tk.END)

# ========================== 主系统 ==========================
class BIMAGISystem:
"""主系统，整合所有组件"""
def __init__(self):
self.security = SecurityManager("bim_agi_password")
self.storage = Storage("bim_agi_data.json")
self.semantic_lib = SemanticLibrary(self.storage, self.security)
self.rule_lib = KnowledgeRuleLibrary(self.storage, self.semantic_lib, self.security)
self.algo_lib = AlgorithmLibrary(self.storage, self.semantic_lib, self.security)
self.model_lib = ModelPropertyLibrary(self.storage, self.semantic_lib, self.rule_lib, self.algo_lib, self.security)
self.contract_lib = CausalContractLibrary(self.storage, self.semantic_lib, self.model_lib, self.security)
self.task_lib = TaskFlowLibrary(self.storage, self.semantic_lib, self.contract_lib, self.security)
self.network = BIMNeuralNetwork()
self.learning_engine = LearningEngine(self.rule_lib, self.model_lib, self.algo_lib)
self.distributed = DistributedSystem()

# 初始化示例数据（桥梁设计场景）
self.init_demo_data()

def init_demo_data(self):
"""按照专著示例初始化数据"""
# 语义库
self.semantic_lib.add_entity("span_length", "跨度长度", "桥梁主跨长度", "float", "米")
self.semantic_lib.add_entity("bridge_type", "桥梁类型", "结构类型", "string")
self.semantic_lib.add_entity("material", "材料", "主要材料", "string")
self.semantic_lib.add_entity("beam_height", "梁高", "梁的高度", "float", "米")
# 算法：计算梁高
def calc_beam_height(span_length):
return {"beam_height": span_length / 20}
self.algo_lib.add_algorithm("algo_beam_height", "梁高计算", calc_beam_height,
[{"term": "span_length", "data_type": "float"}],
[{"term": "beam_height", "data_type": "float"}])
# 规则：跨度>100则预应力混凝土
self.rule_lib.add_rule("rule_bridge_type",
[{"term": "span_length", "operator": "gt", "value": 100}],
[{"type": "set_value", "parameter": "bridge_type", "value": "预应力混凝土"}])
# 模型模板
self.model_lib.add_template("template_bridge", "桥梁模板",
[{"term": "span_length", "data_type": "float"},
{"term": "bridge_type", "data_type": "string"},
{"term": "material", "data_type": "string"},
{"term": "beam_height", "data_type": "float"}],
rules=["rule_bridge_type"],
algorithms=["algo_beam_height"])
# 合约
self.contract_lib.add_contract("contract_create_bridge", "创建桥梁",
precondition=[{"term": "span_length", "operator": "gt", "value": 0}],
action=[{"type": "create_model", "instance_id": "bridge_instance_1",
"template_id": "template_bridge",
"parameters": {"span_length": "$span_length", "material": "$material"}}],
postcondition=[{"term": "bridge_type", "operator": "eq", "value": "预应力混凝土"}])
# 流程
self.task_lib.add_workflow("workflow_bridge_design", "桥梁设计流程",
steps=[{"type": "contract", "name": "创建桥梁模型", "contract_id": "contract_create_bridge"},
{"type": "manual", "name": "专家审核", "description": "请结构专家审核模型合理性"}])

# 创建节点网络
platform = NeuralNode("platform", "协同共享平台")
design = NeuralNode("design", "设计节点")
review = NeuralNode("review", "审核节点")
platform.add_child(design)
platform.add_child(review)
self.network.add_node(platform)
self.network.add_node(design)
self.network.add_node(review)
self.network.add_relation("design", "review") # 设计->审核
# 关联合约到设计节点
design.contracts.append("contract_create_bridge")
# 设置感觉信息（例如传感器数据）
design.set_sensor_data({"temperature": 25, "wind_speed": 12})

def run_gui(self):
root = tk.Tk()
app = BIMAGIGUI(root, self)
root.mainloop()

# ========================== 启动 ==========================
if __name__ == "__main__":
system = BIMAGISystem()
# 启动GUI
system.run_gui()