"""
Pydantic schemas for validating and constraining LLM action outputs
"""
from pydantic import BaseModel, Field, field_validator
from typing import Dict, Any, Optional, Literal, Union
import json
import yaml
import os
[docs]
class BaseAction(BaseModel):
"""Base action schema"""
type: str
[docs]
class OfferAction(BaseModel):
"""Price bargaining offer action"""
type: Literal["offer"] = Field(..., description="Must be exactly 'offer'")
price: float = Field(..., ge=0, description="Price in euros, must be positive")
[docs]
class AcceptAction(BaseModel):
"""Accept action for any game type"""
type: Literal["accept"] = Field(..., description="Must be exactly 'accept'")
[docs]
class CounterAction(BaseModel):
"""Price bargaining counter-offer action"""
type: Literal["counter"] = Field(..., description="Must be exactly 'counter'")
price: float = Field(..., ge=0, description="Price in euros, must be positive")
[docs]
class RejectAction(BaseModel):
"""Reject action for any game type"""
type: Literal["reject"] = Field(..., description="Must be exactly 'reject'")
[docs]
class ResourceProposalAction(BaseModel):
"""Resource allocation proposal with gpu_hours and cpu_hours"""
type: Literal["propose"] = Field(..., description="Must be exactly 'propose'")
gpu_hours: float = Field(..., ge=0, description="GPU hours requested")
cpu_hours: float = Field(..., ge=0, description="CPU hours requested")
[docs]
class ProposeTradeAction(BaseModel):
"""Alternative resource allocation trade proposal"""
type: Literal["propose_trade"] = Field(..., description="Must be exactly 'propose_trade'")
offer: Dict[str, float] = Field(..., description="Resources offered")
request: Dict[str, float] = Field(..., description="Resources requested")
[docs]
class IntegrativeProposalAction(BaseModel):
"""Integrative negotiation proposal with constrained discrete values"""
type: Literal["propose"] = Field(..., description="Must be exactly 'propose'")
server_room: Union[int, float] = Field(..., description="Server room size")
meeting_access: Union[int, float] = Field(..., description="Meeting access hours")
cleaning: str = Field(..., description="Cleaning responsibility")
branding: str = Field(..., description="Branding visibility level")
@field_validator('server_room')
@classmethod
def validate_server_room(cls, v):
"""Validate server room size"""
if v <= 75:
#print(f"⚠️ [VALIDATION] server_room corrected from {v} to 50")
return 50
elif v <= 125:
#print(f"⚠️ [VALIDATION] server_room corrected from {v} to 100")
return 100
else:
#print(f"⚠️ [VALIDATION] server_room corrected from {v} to 150")
return 150
@field_validator('meeting_access')
@classmethod
def validate_meeting_access(cls, v):
"""Validate meeting access hours"""
if v <= 3:
#print(f"⚠️ [VALIDATION] meeting_access corrected from {v} to 2")
return 2
elif v <= 5:
#print(f"⚠️ [VALIDATION] meeting_access corrected from {v} to 4")
return 4
else:
#print(f"⚠️ [VALIDATION] meeting_access corrected from {v} to 7")
return 7
@field_validator('cleaning')
@classmethod
def validate_cleaning(cls, v):
"""Validate cleaning responsibility"""
if str(v).lower() in ['it']:
#print(f"⚠️ [VALIDATION] cleaning corrected from {v} to 'IT'")
return "IT"
elif str(v).lower() in ['shared']:
#print(f"⚠️ [VALIDATION] cleaning corrected from {v} to 'Shared'")
return "Shared"
else:
#print(f"⚠️ [VALIDATION] cleaning corrected from {v} to 'Outsourced'")
return "Outsourced"
@field_validator('branding')
@classmethod
def validate_branding(cls, v):
"""Validate branding visibility"""
if str(v).lower() in ['minimal']:
#print(f"⚠️ [VALIDATION] branding corrected from {v} to 'Minimal'")
return "Minimal"
elif str(v).lower() in ['moderate']:
#print(f"⚠️ [VALIDATION] branding corrected from {v} to 'Moderate'")
return "Moderate"
else:
#print(f"⚠️ [VALIDATION] branding corrected from {v} to 'Prominent'")
return "Prominent"
# Union of all possible negotiation actions across game types
GameAction = Union[
OfferAction,
AcceptAction,
CounterAction,
RejectAction,
ResourceProposalAction,
ProposeTradeAction,
IntegrativeProposalAction
]
"""Union type encompassing all valid negotiation actions.
This type union provides comprehensive coverage of all supported action
types across different negotiation game scenarios. It serves as the
canonical reference for valid action schemas in the validation pipeline
and enables type-safe action processing throughout the platform.
Included Action Types:
- OfferAction: Price-based offers in bargaining scenarios
- AcceptAction: Universal acceptance across all game types
- CounterAction: Price-based counter-offers in bargaining
- RejectAction: Universal rejection across all game types
- ResourceProposalAction: Computing resource allocation proposals
- ProposeTradeAction: Complex resource trading proposals
- IntegrativeProposalAction: Multi-issue integrative negotiations
Usage:
This union is used internally by the validation system and should
not typically be referenced directly in user code. Instead, use
the validate_and_constrain_action() function for action processing.
"""
[docs]
def validate_and_constrain_action(raw_response: str, game_type: str) -> Dict[str, Any]:
"""Validate and constrain LLM response to proper negotiation action format.
This function serves as the primary entry point for converting raw LLM
responses into validated, game-appropriate action dictionaries. It applies
comprehensive validation using Pydantic schemas, performs automatic error
correction, and provides robust fallback handling for malformed responses.
The validation process includes JSON parsing, game-specific schema
validation, automatic value correction, and intelligent error recovery
to maximize the success rate of action parsing from LLM outputs.
Args:
raw_response (str): Raw JSON string response from the LLM, which
may contain formatting issues, invalid values, or non-standard
structure requiring correction and validation.
game_type (str): Type of negotiation game context for validation
('price_bargaining', 'company_car', 'resource_allocation',
'integrative_negotiations'). Determines which validation
schemas and correction rules are applied.
Returns:
Dict[str, Any]: Validated and constrained action dictionary
containing properly formatted action data with all required
fields and valid values according to game-specific rules.
Includes type field and game-appropriate additional fields.
Raises:
ValueError: If the response cannot be parsed, validated, or
corrected into a valid action format despite multiple
correction attempts and fallback strategies.
Example:
>>> response = '{\"type\": \"offer\", \"price\": 28000}'
>>> action = validate_and_constrain_action(response, 'company_car')
>>> print(action)
{'type': 'offer', 'price': 28000.0}
>>> # Auto-correction example
>>> response = '{\"type\": \"propose\", \"server_room\": 75}'
>>> action = validate_and_constrain_action(response, 'integrative_negotiations')
>>> print(action['server_room']) # Auto-corrected to 50
50
Note:
This function includes extensive error handling and automatic
correction capabilities. It integrates with the auto_correct_action
function to handle common LLM output mistakes and provides graceful
degradation when validation fails completely.
"""
import json
try:
# Parse JSON
parsed = json.loads(raw_response.strip())
# Debug: Log the parsed JSON
print(f"[DEBUG] Parsed JSON: {parsed}")
# Pre-validation check for 'type' field
action_type = parsed.get("type", None)
if not action_type:
raise ValueError("Missing 'type' field in action")
# Validate based on game type and action type
action_type = parsed.get("type", "")
if action_type == "accept":
action = AcceptAction(**parsed)
elif action_type == "reject":
action = RejectAction(**parsed)
elif action_type == "offer" and game_type in ["price_bargaining", "company_car"]:
action = OfferAction(**parsed)
elif action_type == "counter" and game_type in ["price_bargaining", "company_car"]:
action = CounterAction(**parsed)
elif action_type == "propose" and game_type == "resource_allocation":
action = ResourceProposalAction(**parsed)
elif action_type == "propose_trade" and game_type == "resource_allocation":
action = ProposeTradeAction(**parsed)
elif action_type == "propose" and game_type == "integrative_negotiations":
action = IntegrativeProposalAction(**parsed)
else:
# Try to auto-correct common mistakes
corrected = auto_correct_action(parsed, game_type)
if corrected:
return corrected
raise ValueError(f"Invalid action type '{action_type}' for game '{game_type}'")
return action.dict()
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON: {e}")
except Exception as e:
# Debug: Log the error and the raw response
print(f"[DEBUG] Validation error: {e}")
print(f"[DEBUG] Raw response: {raw_response}")
raise ValueError(f"Action validation failed: {e}")
[docs]
def auto_correct_action(parsed: Dict[str, Any], game_type: str) -> Optional[Dict[str, Any]]:
"""Auto-correct common LLM mistakes in action format with intelligent fallbacks.
This function implements sophisticated error correction for common mistakes
made by language models when generating negotiation actions. It uses
pattern matching, keyword detection, and game-specific knowledge to
recover valid actions from malformed or non-standard LLM outputs.
Args:
parsed (Dict[str, Any]): Parsed action dictionary that failed
standard validation, potentially containing incorrect field
names, invalid values, or missing required information.
game_type (str): Type of negotiation game for context-specific
correction rules ('price_bargaining', 'company_car',
'resource_allocation', 'integrative_negotiations').
Returns:
Optional[Dict[str, Any]]: Corrected action dictionary if successful
correction is possible, None if the input cannot be meaningfully
corrected into a valid action format. Corrected actions conform
to the appropriate schema for the specified game type.
Correction Strategies:
- Action type normalization (accept/agree/yes → accept)
- Alternative field name handling (amount/value → price)
- Game-specific field mapping (x/y → gpu_hours/cpu_hours)
- Value constraint enforcement for integrative negotiations
- Format conversion (nested proposal structures)
Example:
>>> # Correct alternative acceptance terms
>>> parsed = {'type': 'agree'}
>>> corrected = auto_correct_action(parsed, 'company_car')
>>> print(corrected)
{'type': 'accept'}
>>> # Correct alternative field names
>>> parsed = {'type': 'offer', 'amount': 25000}
>>> corrected = auto_correct_action(parsed, 'price_bargaining')
>>> print(corrected)
{'type': 'offer', 'price': 25000.0}
Note:
This function is designed to be permissive and attempts multiple
correction strategies before giving up. It logs correction actions
for debugging and research purposes when values are automatically
adjusted to meet constraint requirements.
"""
action_type = parsed.get("type", "").lower()
# Auto-correct common acceptance variations
if any(word in action_type for word in ["accept", "agree", "yes"]):
return {"type": "accept"}
# Auto-correct common rejection variations
if any(word in action_type for word in ["reject", "decline", "no"]):
return {"type": "reject"}
# Auto-correct offer variations for price bargaining
if game_type in ["price_bargaining", "company_car"] and any(word in action_type for word in ["offer", "bid", "propose"]):
price = parsed.get("price") or parsed.get("amount") or parsed.get("value")
if price is not None:
return {"type": "offer", "price": float(price)}
# Auto-correct counter variations for price bargaining
if game_type in ["price_bargaining", "company_car"] and any(word in action_type for word in ["counter", "counteroffer"]):
price = parsed.get("price") or parsed.get("amount") or parsed.get("value")
if price is not None:
return {"type": "counter", "price": float(price)}
# Auto-correct resource allocation variations
if game_type == "resource_allocation" and any(word in action_type for word in ["trade", "propose", "offer"]):
# Try standard format first (gpu_hours, cpu_hours)
gpu_hours = parsed.get("gpu_hours") or parsed.get("gpu") or parsed.get("x")
cpu_hours = parsed.get("cpu_hours") or parsed.get("cpu") or parsed.get("y") or parsed.get("bw")
if gpu_hours is not None and cpu_hours is not None:
return {"type": "propose", "gpu_hours": float(gpu_hours), "cpu_hours": float(cpu_hours)}
# Try trade format as fallback
offer = parsed.get("offer") or parsed.get("give") or {}
request = parsed.get("request") or parsed.get("want") or parsed.get("ask") or {}
if offer and request:
return {"type": "propose_trade", "offer": offer, "request": request}
# Auto-correct integrative negotiations variations -> not necessary
if game_type == "integrative_negotiations" and any(word in action_type for word in ["propose", "offer"]):
# Load valid options from game config
config_path = os.path.join(os.path.dirname(__file__), '..', 'configs', 'game_configs.yaml')
try:
with open(config_path, 'r') as f:
config = yaml.safe_load(f)
integrative_config = config.get('integrative_negotiations', {}).get('issues', {})
except Exception:
integrative_config = {}
# Check if it's already in the correct nested format
proposal = parsed.get("proposal")
if proposal and isinstance(proposal, dict):
# Auto-correct invalid values to valid discrete options
corrected_proposal = {}
# Auto-correct server_room
if "server_room" in proposal:
server_room = proposal["server_room"]
valid_options = integrative_config.get('server_room', {}).get('options', [50, 100, 150])
if server_room not in valid_options:
if server_room <= 75:
print(f"⚠️ [VALIDATION] server_room corrected from {server_room} to 50")
corrected_proposal["server_room"] = 50
elif server_room <= 125:
print(f"⚠️ [VALIDATION] server_room corrected from {server_room} to 100")
corrected_proposal["server_room"] = 100
else:
print(f"⚠️ [VALIDATION] server_room corrected from {server_room} to 150")
corrected_proposal["server_room"] = 150
else:
corrected_proposal["server_room"] = server_room
# Auto-correct meeting_access
if "meeting_access" in proposal:
meeting_access = proposal["meeting_access"]
valid_options = integrative_config.get('meeting_access', {}).get('options', [2, 4, 7])
if meeting_access not in valid_options:
if meeting_access <= 3:
print(f"⚠️ [VALIDATION] meeting_access corrected from {meeting_access} to 2")
corrected_proposal["meeting_access"] = 2
elif meeting_access <= 5:
print(f"⚠️ [VALIDATION] meeting_access corrected from {meeting_access} to 4")
corrected_proposal["meeting_access"] = 4
else:
print(f"⚠️ [VALIDATION] meeting_access corrected from {meeting_access} to 7")
corrected_proposal["meeting_access"] = 7
else:
corrected_proposal["meeting_access"] = meeting_access
# Auto-correct cleaning values
if "cleaning" in proposal:
cleaning = str(proposal["cleaning"])
valid_options = integrative_config.get('cleaning', {}).get('options', ["IT", "Shared", "Outsourced"])
if cleaning not in valid_options:
cleaning_lower = cleaning.lower()
if "it" in cleaning_lower:
print(f"⚠️ [VALIDATION] cleaning corrected from {cleaning} to 'IT'")
corrected_proposal["cleaning"] = "IT"
elif "shared" in cleaning_lower:
print(f"⚠️ [VALIDATION] cleaning corrected from {cleaning} to 'Shared'")
corrected_proposal["cleaning"] = "Shared"
else:
print(f"⚠️ [VALIDATION] cleaning corrected from {cleaning} to 'Outsourced'")
corrected_proposal["cleaning"] = "Outsourced"
else:
corrected_proposal["cleaning"] = cleaning
# Auto-correct branding values
if "branding" in proposal:
branding = str(proposal["branding"])
valid_options = integrative_config.get('branding', {}).get('options', ["Minimal", "Moderate", "Prominent"])
if branding not in valid_options:
branding_lower = branding.lower()
if "minimal" in branding_lower:
print(f"⚠️ [VALIDATION] branding corrected from {branding} to 'Minimal'")
corrected_proposal["branding"] = "Minimal"
elif "moderate" in branding_lower:
print(f"⚠️ [VALIDATION] branding corrected from {branding} to 'Moderate'")
corrected_proposal["branding"] = "Moderate"
else:
print(f"⚠️ [VALIDATION] branding corrected from {branding} to 'Prominent'")
corrected_proposal["branding"] = "Prominent"
else:
corrected_proposal["branding"] = branding
# Use original values for any keys not corrected
for key, value in proposal.items():
if key not in corrected_proposal:
corrected_proposal[key] = value
return {"type": "propose", "proposal": corrected_proposal}
# Try to extract proposal from flat format
server_room = parsed.get("server_room")
meeting_access = parsed.get("meeting_access")
cleaning = parsed.get("cleaning")
branding = parsed.get("branding")
if all(x is not None for x in [server_room, meeting_access, cleaning, branding]):
# Apply same corrections to flat format using config
server_options = integrative_config.get('server_room', {}).get('options', [50, 100, 150])
meeting_options = integrative_config.get('meeting_access', {}).get('options', [2, 4, 7])
corrected_server_room = server_room
if server_room not in server_options:
corrected_server_room = 50 if server_room <= 75 else (100 if server_room <= 125 else 150)
print(f"⚠️ [VALIDATION] server_room corrected from {server_room} to {corrected_server_room}")
corrected_meeting_access = meeting_access
if meeting_access not in meeting_options:
corrected_meeting_access = 2 if meeting_access <= 3 else (4 if meeting_access <= 5 else 7)
print(f"⚠️ [VALIDATION] meeting_access corrected from {meeting_access} to {corrected_meeting_access}")
return {"type": "propose", "proposal": {
"server_room": corrected_server_room,
"meeting_access": corrected_meeting_access,
"cleaning": str(cleaning).title(),
"branding": str(branding).title()
}}
return None