《AI Agent智能体开发实践 邓立国 邓淇文著 五大实战案例掌握AI Agent开发 LangChain示例 人工智能技术丛书 清华大学出版社》【摘要 书评 试读】- 京东图书
多Agent协同（Multi-Agent Collaboration，MAS）是指多个具备自主决策能力的智能体（Agent）通过通信、任务分配与协同机制，共同完成复杂目标的技术体系。其核心在于突破单个Agent的能力局限，通过群体智能实现更高阶的自动化。
3.3.1 角色分工：定义不同Agent的职能
1. 多Agent系统概述
多Agent系统是由多个自主或半自主的智能Agent组成的系统，这些Agent通过协作来完成单个Agent难以完成的复杂任务。在多Agent系统中，角色分工是关键，不同的Agent承担不同的职责，通过协同工作提高整体效率。
2. Agent角色分工
1）分析师（Analyst）Agent
职责：数据收集、处理、分析和模式识别。
特点：强大的数据处理和推理能力。
2）执行者（Executor）Agent
职责：执行具体任务和操作。
特点：高效的执行能力和操作接口。
3）协调者（Coordinator）Agent
职责：任务分配、资源调度和冲突解决。
特点：全局视角和决策能力。
4）接口（Interface）Agent
职责：与用户或其他系统交互。
特点：良好的交互能力和信息转换能力。
5）监控（Monitor）Agent
职责：系统状态监测和异常检测。
特点：实时性和预警能力。
3. Python实现案例
【示例3.7】实现一个包含分析师Agent和执行者Agent的简单股票分析系统。
#1. 基础架构 from abc import ABC, abstractmethod from typing import Dict, Any import random import time class Agent(ABC): """Agent抽象基类""" def __init__(self, name: str): self.name = name self.knowledge_base = {} @abstractmethod def perform_task(self, task: Dict[str, Any]) -> Dict[str, Any]: """执行任务并返回结果""" pass def communicate(self, message: str) -> str: """简单的通信方法""" print(f"{self.name} received: {message}") return f"{self.name} acknowledged: {message}" #2. 分析师Agent实现 class AnalystAgent(Agent): """分析师Agent，负责数据分析""" def __init__(self, name: str): super().__init__(name) # 初始化分析工具或模型 self.analysis_tools = ["趋势分析", "技术指标", "模式识别"] def perform_task(self, task: Dict[str, Any]) -> Dict[str, Any]: """执行分析任务""" print(f"{self.name} is analyzing data...") # 模拟分析过程 data = task.get("data", {}) analysis_type = task.get("analysis_type", "趋势分析") # 模拟分析耗时 time.sleep(1) # 生成分析结果 result = { "analysis_type": analysis_type, "result": f"{analysis_type}结果: {random.choice(['买入', '卖出', '持有'])}", "confidence": random.uniform(0.5, 1.0), "factors_considered": random.sample(["价格", "成交量", "市场情绪", "宏观经济"], 2) } print(f"{self.name} completed analysis with result: {result}") return result def train_model(self, dataset): """模拟模型训练方法""" print(f"{self.name} is training model with {len(dataset)} samples...") time.sleep(2) return {"status": "success", "accuracy": random.uniform(0.7, 0.95)} #3. 执行者Agent实现 class ExecutorAgent(Agent): """执行者Agent，负责执行交易操作""" def __init__(self, name: str): super().__init__(name) self.available_actions = ["买入", "卖出", "调整仓位", "取消订单"] def perform_task(self, task: Dict[str, Any]) -> Dict[str, Any]: """执行交易任务""" print(f"{self.name} is executing trade...") action = task.get("action", "买入") symbol = task.get("symbol", "UNKNOWN") quantity = task.get("quantity", 0) # 验证操作是否可用 if action not in self.available_actions: return {"status": "error", "message": f"Unsupported action: {action}"} # 模拟执行耗时 time.sleep(0.5) # 生成执行结果 result = { "status": "success", "action": action, "symbol": symbol, "quantity": quantity, "execution_price": round(random.uniform(100, 200), 2), "timestamp": time.strftime("%Y-%m-%d %H:%M:%S") } print(f"{self.name} executed {action} {quantity} shares of {symbol}") return result def check_balance(self): """检查账户余额""" return { "cash": round(random.uniform(10000, 50000), 2), "positions": {f"STOCK_{i}": random.randint(0, 100) for i in range(1, 4)} } #4. 协调者Agent实现 class CoordinatorAgent(Agent): """协调者Agent，负责任务分配和协调""" def __init__(self, name: str): super().__init__(name) self.registered_agents = {} def register_agent(self, agent: Agent): """注册Agent到系统中""" self.registered_agents[agent.name] = agent print(f"Agent {agent.name} registered with coordinator {self.name}") def perform_task(self, task: Dict[str, Any]) -> Dict[str, Any]: """分配任务给合适的Agent""" task_type = task.get("type", "") if "analysis" in task_type.lower(): agent = self.registered_agents.get("Analyst") elif "execute" in task_type.lower(): agent = self.registered_agents.get("Executor") else: return {"status": "error", "message": "Unknown task type"} if not agent: return {"status": "error", "message": "No suitable agent available"} print(f"{self.name} is delegating task to {agent.name}") return agent.perform_task(task) def broadcast(self, message: str): """广播消息给所有注册的Agent""" responses = {} for name, agent in self.registered_agents.items(): responses[name] = agent.communicate(message) return responses #5. 系统集成与运行示例 def main(): # 创建Agent实例 coordinator = CoordinatorAgent("Coordinator") analyst = AnalystAgent("Analyst") executor = ExecutorAgent("Executor") # 注册Agent coordinator.register_agent(analyst) coordinator.register_agent(executor) # 模拟分析任务 print("\n=== Running Analysis Task ===") analysis_task = { "type": "stock_analysis", "data": {"symbol": "AAPL", "history": "1Y"}, "analysis_type": "技术指标" } analysis_result = coordinator.perform_task(analysis_task) print("Analysis Result:", analysis_result) # 模拟执行任务 print("\n=== Running Execution Task ===") execution_task = { "type": "execute_trade", "action": "买入", "symbol": "AAPL", "quantity": 100, "analysis_reference": analysis_result } execution_result = coordinator.perform_task(execution_task) print("Execution Result:", execution_result) # 测试广播通信 print("\n=== Testing Broadcast Communication ===") broadcast_responses = coordinator.broadcast("System check at " + time.strftime("%H:%M:%S")) for agent, response in broadcast_responses.items(): print(f"{agent}: {response}") # 检查执行者状态 print("\n=== Checking Executor Status ===") print("Account Balance:", executor.check_balance()) if __name__ == "__main__": main()运行代码，输出如下：
Agent Analyst registered with coordinator Coordinator Agent Executor registered with coordinator Coordinator === Running Analysis Task === Coordinator is delegating task to Analyst Analyst is analyzing data... Analyst completed analysis with result: {'analysis_type': '技术指标', 'result': '技术指标结果: 买入', 'confidence': 0.878509121476005, 'factors_considered': ['价格', '成交量']} Analysis Result: {'analysis_type': '技术指标', 'result': '技术指标结果: 买入', 'confidence': 0.878509121476005, 'factors_considered': ['价格', '成交量']} === Running Execution Task === Coordinator is delegating task to Executor Executor is executing trade... Executor executed 买入 100 shares of AAPL Execution Result: {'status': 'success', 'action': '买入', 'symbol': 'AAPL', 'quantity': 100, 'execution_price': 109.74, 'timestamp': '2025-07-03 16:37:13'} === Testing Broadcast Communication === Analyst received: System check at 16:37:13 Executor received: System check at 16:37:13 Analyst: Analyst acknowledged: System check at 16:37:13 Executor: Executor acknowledged: System check at 16:37:13 === Checking Executor Status === Account Balance: {'cash': 19739.02, 'positions': {'STOCK_1': 88, 'STOCK_2': 94, 'STOCK_3': 54}}这个框架提供了多Agent系统的基本结构和实现示例，可以根据具体应用场景进行扩展和优化。
3.3.2 通信协议：基于自然语言或结构化消息
多Agent协同通信协议是实现多个智能体之间高效协作的关键技术，可以基于自然语言或结构化消息进行通信。
1. 自然语言处理
自然语言通信依赖于强大的自然语言处理（NLP）技术，包括语言理解、语义解析和语言生成。例如，使用Transformer架构的模型（如GPT）可以处理复杂的语言交互。
【示例3.8】基于DeepSeek自然语言的Agent通信示例，使用Transformers库实现语言生成和理解。
此示例在运行过程中，经常遇到无法连接到Hugging Face的模型仓库的错误，这是由于网络问题或访问限制引起的，以下是几种解决方案：
- 检查网络连接，确保能访问Hugging Face网站。
- 如果在国内，可以使用国内镜像站点。
提前手动下载模型到本地，然后从本地加载模型。
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline import torch import os # ----------------------------- # 配置：选择模型加载方式 # ----------------------------- USE_MIRROR = True # 是否使用国内镜像 LOCAL_MODEL_PATH = "./local_deepseek_model" # 本地模型路径，如果存在，则从本地加载 LOCAL_CLASSIFIER_PATH = "./local_classifier_model" # 本地分类器模型路径 # 国内镜像设置 if USE_MIRROR: os.environ["HF_ENDPOINT"] = "https://hf-mirror.com" # ----------------------------- # 1. 加载 DeepSeek 模型和 tokenizer # ----------------------------- model_name = "deepseek-ai/deepseek-llm-1.3b-chat" # 检查本地是否有模型 if os.path.exists(LOCAL_MODEL_PATH): print(f"从本地加载模型: {LOCAL_MODEL_PATH}") model_path = LOCAL_MODEL_PATH else: model_path = model_name try: tokenizer = AutoTokenizer.from_pretrained( model_path, trust_remote_code=True, local_files_only=os.path.exists(LOCAL_MODEL_PATH) # 如果是本地路径，则只检查本地文件 ) model = AutoModelForCausalLM.from_pretrained( model_path, torch_dtype=torch.bfloat16 if torch.cuda.is_available() else torch.float32, device_map="auto" if torch.cuda.is_available() else None, trust_remote_code=True, local_files_only=os.path.exists(LOCAL_MODEL_PATH) ) except Exception as e: print(f"模型加载失败: {e}") print("尝试使用更小的模型或检查网络连接") # 可以在这里提供备选模型 exit(1) # 生成管道（用于非流式生成） generator = pipeline( "text-generation", model=model, tokenizer=tokenizer, device=0 if torch.cuda.is_available() else -1 ) # ----------------------------- # 2. 意图理解 # ----------------------------- classifier_name = "uer/roberta-base-finetuned-dianping-chinese" # 检查本地是否有分类器模型 if os.path.exists(LOCAL_CLASSIFIER_PATH): print(f"从本地加载分类器模型: {LOCAL_CLASSIFIER_PATH}") classifier_path = LOCAL_CLASSIFIER_PATH else: classifier_path = classifier_name try: intent_classifier = pipeline( "text-classification", model=classifier_path, tokenizer=classifier_path, local_files_only=os.path.exists(LOCAL_CLASSIFIER_PATH) ) except Exception as e: print(f"分类器模型加载失败: {e}") print("将仅使用关键词意图识别") # 手动关键词意图识别（中文） def detect_intent(text): if any(w in text for w in ["你好", "嗨", "早上好", "下午好"]): return "问候" elif any(w in text for w in ["帮助", "帮忙", "怎么做", "如何"]): return "请求帮助" elif any(w in text for w in ["是", "对", "没错", "提供", "信息", "有"]): return "提供信息" elif any(w in text for w in ["再见", "拜拜", "结束", "退出"]): return "告别" else: return "其他" # ----------------------------- # 3. 定义支持 DeepSeek 格式的 Agent # ----------------------------- class DeepSeekAgent: def __init__(self, name, model, tokenizer): self.name = name self.model = model self.tokenizer = tokenizer self.history = [] # 存储对话历史 def send(self, message): print(f"\033[92m{self.name}: {message}\033[0m") # 绿色输出 return message def receive(self, message): print(f"\033[94m{self.name} 收到: {message}\033[0m") # 蓝色 # 意图识别 intent = detect_intent(message) print(f" 🔍 意图识别: {intent}") # 构造 DeepSeek 的对话输入格式 prompt_lines = [] # 添加历史 for i in range(0, len(self.history), 2): if i + 1 < len(self.history): prompt_lines.append(f"<｜User｜>:{self.history[i]}<｜Assistant｜>: {self.history[i+1]}") else: prompt_lines.append(f"<｜User｜>:{self.history[i]}") # 添加当前消息 prompt_lines.append(f"<｜User｜>:{message}<｜Assistant｜>:") full_prompt = "\n".join(prompt_lines) try: # Tokenize 并生成 inputs = self.tokenizer(full_prompt, return_tensors="pt").to(self.model.device) with torch.no_grad(): outputs = self.model.generate( **inputs, max_new_tokens=128, temperature=0.7, top_p=0.9, do_sample=True, pad_token_id=self.tokenizer.eos_token_id, eos_token_id=self.tokenizer.eos_token_id ) # 解码回复 response = self.tokenizer.decode(outputs[0][inputs['input_ids'].shape[-1]:], skip_special_tokens=True) # 清理输出 reply = response.strip().split("\n")[0] reply = reply.replace("<｜Assistant｜>:", "").strip() # 更新历史 self.history.append(message) self.history.append(reply) return reply except Exception as e: print(f"生成回复时出错: {e}") return "抱歉，我无法生成回复。" # ----------------------------- # 4. 创建两个 Agent # ----------------------------- agent_a = DeepSeekAgent("AgentA", model, tokenizer) agent_b = DeepSeekAgent("AgentB", model, tokenizer) # ----------------------------- # 5. 开始多轮对话 # ----------------------------- print("🗣 基于 DeepSeek 的 Agent 自然语言通信启动...\n") # AgentA 发起 msg = "你好，我是AgentA，我们可以一起讨论一个合作计划吗？" msg = agent_a.send(msg) try: for i in range(4): # 4 轮来回 print("-" * 50) reply = agent_b.receive(msg) msg = agent_b.send(reply) print("-" * 50) reply = agent_a.receive(msg) msg = agent_a.send(reply) except KeyboardInterrupt: print("\n对话已手动终止") except Exception as e: print(f"对话过程中出错: {e}")运行代码，输出如下：
基于 DeepSeek 的 Agent 自然语言通信启动... AgentA: 你好，我是AgentA，我们可以一起讨论一个合作计划吗？ -------------------------------------------------- AgentB 收到: 你好，我是AgentA，我们可以一起讨论一个合作计划吗？ 意图识别: 问候 AgentB: 当然可以，我很乐意与您合作。请告诉我您有什么计划？ -------------------------------------------------- AgentA 收到: 当然可以，我很乐意与您合作。请告诉我您有什么计划？ 意图识别: 提供信息 AgentA: 我想开发一个智能助手，能自动安排日程和发送邮件。 ...2. 基于结构化消息的通信协议
结构化消息使用预定义的格式和字段，便于机器解析和处理，提高通信效率。
常见的结构化消息格式包括：
- FIPA ACL：Foundation for Intelligent Physical Agents的标准。
- JSON-based：自定义JSON格式。
- Protocol Buffers：Google的高效二进制协议。
【示例3.9】一个基于JSON格式的结构化消息通信示例。
import json # 定义结构化消息格式 class Message: def __init__(self, sender, receiver, msg_type, content): self.sender = sender self.receiver = receiver self.msg_type = msg_type self.content = content def to_json(self): return json.dumps(self.__dict__) # Agent A 发送结构化消息 agent_a_msg = Message(sender="Agent A", receiver="Agent B", msg_type="REQUEST", content="计算销售数据") msg_json = agent_a_msg.to_json() print("Agent A 发送：", msg_json) # Agent B 接收并解析消息 agent_b_msg = json.loads(msg_json) print("Agent B 接收：", agent_b_msg) # Agent B 回复结构化消息 agent_b_response = Message(sender="Agent B", receiver="Agent A", msg_type="RESPONSE", content="销售数据已计算完成") response_json = agent_b_response.to_json() print("Agent B 回复：", response_json)运行代码，输出如下：
Agent A 发送： {"sender": "Agent A", "receiver": "Agent B", "msg_type": "REQUEST", "content": "\u8ba1\u7b97\u9500\u552e\u6570\u636e"} Agent B 接收： {'sender': 'Agent A', 'receiver': 'Agent B', 'msg_type': 'REQUEST', 'content': '计算销售数据'} Agent B 回复： {"sender": "Agent B", "receiver": "Agent A", "msg_type": "RESPONSE", "content": "\u9500\u552e\u6570\u636e\u5df2\u8ba1\u7b97\u5b8c\u6210"}3. 实际应用场景
（1）智能工厂：在智能工厂中，多个机器人和设备需要高效协同工作。结构化消息协议（如FIPA-ACL）被广泛用于任务分配、冲突解决和状态更新。
（2）智能助手：在多智能助手场景中，自然语言通信可用于初次交互和复杂任务的协商，而结构化消息则用于后续的高效执行。
自然语言通信适用于灵活性和适应性要求较高的场景，但计算成本较高。结构化消息通信适用于效率和可扩展性要求较高的场景，但需要预定义协议。在实际应用中，可以根据需求结合使用这两种通信方式。
3.3.3 竞争协调：拍卖机制或投票系统
1. 竞争协调的核心价值
在多智能体系统（MAS）中，当智能体目标冲突或资源有限时，需通过竞争协调机制实现高效决策。该机制需满足：
- 公平性：避免单一智能体垄断资源。
- 效率性：最小化协调成本，最大化系统收益。
- 稳定性：抑制策略性操纵行为。
2. 拍卖机制：资源竞争的定价策略
1）核心原理
通过竞价博弈分配稀缺资源，以出价作为竞争信号。其运作遵循：
- 价高者得：原则实现帕累托最优分配。
- 显式规则：（如密封投标、公开叫价）决定最终归属。
- 支付机制：设计（如首价/次价支付）影响竞拍策略。
2）拍卖机制实现
【示例3.10】拍卖机制常用于资源分配和任务分配场景。下面实现一个简单的英式拍卖（公开增价拍卖）。
import random from typing import List, Dict class EnglishAuction: def __init__(self, items: List[str], agents: List[str], initial_prices: Dict[str, float]): """ 初始化英式拍卖 :param items: 拍卖物品列表 :param agents: 参与拍卖的智能体列表 :param initial_prices: 各物品的初始价格 """ self.items = items self.agents = agents self.current_prices = initial_prices.copy() self.winners = {item: None for item in items} self.bids = {item: [] for item in items} def agent_bid(self, agent: str, item: str, bid_price: float) -> bool: """ 智能体出价 :param agent: 出价智能体 :param item: 拍卖物品 :param bid_price: 出价金额 :return: 是否出价成功 """ if bid_price > self.current_prices[item]: self.current_prices[item] = bid_price self.bids[item].append((agent, bid_price)) return True return False def close_auction(self) -> Dict[str, str]: """ 结束拍卖，确定获胜者 :return: 各物品的获胜者 """ for item in self.items: if self.bids[item]: # 最高出价者获胜 self.winners[item] = max(self.bids[item], key=lambda x: x[1])[0] return self.winners def simulate_auction(self, max_rounds: int = 10): """ 模拟拍卖过程 :param max_rounds: 最大轮数 """ for _ in range(max_rounds): for agent in self.agents: for item in self.items: # 模拟智能体随机出价 if random.random() > 0.7: # 30%概率出价 current_price = self.current_prices[item] bid_price = current_price * (1 + random.uniform(0.1, 0.5)) self.agent_bid(agent, item, bid_price) print(f"Round {_+1} current prices: {self.current_prices}") self.close_auction() print("Auction results:", self.winners) # 使用示例 if __name__ == "__main__": items = ["task1", "task2", "resource1"] agents = ["agent1", "agent2", "agent3", "agent4"] initial_prices = {"task1": 10.0, "task2": 15.0, "resource1": 20.0} auction = EnglishAuction(items, agents, initial_prices) auction.simulate_auction()运行代码，输出如下：
Round 1 current prices: {'task1': 10.0, 'task2': 15.0, 'resource1': 27.624252137343362} Round 2 current prices: {'task1': 10.0, 'task2': 15.0, 'resource1': 27.624252137343362} Round 3 current prices: {'task1': 14.150423418969961, 'task2': 25.598419655339914, 'resource1': 48.43936171930753} Round 4 current prices: {'task1': 14.150423418969961, 'task2': 34.189396186994834, 'resource1': 84.86987307269457} Round 5 current prices: {'task1': 14.150423418969961, 'task2': 44.89570715854789, 'resource1': 102.4059754336159} Round 6 current prices: {'task1': 14.150423418969961, 'task2': 55.70639803093716, 'resource1': 265.4076490101352} Round 7 current prices: {'task1': 16.165556762263797, 'task2': 81.8150489398106, 'resource1': 454.164529972914} Round 8 current prices: {'task1': 17.843491163327904, 'task2': 81.8150489398106, 'resource1': 664.0285634582275} Round 9 current prices: {'task1': 25.410115323935496, 'task2': 81.8150489398106, 'resource1': 776.2515227325367} Round 10 current prices: {'task1': 72.47785125080446, 'task2': 158.02764874120223, 'resource1': 1088.1494800815683} Auction results: {'task1': 'agent4', 'task2': 'agent4', 'resource1': 'agent2'}【示例3.11】投票系统实现，投票系统常用于集体决策场景。下面实现一个多数投票和波达计数（Borda Count）投票系统。
from typing import List, Dict from collections import defaultdict class VotingSystem: def __init__(self, candidates: List[str], voters: List[str]): """ 初始化投票系统 :param candidates: 候选选项 :param voters: 投票者列表 """ self.candidates = candidates self.voters = voters self.ballots = [] def submit_ballot(self, voter: str, ranking: List[str]) -> bool: """ 提交投票 :param voter: 投票者 :param ranking: 投票者的偏好排序 :return: 是否提交成功 """ if voter not in self.voters: return False if set(ranking) != set(self.candidates): return False self.ballots.append((voter, ranking)) return True def majority_vote(self) -> str: """ 多数投票制 :return: 获胜者 """ if not self.ballots: return None first_choices = [ballot[1][0] for ballot in self.ballots] vote_counts = defaultdict(int) for candidate in first_choices: vote_counts[candidate] += 1 winner = max(vote_counts.items(), key=lambda x: x[1])[0] return winner def borda_count(self) -> str: """ 波达计数法 :return: 获胜者 """ if not self.ballots: return None candidate_scores = defaultdict(int) num_candidates = len(self.candidates) for _, ranking in self.ballots: for position, candidate in enumerate(ranking): # 波达计数：第一名得n-1分，第二名得n-2分，…，最后一名得0分 candidate_scores[candidate] += (num_candidates - position - 1) winner = max(candidate_scores.items(), key=lambda x: x[1])[0] return winner def simulate_voting(self): """ 模拟投票过程 """ # 模拟投票者提交偏好排序 for voter in self.voters: # 随机生成偏好排序 ranking = self.candidates.copy() random.shuffle(ranking) self.submit_ballot(voter, ranking) print(f"{voter} votes: {ranking}") # 多数投票结果 majority_winner = self.majority_vote() print(f"Majority vote winner: {majority_winner}") # 波达计数结果 borda_winner = self.borda_count() print(f"Borda count winner: {borda_winner}") # 使用示例 if __name__ == "__main__": import random candidates = ["OptionA", "OptionB", "OptionC", "OptionD"] voters = ["Agent1", "Agent2", "Agent3", "Agent4", "Agent5"] voting_system = VotingSystem(candidates, voters) voting_system.simulate_voting()运行代码，输出如下：
Agent1 votes: ['OptionA', 'OptionC', 'OptionB', 'OptionD'] Agent2 votes: ['OptionB', 'OptionC', 'OptionD', 'OptionA'] Agent3 votes: ['OptionD', 'OptionB', 'OptionC', 'OptionA'] Agent4 votes: ['OptionC', 'OptionA', 'OptionB', 'OptionD'] Agent5 votes: ['OptionD', 'OptionC', 'OptionB', 'OptionA'] Majority vote winner: OptionD Borda count winner: OptionC【示例3.12】智能体协调框架，结合拍卖和投票的智能体协调框架示例。
import random from typing import List, Dict from collections import defaultdict class EnglishAuction: def __init__(self, items: List[str], agents: List[str], initial_prices: Dict[str, float]): """ 初始化英式拍卖 :param items: 拍卖物品列表 :param agents: 参与拍卖的智能体列表 :param initial_prices: 各物品的初始价格 """ self.items = items self.agents = agents self.current_prices = initial_prices.copy() self.winners = {item: None for item in items} self.bids = {item: [] for item in items} def agent_bid(self, agent: str, item: str, bid_price: float) -> bool: """ 智能体出价 :param agent: 出价智能体 :param item: 拍卖物品 :param bid_price: 出价金额 :return: 是否出价成功 """ if bid_price > self.current_prices[item]: self.current_prices[item] = bid_price self.bids[item].append((agent, bid_price)) return True return False def close_auction(self) -> Dict[str, str]: """ 结束拍卖，确定获胜者 :return: 各物品的获胜者 """ for item in self.items: if self.bids[item]: # 最高出价者获胜 self.winners[item] = max(self.bids[item], key=lambda x: x[1])[0] return self.winners def simulate_auction(self, max_rounds: int = 10): """ 模拟拍卖过程 :param max_rounds: 最大轮数 """ for round_num in range(max_rounds): for agent in self.agents: for item in self.items: # 模拟智能体随机出价 if random.random() > 0.7: # 30%概率出价 current_price = self.current_prices[item] bid_price = current_price * (1 + random.uniform(0.1, 0.5)) self.agent_bid(agent, item, bid_price) print(f"Round {round_num+1} current prices: {self.current_prices}") self.close_auction() print("Auction results:", self.winners) return self.winners class VotingSystem: def __init__(self, candidates: List[str], voters: List[str]): """ 初始化投票系统 :param candidates: 候选选项 :param voters: 投票者列表 """ self.candidates = candidates self.voters = voters self.ballots = [] def submit_ballot(self, voter: str, ranking: List[str]) -> bool: """ 提交投票 :param voter: 投票者 :param ranking: 投票者的偏好排序 :return: 是否提交成功 """ if voter not in self.voters: return False if set(ranking) != set(self.candidates): return False self.ballots.append((voter, ranking)) return True def majority_vote(self) -> str: """ 多数投票制 :return: 获胜者 """ if not self.ballots: return None first_choices = [ballot[1][0] for ballot in self.ballots] vote_counts = defaultdict(int) for candidate in first_choices: vote_counts[candidate] += 1 winner = max(vote_counts.items(), key=lambda x: x[1])[0] return winner def borda_count(self) -> str: """ 波达计数法 :return: 获胜者 """ if not self.ballots: return None candidate_scores = defaultdict(int) num_candidates = len(self.candidates) for _, ranking in self.ballots: for position, candidate in enumerate(ranking): # 波达计数：第一名得n-1分，第二名得n-2分，…，最后一名得0分 candidate_scores[candidate] += (num_candidates - position - 1) winner = max(candidate_scores.items(), key=lambda x: x[1])[0] return winner def simulate_voting(self): """ 模拟投票过程 """ # 模拟投票者提交偏好排序 for voter in self.voters: # 随机生成偏好排序 ranking = self.candidates.copy() random.shuffle(ranking) self.submit_ballot(voter, ranking) print(f"{voter} votes: {ranking}") # 多数投票结果 majority_winner = self.majority_vote() print(f"Majority vote winner: {majority_winner}") # 波达计数结果 borda_winner = self.borda_count() print(f"Borda count winner: {borda_winner}") return majority_winner, borda_winner class Agent: def __init__(self, name, resources, preferences): self.name = name self.resources = resources self.preferences = preferences # {item: value} def bid_strategy(self, auction_type, item, current_bid): """根据拍卖类型制定出价策略""" if auction_type == "english": # 英式拍卖策略：出价比当前最高价高，但不超过自己的估值 my_value = self.preferences.get(item, 0) if current_bid < my_value: return min(current_bid + 5, my_value) # 简单增量策略 return 0 elif auction_type == "vickrey": # Vickrey拍卖策略：出价等于真实估值 return self.preferences.get(item, 0) return 0 def vote_strategy(self, voting_system, alternatives): """投票策略""" if isinstance(voting_system, MajorityVoting): # 多数投票：选择最偏好的选项 return max(alternatives, key=lambda x: self.preferences.get(x, 0)) elif isinstance(voting_system, BordaVoting): # Borda投票：按偏好排序 return sorted(alternatives, key=lambda x: self.preferences.get(x, 0), reverse=True) return None class MajorityVoting: def __init__(self, alternatives): """ 多数投票系统 :param alternatives: 可选方案列表 """ self.alternatives = alternatives self.votes = {alt: 0 for alt in alternatives} self.voters = set() def cast_vote(self, voter, alternative): """投票""" if alternative not in self.alternatives: raise ValueError("Invalid alternative") self.votes[alternative] += 1 self.voters.add(voter) def get_winner(self): """获取获胜方案""" if not self.voters: return None return max(self.votes.items(), key=lambda x: x[1])[0] def get_vote_distribution(self): """获取投票分布""" return self.votes class BordaVoting: def __init__(self, alternatives): """ Borda投票系统 :param alternatives: 可选方案列表 """ self.alternatives = alternatives self.rankings = [] self.voters = set() def submit_ranking(self, voter, ranking): """提交排序""" if set(ranking) != set(self.alternatives): raise ValueError("Invalid ranking") self.rankings.append((voter, ranking)) self.voters.add(voter) def get_winner(self): """获取获胜方案""" if not self.voters: return None candidate_scores = defaultdict(int) num_candidates = len(self.alternatives) for _, ranking in self.rankings: for position, candidate in enumerate(ranking): candidate_scores[candidate] += (num_candidates - position - 1) return max(candidate_scores.items(), key=lambda x: x[1])[0] class VickreyAuction: def __init__(self, items): self.items = items self.bids = {item: [] for item in items} self.winners = {item: None for item in items} self.winning_prices = {item: 0 for item in items} def submit_bid(self, item, agent, bid): self.bids[item].append((agent, bid)) def determine_winner(self, item): if not self.bids[item]: return sorted_bids = sorted(self.bids[item], key=lambda x: x[1], reverse=True) if len(sorted_bids) >= 2: self.winners[item] = sorted_bids[0][0] self.winning_prices[item] = sorted_bids[1][1] elif len(sorted_bids) == 1: self.winners[item] = sorted_bids[0][0] self.winning_prices[item] = sorted_bids[0][1] def get_status(self, item): return { 'winner': self.winners[item], 'winning_price': self.winning_prices[item] } class CoordinationFramework: def __init__(self, agents, items, voting_alternatives): self.agents = agents self.items = items self.voting_alternatives = voting_alternatives def run_auction(self, auction_type): """运行拍卖""" if auction_type == "english": initial_prices = {item: 0 for item in self.items} auction = EnglishAuction(self.items, [agent.name for agent in self.agents], initial_prices) for round_num in range(10): # 进行10轮拍卖 for agent in self.agents: for item in self.items: current_price = auction.current_prices[item] bid = agent.bid_strategy("english", item, current_price) if bid > 0: auction.agent_bid(agent.name, item, bid) print(f"Round {round_num+1} current prices: {auction.current_prices}") auction.close_auction() return auction elif auction_type == "vickrey": auction = VickreyAuction(self.items) for item in self.items: for agent in self.agents: bid = agent.bid_strategy("vickrey", item, 0) auction.submit_bid(item, agent.name, bid) auction.determine_winner(item) return auction return None def run_voting(self, voting_type): """运行投票""" if voting_type == "majority": voting = MajorityVoting(self.voting_alternatives) for agent in self.agents: vote = agent.vote_strategy(voting, self.voting_alternatives) voting.cast_vote(agent.name, vote) return voting elif voting_type == "borda": voting = BordaVoting(self.voting_alternatives) for agent in self.agents: ranking = agent.vote_strategy(voting, self.voting_alternatives) voting.submit_ranking(agent.name, ranking) return voting return None # 使用示例 if __name__ == "__main__": # 创建智能体 agents = [ Agent("agent1", 200, {"painting": 150, "book": 50, "optionA": 10, "optionB": 5}), Agent("agent2", 300, {"painting": 100, "book": 80, "optionA": 5, "optionB": 8}), Agent("agent3", 250, {"painting": 120, "book": 70, "optionA": 7, "optionB": 6}) ] # 创建协调框架 framework = CoordinationFramework( agents=agents, items=["painting", "book"], voting_alternatives=["optionA", "optionB"] ) # 运行英式拍卖 print("=== English Auction ===") english_result = framework.run_auction("english") print("English Auction Results:") for item in english_result.items: print(f"{item}: {english_result.winners[item]}") # 运行多数投票 print("\n=== Majority Voting ===") majority_result = framework.run_voting("majority") print("Majority Voting Results:") print("Winner:", majority_result.get_winner()) print("Distribution:", majority_result.get_vote_distribution()) # 运行Borda投票 print("\n=== Borda Voting ===") borda_result = framework.run_voting("borda") print("Borda Voting Results:") print("Winner:", borda_result.get_winner()) # 运行Vickrey拍卖 print("\n=== Vickrey Auction ===") vickrey_result = framework.run_auction("vickrey") print("Vickrey Auction Results:") for item in vickrey_result.items: status = vickrey_result.get_status(item) print(f"{item}: Winner={status['winner']}, Price={status['winning_price']}")运行代码，输出如下：
=== English Auction === Round 1 current prices: {'painting': 15, 'book': 15} Round 2 current prices: {'painting': 30, 'book': 30} Round 3 current prices: {'painting': 45, 'book': 45} Round 4 current prices: {'painting': 60, 'book': 60} Round 5 current prices: {'painting': 75, 'book': 70} Round 6 current prices: {'painting': 90, 'book': 75} Round 7 current prices: {'painting': 105, 'book': 80} Round 8 current prices: {'painting': 115, 'book': 80} Round 9 current prices: {'painting': 120, 'book': 80} Round 10 current prices: {'painting': 125, 'book': 80} English Auction Results: painting: agent1 book: agent2 === Majority Voting === Majority Voting Results: Winner: optionA Distribution: {'optionA': 2, 'optionB': 1} === Borda Voting === Borda Voting Results: Winner: optionA === Vickrey Auction === Vickrey Auction Results: painting: Winner=agent1, Price=120 book: Winner=agent2, Price=70以上代码实现了AI智能体竞争协调中的两种主要机制：拍卖机制和投票系统。这两种机制可以用于多智能体系统中的资源分配、决策制定等协调问题。你可以根据具体需求扩展这些基本实现，例如添加更复杂的出价策略、考虑预算约束、实现组合拍卖等。
