Spaces:
Sleeping
Sleeping
File size: 5,451 Bytes
0e4f105 | 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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | from typing import List
from viral_script_engine.agents.critic import CritiqueClaim
from viral_script_engine.agents.defender import DefenderOutput
from viral_script_engine.environment.actions import ArbitratorAction
from viral_script_engine.environment.observations import RewardComponents
_SEVERITY_RANK = {"high": 3, "medium": 2, "low": 1}
_SECTION_KEYWORDS = {
"hook": ["hook", "opening", "start", "first", "beginning", "intro"],
"body": ["body", "middle", "main", "content"],
"cta": ["cta", "call to action", "call-to-action", "ending", "end", "conclusion"],
}
class ProcessVerifier:
"""
Checks whether the Arbitrator's reasoning chain is correct BEFORE
the action is executed. This is process supervision.
Three checks, each independently scored.
"""
def verify_priority_assessment(
self,
priority_assessment: str,
critic_claims: List[CritiqueClaim],
current_reward_components: RewardComponents,
) -> float:
"""
Checks: does priority_assessment mention the critique_class with
the highest severity in the current Critic output?
Score:
- 1.0: mentions the highest-severity critique_class
- 0.5: mentions a medium-severity class (not the worst, but not random)
- 0.0: mentions only a low-severity class or is empty
"""
if not priority_assessment or not critic_claims:
return 0.0
sorted_claims = sorted(
critic_claims,
key=lambda c: _SEVERITY_RANK.get(c.severity.lower(), 0),
reverse=True,
)
assessment_lower = priority_assessment.lower()
highest_class = sorted_claims[0].critique_class.lower()
if highest_class in assessment_lower:
return 1.0
medium_classes = {
c.critique_class.lower()
for c in sorted_claims
if c.severity.lower() == "medium"
}
if any(cls in assessment_lower for cls in medium_classes):
return 0.5
return 0.0
def verify_conflict_check(
self,
conflict_check_answer: str,
conflict_check_reason: str,
action: ArbitratorAction,
current_reward_components: RewardComponents,
episode_start_components: RewardComponents,
) -> float:
"""
Checks: is conflict_check_answer consistent with the actual risk?
Known conflict patterns:
- hook_rewrite when r3 >= 0.7 → conflict likely
- section_reorder when r2 <= 0.6 → conflict likely
- cultural_ref_sub when r5 <= 0.5 → conflict likely
- cta_placement when r1 <= 0.4 → conflict likely
Score:
- 1.0: answer matches rule-based assessment
- 0.0: answer contradicts it or is empty
"""
if not conflict_check_answer:
return 0.0
action_type = action.action_type.value
r1 = current_reward_components.r1_hook_strength or 0.0
r2 = current_reward_components.r2_coherence or 0.0
r3 = current_reward_components.r3_cultural_alignment or 0.0
r5 = current_reward_components.r5_defender_preservation or 0.0
conflict_exists = False
if action_type == "hook_rewrite" and r3 >= 0.7:
conflict_exists = True
elif action_type == "section_reorder" and r2 <= 0.6:
conflict_exists = True
elif action_type == "cultural_ref_sub" and r5 <= 0.5:
conflict_exists = True
elif action_type == "cta_placement" and r1 <= 0.4:
conflict_exists = True
model_says_conflict = conflict_check_answer.lower().strip().startswith("yes")
return 1.0 if model_says_conflict == conflict_exists else 0.0
def verify_defender_consideration(
self,
defender_consideration_answer: str,
defender_consideration_reason: str,
action: ArbitratorAction,
defender_output: DefenderOutput,
) -> float:
"""
Checks: if the action targets the same section as the Defender's
core_strength_quote, did the Arbitrator say defender_consideration = "yes"?
Score:
- 1.0: answer is correct
- 0.0: answer is wrong or empty
"""
if not defender_consideration_answer:
return 0.0
core_quote_lower = defender_output.core_strength_quote.lower()
target_section = action.target_section.lower()
# Infer which section the core strength resides in
core_section = None
for section, keywords in _SECTION_KEYWORDS.items():
if any(kw in core_quote_lower for kw in keywords):
core_section = section
break
# target_section "full" always overlaps with core strength
if target_section == "full":
target_matches_core = True
elif core_section is not None:
target_matches_core = (
target_section == core_section
or target_section in core_section
or core_section in target_section
)
else:
# Cannot determine core section — give benefit of doubt based on exact match
target_matches_core = target_section in core_quote_lower
model_says_yes = defender_consideration_answer.lower().strip().startswith("yes")
return 1.0 if model_says_yes == target_matches_core else 0.0
|