o2o-negosium-original/agent/negotiation/cards/action_card_mapper.py

57 lines
2.0 KiB
Python

"""ActionCardMapper — action_id ↔ card_id 매핑 (config 주입형).
Chat_server 의 매퍼는 JSON 파일에 결합돼 있었다. 여기서는 TenantConfig.action_mapping
(ActionMappingConfig)에서 주입받아 테넌트별로 다른 카드셋을 지원한다.
PoC 는 카드 매핑을 고정한다(action_space_size ↔ Q-Table 차원 정합성 리스크 회피, 계획서).
"""
from typing import Dict, List, Optional
import numpy as np
from tenancy.config import ActionMappingConfig
class ActionCardMapper:
def __init__(self, config: ActionMappingConfig):
self._config = config
self._action_to_card: Dict[int, str] = {}
self._card_to_action: Dict[str, int] = {}
self._rebuild()
def _rebuild(self):
self._action_to_card = {int(a): c for a, c in self._config.action_to_card.items()}
self._card_to_action = {c: a for a, c in self._action_to_card.items()}
def reload(self, config: ActionMappingConfig):
"""카드 동기화/테넌트 reload 시 매핑 교체 (P6)."""
self._config = config
self._rebuild()
@property
def action_space_size(self) -> int:
return len(self._action_to_card)
def get_card_id(self, action_id: int) -> Optional[str]:
return self._action_to_card.get(action_id)
def get_action_id(self, card_id: str) -> Optional[int]:
return self._card_to_action.get(card_id)
def action_ids(self) -> List[int]:
return sorted(self._action_to_card.keys())
def available_mask(self, used_action_ids: Optional[set] = None) -> np.ndarray:
"""중복 방지 마스킹: 이미 사용한 action 은 False. 정책 select 시 곱해 제외한다.
길이 = action_space_size, dtype=bool. used_action_ids 가 None 이면 전부 True.
"""
n = self.action_space_size
mask = np.ones(n, dtype=bool)
if used_action_ids:
for a in used_action_ids:
if 0 <= a < n:
mask[a] = False
return mask