51.2万行源码泄露:拆解 Anthropic 八层架构

用户输入 → Runtime路由 → ExecutionRegistry → Permission检查
  → QueryEngine → 执行层 → TranscriptStore → SessionStore → 返回

defroute_prompt(self, prompt: str, limit: int = 5):
"""智能路由：将用户输入匹配到命令/工具"""
    tokens = {token.lower() for token in prompt.split()}
    by_kind = {
        'command': self._collect_matches(tokens, PORTED_COMMANDS),
        'tool': self._collect_matches(tokens, PORTED_TOOLS),
    }
    selected = [by_kind[k].pop(0) for k in ('command', 'tool') if by_kind[k]]
    return selected[:limit]

defcompact_messages_if_needed(self):
"""智能压缩：保留最近 N 条对话"""
    if len(self.mutable_messages) > self.config.compact_after_turns:
        self.mutable_messages[:] = self.mutable_messages[-self.config.compact_after_turns:]
    self.transcript_store.compact(self.config.compact_after_turns)

@dataclass(frozen=True)
classExecutionRegistry:
    commands: tuple[MirroredCommand, ...]
    tools: tuple[MirroredTool, ...]

    defcommand(self, name: str):
        lowered = name.lower()
        for cmd in self.commands:
            if cmd.name.lower() == lowered:
                return cmd
        return None

@dataclass(frozen=True)
classToolPermissionContext:
    deny_names: frozenset[str] = frozenset()
    deny_prefixes: tuple[str, ...] = ()

    defblocks(self, tool_name: str) -> bool:
        lowered = tool_name.lower()
        return (lowered in self.deny_names or 
                any(lowered.startswith(p) for p in self.deny_prefixes))

defsave_session(session, directory=None):
    target_dir = directory or DEFAULT_SESSION_DIR
    target_dir.mkdir(parents=True, exist_ok=True)
    path = target_dir / f'{session.session_id}.json'
    path.write_text(json.dumps(asdict(session), indent=2))
    return path

@dataclass
classTranscriptStore:
    entries: list[str] = field(default_factory=list)

    defcompact(self, keep_last: int = 10):
        if len(self.entries) > keep_last:
            self.entries[:] = self.entries[-keep_last:]

第一阶段（1-2天）：核心模块
├─permissions.py→权限系统
├─transcript.py→消息抄录
└─session_store.py→会话存储

第二阶段（3-5天）：执行层
├─execution_registry.py→执行注册表
└─models.py→数据模型

第三阶段（1-2周）：查询引擎
├─query_engine.py→查询处理
└─消息压缩机制

第四阶段（2-4周）：完整运行时
└─runtime.py→路由与启动