231 lines
8.0 KiB
Python
231 lines
8.0 KiB
Python
"""AuctionHouse - 拍卖机制，基于竞价选择 Agent"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from agentkit.marketplace.wealth import WealthTracker
|
|
|
|
|
|
@dataclass
|
|
class Bid:
|
|
"""Agent 竞价信息"""
|
|
|
|
agent_name: str
|
|
architecture: str # "react", "rewoo", "plan_exec", "reflexion", "direct"
|
|
estimated_steps: int
|
|
estimated_cost: float # estimated token cost
|
|
confidence: float # 0.0-1.0 confidence in completing the task
|
|
payment_offer: float # how much the agent "charges"
|
|
capabilities: list[str] = field(default_factory=list)
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.estimated_cost < 0:
|
|
raise ValueError(f"estimated_cost must be non-negative, got {self.estimated_cost}")
|
|
if not (0.0 <= self.confidence <= 1.0):
|
|
raise ValueError(f"confidence must be between 0.0 and 1.0, got {self.confidence}")
|
|
if self.payment_offer < 0:
|
|
raise ValueError(f"payment_offer must be non-negative, got {self.payment_offer}")
|
|
|
|
|
|
@dataclass
|
|
class AuctionResult:
|
|
"""拍卖结果"""
|
|
|
|
winner: Bid | None
|
|
all_bids: list[Bid]
|
|
selection_reason: str
|
|
total_bidders: int
|
|
|
|
|
|
class AuctionHouse:
|
|
"""Auction-based agent selection mechanism.
|
|
|
|
Default disabled. Enable via marketplace.auction_enabled: true in config.
|
|
When enabled, Layer 2 routing uses auction instead of capability matching.
|
|
"""
|
|
|
|
def __init__(self, wealth_tracker: WealthTracker | None = None) -> None:
|
|
self._wealth_tracker = wealth_tracker or WealthTracker()
|
|
|
|
async def run_auction(self, task_description: str, bidders: list[Bid]) -> AuctionResult:
|
|
"""Run auction among bidders, select winner.
|
|
|
|
Scoring formula:
|
|
score = (confidence / max(estimated_cost, 0.001)) * wealth_factor
|
|
|
|
wealth_factor = 1.0 + (wealth / 1000.0) # wealth bonus, diminishing returns
|
|
"""
|
|
if not bidders:
|
|
return AuctionResult(
|
|
winner=None,
|
|
all_bids=[],
|
|
selection_reason="No bidders participated",
|
|
total_bidders=0,
|
|
)
|
|
|
|
# Filter out bankrupt agents
|
|
eligible = [
|
|
b for b in bidders
|
|
if not self._wealth_tracker.is_bankrupt(b.agent_name)
|
|
]
|
|
|
|
if not eligible:
|
|
return AuctionResult(
|
|
winner=None,
|
|
all_bids=bidders,
|
|
selection_reason="All bidders are bankrupt",
|
|
total_bidders=len(bidders),
|
|
)
|
|
|
|
# Score each bid
|
|
scored: list[tuple[Bid, float]] = []
|
|
for bid in eligible:
|
|
score = self.score_bid(bid)
|
|
scored.append((bid, score))
|
|
|
|
# Select highest score
|
|
scored.sort(key=lambda x: x[1], reverse=True)
|
|
winner, winner_score = scored[0]
|
|
|
|
return AuctionResult(
|
|
winner=winner,
|
|
all_bids=bidders,
|
|
selection_reason=(
|
|
f"Agent '{winner.agent_name}' won with score {winner_score:.4f} "
|
|
f"(confidence={winner.confidence}, cost={winner.estimated_cost}, "
|
|
f"wealth_factor={self._wealth_tracker.get_wealth_factor(winner.agent_name):.4f})"
|
|
),
|
|
total_bidders=len(bidders),
|
|
)
|
|
|
|
def score_bid(self, bid: Bid) -> float:
|
|
"""Calculate bid score without running full auction"""
|
|
wealth_factor = self._wealth_tracker.get_wealth_factor(bid.agent_name)
|
|
score = (bid.confidence / max(bid.estimated_cost, 0.001)) * wealth_factor
|
|
return score
|
|
|
|
def filter_by_capabilities(
|
|
self, bidders: list[Bid], required_capabilities: list[str]
|
|
) -> list[Bid]:
|
|
"""Filter bidders whose capabilities include ALL required capabilities.
|
|
|
|
Args:
|
|
bidders: List of Bid objects to filter.
|
|
required_capabilities: Capabilities that a bidder must ALL possess.
|
|
|
|
Returns:
|
|
List of Bids whose capabilities list includes every required capability.
|
|
"""
|
|
if not required_capabilities:
|
|
return bidders
|
|
required_set = {cap.lower() for cap in required_capabilities}
|
|
return [
|
|
b for b in bidders
|
|
if required_set.issubset({cap.lower() for cap in b.capabilities})
|
|
]
|
|
|
|
async def run_vickrey_auction(
|
|
self,
|
|
task_description: str,
|
|
bidders: list[Bid],
|
|
required_capabilities: list[str] | None = None,
|
|
) -> AuctionResult:
|
|
"""Run a Vickrey (second-price sealed-bid) auction.
|
|
|
|
In a Vickrey auction each bidder submits a sealed bid (their
|
|
estimated_cost). The lowest estimated_cost wins, but the winner
|
|
*pays* the second-lowest estimated_cost rather than their own.
|
|
This is incentive-compatible: agents' dominant strategy is to bid
|
|
truthfully.
|
|
|
|
Steps:
|
|
1. Filter by required_capabilities (if provided).
|
|
2. Filter out bankrupt agents.
|
|
3. If only 1 eligible bidder → wins, pays 0.
|
|
4. If 2+ eligible bidders → lowest cost wins, pays second-lowest.
|
|
5. Update WealthTracker: winner earns (payment - cost_estimate).
|
|
6. Return AuctionResult with selection_reason.
|
|
|
|
Args:
|
|
task_description: Description of the task being auctioned.
|
|
bidders: List of Bid objects.
|
|
required_capabilities: Optional list of capabilities that bidders
|
|
must possess to be eligible.
|
|
|
|
Returns:
|
|
AuctionResult with the winner and Vickrey outcome details.
|
|
"""
|
|
# 1. Capability filtering
|
|
eligible = (
|
|
self.filter_by_capabilities(bidders, required_capabilities)
|
|
if required_capabilities
|
|
else list(bidders)
|
|
)
|
|
|
|
# 2. Filter out bankrupt agents
|
|
eligible = [
|
|
b for b in eligible
|
|
if not self._wealth_tracker.is_bankrupt(b.agent_name)
|
|
]
|
|
|
|
# No bidders at all
|
|
if not bidders:
|
|
return AuctionResult(
|
|
winner=None,
|
|
all_bids=bidders,
|
|
selection_reason="No bidders participated",
|
|
total_bidders=0,
|
|
)
|
|
|
|
# All eligible bidders filtered out (bankrupt or no capabilities)
|
|
if not eligible:
|
|
return AuctionResult(
|
|
winner=None,
|
|
all_bids=bidders,
|
|
selection_reason="No eligible bidders (bankrupt or missing capabilities)",
|
|
total_bidders=len(bidders),
|
|
)
|
|
|
|
# 3. Sort by estimated_cost ascending (lowest cost = best bid)
|
|
# Apply minimum cost floor to prevent zero-cost bid manipulation
|
|
MIN_COST = 0.001
|
|
for b in eligible:
|
|
b.estimated_cost = max(b.estimated_cost, MIN_COST)
|
|
eligible = sorted(eligible, key=lambda b: b.estimated_cost)
|
|
winner = eligible[0]
|
|
|
|
# 4. Determine payment (second-price rule)
|
|
if len(eligible) == 1:
|
|
# Single bidder: payment equals their own cost (no profit, no loss)
|
|
payment = winner.estimated_cost
|
|
else:
|
|
payment = eligible[1].estimated_cost
|
|
|
|
# 5. Update WealthTracker
|
|
profit = payment - winner.estimated_cost
|
|
self._wealth_tracker.reward(winner.agent_name, profit)
|
|
|
|
# 6. Build selection_reason
|
|
if len(eligible) == 1:
|
|
reason = (
|
|
f"Vickrey auction: Agent '{winner.agent_name}' won as sole eligible bidder "
|
|
f"(cost={winner.estimated_cost}, payment={payment:.4f}, profit=0)"
|
|
)
|
|
else:
|
|
second = eligible[1]
|
|
reason = (
|
|
f"Vickrey auction: Agent '{winner.agent_name}' won with lowest cost "
|
|
f"({winner.estimated_cost}), pays second-lowest cost ({payment}) from "
|
|
f"'{second.agent_name}'; profit={profit:.4f}"
|
|
)
|
|
|
|
return AuctionResult(
|
|
winner=winner,
|
|
all_bids=bidders,
|
|
selection_reason=reason,
|
|
total_bidders=len(bidders),
|
|
)
|