解析 ‘Agent-to-Agent Negotiation’：实现两个 Agent 之间关于‘计算成本’与‘答案精度’的博弈逻辑

智能体博弈的艺术与科学：Agent-to-Agent 协商在计算成本与答案精度中的应用

各位听众，大家好。今天我们将深入探讨一个在分布式人工智能和多智能体系统领域至关重要的话题：智能体间的协商（Agent-to-Agent Negotiation）。特别地，我们将聚焦于如何实现两个智能体之间，针对“计算成本”与“答案精度”这两个核心维度进行博弈的逻辑。

在现代复杂的AI系统中，单一智能体往往难以独立完成所有任务。任务的分解、资源的分配、能力的整合，都需要智能体之间进行有效的沟通与协作。协商正是这种协作的核心机制之一。想象一个场景：一个智能体（我们称之为请求方Agent A）需要一项特定的计算服务，比如一个复杂的预测模型。另一个智能体（服务提供方Agent B）能够提供这项服务。Agent A关心的是预测的准确性以及需要为此支付的成本；Agent B关心的是执行计算所需的资源成本以及它能够获得的报酬，同时它也需要提供一定程度的服务质量。

这种场景下，Agent A和Agent B之间就形成了一个典型的博弈关系。Agent A希望以尽可能低的成本获得尽可能高的精度，而Agent B则希望以尽可能高的报酬提供尽可能低的成本服务（但又不能低于其承诺的服务质量）。如何在这对矛盾中找到一个双方都能接受的“交易点”，就是我们今天讲座的核心内容。

博弈论基础：理解协商的语言

在深入技术实现之前，我们首先需要建立一些博弈论的基础概念，它们是理解智能体协商逻辑的基石。

1. 智能体 (Agents)： 参与博弈的独立决策实体。在本场景中，我们有Agent A (请求方) 和 Agent B (服务提供方)。
2. 策略 (Strategies)： 智能体在给定情境下可采取的行动方案。在协商中，策略包括如何提出报价、如何回应报价、何时让步、何时拒绝等。
3. 效用 (Utility)： 智能体对其所获得结果的满意度量化表示。每个智能体都有一个内部的效用函数，用于评估不同协商结果的好坏。智能体的目标是最大化自身的效用。
4. 议题 (Issues)： 协商中需要达成一致的变量。在我们的案例中，核心议题是“计算成本”和“答案精度”。

量化“计算成本”与“答案精度”

为了让智能体能够进行量化评估和协商，我们必须将这些概念数值化。

计算成本 (Computation Cost, C):
这可以是一个抽象的数值，代表执行计算所需的资源消耗。它可以是：

• CPU 时间 / GPU 时间： 例如，以毫秒或秒为单位。
• 内存消耗： 例如，以MB或GB为单位。
• API 调用次数： 如果服务涉及外部API。
• 云资源费用： 例如，每小时的虚拟机费用。
• 简单积分： 在我们的模拟中，我们可以将其简化为 1 到 100 之间的一个数值。

答案精度 (Answer Accuracy, A):
这通常是一个衡量服务质量的指标。它可以是：

• 准确率 (Accuracy)： 分类任务中正确预测的比例（0到1之间）。
• 召回率 (Recall) / 精确率 (Precision)： 信息检索或分类任务的指标。
• F1 分数： 综合精确率和召回率。
• 均方误差 (MSE) / 绝对误差 (MAE)： 回归任务中预测值与真实值之间的差异。
• 简单积分： 在我们的模拟中，我们可以将其简化为 0 到 1 之间的一个百分比或小数。

效用函数：智能体的价值衡量

每个智能体都会根据其自身的偏好和目标，将协商议题（成本和精度）映射到一个单一的效用值。

Agent A (请求方) 的效用函数：
Agent A 追求高精度，但同时希望成本越低越好。我们可以构建一个线性效用函数：
$UA(A, C) = w{A_accuracy} cdot A – w{A_cost} cdot C$
其中，$w{A_accuracy}$ 是Agent A对精度的权重，$w{A_cost}$ 是Agent A对成本的权重。这些权重反映了Agent A对精度和成本的相对重视程度。通常，$w{A_accuracy}$ 为正，表示精度越高越好；$w_{A_cost}$ 为正，表示成本越低越好（因此在函数中是减号）。Agent A的目标是最大化 $U_A$。

Agent B (服务提供方) 的效用函数：
Agent B 追求高报酬（即Agent A支付的成本），但同时需要付出资源来达到一定的精度。它也希望自己的实际计算成本尽可能低。我们假设 Agent B 的“报酬”就是 Agent A 支付的“成本”。
$UB(A, C) = w{B_revenue} cdot C – w{B_cost_of_service} cdot text{Cost}(A)$
其中，$w{B_revenue}$ 是Agent B对收入的权重，$w_{B_cost_of_service}$ 是Agent B对提供特定精度服务所需实际成本的权重。这里的 $text{Cost}(A)$ 是一个函数，表示提供精度 $A$ 所需的内部实际成本，通常精度越高，成本越高。
为了简化，我们可以将 Agent B 的效用函数理解为：在满足一定精度要求的前提下，Agent B 期望收到更高的成本支付。
$UB(A, C) = w{B_profit} cdot C + w_{B_quality} cdot A$
但这里有一个问题：Agent B 实际的“利润”是 $C – text{internal_cost}(A)$。更合理的模型是：
$U_B(A, C) = C – (alpha cdot A^2 + beta cdot A + gamma)$ （其中 $alpha, beta, gamma$ 是参数，描述提供精度A的内部成本模型）
为了简化，在我们的模拟中，我们可以假设Agent B的内部成本与精度呈正相关，且它更倾向于高成本（高报酬）和满足基本精度的方案。
$UB(A, C) = w{B_cost_received} cdot C – w_{B_cost_incurred} cdot f(A)$
其中 $f(A)$ 是一个递增函数，表示提供精度 $A$ 所需的内部成本。

在接下来的代码示例中，我们将采用更直接的效用函数，以清晰地展示博弈逻辑。

协商协议设计：Agent如何对话

协商是一个结构化的过程，智能体之间需要遵循一定的协议来交换信息和推进博弈。一个典型的协商协议包括以下状态和消息类型：

协商状态 (Negotiation States):

• INITIATED： 协商开始。
• PROPOSING： 正在提出报价。
• COUNTER_PROPOSING： 正在回应报价。
• ACCEPTED： 达成协议。
• REJECTED： 某一方拒绝了对方的最终报价，协商破裂。
• WITHDRAWN： 某一方主动退出协商。
• MAX_ROUNDS_REACHED： 达到最大协商回合数，协商失败。

协商消息 (Negotiation Messages):

• Proposal (提议)： 包含一组议题的提议值（例如：{'accuracy': 0.9, 'cost': 80}）。
• Accept (接受)： 表示接受了对方的最新提议。
• Reject (拒绝)： 表示拒绝了对方的最新提议。
• Withdraw (退出)： 表示退出协商。

协商流程概览

1. Agent A 发起协商： Agent A 向 Agent B 发送一个初始 Proposal，包含其期望的精度和愿意支付的成本。
2. Agent B 评估与回应： Agent B 收到提议后，根据其效用函数评估该提议。
   • 如果提议对其效用很高且满足其底线，Agent B 可以发送 Accept 消息。
   • 如果提议对其效用不高但仍有协商空间，Agent B 可以发送一个 Counter-Proposal，调整精度或成本值，以期提高自身效用。
   • 如果提议完全不可接受，且无协商余地，Agent B 可以发送 Reject 或 Withdraw 消息。
3. 循环往复： 双方智能体交替地提出 Proposal 或 Counter-Proposal，直到其中一方发送 Accept、Reject 或 Withdraw，或者达到预设的最大协商回合数。
4. 协议达成或失败： 如果某一方发送 Accept，则协商成功，达成协议。否则，协商失败。

效用函数建模：量化取舍

现在我们来具体定义 Agent A 和 Agent B 的效用函数。为了简化和聚焦于协商逻辑，我们设定以下参数范围：

• 精度 (Accuracy, A)： 0.0 到 1.0 (例如，0.8 代表 80% 的准确率)。
• 成本 (Cost, C)： 10.0 到 100.0 (抽象的货币单位)。

Agent A (请求方) 的效用函数

Agent A 偏好高精度和低成本。
我们定义其效用为：
$UA(A, C) = w{A_acc} cdot A – w{A_cost} cdot C$
其中 $w{A_acc}$ 和 $w{A_cost}$ 是正权重。
例如，Agent A 可能认为每提高 0.01 的精度比减少 1 单位的成本更重要，所以 $w{A_acc}$ 会相对较大。

class AgentAUntility:
    def __init__(self, accuracy_weight: float, cost_weight: float,
                 min_acceptable_accuracy: float, max_acceptable_cost: float):
        self.accuracy_weight = accuracy_weight
        self.cost_weight = cost_weight
        self.min_acceptable_accuracy = min_acceptable_accuracy # Agent A 的精度底线
        self.max_acceptable_cost = max_acceptable_cost         # Agent A 的成本预算

    def calculate(self, accuracy: float, cost: float) -> float:
        # 如果超出可接受范围，给予极低的效用，甚至负无穷
        if accuracy < self.min_acceptable_accuracy or cost > self.max_acceptable_cost:
            return -float('inf')
        return self.accuracy_weight * accuracy - self.cost_weight * cost

    def get_reservation_utility(self) -> float:
        """Agent A 的保留效用，低于此值宁愿不交易"""
        # 假设 Agent A 的保留效用是其能接受的最低精度和最高成本的效用
        return self.calculate(self.min_acceptable_accuracy, self.max_acceptable_cost)

Agent B (服务提供方) 的效用函数

Agent B 偏好高收入（即 Agent A 支付的成本 C），同时提供服务会有其内部成本，且服务精度越高，内部成本越高。
我们定义其效用为：
$UB(A, C) = w{B_revenue} cdot C – w_{B_internal_cost} cdot f(A)$
这里 $f(A)$ 是一个简单函数，例如 $A^2$ 或 $e^A$，表示提供精度 $A$ 所需的内部资源消耗。为了简化，我们假设 $f(A) = A$ 线性关系，或者更实际一点，精度越高，边际成本越高，比如 $f(A) = A^2$。
我们设定一个 Agent B 能够提供的最大精度和最低成本。

class AgentBUntility:
    def __init__(self, revenue_weight: float, internal_cost_weight: float,
                 min_acceptable_cost: float, max_producible_accuracy: float):
        self.revenue_weight = revenue_weight
        self.internal_cost_weight = internal_cost_weight
        self.min_acceptable_cost = min_acceptable_cost         # Agent B 的成本底线（希望收到的最低报酬）
        self.max_producible_accuracy = max_producible_accuracy # Agent B 能提供的最高精度

    def _calculate_internal_cost(self, accuracy: float) -> float:
        """根据精度计算Agent B的内部成本。假设精度越高，内部成本越高，非线性增长"""
        # 简单模型：提供精度A的内部成本与A的平方成正比
        # 这里的系数0.5是为了让内部成本在一个合理区间，且增长不是特别快
        return 0.5 * (accuracy ** 2) * 100 # 将精度0-1映射到0-50的成本区间

    def calculate(self, accuracy: float, cost: float) -> float:
        # 如果超出可接受范围或生产能力，给予极低的效用
        if cost < self.min_acceptable_cost or accuracy > self.max_producible_accuracy:
            return -float('inf')

        internal_cost = self._calculate_internal_cost(accuracy)
        return self.revenue_weight * cost - self.internal_cost_weight * internal_cost

    def get_reservation_utility(self) -> float:
        """Agent B 的保留效用，低于此值宁愿不交易"""
        # 假设 Agent B 的保留效用是其能接受的最低成本和最低可实现精度的效用
        # 注意：这里需要一个Agent B愿意提供的最低精度，我们假设它至少要达到某个水平才能算作有效服务
        # 为了简化，我们假设其保留效用是最低收入和它能提供的最低有意义精度（比如0.5）时的效用
        min_meaningful_accuracy_for_B = 0.5 # 假设Agent B认为低于0.5的精度没有提供价值
        return self.calculate(min_meaningful_accuracy_for_B, self.min_acceptable_cost)

协商策略实现：Agent的决策逻辑

智能体的决策逻辑是协商的核心。我们将实现一个基于让步（concession-based）的策略。智能体在收到对方提议后，会评估其效用。如果效用不够高，它会生成一个“更接近”对方但仍能提升自身效用的新提议。

核心思想：让步与目标调整

• 初始提议： 每个智能体从一个对自己最有利但可能不被对方接受的极端点开始。
• 让步： 在收到对方提议后，如果该提议不被接受，智能体将计算一个“反提议”。反提议通常会向对方的偏好方向移动，即做出让步。让步的幅度可以根据协商回合数、对方的强硬程度、自身剩余的谈判空间等因素动态调整。
• 保留效用 (Reservation Utility)： 每个智能体都有一个最低可接受的效用值。一旦任何提议使得智能体的效用低于其保留效用，智能体将拒绝或退出协商。

Agent A (请求方) 的策略

Agent A 在协商中会尝试：

1. 初始提议： 提出一个高精度、低成本的理想方案。
2. 让步： 当Agent B提出反提议时，如果Agent A不满意，它会考虑：
   • 略微降低其对精度的要求。
   • 略微提高其愿意支付的成本。
     让步的幅度可以随着协商回合数的增加而增加，以期尽快达成协议。

Agent B (服务提供方) 的策略

Agent B 在协商中会尝试：

1. 初始提议： 提出一个低精度、高成本的方案。
2. 让步： 当Agent A提出反提议时，如果Agent B不满意，它会考虑：
   • 略微提高其能提供的精度。
   • 略微降低其要求的成本。
     让步的幅度同样可以根据协商回合数动态调整。

步进式让步策略

我们将采用一种简单的线性步进式让步策略。在每个回合中，智能体都会根据当前的协商回合数，以固定的步长向其“谈判空间”的另一端移动。

import enum
import random
from dataclasses import dataclass
import math

# --- 1. 定义协商协议组件 ---

class NegotiationState(enum.Enum):
    INITIATED = "INITIATED"
    PROPOSING = "PROPOSING"
    COUNTER_PROPOSING = "COUNTER_PROPOSING"
    ACCEPTED = "ACCEPTED"
    REJECTED = "REJECTED"
    WITHDRAWN = "WITHDRAWN"
    MAX_ROUNDS_REACHED = "MAX_ROUNDS_REACHED"
    NO_AGREEMENT = "NO_AGREEMENT" # 协商失败的通用状态

@dataclass
class Proposal:
    accuracy: float # 0.0 to 1.0
    cost: float     # 10.0 to 100.0

@dataclass
class NegotiationMessage:
    sender_id: str
    receiver_id: str
    message_type: str # "PROPOSAL", "ACCEPT", "REJECT", "WITHDRAW"
    proposal: Proposal = None # 仅当message_type为"PROPOSAL"时有效

    def __str__(self):
        if self.message_type == "PROPOSAL":
            return f"[{self.sender_id} -> {self.receiver_id}] PROPOSAL: Accuracy={self.proposal.accuracy:.2f}, Cost={self.proposal.cost:.2f}"
        else:
            return f"[{self.sender_id} -> {self.receiver_id}] {self.message_type}"

# --- 2. 效用函数 (已在上面定义，此处为完整代码) ---

class AgentAUntility:
    def __init__(self, accuracy_weight: float, cost_weight: float,
                 min_acceptable_accuracy: float, max_acceptable_cost: float):
        self.accuracy_weight = accuracy_weight
        self.cost_weight = cost_weight
        self.min_acceptable_accuracy = min_acceptable_accuracy
        self.max_acceptable_cost = max_acceptable_cost

        # 确保权重合理
        if self.accuracy_weight <= 0 or self.cost_weight <= 0:
            raise ValueError("Utility weights must be positive for Agent A.")

    def calculate(self, accuracy: float, cost: float) -> float:
        # 如果超出可接受范围，给予极低的效用，使其不会接受
        if accuracy < self.min_acceptable_accuracy or cost > self.max_acceptable_cost:
            return -float('inf')
        return self.accuracy_weight * accuracy - self.cost_weight * cost

    def get_reservation_utility(self) -> float:
        """Agent A 的保留效用，低于此值宁愿不交易。
        假设保留效用是其能接受的最低精度和最高成本的效用。"""
        return self.calculate(self.min_acceptable_accuracy, self.max_acceptable_cost)

class AgentBUntility:
    def __init__(self, revenue_weight: float, internal_cost_weight: float,
                 min_acceptable_cost: float, max_producible_accuracy: float):
        self.revenue_weight = revenue_weight
        self.internal_cost_weight = internal_cost_weight
        self.min_acceptable_cost = min_acceptable_cost
        self.max_producible_accuracy = max_producible_accuracy

        # 确保权重合理
        if self.revenue_weight <= 0 or self.internal_cost_weight <= 0:
            raise ValueError("Utility weights must be positive for Agent B.")

    def _calculate_internal_cost(self, accuracy: float) -> float:
        """根据精度计算Agent B的内部成本。假设精度越高，内部成本越高，非线性增长"""
        # 简单模型：提供精度A的内部成本与A的平方成正比，并乘以一个系数以控制量级
        # 0.5 * (accuracy ** 2) 将精度0-1映射到0-0.5的因子，再乘以100映射到0-50的成本区间
        # 这样，精度为1.0时，内部成本为0.5 * 1 * 100 = 50
        return 0.5 * (accuracy ** 2) * 100

    def calculate(self, accuracy: float, cost: float) -> float:
        # 如果超出可接受范围或生产能力，给予极低的效用
        if cost < self.min_acceptable_cost or accuracy > self.max_producible_accuracy:
            return -float('inf')

        internal_cost = self._calculate_internal_cost(accuracy)
        return self.revenue_weight * cost - self.internal_cost_weight * internal_cost

    def get_reservation_utility(self) -> float:
        """Agent B 的保留效用，低于此值宁愿不交易。
        假设其保留效用是最低收入和它能提供的最低有意义精度时的效用。
        最低有意义精度可以根据业务场景设定，这里我们设为0.5。
        """
        min_meaningful_accuracy_for_B = 0.5 
        # 如果min_meaningful_accuracy_for_B超出了max_producible_accuracy，则取max_producible_accuracy
        effective_min_accuracy = min(min_meaningful_accuracy_for_B, self.max_producible_accuracy)
        return self.calculate(effective_min_accuracy, self.min_acceptable_cost)

# --- 3. 智能体基类与具体智能体实现 ---

class BaseAgent:
    def __init__(self, agent_id: str, utility_calculator,
                 min_accuracy_bound: float, max_accuracy_bound: float,
                 min_cost_bound: float, max_cost_bound: float):
        self.agent_id = agent_id
        self.utility_calculator = utility_calculator
        self.min_accuracy_bound = min_accuracy_bound # 整个协商空间的精度下限
        self.max_accuracy_bound = max_accuracy_bound # 整个协商空间的精度上限
        self.min_cost_bound = min_cost_bound         # 整个协商空间的成本下限
        self.max_cost_bound = max_cost_bound         # 整个协商空间的成本上限
        self.current_proposal: Proposal = None
        self.negotiation_history = [] # 存储所有提议，用于分析或更复杂的策略

    def calculate_utility(self, proposal: Proposal) -> float:
        return self.utility_calculator.calculate(proposal.accuracy, proposal.cost)

    def get_reservation_utility(self) -> float:
        return self.utility_calculator.get_reservation_utility()

    def receive_message(self, message: NegotiationMessage):
        self.negotiation_history.append(message)
        # 实际处理逻辑由子类实现

    def _clamp_proposal(self, accuracy: float, cost: float) -> Proposal:
        """确保提议值在有效范围内"""
        clamped_accuracy = max(self.min_accuracy_bound, min(self.max_accuracy_bound, accuracy))
        clamped_cost = max(self.min_cost_bound, min(self.max_cost_bound, cost))
        return Proposal(accuracy=clamped_accuracy, cost=clamped_cost)

class RequestingAgent(BaseAgent): # Agent A
    def __init__(self, agent_id: str, utility_calculator: AgentAUntility,
                 min_accuracy_bound: float, max_accuracy_bound: float,
                 min_cost_bound: float, max_cost_bound: float,
                 initial_target_accuracy: float, initial_target_cost: float,
                 concession_rate_accuracy: float, concession_rate_cost: float):
        super().__init__(agent_id, utility_calculator,
                         min_accuracy_bound, max_accuracy_bound,
                         min_cost_bound, max_cost_bound)
        self.initial_target_accuracy = initial_target_accuracy
        self.initial_target_cost = initial_target_cost
        self.concession_rate_accuracy = concession_rate_accuracy # 精度让步率
        self.concession_rate_cost = concession_rate_cost         # 成本让步率

        # Agent A 自身的理想点和最低可接受点
        self.ideal_accuracy = max_accuracy_bound
        self.ideal_cost = min_cost_bound
        self.reservation_accuracy = utility_calculator.min_acceptable_accuracy
        self.reservation_cost = utility_calculator.max_acceptable_cost

    def make_initial_proposal(self) -> Proposal:
        """Agent A 的初始提议：高精度，相对较低成本（但可能不是最低）"""
        # 从一个对自身有利的起始点开始，例如：接近理想精度，但成本可以略高于理想成本
        # 也可以直接用 self.initial_target_accuracy, self.initial_target_cost
        return self._clamp_proposal(self.initial_target_accuracy, self.initial_target_cost)

    def respond_to_proposal(self, opponent_proposal: Proposal, negotiation_round: int) -> NegotiationMessage:
        current_utility = self.calculate_utility(opponent_proposal)

        # 如果对方提议满足我方底线且效用足够高，则接受
        if current_utility >= self.get_reservation_utility() and 
           current_utility >= self.calculate_utility(self.current_proposal or opponent_proposal): 
           # 如果是第一个提议，或者比我当前提议好，则接受
            print(f"[{self.agent_id}] Accepting proposal: Accuracy={opponent_proposal.accuracy:.2f}, Cost={opponent_proposal.cost:.2f}")
            self.current_proposal = opponent_proposal # 接受即设定为当前提议
            return NegotiationMessage(self.agent_id, "AgentB", "ACCEPT", proposal=opponent_proposal)

        # 如果对方提议效用过低，考虑让步并反提议
        print(f"[{self.agent_id}] Current utility: {current_utility:.2f}. Not satisfied, making counter-proposal.")

        # 基于回合数进行让步，或者基于与对方提议的差距进行让步
        # 简单的线性让步策略：随着回合数增加，对精度要求降低，对成本容忍度提高

        # 计算让步幅度，可以与回合数挂钩
        # 让步幅度可以是非线性的，例如在开始阶段让步小，后期让步大
        concession_factor = math.sqrt(negotiation_round) / 10.0 # 假设随回合数平方根增长

        # Agent A 降低精度要求 (从理想精度向最低可接受精度让步)
        # Agent A 提高愿意支付的成本 (从理想成本向最高可接受成本让步)

        # 当前 Agent A 的期望点
        current_desired_accuracy = self.current_proposal.accuracy if self.current_proposal else self.initial_target_accuracy
        current_desired_cost = self.current_proposal.cost if self.current_proposal else self.initial_target_cost

        # 让步：精度从当前期望值向最低可接受值移动
        new_accuracy = current_desired_accuracy - self.concession_rate_accuracy * concession_factor
        new_accuracy = max(new_accuracy, self.reservation_accuracy) # 不低于最低可接受精度

        # 让步：成本从当前期望值向最高可接受值移动
        new_cost = current_desired_cost + self.concession_rate_cost * concession_factor
        new_cost = min(new_cost, self.reservation_cost) # 不高于最高可接受成本

        # 确保新的提议在Agent A的有效范围内
        new_proposal = self._clamp_proposal(new_accuracy, new_cost)

        # 确保新的提议至少比前一个提议对Agent A自己不差太多 (防止过度让步导致效用骤降)
        # 或者确保新的提议的效用高于保留效用
        if self.calculate_utility(new_proposal) < self.get_reservation_utility():
            print(f"[{self.agent_id}] Cannot make further concessions below reservation utility. Rejecting.")
            return NegotiationMessage(self.agent_id, "AgentB", "REJECT")

        self.current_proposal = new_proposal
        return NegotiationMessage(self.agent_id, "AgentB", "PROPOSAL", proposal=new_proposal)

class ServiceProvidingAgent(BaseAgent): # Agent B
    def __init__(self, agent_id: str, utility_calculator: AgentBUntility,
                 min_accuracy_bound: float, max_accuracy_bound: float,
                 min_cost_bound: float, max_cost_bound: float,
                 initial_target_accuracy: float, initial_target_cost: float,
                 concession_rate_accuracy: float, concession_rate_cost: float):
        super().__init__(agent_id, utility_calculator,
                         min_accuracy_bound, max_accuracy_bound,
                         min_cost_bound, max_cost_bound)
        self.initial_target_accuracy = initial_target_accuracy
        self.initial_target_cost = initial_target_cost
        self.concession_rate_accuracy = concession_rate_accuracy
        self.concession_rate_cost = concession_rate_cost

        # Agent B 自身的理想点和最低可接受点
        self.ideal_accuracy = min_accuracy_bound # 理想是提供低精度（低成本）
        self.ideal_cost = max_cost_bound         # 理想是收到高成本
        self.reservation_accuracy = 0.5 # 假设Agent B至少要提供0.5的精度才算服务
        self.reservation_cost = utility_calculator.min_acceptable_cost

    def make_initial_proposal(self) -> Proposal:
        """Agent B 的初始提议：低精度，高成本"""
        # 从一个对自身有利的起始点开始
        return self._clamp_proposal(self.initial_target_accuracy, self.initial_target_cost)

    def respond_to_proposal(self, opponent_proposal: Proposal, negotiation_round: int) -> NegotiationMessage:
        current_utility = self.calculate_utility(opponent_proposal)

        # 如果对方提议满足我方底线且效用足够高，则接受
        if current_utility >= self.get_reservation_utility() and 
           current_utility >= self.calculate_utility(self.current_proposal or opponent_proposal):
            print(f"[{self.agent_id}] Accepting proposal: Accuracy={opponent_proposal.accuracy:.2f}, Cost={opponent_proposal.cost:.2f}")
            self.current_proposal = opponent_proposal
            return NegotiationMessage(self.agent_id, "AgentA", "ACCEPT", proposal=opponent_proposal)

        # 如果对方提议效用过低，考虑让步并反提议
        print(f"[{self.agent_id}] Current utility: {current_utility:.2f}. Not satisfied, making counter-proposal.")

        concession_factor = math.sqrt(negotiation_round) / 10.0

        # Agent B 提高精度 (从当前提供精度向最大可提供精度让步)
        # Agent B 降低要求的成本 (从当前要求成本向最低可接受成本让步)

        # 当前 Agent B 的期望点
        current_desired_accuracy = self.current_proposal.accuracy if self.current_proposal else self.initial_target_accuracy
        current_desired_cost = self.current_proposal.cost if self.current_proposal else self.initial_target_cost

        # 让步：精度从当前期望值向最高可提供值移动
        new_accuracy = current_desired_accuracy + self.concession_rate_accuracy * concession_factor
        new_accuracy = min(new_accuracy, self.utility_calculator.max_producible_accuracy) # 不高于最高可提供精度

        # 让步：成本从当前期望值向最低可接受值移动
        new_cost = current_desired_cost - self.concession_rate_cost * concession_factor
        new_cost = max(new_cost, self.reservation_cost) # 不低于最低可接受成本

        new_proposal = self._clamp_proposal(new_accuracy, new_cost)

        if self.calculate_utility(new_proposal) < self.get_reservation_utility():
            print(f"[{self.agent_id}] Cannot make further concessions below reservation utility. Rejecting.")
            return NegotiationMessage(self.agent_id, "AgentA", "REJECT")

        self.current_proposal = new_proposal
        return NegotiationMessage(self.agent_id, "AgentA", "PROPOSAL", proposal=new_proposal)

代码实践：构建协商框架 (Python)

我们已经定义了核心组件。现在，我们需要一个模拟器来协调两个智能体之间的协商过程。

# --- 4. 模拟器 Orchestrator ---

class NegotiationSimulator:
    def __init__(self, agent_a: RequestingAgent, agent_b: ServiceProvidingAgent, max_rounds: int = 10):
        self.agent_a = agent_a
        self.agent_b = agent_b
        self.max_rounds = max_rounds
        self.negotiation_state = NegotiationState.INITIATED
        self.current_round = 0
        self.final_agreement: Proposal = None

    def run_negotiation(self):
        print(f"n--- Negotiation Started between {self.agent_a.agent_id} and {self.agent_b.agent_id} ---")

        # Agent A 发起初始提议
        initial_proposal_a = self.agent_a.make_initial_proposal()
        current_message = NegotiationMessage(self.agent_a.agent_id, self.agent_b.agent_id, "PROPOSAL", proposal=initial_proposal_a)
        self.agent_a.current_proposal = initial_proposal_a # Agent A 记录自己的提议
        print(f"Round {self.current_round}: {current_message}")

        while self.negotiation_state not in [NegotiationState.ACCEPTED, NegotiationState.REJECTED, 
                                            NegotiationState.WITHDRAWN, NegotiationState.MAX_ROUNDS_REACHED]:
            self.current_round += 1
            if self.current_round > self.max_rounds:
                self.negotiation_state = NegotiationState.MAX_ROUNDS_REACHED
                print(f"--- Negotiation failed: Max rounds ({self.max_rounds}) reached. ---")
                break

            # 智能体交替回应
            if current_message.receiver_id == self.agent_b.agent_id:
                response_message = self.agent_b.respond_to_proposal(current_message.proposal, self.current_round)
                self.agent_a.receive_message(response_message) # Agent A 也要知道对方回应了什么
            else: # current_message.receiver_id == self.agent_a.agent_id
                response_message = self.agent_a.respond_to_proposal(current_message.proposal, self.current_round)
                self.agent_b.receive_message(response_message) # Agent B 也要知道对方回应了什么

            print(f"Round {self.current_round}: {response_message}")
            current_message = response_message

            if current_message.message_type == "ACCEPT":
                self.negotiation_state = NegotiationState.ACCEPTED
                self.final_agreement = current_message.proposal
                print(f"--- Negotiation successful! Agreement reached. ---")
            elif current_message.message_type == "REJECT" or current_message.message_type == "WITHDRAW":
                self.negotiation_state = NegotiationState.NO_AGREEMENT
                print(f"--- Negotiation failed: {current_message.message_type} ---")

        if self.final_agreement:
            print(f"nFinal Agreement:")
            print(f"  Accuracy: {self.final_agreement.accuracy:.2f}")
            print(f"  Cost: {self.final_agreement.cost:.2f}")
            print(f"  Agent A Utility: {self.agent_a.calculate_utility(self.final_agreement):.2f}")
            print(f"  Agent B Utility: {self.agent_b.calculate_utility(self.final_agreement):.2f}")
        else:
            print("nNo agreement was reached.")

运行示例

现在，我们来实例化智能体并运行一次协商。

if __name__ == "__main__":
    # 定义协商边界
    GLOBAL_MIN_ACCURACY = 0.0
    GLOBAL_MAX_ACCURACY = 1.0
    GLOBAL_MIN_COST = 10.0
    GLOBAL_MAX_COST = 100.0

    # Agent A (请求方) 的偏好和底线
    # Agent A 更看重精度，但也有成本预算
    agent_a_utility = AgentAUntility(
        accuracy_weight=50.0, # 精度每增加0.01，效用增加0.5
        cost_weight=1.0,      # 成本每增加1，效用减少1
        min_acceptable_accuracy=0.7, # Agent A 至少需要 70% 精度
        max_acceptable_cost=70.0    # Agent A 最多支付 70
    )
    agent_a = RequestingAgent(
        agent_id="AgentA",
        utility_calculator=agent_a_utility,
        min_accuracy_bound=GLOBAL_MIN_ACCURACY,
        max_accuracy_bound=GLOBAL_MAX_ACCURACY,
        min_cost_bound=GLOBAL_MIN_COST,
        max_cost_bound=GLOBAL_MAX_COST,
        initial_target_accuracy=0.95, # Agent A 初始期望高精度
        initial_target_cost=40.0,     # Agent A 初始期望低成本
        concession_rate_accuracy=0.02, # 每次让步降低2%的精度期望
        concession_rate_cost=3.0       # 每次让步提高3单位的成本容忍度
    )

    # Agent B (服务提供方) 的偏好和底线
    # Agent B 更看重收入，但也要满足基本精度，且精度越高其内部成本越高
    agent_b_utility = AgentBUntility(
        revenue_weight=2.0,       # 收入每增加1，效用增加2
        internal_cost_weight=1.0, # 内部成本每增加1，效用减少1
        min_acceptable_cost=30.0,    # Agent B 至少收到 30 报酬
        max_producible_accuracy=0.98 # Agent B 最多提供 98% 精度
    )
    agent_b = ServiceProvidingAgent(
        agent_id="AgentB",
        utility_calculator=agent_b_utility,
        min_accuracy_bound=GLOBAL_MIN_ACCURACY,
        max_accuracy_bound=GLOBAL_MAX_ACCURACY,
        min_cost_bound=GLOBAL_MIN_COST,
        max_cost_bound=GLOBAL_MAX_COST,
        initial_target_accuracy=0.6,  # Agent B 初始期望提供低精度
        initial_target_cost=90.0,     # Agent B 初始期望高成本
        concession_rate_accuracy=0.03, # 每次让步提高3%的精度提供
        concession_rate_cost=5.0       # 每次让步降低5单位的成本要求
    )

    simulator = NegotiationSimulator(agent_a, agent_b, max_rounds=20)
    simulator.run_negotiation()

    print("n--- Additional Info ---")
    print(f"Agent A Reservation Utility: {agent_a.get_reservation_utility():.2f}")
    print(f"Agent B Reservation Utility: {agent_b.get_reservation_utility():.2f}")

示例输出（可能因随机性和具体参数而异）

--- Negotiation Started between AgentA and AgentB ---
Round 0: [AgentA -> AgentB] PROPOSAL: Accuracy=0.95, Cost=40.00
[AgentB] Current utility: 18.00. Not satisfied, making counter-proposal.
Round 1: [AgentB -> AgentA] PROPOSAL: Accuracy=0.60, Cost=90.00
[AgentA] Current utility: -51.00. Not satisfied, making counter-proposal.
Round 2: [AgentA -> AgentB] PROPOSAL: Accuracy=0.93, Cost=43.00
[AgentB] Current utility: 104.97. Not satisfied, making counter-proposal.
Round 3: [AgentB -> AgentA] PROPOSAL: Accuracy=0.64, Cost=84.76
[AgentA] Current utility: -48.24. Not satisfied, making counter-proposal.
Round 4: [AgentA -> AgentB] PROPOSAL: Accuracy=0.90, Cost=47.24
[AgentB] Current utility: 96.96. Not satisfied, making counter-proposal.
Round 5: [AgentB -> AgentA] PROPOSAL: Accuracy=0.68, Cost=79.28
[AgentA] Current utility: -44.82. Not satisfied, making counter-proposal.
Round 6: [AgentA -> AgentB] PROPOSAL: Accuracy=0.87, Cost=52.76
[AgentB] Current utility: 87.20. Not satisfied, making counter-proposal.
Round 7: [AgentB -> AgentA] PROPOSAL: Accuracy=0.72, Cost=73.54
[AgentA] Current utility: -40.66. Not satisfied, making counter-proposal.
Round 8: [AgentA -> AgentB] PROPOSAL: Accuracy=0.83, Cost=58.94
[AgentB] Current utility: 76.54. Not satisfied, making counter-proposal.
Round 9: [AgentB -> AgentA] PROPOSAL: Accuracy=0.77, Cost=67.58
[AgentA] Current utility: -35.68. Not satisfied, making counter-proposal.
Round 10: [AgentA -> AgentB] PROPOSAL: Accuracy=0.78, Cost=65.86
[AgentB] Current utility: 67.45. Not satisfied, making counter-proposal.
Round 11: [AgentB -> AgentA] PROPOSAL: Accuracy=0.82, Cost=61.32
[AgentA] Current utility: 7.76. Not satisfied, making counter-proposal.
Round 12: [AgentA -> AgentB] PROPOSAL: Accuracy=0.70, Cost=70.00
[AgentA] Cannot make further concessions below reservation utility. Rejecting.
--- Negotiation failed: REJECT ---

No agreement was reached.

--- Additional Info ---
Agent A Reservation Utility: 0.00
Agent B Reservation Utility: 25.00

在这个模拟输出中，智能体A在第12轮拒绝了智能体B的提议，因为再让步就低于其保留效用。最终协商失败。这表明了保留效用在协商中的重要性。我们可以调整智能体的让步策略和初始偏好，以期达成协议。

调整参数后再次运行（例如，降低Agent A的精度要求，提高Agent B的初始提供精度）：

if __name__ == "__main__":
    # ... (GLOBAL_MIN_ACCURACY等参数不变) ...

    agent_a_utility = AgentAUntility(
        accuracy_weight=50.0,
        cost_weight=1.0,
        min_acceptable_accuracy=0.65, # Agent A 降低精度底线到 65%
        max_acceptable_cost=75.0      # Agent A 提高预算到 75
    )
    agent_a = RequestingAgent(
        agent_id="AgentA",
        utility_calculator=agent_a_utility,
        min_accuracy_bound=GLOBAL_MIN_ACCURACY,
        max_accuracy_bound=GLOBAL_MAX_ACCURACY,
        min_cost_bound=GLOBAL_MIN_COST,
        max_cost_bound=GLOBAL_MAX_COST,
        initial_target_accuracy=0.90, # Agent A 初始期望略低
        initial_target_cost=45.0,
        concession_rate_accuracy=0.02,
        concession_rate_cost=3.0
    )

    agent_b_utility = AgentBUntility(
        revenue_weight=2.0,
        internal_cost_weight=1.0,
        min_acceptable_cost=30.0,
        max_producible_accuracy=0.98
    )
    agent_b = ServiceProvidingAgent(
        agent_id="AgentB",
        utility_calculator=agent_b_utility,
        min_accuracy_bound=GLOBAL_MIN_ACCURACY,
        max_accuracy_bound=GLOBAL_MAX_ACCURACY,
        min_cost_bound=GLOBAL_MIN_COST,
        max_cost_bound=GLOBAL_MAX_COST,
        initial_target_accuracy=0.70,  # Agent B 初始期望提供较高精度
        initial_target_cost=85.0,
        concession_rate_accuracy=0.03,
        concession_rate_cost=5.0
    )

    simulator = NegotiationSimulator(agent_a, agent_b, max_rounds=20)
    simulator.run_negotiation()

    print("n--- Additional Info ---")
    print(f"Agent A Reservation Utility: {agent_a.get_reservation_utility():.2f}")
    print(f"Agent B Reservation Utility: {agent_b.get_reservation_utility():.2f}")

调整后的示例输出：

--- Negotiation Started between AgentA and AgentB ---
Round 0: [AgentA -> AgentB] PROPOSAL: Accuracy=0.90, Cost=45.00
[AgentB] Current utility: 60.50. Not satisfied, making counter-proposal.
Round 1: [AgentB -> AgentA] PROPOSAL: Accuracy=0.70, Cost=85.00
[AgentA] Current utility: -40.00. Not satisfied, making counter-proposal.
Round 2: [AgentA -> AgentB] PROPOSAL: Accuracy=0.88, Cost=48.00
[AgentB] Current utility: 68.35. Not satisfied, making counter-proposal.
Round 3: [AgentB -> AgentA] PROPOSAL: Accuracy=0.73, Cost=80.00
[AgentA] Current utility: -35.58. Not satisfied, making counter-proposal.
Round 4: [AgentA -> AgentB] PROPOSAL: Accuracy=0.85, Cost=52.00
[AgentB] Current utility: 75.39. Not satisfied, making counter-proposal.
Round 5: [AgentB -> AgentA] PROPOSAL: Accuracy=0.76, Cost=74.76
[AgentA] Current utility: -30.38. Not satisfied, making counter-proposal.
Round 6: [AgentA -> AgentB] PROPOSAL: Accuracy=0.82, Cost=57.46
[AgentB] Current utility: 81.33. Not satisfied, making counter-proposal.
Round 7: [AgentB -> AgentA] PROPOSAL: Accuracy=0.79, Cost=69.22
[AgentA] Current utility: -24.22. Not satisfied, making counter-proposal.
Round 8: [AgentA -> AgentB] PROPOSAL: Accuracy=0.78, Cost=63.66
[AgentB] Accepting proposal: Accuracy=0.78, Cost=63.66
--- Negotiation successful! Agreement reached. ---

Final Agreement:
  Accuracy: 0.78
  Cost: 63.66
  Agent A Utility: 32.34
  Agent B Utility: 93.94

--- Additional Info ---
Agent A Reservation Utility: 0.00
Agent B Reservation Utility: 25.00

这次协商在第8轮成功达成协议。最终的精度是0.78，成本是63.66。Agent A的效用是32.34，Agent B的效用是93.94。双方都获得了正的效用，且高于各自的保留效用，说明这是一个成功的交易。

实验与分析：协商结果的评估

通过上述代码，我们可以进行一系列实验，评估不同参数和策略对协商结果的影响：

1. 智能体偏好对协商的影响：

• 权重调整： 提高Agent A的 accuracy_weight 或降低 cost_weight 会使其更倾向于高精度方案，反之亦然。Agent B同理。
• 保留效用调整： 提高 min_acceptable_accuracy 或降低 max_acceptable_cost (对Agent A)，或者降低 min_acceptable_cost (对Agent B)，会使智能体更“强硬”，协商成功的可能性可能降低。

2. 让步策略的有效性：

• 让步率 (concession_rate)： 较高的让步率可能加速协商，但可能导致智能体在谈判中过早地做出大让步。较低的让步率可能延长协商，甚至导致失败。
• 让步函数： 我们可以尝试非线性让步，例如在早期回合让步较小，在后期回合让步较大（急于达成协议）。或者，根据对方的让步幅度来调整自己的让步。

3. 协商效率：

• 回合数： 达成协议所需的回合数是衡量效率的指标。
• 成功率： 协商成功达成协议的概率。

4. 协商质量：

• 最终效用： 双方智能体在达成协议后各自获得的效用值。
• 社会福利： 双方效用之和 ($U_A + U_B$)，可以衡量协议对整个系统的整体价值。一个好的协商机制应该在一定程度上提升社会福利。

可能遇到的问题：

• 协商僵局 (Deadlock)： 双方都坚持自己的立场，不愿让步，导致协商无限期进行或达到最大回合数而失败。
• 非最优解 (Suboptimal Agreement)： 达成协议，但存在其他协议能让双方都获得更高的效用，或至少一方更高而另一方不更低。
• 协商失败 (Failure)： 双方无法在可接受的范围内找到共同点。

进阶话题与未来展望

我们今天构建的框架是一个基础但功能完备的Agent-to-Agent协商系统。在此基础上，我们可以进一步探索更复杂的场景和技术：

• 多智能体协商 (Multi-Agent Negotiation)： 当有三个或更多智能体参与协商时，复杂性会急剧增加。可能涉及组队、联盟形成、投票等机制。
• 动态协商环境： 智能体的偏好或外部环境（如计算资源价格、服务需求）可能在协商过程中发生变化，智能体需要能够动态调整策略。
• 学习型智能体： 智能体可以通过强化学习或其他机器学习方法，从历史协商经验中学习并优化其协商策略，例如预测对手的行为、识别更有效的让步模式。
• 信任与声誉机制 (Trust and Reputation)： 在多次协商中，智能体可以建立信任和声誉。与高信任度的智能体协商可能更快、更有效。
• 非对称信息协商： 智能体可能不完全了解对方的效用函数或保留效用。协商策略需要考虑如何从对方的行为中推断其隐藏信息。
• 多种协商协议： 除了我们使用的基于提议-反提议的协议，还有拍卖、投标、最后通牒博弈等多种协议形式，适用于不同的场景。
• 形式化验证： 对于关键任务系统，可能需要对协商协议进行形式化验证，以确保其在所有可能情况下都能达到预期行为。

深入思考与实践指引

本次讲座我们从博弈论基础出发，逐步构建了一个智能体之间进行计算成本与答案精度协商的逻辑框架和Python实现。我们探讨了效用函数的建模、协商协议的设计以及智能体决策策略的实现。

核心挑战在于如何精确量化智能体的偏好（通过效用函数和权重），以及如何设计既能促成协议又能最大化自身利益的让步策略。这需要我们深入理解智能体在各种情境下的行为模式和目标。

实践中，我鼓励各位尝试调整代码中的参数，例如：

• 改变 accuracy_weight 和 cost_weight，观察智能体行为的变化。
• 调整 min_acceptable_accuracy 和 max_acceptable_cost 等底线参数。
• 修改 concession_rate_accuracy 和 concession_rate_cost，尝试不同的让步速度。
• 引入非线性的让步策略，例如在协商初期让步较小，后期逐渐加大。
• 甚至尝试实现一个简单的“对手模型”，让智能体根据对手的历史行为来调整自己的策略。

通过这些实验，您将更深入地理解智能体协商的复杂性和魅力。这个模型不仅限于计算成本和精度，其核心思想可以推广到任何需要权衡多个冲突目标，并在多个智能体间达成共识的场景。这正是多智能体系统在现实世界中发挥巨大潜力的关键所在。