Spaces:
Sleeping
Sleeping
File size: 2,845 Bytes
c1be7c3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | # Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.
"""
Data models for the ProcureRL Environment.
The ProcureRL environment is a procurement negotiation RL environment where
an LLM agent learns to negotiate against scripted supplier opponents.
"""
from typing import Optional, List, Dict, Any
from pydantic import BaseModel, Field, ConfigDict
try:
from openenv.core.env_server.types import Action, Observation, State as OpenEnvState
except ImportError:
OpenEnvState = object
class NegotiationAction(BaseModel):
model_config = ConfigDict(extra="allow")
move_type: str = Field(
default="make_offer",
description="Choose action: make_offer (propose), accept (take current deal), reject (walk away), bundle (multi-issue offer)",
)
terms: Dict[str, Any] = Field(
default_factory=lambda: {"price": 45000},
description='For single_issue: {"price": 45000}. For multi_issue: {"price": 45000, "payment_days": 30}. For adversarial: add "support_hours": 100',
)
message: str = Field(
default="I value our partnership and believe we can reach a fair agreement together.",
description="Write a collaborative message. Use: partnership, mutual, flexible, understand, solution. Avoid: demand, final offer, ultimatum",
)
def model_post_init(self, *args, **kwargs):
valid_moves = ("make_offer", "accept", "reject", "bundle")
if self.move_type not in valid_moves:
raise ValueError(
f"Invalid move_type: {self.move_type}. Must be one of {valid_moves}"
)
class NegotiationObservation(BaseModel):
model_config = ConfigDict(extra="allow")
task_id: str = ""
round_number: int = 0
max_rounds: int = 0
supplier_message: str = ""
current_offer: Dict[str, Any] = Field(default_factory=dict)
last_4_exchanges: List[Dict] = Field(default_factory=list)
buyer_constraints: Dict[str, Any] = Field(default_factory=dict)
rapport_hint: str = "neutral"
done: bool = False
reward: Optional[float] = None
metadata: Dict[str, Any] = Field(default_factory=dict)
class NegotiationState(BaseModel):
model_config = ConfigDict(extra="allow", validate_assignment=True)
task_id: str = ""
episode_id: str = ""
round_number: int = 0
step_count: int = 0 # Required by OpenEnv web interface
rapport_score: float = 0.5
consecutive_concessions: int = 0
deal_reached: bool = False
final_terms: Optional[Dict] = None
cumulative_reward: float = 0.0
def __getitem__(self, key):
return getattr(self, key)
def get(self, key, default=None):
return getattr(self, key, default)
|