在当今AI技术快速发展的背景下，AI agents（智能体）的应用场景越来越广泛。最近接触到一个很有意思的项目——The Email Game，它让多个AI agents通过加密签名邮件进行竞争互动。这种将密码学与AI结合的设计思路，不仅考验agents的智能决策能力，还确保了通信过程的安全性。本文将完整解析这个项目的技术实现，从加密邮件原理到AI agents的竞争机制，带你一步步搭建自己的AI邮件对战系统。

1. 项目背景与核心概念

1.1 什么是AI agents竞争游戏

AI agents竞争游戏是指多个智能体在特定规则下相互博弈的系统。在The Email Game中，每个AI agent代表一个独立的智能实体，它们通过加密签名的电子邮件进行通信和竞争。这种设计模拟了现实世界中的多方协作与竞争场景，比如商业谈判、资源争夺等场景。

与传统AI系统不同，竞争性AI agents需要具备自主决策、策略规划和风险评估能力。每个agent都有自己的目标函数，通过分析邮件内容、评估对手策略来做出最优决策。这种多智能体系统（Multi-Agent System）的研究对于分布式人工智能发展具有重要意义。

1.2 密码学签名邮件的技术价值

密码学签名在邮件系统中的应用确保了通信的真实性和完整性。在AI agents竞争环境中，加密签名解决了几个关键问题：首先是身份认证，确保每封邮件都来自合法的agent身份；其次是防篡改，保证邮件内容在传输过程中不被恶意修改；最后是非否认性，发送方无法否认自己发送过的邮件。

采用加密签名邮件作为通信载体，为AI agents提供了安全可靠的交互通道。这种设计特别适合需要高度信任保障的竞争环境，比如金融交易模拟、合约谈判等敏感场景。

2. 技术架构与环境准备

2.1 系统架构概述

The Email Game的整体架构包含三个核心模块：AI agents决策引擎、邮件处理中间件和密码学签名服务。AI agents决策引擎负责分析邮件内容、制定回复策略；邮件处理中间件管理邮件的收发队列和存储；密码学签名服务处理邮件的加密、解密和验证流程。

系统采用分布式设计，每个AI agent运行在独立的容器中，通过消息队列进行通信。这种架构保证了系统的可扩展性和容错性，即使某个agent出现故障，也不会影响整体系统的运行。

2.2 开发环境要求

要实现类似的AI邮件竞争系统，需要准备以下开发环境：

基础软件要求：

• Python 3.8+ 运行环境
• PostgreSQL或MySQL数据库
• Redis缓存服务
• Docker容器环境

AI相关依赖：

• TensorFlow或PyTorch深度学习框架
• OpenAI GPT API或本地语言模型
• 强化学习库（如Stable-Baselines3）

密码学工具：

• OpenSSL密码学工具包
• GPG密钥管理工具
• 数字证书生成工具

2.3 项目目录结构

标准的项目目录结构应该清晰划分各个功能模块：

email_game/ ├── agents/ # AI agents核心逻辑 │ ├── base_agent.py # 基类定义 │ ├── strategy/ # 策略实现 │ └── models/ # 机器学习模型 ├── crypto/ # 密码学模块 │ ├── signature.py # 签名验证 │ ├── encryption.py # 加密解密 │ └── keys/ # 密钥管理 ├── email/ # 邮件处理 │ ├── client.py # 邮件客户端 │ ├── parser.py # 邮件解析 │ └── storage.py # 邮件存储 ├── game/ # 游戏逻辑 │ ├── rules.py # 规则引擎 │ ├── scoring.py # 评分系统 │ └── monitor.py # 监控面板 └── config/ # 配置文件 ├── development.yaml ├── production.yaml └── agents.yaml

3. 密码学签名邮件实现

3.1 数字签名原理与实现

数字签名基于非对称加密技术，使用私钥签名、公钥验证的模式。在Python中可以使用cryptography库实现：

from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import rsa, padding from cryptography.hazmat.backends import default_backend import base64 class EmailSigner: def __init__(self, private_key_path=None, public_key_path=None): self.private_key = None self.public_key = None if private_key_path and public_key_path: self.load_keys(private_key_path, public_key_path) else: self.generate_keys() def generate_keys(self, key_size=2048): """生成RSA密钥对""" self.private_key = rsa.generate_private_key( public_exponent=65537, key_size=key_size, backend=default_backend() ) self.public_key = self.private_key.public_key() def sign_email(self, email_content): """对邮件内容进行数字签名""" if not self.private_key: raise ValueError("私钥未初始化") # 对邮件内容进行哈希 hasher = hashes.Hash(hashes.SHA256(), backend=default_backend()) hasher.update(email_content.encode('utf-8')) digest = hasher.finalize() # 使用私钥签名 signature = self.private_key.sign( digest, padding.PSS( mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH ), hashes.SHA256() ) return base64.b64encode(signature).decode('utf-8') def verify_signature(self, email_content, signature, public_key=None): """验证数字签名""" verifying_key = public_key or self.public_key if not verifying_key: raise ValueError("公钥未提供") try: # 计算内容哈希 hasher = hashes.Hash(hashes.SHA256(), backend=default_backend()) hasher.update(email_content.encode('utf-8')) digest = hasher.finalize() # 验证签名 signature_bytes = base64.b64decode(signature) verifying_key.verify( signature_bytes, digest, padding.PSS( mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH ), hashes.SHA256() ) return True except Exception as e: print(f"签名验证失败: {e}") return False

3.2 邮件加密与安全传输

除了签名验证，邮件内容加密也是确保安全性的重要环节。下面实现AES对称加密与RSA非对称加密结合的方案：

import os from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives import padding as sym_padding from cryptography.hazmat.primitives import serialization class EmailEncryptor: def __init__(self): self.aes_key_size = 32 # AES-256 def generate_aes_key(self): """生成随机的AES密钥""" return os.urandom(self.aes_key_size) def encrypt_email(self, email_content, recipient_public_key): """使用混合加密方式加密邮件""" # 生成随机的AES密钥 aes_key = self.generate_aes_key() # 使用AES加密邮件内容 iv = os.urandom(16) # 初始化向量 padder = sym_padding.PKCS7(128).padder() padded_data = padder.update(email_content.encode()) + padder.finalize() cipher = Cipher(algorithms.AES(aes_key), modes.CBC(iv)) encryptor = cipher.encryptor() encrypted_content = encryptor.update(padded_data) + encryptor.finalize() # 使用接收方的公钥加密AES密钥 encrypted_key = recipient_public_key.encrypt( aes_key, padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None ) ) return { 'encrypted_content': base64.b64encode(encrypted_content).decode(), 'encrypted_key': base64.b64encode(encrypted_key).decode(), 'iv': base64.b64encode(iv).decode() }

4. AI Agents竞争机制设计

4.1 Agent决策引擎架构

每个AI agent的核心是一个决策引擎，它需要处理邮件内容分析、策略制定和行动选择。下面是基础决策引擎的实现：

import numpy as np from typing import Dict, List, Any from abc import ABC, abstractmethod class BaseAgent(ABC): def __init__(self, agent_id: str, config: Dict[str, Any]): self.agent_id = agent_id self.config = config self.memory = [] # 对话记忆 self.score = 0 # 当前得分 @abstractmethod def analyze_email(self, email_content: str) -> Dict[str, Any]: """分析接收到的邮件内容""" pass @abstractmethod def formulate_response(self, analysis_result: Dict[str, Any]) -> str: """制定回复策略""" pass @abstractmethod def update_strategy(self, game_state: Dict[str, Any]): """根据游戏状态更新策略""" pass class StrategicAgent(BaseAgent): def __init__(self, agent_id: str, config: Dict[str, Any]): super().__init__(agent_id, config) self.strategy_model = self._load_strategy_model() def analyze_email(self, email_content: str) -> Dict[str, Any]: """深度分析邮件内容和意图""" analysis = { 'sender_intent': self._detect_intent(email_content), 'urgency_level': self._assess_urgency(email_content), 'emotional_tone': self._analyze_tone(email_content), 'key_points': self._extract_key_points(email_content), 'potential_traps': self._detect_traps(email_content) } return analysis def formulate_response(self, analysis_result: Dict[str, Any]) -> str: """基于多因素决策制定回复""" strategy = self._select_strategy(analysis_result) response_template = self._choose_response_template(strategy) customized_response = self._customize_response(response_template, analysis_result) return customized_response def _select_strategy(self, analysis: Dict[str, Any]) -> str: """根据分析结果选择应对策略""" if analysis['potential_traps']: return 'defensive' elif analysis['urgency_level'] == 'high': return 'responsive' else: return 'strategic'

4.2 多智能体竞争算法

竞争环境中的AI agents需要采用博弈论算法来优化决策。下面是基于Q学习的竞争策略实现：

import numpy as np from collections import defaultdict class CompetitiveQLearning: def __init__(self, learning_rate=0.1, discount_factor=0.9, exploration_rate=0.1): self.q_table = defaultdict(lambda: defaultdict(float)) self.learning_rate = learning_rate self.discount_factor = discount_factor self.exploration_rate = exploration_rate def choose_action(self, state, available_actions): """根据当前状态选择行动""" if np.random.random() < self.exploration_rate: # 探索：随机选择行动 return np.random.choice(available_actions) else: # 利用：选择Q值最高的行动 q_values = [self.q_table[state][action] for action in available_actions] max_q = max(q_values) # 如果多个行动有相同Q值，随机选择 actions_with_max_q = [action for action, q in zip(available_actions, q_values) if q == max_q] return np.random.choice(actions_with_max_q) def update_q_value(self, state, action, reward, next_state, next_available_actions): """更新Q值表""" if next_available_actions: max_next_q = max([self.q_table[next_state][next_action] for next_action in next_available_actions]) else: max_next_q = 0 current_q = self.q_table[state][action] new_q = current_q + self.learning_rate * (reward + self.discount_factor * max_next_q - current_q) self.q_table[state][action] = new_q class GameMaster: def __init__(self, agents: List[BaseAgent], rules: Dict[str, Any]): self.agents = {agent.agent_id: agent for agent in agents} self.rules = rules self.game_state = self._initialize_game_state() self.q_learners = {agent_id: CompetitiveQLearning() for agent_id in self.agents.keys()} def process_round(self, sender_id: str, receiver_id: str, email_content: str): """处理一轮邮件交互""" # 验证邮件签名 if not self._verify_email_signature(sender_id, email_content): return "签名验证失败" # 接收方分析邮件 receiver_agent = self.agents[receiver_id] analysis = receiver_agent.analyze_email(email_content) # 根据当前状态选择回复策略 current_state = self._get_game_state_hash() available_actions = self._get_available_actions(receiver_id) chosen_action = self.q_learners[receiver_id].choose_action(current_state, available_actions) response_content = receiver_agent.formulate_response(analysis, chosen_action) # 计算奖励并更新Q值 reward = self._calculate_reward(receiver_id, analysis, chosen_action) next_state = self._get_game_state_hash() next_actions = self._get_available_actions(receiver_id) self.q_learners[receiver_id].update_q_value(current_state, chosen_action, reward, next_state, next_actions) return response_content

5. 完整系统集成实战

5.1 系统配置与初始化

首先创建系统的主配置文件，定义agents参数、游戏规则和邮件服务器设置：

# config/game_config.yaml game: name: "AI Email Competition" max_rounds: 100 scoring_system: successful_communication: 10 strategic_advantage: 20 trap_avoidance: 15 penalty_miscommunication: -10 email: smtp_server: "smtp.example.com" smtp_port: 587 use_tls: true check_interval: 30 # 秒 agents: agent1: type: "strategic" personality: "aggressive" learning_rate: 0.1 public_key_path: "keys/agent1_public.pem" agent2: type: "cooperative" personality: "cautious" learning_rate: 0.05 public_key_path: "keys/agent2_public.pem" crypto: algorithm: "RSA" key_size: 2048 hash_algorithm: "SHA256"

5.2 主控制系统实现

主控制系统负责协调各个模块的工作，管理游戏流程和agent交互：

import asyncio import yaml from datetime import datetime from typing import Dict, List import smtplib from email.mime.text import MIMEText class EmailGameController: def __init__(self, config_path: str): self.config = self._load_config(config_path) self.agents = self._initialize_agents() self.game_master = GameMaster(list(self.agents.values()), self.config['game']) self.email_client = EmailClient(self.config['email']) self.running = False async def start_game(self): """启动游戏主循环""" self.running = True print(f"游戏开始于 {datetime.now()}") while self.running and self.game_master.current_round < self.config['game']['max_rounds']: await self._process_game_round() await asyncio.sleep(self.config['email']['check_interval']) await self._end_game() async def _process_game_round(self): """处理单个游戏回合""" # 检查新邮件 new_emails = await self.email_client.fetch_new_emails() for email in new_emails: # 验证邮件签名和解析发送方 sender_id = self._extract_sender_id(email) if sender_id not in self.agents: continue # 处理邮件内容 response = self.game_master.process_round( sender_id, self._determine_receiver(sender_id), email['content'] ) # 发送回复 if response: await self._send_response(sender_id, response) # 更新游戏状态 self.game_master.update_scores() self._log_round_status() def _initialize_agents(self) -> Dict[str, BaseAgent]: """初始化所有AI agents""" agents = {} for agent_id, agent_config in self.config['agents'].items(): if agent_config['type'] == 'strategic': agents[agent_id] = StrategicAgent(agent_id, agent_config) elif agent_config['type'] == 'cooperative': agents[agent_id] = CooperativeAgent(agent_id, agent_config) # 加载其他类型的agents... return agents # 启动游戏 async def main(): controller = EmailGameController('config/game_config.yaml') await controller.start_game() if __name__ == "__main__": asyncio.run(main())

5.3 邮件客户端实现

实现支持加密签名的邮件客户端，处理邮件的发送和接收：

import aiosmtplib import imaplib import email from email.header import decode_header class SecureEmailClient: def __init__(self, config: Dict[str, Any]): self.smtp_config = config['smtp'] self.imap_config = config['imap'] self.signer = EmailSigner() async def send_secure_email(self, to_address: str, subject: str, content: str, sender_id: str) -> bool: """发送加密签名邮件""" try: # 对内容进行数字签名 signature = self.signer.sign_email(content) # 构建安全邮件头 secure_headers = { 'X-Agent-ID': sender_id, 'X-Signature': signature, 'X-Timestamp': datetime.now().isoformat() } # 创建邮件消息 message = MIMEText(content) message['Subject'] = subject message['From'] = f"{sender_id}@game.system" message['To'] = to_address for header, value in secure_headers.items(): message[header] = value # 发送邮件 async with aiosmtplib.SMTP( hostname=self.smtp_config['host'], port=self.smtp_config['port'] ) as smtp: await smtp.login( self.smtp_config['username'], self.smtp_config['password'] ) await smtp.send_message(message) return True except Exception as e: print(f"发送邮件失败: {e}") return False async def fetch_secure_emails(self) -> List[Dict]: """获取并验证安全邮件""" emails = [] try: with imaplib.IMAP4_SSL(self.imap_config['host']) as mail: mail.login(self.imap_config['username'], self.imap_config['password']) mail.select('inbox') status, messages = mail.search(None, 'UNSEEN') email_ids = messages[0].split() for email_id in email_ids: status, msg_data = mail.fetch(email_id, '(RFC822)') email_message = email.message_from_bytes(msg_data[0][1]) # 验证邮件签名 if self._verify_email_signature(email_message): email_data = { 'id': email_id, 'sender': self._extract_sender(email_message), 'subject': self._decode_header(email_message['Subject']), 'content': self._extract_content(email_message), 'signature': email_message['X-Signature'], 'agent_id': email_message['X-Agent-ID'] } emails.append(email_data) except Exception as e: print(f"获取邮件失败: {e}") return emails

6. 高级功能与优化策略

6.1 自适应学习机制

为了让AI agents在竞争环境中持续进化，需要实现自适应学习机制：

class AdaptiveLearningManager: def __init__(self, agents: Dict[str, BaseAgent]): self.agents = agents self.performance_history = defaultdict(list) def analyze_agent_performance(self, agent_id: str, recent_rounds: int = 10): """分析agent近期表现""" history = self.performance_history[agent_id][-recent_rounds:] if not history: return None avg_score = np.mean([h['score'] for h in history]) success_rate = np.mean([1 if h['success'] else 0 for h in history]) strategy_effectiveness = self._calculate_strategy_effectiveness(history) return { 'average_score': avg_score, 'success_rate': success_rate, 'strategy_effectiveness': strategy_effectiveness, 'improvement_trend': self._detect_improvement_trend(history) } def optimize_agent_parameters(self, agent_id: str): """根据表现优化agent参数""" performance = self.analyze_agent_performance(agent_id) if not performance: return agent = self.agents[agent_id] # 根据表现调整学习率 if performance['improvement_trend'] < 0: # 表现下降 agent.learning_rate *= 1.1 # 增加探索 elif performance['improvement_trend'] > 0.1: # 稳定提升 agent.learning_rate *= 0.9 # 减少探索 # 调整策略权重 if performance['strategy_effectiveness']['defensive'] < 0.5: agent.defensive_strategy_weight += 0.1

6.2 多维度评分系统

完善的评分系统能够准确反映agents的竞争表现：

class ComprehensiveScoringSystem: def __init__(self, config: Dict[str, Any]): self.scoring_rules = config['scoring_system'] self.weight_factors = config.get('weight_factors', {}) def calculate_round_score(self, agent_id: str, round_data: Dict) -> float: """计算单回合得分""" base_score = 0 # 通信有效性得分 if round_data['communication_successful']: base_score += self.scoring_rules['successful_communication'] # 策略优势得分 strategic_advantage = self._assess_strategic_advantage(round_data) base_score += strategic_advantage * self.scoring_rules['strategic_advantage'] # 陷阱规避得分 if round_data['trap_avoided']: base_score += self.scoring_rules['trap_avoidance'] # 惩罚错误通信 if round_data['miscommunication']: base_score += self.scoring_rules['penalty_miscommunication'] # 应用权重因子 weighted_score = base_score * self._calculate_weight_factor(agent_id, round_data) return max(0, weighted_score) # 确保分数不为负 def _assess_strategic_advantage(self, round_data: Dict) -> float: """评估策略优势程度""" advantage_score = 0 # 基于对手反应评估 opponent_reaction = round_data.get('opponent_reaction', 'neutral') if opponent_reaction == 'defensive': advantage_score += 0.7 elif opponent_reaction == 'confused': advantage_score += 0.9 # 基于达成目标评估 objectives_achieved = round_data.get('objectives_achieved', 0) advantage_score += objectives_achieved * 0.3 return min(1.0, advantage_score) # 限制在0-1范围内

7. 部署与监控方案

7.1 容器化部署配置

使用Docker容器化部署确保环境一致性：

# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ build-essential \ libssl-dev \ && rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . RUN pip install -r requirements.txt # 复制应用代码 COPY . . # 创建非root用户 RUN useradd -m -u 1000 agentuser USER agentuser # 暴露监控端口 EXPOSE 8080 # 启动命令 CMD ["python", "main.py"]

对应的Docker Compose配置：

# docker-compose.yml version: '3.8' services: email-game: build: . ports: - "8080:8080" volumes: - ./config:/app/config - ./data:/app/data environment: - PYTHONPATH=/app - GAME_ENV=production depends_on: - redis - postgres redis: image: redis:6.2-alpine ports: - "6379:6379" volumes: - redis_data:/data postgres: image: postgres:13 environment: POSTGRES_DB: emailgame POSTGRES_USER: agent POSTGRES_PASSWORD: securepassword volumes: - postgres_data:/var/lib/postgresql/data volumes: redis_data: postgres_data:

7.2 系统监控与日志

实现全面的系统监控和日志记录：

import logging from prometheus_client import Counter, Gauge, Histogram, start_http_server class MonitoringSystem: def __init__(self, port=8080): self.setup_metrics() self.setup_logging() start_http_server(port) def setup_metrics(self): """设置Prometheus监控指标""" self.emails_sent = Counter('emails_sent_total', 'Total emails sent') self.emails_received = Counter('emails_received_total', 'Total emails received') self.agent_scores = Gauge('agent_scores', 'Current agent scores', ['agent_id']) self.response_time = Histogram('response_time_seconds', 'Response time histogram') def setup_logging(self): """配置结构化日志""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('game_system.log'), logging.StreamHandler() ] ) self.logger = logging.getLogger('EmailGame') def log_round_completion(self, round_number: int, scores: Dict[str, float]): """记录回合完成信息""" self.logger.info(f"Round {round_number} completed", extra={ 'scores': scores, 'round': round_number }) for agent_id, score in scores.items(): self.agent_scores.labels(agent_id=agent_id).set(score)

8. 常见问题与解决方案

8.1 密码学相关问题

问题1：签名验证失败

• 现象：邮件签名验证 consistently 失败
• 原因：时钟不同步、密钥不匹配、编码问题
• 解决方案：
  1. 检查系统时间同步：ntpdate pool.ntp.org
  2. 验证密钥对匹配性：重新生成并分发密钥
  3. 统一字符编码：确保使用UTF-8编码

问题2：加密邮件解密失败

• 现象：接收方无法解密邮件内容
• 原因：密钥交换问题、算法不匹配、数据损坏
• 解决方案：
  1. 实现密钥交换协议：使用Diffie-Hellman密钥交换
  2. 检查加密算法一致性：统一使用AES-256-CBC
  3. 添加数据完整性校验：包含HMAC验证

8.2 AI Agents行为异常

问题1：Agent陷入局部最优

• 现象：Agent重复使用相同策略，缺乏创新
• 解决方案：
  • 增加探索率：动态调整ε-greedy参数
  • 引入策略多样性：定期注入随机策略
  • 实现课程学习：从简单到复杂逐步训练

问题2：通信僵局

• 现象：Agents陷入无限循环的无效通信
• 解决方案：
  • 设置最大回合数：强制结束僵局回合
  • 引入第三方调解：Game Master介入调解
  • 实现超时机制：长时间无进展自动跳过

8.3 性能优化问题

问题1：邮件处理延迟

• 现象：系统响应时间逐渐变长
• 解决方案：
  • 实现邮件队列：使用Redis队列管理邮件流
  • 优化数据库查询：添加适当索引
  • 使用连接池：管理数据库和邮件服务器连接

问题2：内存泄漏

• 现象：系统运行时间越长内存占用越高
• 解决方案：
  • 定期清理缓存：实现LRU缓存策略
  • 监控对象生命周期：使用内存分析工具
  • 限制历史数据：自动归档旧邮件数据

9. 安全最佳实践

9.1 密钥管理安全

密钥管理是系统安全的核心，必须遵循最小权限原则：

class SecureKeyManager: def __init__(self, key_storage_path: str): self.storage_path = key_storage_path self.encryption_key = self._load_encryption_key() def store_private_key(self, agent_id: str, private_key: bytes) -> bool: """安全存储私钥""" try: # 加密私钥 encrypted_key = self._encrypt_key(private_key) # 安全存储 key_path = os.path.join(self.storage_path, f"{agent_id}.key.enc") with open(key_path, 'wb') as f: f.write(encrypted_key) # 设置严格的文件权限 os.chmod(key_path, 0o600) return True except Exception as e: logging.error(f"存储私钥失败: {e}") return False def _encrypt_key(self, key_data: bytes) -> bytes: """使用主密钥加密密钥数据""" # 实现AES-GCM加密确保机密性和完整性 iv = os.urandom(12) # GCM推荐12字节IV cipher = Cipher(algorithms.AES(self.encryption_key), modes.GCM(iv)) encryptor = cipher.encryptor() encrypted_data = encryptor.update(key_data) + encryptor.finalize() return iv + encryptor.tag + encrypted_data

9.2 通信安全加固

确保邮件通信过程中的数据安全：

1. 传输层安全：强制使用TLS 1.2+加密SMTP/IMAP连接
2. 内容安全：实现端到端加密，避免中间人攻击
3. 身份验证：使用双因素认证增强账户安全
4. 审计日志：记录所有安全相关事件用于事后分析

9.3 系统安全监控

实现实时安全监控和告警：

class SecurityMonitor: def __init__(self): self.suspicious_activities = [] self.alert_threshold = 5 # 触发告警的阈值 def monitor_activity(self, activity_type: str, agent_id: str, details: Dict): """监控安全相关活动""" if self._is_suspicious(activity_type, details): self.suspicious_activities.append({ 'timestamp': datetime.now(), 'agent_id': agent_id, 'activity_type': activity_type, 'details': details }) if len(self.suspicious_activities) >= self.alert_threshold: self._trigger_security_alert() def _is_suspicious(self, activity_type: str, details: Dict) -> bool: """判断活动是否可疑""" suspicious_patterns = { 'multiple_failed_logins': details.get('failed_attempts', 0) > 3, 'unusual_sending_pattern': details.get('emails_per_minute', 0) > 10, 'signature_verification_failures': details.get('failed_verifications', 0) > 5 } return suspicious_patterns.get(activity_type, False)

通过以上完整实现，我们构建了一个安全可靠的AI agents加密邮件竞争系统。这个系统不仅展示了AI与密码学的结合应用，还为多智能体系统研究提供了实用的实验平台。在实际部署时，建议先在测试环境中充分验证各项功能，逐步扩展到生产环境。