diff --git a/agent/eval_harness/simulator.py b/agent/eval_harness/simulator.py index 5b89dc1..7ca4705 100644 --- a/agent/eval_harness/simulator.py +++ b/agent/eval_harness/simulator.py @@ -32,7 +32,7 @@ class EpisodeResult: def run_episode(policy, buyer: HeuristicBuyer, scenario: Scenario, state_cfg: StateConfig, reward_cfg: RewardConfig, action_space_size: int, max_turns: int = 5, learn: bool = True) -> EpisodeResult: - rc = RewardCalculator(reward_cfg) + rc = RewardCalculator(reward_cfg, state_cfg) episode = EpisodeState() total_r = 0.0 # 협력사는 목표가보다 높게 시작(협상 여지). KT 가 카드로 앵커가 이하까지 끌어내린다. diff --git a/agent/negotiation/qtable/domain/service/reward_calculator.py b/agent/negotiation/qtable/domain/service/reward_calculator.py index 641d852..19427fe 100644 --- a/agent/negotiation/qtable/domain/service/reward_calculator.py +++ b/agent/negotiation/qtable/domain/service/reward_calculator.py @@ -1,16 +1,22 @@ -"""RewardCalculator — 협상 라운드 보상 계산 (우리 자체 공식, RewardConfig 주입). +"""RewardCalculator — 협상 라운드 보상 계산 (설계 명세 v4 정합, RewardConfig/StateConfig 주입). 클린룸: 보상 공식의 '형태'(가격보상 + 종료보상 - 페널티, 동적 가중치)는 기능적 아이디어이고, -아래 산식은 우리 자체 설계다(특정 고객 산식 복제 아님). 모든 계수는 RewardConfig 에서 주입. +아래 산식은 우리 자체 설계다(특정 고객 산식 복제 아님). 모든 계수는 config 에서 주입. -PoC 의도: 학습 루프가 협상 성과(타결 여부·타결가·턴 수)를 보상으로 흡수하는지 검증. -산식은 결정론적이며 RewardConfig 로 튜닝 가능하다. +산식 (q-table 상세 설명 v4 식 (8)~(14)): + R = W × R_price + (1 − W) × R_end − R_penalty + R_price 3단계: P < anchor → 1 + β·(anchor−P)/anchor (초과달성 보너스) + anchor ≤ P ≤ target → (target−P)/(target−anchor) + P > target → 0 + W = clip(Σ wᵢ·Sᵢ, min_weight, max_weight) — Sᵢ 는 각 state 차원의 정규화 가중치 + R_penalty = λ × round """ from dataclasses import dataclass from negotiation.qtable.domain.model.snapshot import NegotiationOutcome, NegotiationSnapshot -from tenancy.config import RewardConfig +from negotiation.qtable.domain.service.state_calculator import build_state +from tenancy.config import RewardConfig, StateConfig @dataclass(frozen=True) @@ -29,19 +35,24 @@ def _clip(v: float, lo: float, hi: float) -> float: class RewardCalculator: - def __init__(self, config: RewardConfig): + def __init__(self, config: RewardConfig, state_config: StateConfig): self._cfg = config + self._state_cfg = state_config def _price_reward(self, snapshot: NegotiationSnapshot) -> float: """KT 구매자 관점: 협력사 제시가가 낮을수록 보상↑ (anchor 앵커링가 < target 목표 매입가). - progress = clip((target - input) / (target - anchor), 0, 1). - 제시가가 목표가면 0, 앵커링가까지 내려오면 1, 더 낮으면 1로 캡. + 3단계 차등 (식 9~11): 앵커 미만이면 기본 1.0 + β 보너스, 범위 내 선형, 목표 초과면 0. """ - band = snapshot.target_price - snapshot.anchor_price - if band <= 0: + anchor, target, price = snapshot.anchor_price, snapshot.target_price, snapshot.input_price + band = target - anchor + if band <= 0 or anchor <= 0: return 0.0 - return _clip((snapshot.target_price - snapshot.input_price) / band, 0.0, 1.0) + if price < anchor: + return 1.0 + self._cfg.beta * (anchor - price) / anchor + if price <= target: + return (target - price) / band + return 0.0 def _end_reward(self, outcome: NegotiationOutcome) -> float: if outcome == NegotiationOutcome.SUCCESS: @@ -51,20 +62,28 @@ class RewardCalculator: return self._cfg.ongoing_reward def _dynamic_weight(self, snapshot: NegotiationSnapshot) -> float: - """라운드가 진행될수록 가중치를 beta 만큼 감쇠해 [min_weight, max_weight] 로 클립. + """5개 state 차원의 정규화 가중치(Sᵢ) × 메타 가중치(wᵢ) 가중합 (식 12~13). - w = max_weight - beta * round_number → 초반 라운드일수록 가격보상을 더 크게 본다. - (w1~w5 는 차원별 가중치로 P-후속 단계에서 state 차원 중요도에 결합 예정.) + W_raw = w1·S_revenue + w2·S_dist + w3·S_partner + w4·S_accept + w5·S_pricezone, + [min_weight, max_weight] 로 클립. W 가 클수록 가격보상, 작을수록 종료보상 비중↑. """ - w = self._cfg.max_weight - self._cfg.beta * max(0, snapshot.round_number) - return _clip(w, self._cfg.min_weight, self._cfg.max_weight) + st = build_state(snapshot, self._state_cfg) + s = self._state_cfg + w_raw = ( + self._cfg.w1 * s.revenue.weights[st.revenue_idx] + + self._cfg.w2 * s.distribution.weights[st.distribution_idx] + + self._cfg.w3 * s.partner.weights[st.partner_idx] + + self._cfg.w4 * s.acceptance.weights[st.acceptance_idx] + + self._cfg.w5 * s.price_zone.weights[st.price_zone_idx] + ) + return _clip(w_raw, self._cfg.min_weight, self._cfg.max_weight) def calculate(self, snapshot: NegotiationSnapshot) -> RewardBreakdown: price_reward = self._price_reward(snapshot) end_reward = self._end_reward(snapshot.outcome) weight = self._dynamic_weight(snapshot) penalty = self._cfg.penalty_lambda * max(0, snapshot.round_number) - total = weight * price_reward + end_reward - penalty + total = weight * price_reward + (1.0 - weight) * end_reward - penalty return RewardBreakdown( price_reward=price_reward, end_reward=end_reward, diff --git a/agent/negotiation/qtable/domain/service/state_calculator.py b/agent/negotiation/qtable/domain/service/state_calculator.py index cd6a171..5c4e650 100644 --- a/agent/negotiation/qtable/domain/service/state_calculator.py +++ b/agent/negotiation/qtable/domain/service/state_calculator.py @@ -45,9 +45,11 @@ def _partner_bucket(count: int) -> int: def _price_zone_bucket(input_price: float, anchor_price: float, target_price: float) -> int: - """입력가격 구간 (KT 구매자 관점, anchor=협력사 기준가 ≥ target=KT 목표 매입가). + """입력가격 구간 (KT 구매자 관점, anchor 앵커링가 < target 목표 매입가). - 협력사 제시가가 앵커가 이하면 우선협상 가능 구간(0), 초과면 추가 협상 구간(1). + 협력사 제시가가 앵커링가 이하면 우선협상(즉시 타결) 구간(0), 초과면 협상 지속 구간(1). + 설계 명세 v4 는 target 을 경계로 두지만, 확정 경제모델(제시가≤anchor→우선협상 타결)의 + 실제 의사결정 경계는 anchor 이므로 의도적으로 anchor 를 경계로 사용한다. """ if anchor_price <= 0 or target_price <= 0: raise ValueError("anchor_price/target_price must be positive") diff --git a/agent/services/chat_service.py b/agent/services/chat_service.py index ca9429f..ef24364 100644 --- a/agent/services/chat_service.py +++ b/agent/services/chat_service.py @@ -135,7 +135,7 @@ class ChatService: decision = policy.select(ctx) session.used_action_ids.add(decision.action_id) card_id = engine.mapper.get_card_id(decision.action_id) - reward = RewardCalculator(engine.config.reward).calculate(snap) + reward = RewardCalculator(engine.config.reward, engine.config.state).calculate(snap) policy.update(Transition(state_index=idx, action_id=decision.action_id, reward=reward.total, done=False)) await QTablePolicyStore.persist_cell(repo, version_id, policy, idx, decision.action_id) session.context["last_state"] = idx @@ -165,7 +165,7 @@ class ChatService: async def _terminal_learn(self, engine: TenantEngine, session: ChatSession, outcome: str, res: Res_Chat): oc = NegotiationOutcome.SUCCESS if outcome == "success" else NegotiationOutcome.FAILURE snap = self._snapshot(session, oc) - reward = RewardCalculator(engine.config.reward).calculate(snap) + reward = RewardCalculator(engine.config.reward, engine.config.state).calculate(snap) res.reward_total = reward.total last_state = session.context.get("last_state") last_action = session.context.get("last_action") diff --git a/agent/services/negotiation_service.py b/agent/services/negotiation_service.py index 0450536..1b0744d 100644 --- a/agent/services/negotiation_service.py +++ b/agent/services/negotiation_service.py @@ -59,7 +59,7 @@ class NegotiationService: decision.card_id = engine.mapper.get_card_id(decision.action_id) # 4) 보상 - reward = RewardCalculator(engine.config.reward).calculate(snap) + reward = RewardCalculator(engine.config.reward, engine.config.state).calculate(snap) # 5) 학습: Q-learning 온라인 갱신 + touched 셀 영속화 updated_q = decision.q_value diff --git a/agent/tenancy/config.py b/agent/tenancy/config.py index dc419e2..0028242 100644 --- a/agent/tenancy/config.py +++ b/agent/tenancy/config.py @@ -55,8 +55,8 @@ class AcceptanceConfig(BaseModel): class PriceZoneConfig(BaseModel): - """입력가격 구간 (KT 구매자: anchor=협력사 기준가 ≥ target=목표 매입가). - zone0: 제시가 ≤ anchor(우선협상 가능), zone1: > anchor(협상 지속).""" + """입력가격 구간 (KT 구매자: anchor 앵커링가 < target 목표 매입가). + zone0: 제시가 ≤ anchor(우선협상 = 즉시 타결), zone1: > anchor(협상 지속).""" weights: List[float] = [1.0, 0.5] # at_or_below_anchor, above_anchor descriptions: List[str] = ["at_or_below_anchor", "above_anchor"] @@ -88,12 +88,12 @@ class RewardConfig(BaseModel): 기본값은 플랫폼 중립값(균등 가중치)이다 — 특정 고객 튜닝값 복제 아님(CLEANROOM.md). """ - beta: float = 0.2 + beta: float = 0.2 # 앵커 초과달성 보너스 계수 (명세 v4 식(9): P