def perceive(self, environment):
“””感知环境状态”””
pass

def decide(self, perceptions):
“””基于感知做出决策”””
pass

def act(self, decision):
“””执行决策”””
pass

错误的做法：共享状态

class BadDesign:
def __init__(self):
self.shared_state = {} # 所有智能体共享

# 正确的做法：消息传递

class GoodDesign:

    def __init__(self):

        self.message_queue = MessageQueue()

def register(self, agent_id, agent, capabilities):
self.agents[agent_id] = agent
self.capabilities[agent_id] = capabilities

def unregister(self, agent_id):
if agent_id in self.agents:
del self.agents[agent_id]
del self.capabilities[agent_id]

@staticmethod
def direct_message(sender, receiver, content):
“””点对点通信”””
return {
“mode”: “direct”,
“sender”: sender,
“receiver”: receiver,
“content”: content
}

@staticmethod
def broadcast(sender, content, scope=”all”):
“””广播通信”””
return {
“mode”: “broadcast”,
“sender”: sender,
“scope”: scope,
“content”: content
}

def announce_task(self, task_id, task_description, manager):
“””发布任务公告”””
self.tasks[task_id] = {
“description”: task_description,
“manager”: manager,
“status”: “announced”,
“bids”: []
}
# 广播任务公告
self.broadcast(task_id, task_description)

def submit_bid(self, task_id, bidder, bid_details):
“””提交投标”””
if task_id in self.tasks:
self.tasks[task_id][“bids”].append({
“bidder”: bidder,
“details”: bid_details
})

def write(self, section, data, author):
“””写入数据到黑板”””
if section not in self.sections:
self.sections[section] = []

entry = {
“data”: data,
“author”: author,
“timestamp”: datetime.now().isoformat()
}
self.sections[section].append(entry)

# 通知订阅者
self.notify_subscribers(section, entry)

def read(self, section):
“””从黑板读取数据”””
return self.sections.get(section, [])

def subscribe(self, section, agent_id):
“””订阅黑板区域”””
if section not in self.subscribers:
self.subscribers[section] = []
self.subscribers[section].append(agent_id)

def allocate(self, task):
“””基于能力分配任务”””
required_capabilities = task.get(“required_capabilities”, [])
candidates = []

for agent_id, capabilities in self.registry.capabilities.items():
score = self.calculate_match_score(
required_capabilities,
capabilities
)
if score > 0:
candidates.append((agent_id, score))

# 选择匹配度最高的智能体
if candidates:
best_agent = max(candidates, key=lambda x: x[1])
return best_agent[0]

return None

def calculate_match_score(self, required, available):
“””计算能力匹配分数”””
if not required:
return 1.0

def start_auction(self, task_id, task, auctioneer):
“””启动拍卖”””
self.auctions[task_id] = {
“task”: task,
“auctioneer”: auctioneer,
“bids”: {},
“status”: “open”,
“deadline”: datetime.now() + timedelta(minutes=5)
}

def place_bid(self, task_id, bidder, bid_amount):
“””放置投标”””
if task_id in self.auctions:
auction = self.auctions[task_id]
if auction[“status”] == “open”:
auction[“bids”][bidder] = bid_amount

def close_auction(self, task_id):
“””关闭拍卖并确定获胜者”””
if task_id in self.auctions:
auction = self.auctions[task_id]
auction[“status”] = “closed”

if auction[“bids”]:
winner = min(auction[“bids”].items(), key=lambda x: x[1])
return {
“winner”: winner[0],
“amount”: winner[1]
}

def register_agent(self, agent_id, capacity):
“””注册智能体及其容量”””
self.agent_loads[agent_id] = 0
self.agent_capacities[agent_id] = capacity

def get_load_ratio(self, agent_id):
“””获取负载率”””
if agent_id in self.agent_capacities:
return self.agent_loads[agent_id] / self.agent_capacities[agent_id]
return float(“inf”)

def find_best_agent(self):
“””找到负载最低的智能体”””
if not self.agent_loads:
return None

return min(
self.agent_loads.keys(),
key=lambda x: self.get_load_ratio(x)
)

def assign_task(self, agent_id, task_cost):
“””分配任务并更新负载”””
if agent_id in self.agent_loads:
self.agent_loads[agent_id] += task_cost

def complete_task(self, agent_id, task_cost):
“””完成任务并释放负载”””
if agent_id in self.agent_loads:
self.agent_loads[agent_id] = max(0, self.agent_loads[agent_id]

• task_cost)

def submit_proposal(self, negotiation_id, agent_id, proposal):
“””提交提案”””
if negotiation_id not in self.proposals:
self.proposals[negotiation_id] = []

self.proposals[negotiation_id].append({
“agent”: agent_id,
“proposal”: proposal,
“timestamp”: datetime.now().isoformat()
})

def submit_argument(self, negotiation_id, agent_id, argument):
“””提交论据”””
if negotiation_id not in self.arguments:
self.arguments[negotiation_id] = []

self.arguments[negotiation_id].append({
“agent”: agent_id,
“argument”: argument,
“supports”: argument.get(“supports_proposal”),
“strength”: argument.get(“strength”, 1)
})

def evaluate_proposals(self, negotiation_id):
“””评估提案”””
if negotiation_id not in self.proposals:
return None

proposal_scores = {}
for prop in self.proposals[negotiation_id]:
agent_id = prop[“agent”]
score = self.calculate_score(negotiation_id, agent_id)
proposal_scores[agent_id] = score

best_proposal = max(proposal_scores.items(), key=lambda x: x[1])
return best_proposal

def calculate_score(self, negotiation_id, agent_id):
“””计算提案得分”””
base_score = 1.0

if negotiation_id in self.arguments:
for arg in self.arguments[negotiation_id]:
if arg[“agent”] == agent_id:
base_score += arg[“strength”]
elif arg[“supports”] == agent_id:
base_score += arg[“strength”] * 0.5

def start_vote(self, vote_id, options, voters):
“””启动投票”””
self.votes[vote_id] = {
“options”: options,
“results”: {opt: 0 for opt in options},
“voters”: voters,
“status”: “open”
}

def cast_vote(self, vote_id, voter, option):
“””投票”””
if vote_id in self.votes:
vote = self.votes[vote_id]
if voter in vote[“voters”] and option in vote[“options”]:
vote[“results”][option] += 1

def add_robot(self, robot_id, capacity, location):
“””添加机器人”””
self.robots[robot_id] = {
“capacity”: capacity,
“location”: location,
“status”: “idle”,
“current_task”: None
}

def create_pick_task(self, item_id, quantity, destination):
“””创建拣货任务”””
task_id = f”pick_{len(self.tasks)}”
self.tasks[task_id] = {
“type”: “pick”,
“item_id”: item_id,
“quantity”: quantity,
“destination”: destination,
“status”: “pending”,
“assigned_robot”: None
}
return task_id

def coordinate_robots(self, task_id):
“””协调机器人执行任务”””
task = self.tasks.get(task_id)
if not task:
return None

# 找到最近的空闲机器人
best_robot = None
min_distance = float(“inf”)

6.2 分布式监控系统

def register_monitor(self, monitor_id, metrics):
"""注册监控节点"""
self.monitors[monitor_id] = {
"metrics": metrics,
"status": "active",
"last_heartbeat": datetime.now()
}

def report_metric(self, monitor_id, metric_name, value):
"""上报指标"""
if monitor_id in self.monitors:
if metric_name in self.thresholds:
threshold = self.thresholds[metric_name]
if value > threshold[“warning”]:
self.create_alert(monitor_id, metric_name, value, “warning”)
if value > threshold[“critical”]:
self.create_alert(monitor_id, metric_name, value, “critical”)

def create_alert(self, monitor_id, metric_name, value, severity):
“””创建告警”””
alert = {
“id”: f”alert_{len(self.alerts)}”,
“monitor”: monitor_id,
“metric”: metric_name,
“value”: value,
“severity”: severity,
“timestamp”: datetime.now().isoformat(),
“status”: “new”
}
self.alerts.append(alert)
self.broadcast_alert(alert)