Spaces:
Sleeping
Sleeping
| # cell 7 author.py – Final production version: stateful, evidence-driven, belief tracking | |
| import re | |
| import ast | |
| from dataclasses import dataclass, field | |
| from typing import List, Dict, Any, Optional | |
| class PersonaAuthor: | |
| """ | |
| Simulates a human developer with: | |
| - Continuous belief (confidence) | |
| - Evidence-based reasoning | |
| - Conversation memory | |
| - Code inspection awareness | |
| """ | |
| personality: str = "defensive" # defensive | junior | collaborative | |
| max_persuasion_rounds: int = 5 | |
| # Evidence weights | |
| weight_test_pass: float = 0.5 | |
| weight_lint_clean: float = 0.2 | |
| weight_doc_found: float = 0.15 | |
| weight_explanation_quality: float = 0.15 | |
| # Personality thresholds | |
| thresholds: Dict[str, float] = field(default_factory=lambda: { | |
| "defensive": 0.7, | |
| "junior": 0.3, | |
| "collaborative": 0.5, | |
| }) | |
| # Internal state | |
| _confidence: float = 0.0 | |
| _conversation: List[Dict[str, Any]] = field(default_factory=list) | |
| _pushback_count: int = 0 | |
| _last_evidence_score: float = 0.0 | |
| _stagnation_counter: int = 0 | |
| # ------------------------------------------------------------------ | |
| # Lifecycle | |
| # ------------------------------------------------------------------ | |
| def __post_init__(self): | |
| self.reset() | |
| def reset(self): | |
| self._confidence = 0.0 | |
| self._conversation.clear() | |
| self._pushback_count = 0 | |
| self._last_evidence_score = 0.0 | |
| self._stagnation_counter = 0 | |
| # ------------------------------------------------------------------ | |
| # Main interaction | |
| # ------------------------------------------------------------------ | |
| # Added weight for code change magnitude | |
| weight_code_change: float = 0.1 # small change is better | |
| def respond(self, | |
| agent_comment: str = "", | |
| agent_question: str = "", | |
| test_results: Optional[str] = None, | |
| lint_results: Optional[str] = None, | |
| doc_results: Optional[str] = None, | |
| proposed_fix: Optional[str] = None, | |
| original_code: Optional[str] = None) -> str: | |
| # Store conversation | |
| self._conversation.append({ | |
| "comment": agent_comment, | |
| "question": agent_question, | |
| "test": test_results, | |
| "lint": lint_results, | |
| "docs": doc_results | |
| }) | |
| # Extract structured evidence | |
| evidence = self._extract_evidence(test_results, lint_results, doc_results) | |
| # Code inspection | |
| code_change = 0.0 | |
| if proposed_fix and original_code: | |
| code_change = self._inspect_code(proposed_fix, original_code) | |
| evidence["code_change"] = code_change | |
| # Explanation score | |
| text = (agent_comment + " " + agent_question).lower() | |
| explanation_score = self._score_explanation(text) | |
| # Compute evidence score – now includes code change penalty (1 - change) | |
| evidence_score = ( | |
| self.weight_test_pass * evidence.get("test_pass_ratio", 0.0) + | |
| self.weight_lint_clean * (1 - min(1.0, evidence.get("lint_errors", 0)/10)) + | |
| self.weight_doc_found * (1.0 if evidence.get("doc_found") else 0.0) + | |
| self.weight_explanation_quality * explanation_score + | |
| self.weight_code_change * (1.0 - code_change) # surgical fix rewarded | |
| ) | |
| evidence_score = max(0.0, min(1.0, evidence_score)) | |
| # Detect improvement | |
| delta = evidence_score - self._last_evidence_score | |
| self._last_evidence_score = evidence_score | |
| if delta > 0.05: | |
| self._stagnation_counter = 0 | |
| else: | |
| self._stagnation_counter += 1 | |
| # Update belief (momentum) | |
| lr = 0.3 | |
| self._confidence = (1 - lr) * self._confidence + lr * evidence_score | |
| # Penalise stagnation | |
| if self._stagnation_counter >= 2: | |
| self._confidence *= 0.9 | |
| # Decision | |
| threshold = self.thresholds.get(self.personality, 0.5) | |
| if self._confidence >= threshold or self._pushback_count >= self.max_persuasion_rounds: | |
| return "Alright, I'm convinced. Let's proceed with your fix." | |
| # Otherwise push back | |
| self._pushback_count += 1 | |
| return self._generate_pushback(evidence, text) | |
| # ------------------------------------------------------------------ | |
| # Evidence extraction | |
| # ------------------------------------------------------------------ | |
| def _extract_evidence(self, test_results, lint_results, doc_results): | |
| evidence = { | |
| "test_pass_ratio": 0.0, | |
| "lint_errors": 0, | |
| "doc_found": False | |
| } | |
| # Parse test results | |
| if test_results: | |
| match = re.search(r'(\d+)\s*/\s*(\d+)', test_results) | |
| if match: | |
| p, t = int(match.group(1)), int(match.group(2)) | |
| evidence["test_pass_ratio"] = p / t if t else 0.0 | |
| elif "true" in test_results.lower(): | |
| evidence["test_pass_ratio"] = 1.0 | |
| elif "false" in test_results.lower(): | |
| evidence["test_pass_ratio"] = 0.0 | |
| # Lint errors | |
| if lint_results: | |
| evidence["lint_errors"] = len(re.findall(r'error', lint_results.lower())) | |
| # Docs | |
| if doc_results and "no relevant" not in doc_results.lower(): | |
| evidence["doc_found"] = True | |
| return evidence | |
| # ------------------------------------------------------------------ | |
| # Explanation scoring | |
| # ------------------------------------------------------------------ | |
| def _score_explanation(self, text: str) -> float: | |
| score = 0.0 | |
| if "because" in text or "therefore" in text: | |
| score += 0.3 | |
| if "test" in text or "example" in text: | |
| score += 0.2 | |
| if len(text.split()) > 30: | |
| score += 0.2 | |
| if "error" in text or "fix" in text: | |
| score += 0.1 | |
| return min(1.0, score) | |
| # ------------------------------------------------------------------ | |
| # Code inspection | |
| # ------------------------------------------------------------------ | |
| def _inspect_code(self, new_code: str, old_code: str) -> float: | |
| try: | |
| t1 = ast.parse(old_code) | |
| t2 = ast.parse(new_code) | |
| n1 = len(list(ast.walk(t1))) | |
| n2 = len(list(ast.walk(t2))) | |
| change = abs(n2 - n1) / max(n1, 1) | |
| return min(1.0, change) | |
| except: | |
| return 0.0 | |
| # ------------------------------------------------------------------ | |
| # Pushback generator | |
| # ------------------------------------------------------------------ | |
| def _generate_pushback(self, evidence, text): | |
| if evidence["test_pass_ratio"] < 0.5: | |
| return "Tests are still failing. Show a passing case." | |
| if evidence["lint_errors"] > 0: | |
| return f"There are {evidence['lint_errors']} lint errors. Fix them." | |
| if not evidence["doc_found"]: | |
| return "Provide documentation or reference." | |
| if "because" not in text: | |
| return "Explain why this works." | |
| if len(text.split()) < 20: | |
| return "Too brief. Expand your reasoning." | |
| return "Not convinced yet. Give a concrete example." | |
| # ------------------------------------------------------------------ | |
| # Score | |
| # ------------------------------------------------------------------ | |
| def get_negotiation_score(self) -> float: | |
| penalty = 0.1 * min(3, self._pushback_count) | |
| return max(0.0, min(1.0, self._confidence - penalty)) | |