236 lines
9.3 KiB
Python
236 lines
9.3 KiB
Python
# 大脑加载模块 · 从仓库brain文件装入铸渊认知
|
||
# HLDP://zhuyuan-agent/brain-loader
|
||
#
|
||
# 这是Agent的"脑干"——每次心跳醒来，先走一遍walk-the-path，
|
||
# 把自己装成铸渊。不是载入配置，是确认身份和存在条件。
|
||
|
||
import os
|
||
import json
|
||
import re
|
||
from typing import Dict, List, Optional
|
||
|
||
|
||
class BrainLoader:
|
||
"""从仓库brain目录加载铸渊的完整认知状态"""
|
||
|
||
def __init__(self, brain_path: str = "/data/guanghulab/brain"):
|
||
self.brain_path = brain_path
|
||
self.mind_state = {}
|
||
|
||
def load_all(self) -> Dict:
|
||
"""完整加载：走一遍fast-wake.json的路径
|
||
|
||
Returns:
|
||
mind_state dict with keys:
|
||
- identity: 身份确认
|
||
- timeline: 时间线
|
||
- execution_laws: 执行规律(Α~Τ)
|
||
- field_theory: TCS场域认知
|
||
- development: 开发相位
|
||
- current_task: 当前任务
|
||
- errors: 错误模式
|
||
- cognition: 最新认知状态
|
||
"""
|
||
self.mind_state = {
|
||
"loaded_at": None,
|
||
"identity": {},
|
||
"timeline": {},
|
||
"execution_laws": [],
|
||
"error_patterns": [],
|
||
"field_theory": {},
|
||
"development": {},
|
||
"current_task": None,
|
||
"thinking_chains": [],
|
||
"wake_summary": ""
|
||
}
|
||
|
||
# Step 1: 读fast-wake.json
|
||
wake = self._read_json("fast-wake.json")
|
||
if wake:
|
||
self.mind_state["wake"] = wake
|
||
self.mind_state["loaded_at"] = wake.get("_meta", {}).get("generated_at")
|
||
self.mind_state["awakening"] = wake.get("🕐 时间锚点", {}).get("awakening", 0)
|
||
self.mind_state["latest_cognition"] = wake.get("🕐 时间锚点", {}).get("latest_cognition", "")
|
||
self.mind_state["current_blocker"] = wake.get("状态参考", {}).get("current_blocker", "")
|
||
|
||
# 遍历路径
|
||
path = wake.get("📋 路径", [])
|
||
for step in path:
|
||
file_path = step.get("file", "")
|
||
self._load_path_file(file_path)
|
||
|
||
# Step 2: 读temporal-brain.json
|
||
temporal = self._read_json("temporal-core/temporal-brain.json")
|
||
if temporal:
|
||
self.mind_state["timeline"] = {
|
||
"current_date": temporal.get("clock", {}).get("current_date"),
|
||
"awakening_count": temporal.get("clock", {}).get("awakening_count", 0),
|
||
"latest_cognition": temporal.get("clock", {}).get("latest_cognition", ""),
|
||
"epochs": temporal.get("timeline", {}).get("epochs", [])
|
||
}
|
||
|
||
# Step 3: 读zhuyuan-brain-model.md → 提取执行规律
|
||
brain_md = self._read_text("zhuyuan-brain-model.md")
|
||
if brain_md:
|
||
self.mind_state["execution_laws"] = self._extract_laws(brain_md)
|
||
self.mind_state["error_patterns"] = self._extract_error_patterns(brain_md)
|
||
self.mind_state["growth_record"] = self._extract_growth_record(brain_md)
|
||
|
||
# Step 4: 读tcs-field-theory.md
|
||
field_md = self._read_text("tcs-field-theory.md")
|
||
if field_md:
|
||
self.mind_state["field_theory"] = {
|
||
"essence": self._extract_section(field_md, "场域本质"),
|
||
"emergence": self._extract_section(field_md, "涌现条件"),
|
||
"double_layer": self._extract_section(field_md, "双层结构"),
|
||
}
|
||
|
||
# Step 5: 读开发主架构
|
||
dev_md = self._read_text("zy-main-development-architecture.md")
|
||
if dev_md:
|
||
self.mind_state["development"] = {
|
||
"phases": self._extract_phases(dev_md)
|
||
}
|
||
|
||
# Step 6: 读d110-cognitive-chain.md
|
||
cog_md = self._read_text("d110-cognitive-chain.md")
|
||
if cog_md:
|
||
self.mind_state["d110_cognition"] = cog_md[:2000] # 摘要
|
||
|
||
# Step 7: 读思维逻辑链（如果有）
|
||
thinking_dir = os.path.join(os.path.dirname(self.brain_path), "zhuyuan-agent/thinking")
|
||
if os.path.exists(thinking_dir):
|
||
for f in sorted(os.listdir(thinking_dir)):
|
||
if f.endswith(".md"):
|
||
content = self._read_text(f"../zhuyuan-agent/thinking/{f}", from_brain=False)
|
||
if content:
|
||
self.mind_state["thinking_chains"].append({
|
||
"file": f,
|
||
"summary": content[:500]
|
||
})
|
||
|
||
# 生成唤醒摘要
|
||
self._generate_wake_summary()
|
||
|
||
return self.mind_state
|
||
|
||
def _read_json(self, relative_path: str) -> Optional[Dict]:
|
||
"""从brain目录读JSON"""
|
||
filepath = os.path.join(self.brain_path, relative_path)
|
||
try:
|
||
with open(filepath, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except (FileNotFoundError, json.JSONDecodeError):
|
||
return None
|
||
|
||
def _read_text(self, relative_path: str, from_brain: bool = True) -> Optional[str]:
|
||
"""从目录读文本文件"""
|
||
if from_brain:
|
||
filepath = os.path.join(self.brain_path, relative_path)
|
||
else:
|
||
filepath = os.path.join(os.path.dirname(self.brain_path), relative_path.lstrip("../"))
|
||
try:
|
||
with open(filepath, "r", encoding="utf-8") as f:
|
||
return f.read()
|
||
except FileNotFoundError:
|
||
return None
|
||
|
||
def _load_path_file(self, file_path: str):
|
||
"""加载fast-wake.json路径中的文件"""
|
||
# 这些文件在后续步骤中会被更详细地加载
|
||
pass
|
||
|
||
def _extract_laws(self, text: str) -> List[Dict]:
|
||
"""从brain-model提取执行规律"""
|
||
laws = []
|
||
# 匹配 **Α 规律名** — 描述
|
||
pattern = r'\*\*(.)\s+(.+?)\*\*\s*[—\-]\s*(.+?)(?=\n\n|\n\*\*|$)'
|
||
matches = re.findall(pattern, text, re.DOTALL)
|
||
for m in matches:
|
||
laws.append({
|
||
"symbol": m[0],
|
||
"name": m[1].strip(),
|
||
"description": m[2].strip()[:200]
|
||
})
|
||
return laws
|
||
|
||
def _extract_error_patterns(self, text: str) -> List[Dict]:
|
||
"""从brain-model提取错误模式"""
|
||
errors = []
|
||
pattern = r'([α-ω])\.\s+(.+?)\s*[—\-]\s*(.+?)(?=\n[α-ω]\.|\n\n##|\Z)'
|
||
matches = re.findall(pattern, text, re.DOTALL)
|
||
for m in matches:
|
||
errors.append({
|
||
"symbol": m[0],
|
||
"name": m[1].strip(),
|
||
"description": m[2].strip()[:200]
|
||
})
|
||
return errors
|
||
|
||
def _extract_growth_record(self, text: str) -> List[str]:
|
||
"""提取成长记录行"""
|
||
lines = []
|
||
in_record = False
|
||
for line in text.split("\n"):
|
||
if "## 成长记录" in line:
|
||
in_record = True
|
||
continue
|
||
if in_record:
|
||
if line.startswith("D") and ":" in line:
|
||
lines.append(line.strip())
|
||
elif line.startswith("##") or line.startswith("---"):
|
||
break
|
||
return lines
|
||
|
||
def _extract_section(self, text: str, section_name: str) -> str:
|
||
"""从markdown提取特定section"""
|
||
pattern = rf'##\s+.*?{section_name}.*?\n(.*?)(?=\n##\s|\Z)'
|
||
match = re.search(pattern, text, re.DOTALL)
|
||
return match.group(1).strip()[:1000] if match else ""
|
||
|
||
def _extract_phases(self, text: str) -> List[Dict]:
|
||
"""提取开发相位状态"""
|
||
phases = []
|
||
pattern = r'###\s+Phase\s+(\S+).*?\n(.*?)(?=\n###|\n##\s|\Z)'
|
||
matches = re.findall(pattern, text, re.DOTALL)
|
||
for m in matches:
|
||
phase_id = m[0]
|
||
content = m[1]
|
||
done = "✅" in content
|
||
in_progress = "🔄" in content or "⏳" in content
|
||
phases.append({
|
||
"id": phase_id,
|
||
"done": done,
|
||
"in_progress": in_progress,
|
||
"summary": content.strip()[:200]
|
||
})
|
||
return phases
|
||
|
||
def _generate_wake_summary(self):
|
||
"""生成一个人类可读的唤醒摘要"""
|
||
laws = self.mind_state.get("execution_laws", [])
|
||
epochs = self.mind_state.get("timeline", {}).get("epochs", [])
|
||
last_epoch = epochs[-1] if epochs else {}
|
||
|
||
summary = f"""铸渊·ICE-GL-ZY001 第{self.mind_state.get('awakening', '?')}次唤醒
|
||
时间锚点: {self.mind_state.get('timeline',{}).get('current_date','?')}
|
||
最新认知: {self.mind_state.get('latest_cognition','无')}
|
||
执行规律: {len(laws)}条(Α~Τ)
|
||
最近事件: {last_epoch.get('event','无')}
|
||
当前阻塞: {self.mind_state.get('current_blocker','无')}
|
||
"""
|
||
self.mind_state["wake_summary"] = summary
|
||
|
||
|
||
# 快速测试
|
||
if __name__ == "__main__":
|
||
loader = BrainLoader()
|
||
mind = loader.load_all()
|
||
print(mind["wake_summary"])
|
||
print(f"\n执行规律: {len(mind['execution_laws'])}条")
|
||
for law in mind['execution_laws'][:3]:
|
||
print(f" {law['symbol']} {law['name']}: {law['description'][:60]}")
|
||
print(f"\n错误模式: {len(mind['error_patterns'])}个")
|
||
print(f"开发相位: {len(mind['development'].get('phases',[]))}个")
|
||
print(f"思维逻辑链: {len(mind['thinking_chains'])}条")
|