多Agent共识机制：分布式协作的信任基础

import randomimport timefrom enum import Enumfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Set, Tuple  class MessageType(Enum):    PRE_PREPARE = "pre-prepare"    PREPARE = "prepare"    COMMIT = "commit"    REQUEST_VOTE = "request_vote"    APPEND_ENTRIES = "append_entries"  class NodeState(Enum):    FOLLOWER = "follower"    CANDIDATE = "candidate"    LEADER = "leader"    BYZANTINE = "byzantine"  @dataclassclass Message:    msg_type: MessageType    sender_id: int    term: int    sequence: int    digest: str    payload: dict = field(default_factory=dict)  @dataclassclass ConsensusResult:    success: bool    latency_ms: float    round_num: int    participating_nodes: int    faulty_nodes: int  class PBFTNode:    """PBFT共识节点实现，支持三阶段协议：pre-prepare、prepare、commit"""     def __init__(self, node_id: int, total_nodes: int, is_byzantine: bool = False):        self.node_id = node_id        self.total_nodes = total_nodes        self.is_byzantine = is_byzantine        self.pre_prepares: Dict[int, Message] = {}        self.prepares: Dict[int, List[Message]] = {}        self.commits: Dict[int, List[Message]] = {}        self.log: List[dict] = []        self.view: int = 0     def _quorum(self) -> int:        """计算拜占庭容错所需的法定数量: 2f+1, 其中 n >= 3f+1"""        f = (self.total_nodes - 1) // 3        return 2 * f + 1     def handle_pre_prepare(self, msg: Message) -> Optional[Message]:        if self.is_byzantine:            return Message(MessageType.PREPARE, self.node_id, msg.term,                           msg.sequence, "fake_digest")        self.pre_prepares[msg.sequence] = msg        return Message(MessageType.PREPARE, self.node_id, msg.term,                       msg.sequence, msg.digest)     def handle_prepare(self, msg: Message) -> Optional[Message]:        seq = msg.sequence        if seq not in self.prepares:            self.prepares[seq] = []        self.prepares[seq].append(msg)        if len(self.prepares[seq]) >= self._quorum() - 1:            return Message(MessageType.COMMIT, self.node_id, msg.term,                           seq, msg.digest)        return None     def handle_commit(self, msg: Message) -> bool:        seq = msg.sequence        if seq not in self.commits:            self.commits[seq] = []        self.commits[seq].append(msg)        if len(self.commits[seq]) >= self._quorum():            self.log.append({"seq": seq, "action": "committed"})            return True        return False  class RaftNode:    """Raft共识节点实现，支持领导者选举与日志复制"""     def __init__(self, node_id: int):        self.node_id = node_id        self.current_term = 0        self.voted_for: Optional[int] = None        self.state = NodeState.FOLLOWER        self.votes_received: Set[int] = set()        self.log: List[dict] = []     def request_votes(self, candidate_id: int, term: int) -> bool:        if term > self.current_term:            self.current_term = term            self.voted_for = None            self.state = NodeState.FOLLOWER        if term >= self.current_term and self.voted_for is None:            self.voted_for = candidate_id            return True        return False     def start_election(self, cluster: List["RaftNode"]) -> bool:        self.current_term += 1        self.state = NodeState.CANDIDATE        self.voted_for = self.node_id        self.votes_received = {self.node_id}        for node in cluster:            if node.node_id != self.node_id:                if node.request_votes(self.node_id, self.current_term):                    self.votes_received.add(node.node_id)        if len(self.votes_received) > len(cluster) // 2:            self.state = NodeState.LEADER            return True        self.state = NodeState.FOLLOWER        return False     def send_heartbeat(self, cluster: List["RaftNode"]) -> int:        if self.state != NodeState.LEADER:            return 0        ack_count = 1        for node in cluster:            if node.node_id != self.node_id:                node.current_term = self.current_term                ack_count += 1        return ack_count  class ConsensusSimulator:    """共识机制模拟器，支持PBFT和Raft，含拜占庭节点模拟"""     def __init__(self, pbft_count: int = 4, raft_count: int = 5,                 byzantine_count: int = 1):        self.pbft_nodes = [            PBFTNode(i, pbft_count, i < byzantine_count)            for i in range(pbft_count)        ]        self.raft_nodes = [RaftNode(i) for i in range(raft_count)]        self.results: List[ConsensusResult] = []     def run_pbft_round(self, round_num: int) -> ConsensusResult:        start = time.time()        seq = round_num        digest = f"request_{round_num}"        primary = self.pbft_nodes[round_num % len(self.pbft_nodes)]        pre_prep = Message(MessageType.PRE_PREPARE, primary.node_id,                           0, seq, digest)        prepare_msgs = []        for node in self.pbft_nodes:            resp = node.handle_pre_prepare(pre_prep)            if resp:                prepare_msgs.append(resp)        commit_msgs = []        for msg in prepare_msgs:            for node in self.pbft_nodes:                resp = node.handle_prepare(msg)                if resp:                    commit_msgs.append(resp)        committed = False        for msg in commit_msgs:            for node in self.pbft_nodes:                if node.handle_commit(msg):                    committed = True        latency = (time.time() - start) * 1000        faulty = sum(1 for n in self.pbft_nodes if n.is_byzantine)        result = ConsensusResult(committed, latency, round_num,                                 len(self.pbft_nodes), faulty)        self.results.append(result)        return result     def run_raft_election(self) -> Tuple[Optional[RaftNode], float]:        start = time.time()        candidate = self.raft_nodes[0]        elected = candidate.start_election(self.raft_nodes)        latency = (time.time() - start) * 1000        if elected:            candidate.send_heartbeat(self.raft_nodes)        return (candidate if elected else None), latency     def run_simulation(self, rounds: int = 10) -> dict:        print(f"=== 共识机制模拟 ({rounds} 轮) ===")        pbft_success = 0        total_pbft_latency = 0.0        for r in range(1, rounds + 1):            result = self.run_pbft_round(r)            if result.success:                pbft_success += 1            total_pbft_latency += result.latency_ms        raft_leader, raft_latency = self.run_raft_election()        print(f"PBFT成功率: {pbft_success}/{rounds}")        print(f"PBFT平均延迟: {total_pbft_latency / rounds:.3f}ms")        print(f"Raft领导者: Node {raft_leader.node_id if raft_leader else 'None'}")        print(f"Raft选举延迟: {raft_latency:.3f}ms")        return {            "pbft_success_rate": pbft_success / rounds,            "pbft_avg_latency_ms": total_pbft_latency / rounds,            "raft_leader_id": raft_leader.node_id if raft_leader else None,            "raft_election_latency_ms": raft_latency,        }  if __name__ == "__main__":    sim = ConsensusSimulator(pbft_count=4, raft_count=5, byzantine_count=1)    stats = sim.run_simulation(rounds=10)    print(f"\n模拟统计: {json.dumps(stats, indent=2, ensure_ascii=False)}")