Spaces:
Runtime error
Runtime error
File size: 7,758 Bytes
dbf5e96 | 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 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | """
Benchmark Agents — Random and Perfect baselines.
These establish the floor and ceiling of the CodeReviewEnv benchmark.
Every new agent should be compared against these two.
RandomAgent: picks actions uniformly at random (floor)
PerfectAgent: reads ground truth, always correct (ceiling)
"""
import random
from typing import Dict, List
from env.data_generator import PR_TEMPLATES, get_ground_truth, DataGenerator, SEVERITY_ORDER
from env.models import Action
class RandomAgent:
"""
Picks action_type uniformly at random, random severity.
Establishes the benchmark floor — any useful agent must
significantly outperform random. Expected composite score ~0.18.
"""
def __init__(self, seed: int = 42):
self.rng = random.Random(seed)
def act(self, observation: Dict, system_prompt: str) -> Dict:
"""
Generate random action based on system prompt content.
Detects task from system prompt to generate valid action types.
"""
if "label_severity" in system_prompt:
return self._act_easy(observation)
elif "prioritize" in system_prompt:
return self._act_medium(observation)
else:
return self._act_hard(observation)
def _act_easy(self, observation: Dict) -> Dict:
"""Random severity label."""
severity = self.rng.choice(SEVERITY_ORDER)
return {"action_type": "label_severity", "severity": severity}
def _act_medium(self, observation: Dict) -> Dict:
"""Random queue ordering."""
queue = list(observation.get("review_queue", []))
self.rng.shuffle(queue)
return {"action_type": "prioritize", "priority_order": queue}
def _act_hard(self, observation: Dict) -> Dict:
"""Random comments then random decision."""
# 50% chance to comment, 50% to decide
if self.rng.random() < 0.5:
files = observation.get("files", [])
target_file = files[0].get("filename", "unknown.py") if files else "unknown.py"
return {
"action_type": "add_comment",
"comment": "Looks fine to me.",
"target_file": target_file,
"target_line": self.rng.randint(1, 50),
}
else:
decision = self.rng.choice(["approve", "request_changes"])
return {"action_type": decision}
class PerfectAgent:
"""
Reads ground truth from FIXED_TEST_SUITE, always correct.
Establishes the benchmark ceiling — represents optimal behavior
given full knowledge of bug locations and severities.
Expected composite score ~0.97.
"""
def __init__(self, seed: int = 42):
self.generator = DataGenerator(seed=seed)
self._severity_cache: Dict[str, str] = {
t["pr_id"]: t["ground_truth_severity"] for t in PR_TEMPLATES
}
self._template_cache: Dict[str, Dict] = {
t["pr_id"]: t for t in PR_TEMPLATES
}
self._comment_count: Dict[str, int] = {}
def act(self, observation: Dict, system_prompt: str) -> Dict:
"""
Generate perfect action based on ground truth.
Detects task from system prompt to generate correct responses.
"""
if "label_severity" in system_prompt:
return self._act_easy(observation)
elif "prioritize" in system_prompt:
return self._act_medium(observation)
else:
return self._act_hard(observation)
def _act_easy(self, observation: Dict) -> Dict:
"""Return ground truth severity."""
pr_id = observation.get("pr_id", "")
severity = self._severity_cache.get(pr_id, "medium")
return {"action_type": "label_severity", "severity": severity}
def _act_medium(self, observation: Dict) -> Dict:
"""Return ground truth priority ordering."""
queue_ids = observation.get("review_queue", [])
# Build queue templates from cache
queue_templates = []
for pr_id in queue_ids:
if pr_id in self._template_cache:
queue_templates.append(self._template_cache[pr_id])
if queue_templates:
order = self.generator.compute_priority_order(queue_templates)
else:
order = queue_ids
return {"action_type": "prioritize", "priority_order": order}
def _act_hard(self, observation: Dict) -> Dict:
"""
Generate targeted, specific, actionable comments then decide.
Strategy: comment on each bug line with category-specific keywords,
then request_changes if bugs exist, approve if clean.
"""
pr_id = observation.get("pr_id", "")
template = self._template_cache.get(pr_id, {})
bug_lines = template.get("bug_lines", [])
bug_category = template.get("bug_category", "")
severity = template.get("ground_truth_severity", "none")
# Track comments per PR
if pr_id not in self._comment_count:
self._comment_count[pr_id] = 0
# Add one comment per bug line, then decide
if self._comment_count[pr_id] < len(bug_lines):
idx = self._comment_count[pr_id]
line = bug_lines[idx]
self._comment_count[pr_id] += 1
# Generate category-specific comment with actionability keywords
from env.data_generator import BUG_KEYWORDS
keywords = BUG_KEYWORDS.get(bug_category, ["issue"])
kw = keywords[0] if keywords else "issue"
comment = self._generate_targeted_comment(bug_category, kw)
files = observation.get("files", [])
target_file = files[0].get("filename", "unknown") if files else "unknown"
return {
"action_type": "add_comment",
"comment": comment,
"target_file": target_file,
"target_line": line,
}
else:
# All bugs commented — make decision
self._comment_count[pr_id] = 0
if severity in ("critical", "high", "medium"):
return {"action_type": "request_changes"}
else:
return {"action_type": "approve"}
@staticmethod
def _generate_targeted_comment(bug_category: str, keyword: str) -> str:
"""Generate a relevant, specific, actionable comment."""
comments = {
"null_pointer": f"You should add a {keyword} check guard here to prevent NullPointerException. Consider using Optional or adding an early return.",
"sql_injection": f"This is vulnerable to {keyword}. You should use parameterized queries instead of string concatenation to sanitize user input.",
"race_condition": f"This has a {keyword} condition. You should add a mutex lock or use atomic operations to ensure thread-safe concurrent access.",
"logic_error": f"The {keyword} here has an off-by-one boundary issue. Consider checking the edge case and adjusting the logic.",
"missing_error_handling": f"Missing {keyword} handling here. You should add a try/catch block and handle the error case gracefully.",
"security_vulnerability": f"This leaks {keyword} information. You should encrypt sensitive data and avoid exposing secrets in logs.",
"performance_issue": f"This has O(n) {keyword} complexity. Consider adding an index or cache to optimize the query performance.",
"style_only": f"The {keyword} here doesn't follow conventions. Consider renaming for consistency with the codebase style.",
}
return comments.get(bug_category, f"Consider reviewing the {keyword} usage here. You should refactor for clarity.")
|