101 lines
3.2 KiB
Python
101 lines
3.2 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)
|
|
|
|
|
|
@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
|