协商博弈与决策均衡完整实现
import numpy as np
from typing import Dict, List, Any, Optional, Tuple, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import math
import random
from collections import defaultdict
class NegotiationType(Enum):
"""协商类型"""
SINGLE_ISSUE = "single_issue" # 单议题
MULTI_ISSUE = "multi_issue" # 多议题
AUCTION = "auction" # 拍卖
BARGAINING = "bargaining" # 讨价还价
class GameType(Enum):
"""博弈类型"""
COOPERATIVE = "cooperative" # 合作博弈
NON_COOPERATIVE = "non_cooperative" # 非合作博弈
ZERO_SUM = "zero_sum" # 零和博弈
CONSTANT_SUM = "constant_sum" # 常和博弈
class StrategyType(Enum):
"""策略类型"""
NASH = "nash" # 纳什策略
MAXIMIN = "maximin" # 最大最小策略
TIT_FOR_TAT = "tit_for_tat" # 以牙还牙
DEFECT = "defect" # 背叛
COOPERATE = "cooperate" # 合作
@dataclass
class Utility:
"""效用函数"""
agent_id: str
function_type: str # linear, quadratic, etc.
parameters: Dict[str, float]
def evaluate(self, outcome: Dict[str, float]) -> float:
"""计算效用值"""
if self.function_type == "linear":
return sum(outcome.get(k, 0) * v for k, v in self.parameters.items())
elif self.function_type == "quadratic":
return sum(v * outcome.get(k, 0)**2 for k, v in self.parameters.items())
return 0.0
@dataclass
class Agent:
"""智能体"""
id: str
utility: Utility
strategy: StrategyType = StrategyType.NASH
belief: Dict[str, Any] = field(default_factory=dict)
history: List[Dict[str, Any]] = field(default_factory=list)
@dataclass
class NegotiationIssue:
"""协商议题"""
id: str
name: str
type: str # continuous, discrete
range: Tuple[float, float] = (0, 1)
weight: float = 1.0
@dataclass
class Offer:
"""出价"""
agent_id: str
values: Dict[str, float]
timestamp: datetime = field(default_factory=datetime.now)
utility: float = 0.0
class NegotiationProtocol:
"""
协商协议
支持:
1. 轮流出价
2. 同时出价
3. 拍卖机制
4. 讨价还价
"""
def __init__(self, protocol_type: str = "alternating"):
self.protocol_type = protocol_type
self.rounds = 0
self.max_rounds = 100
self.offers: List[Offer] = []
self.agreement: Optional[Dict[str, float]] = None
def start_negotiation(self, agents: List[Agent],
issues: List[NegotiationIssue]):
"""开始协商"""
self.rounds = 0
self.offers = []
self.agreement = None
print(f"开始协商，{len(agents)}个智能体，{len(issues)}个议题")
def make_offer(self, agent: Agent, issues: List[NegotiationIssue],
strategy: StrategyType) -> Offer:
"""生成出价"""
values = {}
for issue in issues:
if strategy == StrategyType.NASH:
# 纳什均衡出价
value = (issue.range[0] + issue.range[1]) / 2
elif strategy == StrategyType.MAXIMIN:
# 最大最小策略
value = issue.range[0] + 0.3 * (issue.range[1] - issue.range[0])
else:
value = random.uniform(issue.range[0], issue.range[1])
values[issue.id] = value
offer = Offer(agent_id=agent.id, values=values)
offer.utility = agent.utility.evaluate(values)
self.offers.append(offer)
return offer
def check_agreement(self, offers: List[Offer],
threshold: float = 0.8) -> Optional[Dict[str, float]]:
"""检查是否达成协议"""
if len(offers) < 2:
return None
# 检查出价是否接近
values_list = [offer.values for offer in offers[-2:]]
agreement = {}
for key in values_list[0].keys():
vals = [v[key] for v in values_list]
if max(vals) - min(vals) < 0.05: # 差异小于 5%
agreement[key] = sum(vals) / len(vals)
else:
return None
self.agreement = agreement
return agreement
def run_negotiation(self, agents: List[Agent],
issues: List[NegotiationIssue]) -> Optional[Dict[str, float]]:
"""运行协商过程"""
self.start_negotiation(agents, issues)
for round_num in range(self.max_rounds):
self.rounds = round_num + 1
# 每个智能体出价
offers = []
for agent in agents:
offer = self.make_offer(agent, issues, agent.strategy)
offers.append(offer)
agent.history.append({"round": round_num, "offer": offer.values})
# 检查是否达成协议
agreement = self.check_agreement(offers)
if agreement:
print(f"第{round_num+1}轮达成协议：{agreement}")
return agreement
print("协商失败，未达成协议")
return None
class GameSolver:
"""
博弈求解器
支持:
1. 纳什均衡求解
2. 合作博弈求解
3. 演化稳定策略
4. 机制设计
"""
def __init__(self):
self.equilibria: List[Dict[str, Any]] = []
def find_nash_equilibrium(self, payoff_matrix: np.ndarray) -> List[Tuple]:
"""求解纳什均衡（简化版）"""
n_players, n_strategies = payoff_matrix.shape[0], payoff_matrix.shape[1]
equilibria = []
# 简化：检查纯策略纳什均衡
for s1 in range(n_strategies):
for s2 in range(n_strategies):
# 检查是否是纳什均衡
is_nash = True
# 检查玩家 1 是否有动机偏离
for s1_alt in range(n_strategies):
if payoff_matrix[s1_alt, s2] > payoff_matrix[s1, s2]:
is_nash = False
break
if not is_nash:
continue
# 检查玩家 2 是否有动机偏离
for s2_alt in range(n_strategies):
if payoff_matrix[s1, s2_alt] > payoff_matrix[s1, s2]:
is_nash = False
break
if is_nash:
equilibria.append((s1, s2))
self.equilibria = equilibria
return equilibria
def prisoner_dilemma(self) -> np.ndarray:
"""囚徒困境收益矩阵"""
# C=合作，D=背叛
# (C,C)=3,3 (C,D)=0,5 (D,C)=5,0 (D,D)=1,1
return np.array([
[3, 0], # 玩家 1 合作
[5, 1] # 玩家 1 背叛
])
def battle_of_sexes(self) -> np.ndarray:
"""性别战博弈"""
return np.array([
[3, 0], # 都选 A
[0, 2] # 都选 B
])
class MechanismDesigner:
"""
机制设计器
支持:
1. 拍卖机制
2. 投票机制
3. 匹配机制
4. 激励兼容
"""
def __init__(self):
self.mechanisms: Dict[str, Any] = {}
def design_auction(self, auction_type: str,
bidders: List[Agent],
item_value: float) -> Dict[str, Any]:
"""设计拍卖机制"""
if auction_type == "first_price":
return self._first_price_auction(bidders, item_value)
elif auction_type == "second_price":
return self._second_price_auction(bidders, item_value)
elif auction_type == "english":
return self._english_auction(bidders, item_value)
else:
raise ValueError(f"未知拍卖类型：{auction_type}")
def _first_price_auction(self, bidders: List[Agent],
item_value: float) -> Dict[str, Any]:
"""第一价格密封拍卖"""
bids = []
for bidder in bidders:
# 简化：出价略低于真实价值
bid = item_value * random.uniform(0.7, 0.95)
bids.append((bidder.id, bid))
# 最高价者获胜，支付出价
winner_id, winning_bid = max(bids, key=lambda x: x[1])
return {
"type": "first_price",
"winner": winner_id,
"price": winning_bid,
"bids": bids
}
def _second_price_auction(self, bidders: List[Agent],
item_value: float) -> Dict[str, Any]:
"""第二价格密封拍卖（Vickrey 拍卖）"""
bids = []
for bidder in bidders:
# 真实出价是占优策略
bid = item_value
bids.append((bidder.id, bid))
# 最高价者获胜，支付第二高价
sorted_bids = sorted(bids, key=lambda x: x[1], reverse=True)
winner_id = sorted_bids[0][0]
price = sorted_bids[1][1] if len(sorted_bids) > 1 else 0
return {
"type": "second_price",
"winner": winner_id,
"price": price,
"bids": bids
}
def _english_auction(self, bidders: List[Agent],
item_value: float) -> Dict[str, Any]:
"""英式拍卖（公开增价）"""
current_price = 0
increment = item_value * 0.1
active_bidders = set(b.id for b in bidders)
while len(active_bidders) > 1:
current_price += increment
# 简化：价值低于当前价格的 bidder 退出
active_bidders = {
b.id for b in bidders
if item_value >= current_price and b.id in active_bidders
}
winner = list(active_bidders)[0] if active_bidders else None
return {
"type": "english",
"winner": winner,
"price": current_price
}
# 使用示例
if __name__ == "__main__":
print("=== 多智能体协商、博弈与决策均衡 ===\n")
print("=== 创建协商系统 ===")
# 创建协商协议
protocol = NegotiationProtocol(protocol_type="alternating")
# 创建议题
issues = [
NegotiationIssue(id="price", name="价格", type="continuous", range=(0, 100), weight=0.5),
NegotiationIssue(id="quality", name="质量", type="continuous", range=(0, 10), weight=0.3),
NegotiationIssue(id="delivery", name="交付时间", type="continuous", range=(1, 30), weight=0.2),
]
# 创建智能体
agents = [
Agent(
id="buyer",
utility=Utility("buyer", "linear", {"price": -0.01, "quality": 0.1, "delivery": -0.02}),
strategy=StrategyType.NASH
),
Agent(
id="seller",
utility=Utility("seller", "linear", {"price": 0.01, "quality": -0.05, "delivery": 0.01}),
strategy=StrategyType.MAXIMIN
)
]
print(f"创建{len(agents)}个智能体，{len(issues)}个议题")
print(f"\n=== 运行协商 ===")
# 运行协商
agreement = protocol.run_negotiation(agents, issues)
if agreement:
print(f"\n达成协议:")
for issue_id, value in agreement.items():
print(f" {issue_id}: {value:.2f}")
# 计算各方效用
for agent in agents:
utility = agent.utility.evaluate(agreement)
print(f" {agent.id}效用：{utility:.2f}")
else:
print("协商失败")
print(f"\n=== 博弈求解 ===")
# 创建博弈求解器
solver = GameSolver()
# 囚徒困境
print("\n囚徒困境:")
pd_matrix = solver.prisoner_dilemma()
print(f"收益矩阵:\n{pd_matrix}")
nash_eq = solver.find_nash_equilibrium(pd_matrix)
print(f"纳什均衡：{nash_eq}")
print("解释：(背叛，背叛) 是唯一纳什均衡，尽管(合作，合作) 对双方都更好")
# 性别战
print("\n性别战博弈:")
bos_matrix = solver.battle_of_sexes()
print(f"收益矩阵:\n{bos_matrix}")
nash_eq = solver.find_nash_equilibrium(bos_matrix)
print(f"纳什均衡：{nash_eq}")
print("解释：存在两个纯策略纳什均衡，需要协调")
print(f"\n=== 机制设计 ===")
# 创建机制设计器
designer = MechanismDesigner()
# 创建竞拍者
bidders = [
Agent(id="bidder1", utility=Utility("bidder1", "linear", {"value": 1.0})),
Agent(id="bidder2", utility=Utility("bidder2", "linear", {"value": 1.0})),
Agent(id="bidder3", utility=Utility("bidder3", "linear", {"value": 1.0})),
]
item_value = 100.0
print(f"\n物品价值：{item_value}")
# 第一价格拍卖
result1 = designer.design_auction("first_price", bidders, item_value)
print(f"\n第一价格拍卖:")
print(f" 获胜者：{result1['winner']}")
print(f" 成交价：{result1['price']:.2f}")
# 第二价格拍卖
result2 = designer.design_auction("second_price", bidders, item_value)
print(f"\n第二价格拍卖 (Vickrey):")
print(f" 获胜者：{result2['winner']}")
print(f" 成交价：{result2['price']:.2f}")
print(f" 特点：真实出价是占优策略")
# 英式拍卖
result3 = designer.design_auction("english", bidders, item_value)
print(f"\n英式拍卖:")
print(f" 获胜者：{result3['winner']}")
print(f" 成交价：{result3['price']:.2f}")
print(f"\n关键观察:")
print("1. 协商：通过轮流出价和妥协达成协议")
print("2. 博弈：纳什均衡是稳定策略组合")
print("3. 囚徒困境：个体理性导致集体非理性")
print("4. 机制设计：不同拍卖机制导致不同结果")
print("5. 第二价格拍卖：激励兼容，真实出价最优")
print("\n均衡的核心：协商 + 博弈 + 决策 + 机制 = 群体理性")