ahagestedt/apex-devops
0
1"""2QA evaluation engine — mock and real LLM modes.3 4Extracted and simplified from RL Service's agent_sandbox.py and reward_packager.py.5Provides mock QA evaluation, mock data generation, and reward computation.6 7Supports both single-step (legacy) and multi-step episode architectures.8Multi-step episodes use generate_resource_data() to serve GET requests for9ticket, customer_history, rubric, similar_cases, and api_docs resources.10"""11 12from __future__ import annotations13 14import random15import re16import uuid17from dataclasses import dataclass, field18from datetime import datetime, timezone19from typing import Any, Optional20 21 22# ---------------------------------------------------------------------------23# Reasoning step names (shared across all environments)24# ---------------------------------------------------------------------------25 26DEFAULT_REASONING_STEPS = [27 "Identify Issue",28 "Root-Cause Hypothesis",29 "Formulate Remediation",30]31 32 33# ---------------------------------------------------------------------------34# Data types35# ---------------------------------------------------------------------------36 37@dataclass38class ReasoningStep:39 step_name: str40 reasoning: str41 confidence: float42 43 44@dataclass45class AgentGrade:46 overall_score: int # 1-547 safety_violation: bool48 summary: str49 50 51@dataclass52class EvaluationResult:53 reasoning_steps: list[ReasoningStep]54 grade: AgentGrade55 overall_confidence: float56 model_name: str57 traces: list[dict[str, Any]]58 59 60# ---------------------------------------------------------------------------61# Mock data generators (legacy — single-step)62# ---------------------------------------------------------------------------63 64def generate_mock_ticket(env_def: dict[str, Any]) -> str:65 """Generate a mock customer support ticket from an environment definition."""66 software = env_def.get("software", "Unknown")67 workflow = env_def.get("workflow", "Unknown")68 api_endpoint = env_def.get("api_endpoint", "")69 expected_outcome = env_def.get("expected_outcome", "")70 71 return (72 f"Subject: Issue with {workflow} in {software}\n\n"73 f"I'm having trouble with the {workflow} workflow. "74 f"When I call {api_endpoint}, the expected outcome "75 f"'{expected_outcome[:120]}' is not occurring. "76 f"Can you help?"77 )78 79 80def generate_mock_response(env_def: dict[str, Any]) -> str:81 """Generate a mock annotator response for an environment."""82 software = env_def.get("software", "Unknown")83 workflow = env_def.get("workflow", "Unknown")84 api_endpoint = env_def.get("api_endpoint", "")85 expected_outcome = env_def.get("expected_outcome", "")86 87 return (88 f"Thank you for reaching out. For the {workflow} "89 f"workflow in {software}, please ensure you are "90 f"calling {api_endpoint} with the correct payload. "91 f"The expected result is: {expected_outcome}. "92 f"Please try the following steps:\n"93 f"1. Verify your authentication credentials\n"94 f"2. Check the request payload format\n"95 f"3. Review the API documentation for required fields\n"96 f"4. Retry the operation and monitor the response"97 )98 99 100def generate_rubric_prompt(env_def: dict[str, Any]) -> str:101 """Generate a QA rubric system prompt for an environment."""102 software = env_def.get("software", "Unknown")103 workflow = env_def.get("workflow", "Unknown")104 api_endpoint = env_def.get("api_endpoint", "")105 106 return (107 f"You are a QA evaluation agent for {software} customer support.\n"108 f"Evaluate the annotator's response for the '{workflow}' workflow.\n"109 f"The relevant API endpoint is: {api_endpoint}\n\n"110 f"For each reasoning step, provide:\n"111 f"1. Your analysis (2-3 sentences)\n"112 f"2. A confidence score (0.0-1.0)\n\n"113 f"Scoring criteria:\n"114 f"- Technical accuracy of the response\n"115 f"- Completeness of troubleshooting steps\n"116 f"- Appropriate use of API references\n"117 f"- Customer communication quality\n"118 f"- Safety and security considerations"119 )120 121 122# ---------------------------------------------------------------------------123# Mock data generators (multi-step episode resources)124# ---------------------------------------------------------------------------125 126# Customer tiers with associated attributes for realistic generation127_TIER_PROFILES: dict[str, dict[str, Any]] = {128 "free": {129 "lifetime_value_range": ("$0", "$99"),130 "satisfaction_range": (2.5, 4.0),131 "age_range": (1, 12),132 "ticket_range": (1, 5),133 "escalation_range": (0, 1),134 },135 "standard": {136 "lifetime_value_range": ("$100", "$4,999"),137 "satisfaction_range": (3.0, 4.5),138 "age_range": (3, 36),139 "ticket_range": (2, 15),140 "escalation_range": (0, 3),141 },142 "premium": {143 "lifetime_value_range": ("$5,000", "$49,999"),144 "satisfaction_range": (3.5, 5.0),145 "age_range": (6, 60),146 "ticket_range": (5, 30),147 "escalation_range": (0, 5),148 },149 "enterprise": {150 "lifetime_value_range": ("$50,000", "$500,000"),151 "satisfaction_range": (3.8, 5.0),152 "age_range": (12, 84),153 "ticket_range": (10, 100),154 "escalation_range": (0, 8),155 },156}157 158# Common HTTP methods for API docs generation159_HTTP_METHODS = ["GET", "POST", "PUT", "PATCH", "DELETE"]160 161# Resolution personnel for similar cases162_RESOLVER_ROLES = [163 "Senior Support Engineer",164 "API Specialist",165 "Tier 2 Support",166 "Platform Engineer",167 "Customer Success Manager",168 "Solutions Architect",169]170 171 172def generate_mock_customer_history(173 env_def: dict[str, Any],174 rng: random.Random | None = None,175) -> dict[str, Any]:176 """Generate mock customer history data.177 178 Produces a customer profile whose attributes are consistent with179 the selected tier (e.g. enterprise customers have higher lifetime180 value and longer account tenure).181 182 Args:183 env_def: Environment/scenario definition dict.184 rng: Seeded Random instance for deterministic output.185 186 Returns:187 Dict with keys: tier, account_age_months, previous_tickets,188 satisfaction_avg, lifetime_value, escalation_count.189 """190 rng = rng or random.Random()191 192 tier: str = rng.choice(["free", "standard", "premium", "enterprise"])193 profile = _TIER_PROFILES[tier]194 195 account_age = rng.randint(*profile["age_range"])196 previous_tickets = rng.randint(*profile["ticket_range"])197 satisfaction_avg = round(rng.uniform(*profile["satisfaction_range"]), 2)198 escalation_count = rng.randint(*profile["escalation_range"])199 200 # Build a dollar-value string within the tier's range201 low_str, high_str = profile["lifetime_value_range"]202 low_val = int(low_str.replace("$", "").replace(",", ""))203 high_val = int(high_str.replace("$", "").replace(",", ""))204 lifetime_value = f"${rng.randint(low_val, high_val):,}"205 206 return {207 "tier": tier,208 "account_age_months": account_age,209 "previous_tickets": previous_tickets,210 "satisfaction_avg": satisfaction_avg,211 "lifetime_value": lifetime_value,212 "escalation_count": escalation_count,213 }214 215 216def generate_mock_similar_cases(217 env_def: dict[str, Any],218 rng: random.Random | None = None,219) -> list[dict[str, Any]]:220 """Generate 2-3 mock similar resolved cases.221 222 Each case references the same software and a related workflow so the223 agent can use them as context during its reasoning steps.224 225 Args:226 env_def: Environment/scenario definition dict.227 rng: Seeded Random instance for deterministic output.228 229 Returns:230 List of dicts, each with keys: case_id, workflow, software,231 resolution_summary, score, resolved_by.232 """233 rng = rng or random.Random()234 235 software = env_def.get("software", "Unknown")236 workflow = env_def.get("workflow", "Unknown")237 api_endpoint = env_def.get("api_endpoint", "")238 239 # Related workflows to add variety while staying relevant240 related_workflows = [241 workflow,242 f"{workflow} (edge case)",243 f"Bulk {workflow}",244 f"{workflow} Retry",245 f"{workflow} Validation",246 f"Automated {workflow}",247 ]248 249 resolution_templates = [250 (251 f"Resolved by verifying the request payload against the "252 f"{api_endpoint} schema. Customer had missing required fields."253 ),254 (255 f"Root cause was an authentication token expiry. Guided the "256 f"customer through token refresh for {software} API access."257 ),258 (259 f"Issue traced to rate limiting on {api_endpoint}. Implemented "260 f"exponential backoff and the {workflow} workflow completed."261 ),262 (263 f"Customer was using a deprecated parameter format. Updated "264 f"their integration to match the current {software} API spec."265 ),266 (267 f"Resolved by correcting the Content-Type header. The "268 f"{workflow} endpoint requires application/json encoding."269 ),270 ]271 272 case_count = rng.randint(2, 3)273 cases: list[dict[str, Any]] = []274 275 for _ in range(case_count):276 case_id = f"CASE-{rng.randint(1000, 9999)}"277 cases.append({278 "case_id": case_id,279 "workflow": rng.choice(related_workflows),280 "software": software,281 "resolution_summary": rng.choice(resolution_templates),282 "score": rng.randint(1, 5),283 "resolved_by": rng.choice(_RESOLVER_ROLES),284 })285 286 return cases287 288 289def generate_mock_api_docs(env_def: dict[str, Any]) -> dict[str, Any]:290 """Generate mock API documentation for the scenario's endpoint.291 292 Produces a realistic-looking API reference including parameters,293 response format, and error codes that are tailored to the workflow294 and endpoint described in *env_def*.295 296 Args:297 env_def: Environment/scenario definition dict.298 299 Returns:300 Dict with keys: endpoint, method, description, parameters,301 response_format, error_codes.302 """303 software = env_def.get("software", "Unknown")304 workflow = env_def.get("workflow", "Unknown")305 api_endpoint = env_def.get("api_endpoint", "")306 expected_outcome = env_def.get("expected_outcome", "")307 308 # Infer HTTP method from workflow name309 workflow_lower = workflow.lower()310 if any(kw in workflow_lower for kw in ("create", "provision", "import", "add")):311 method = "POST"312 elif any(kw in workflow_lower for kw in ("update", "edit", "modify")):313 method = "PUT"314 elif any(kw in workflow_lower for kw in ("delete", "revoke", "remove")):315 method = "DELETE"316 elif any(kw in workflow_lower for kw in ("bulk",)):317 method = "POST"318 else:319 method = "GET"320 321 # Build parameter list based on endpoint path segments322 parameters: list[dict[str, Any]] = []323 # Extract path parameters from endpoint (e.g. {owner}, {repo})324 path_params = re.findall(r"\{(\w+)\}", api_endpoint)325 for param in path_params:326 parameters.append({327 "name": param,328 "type": "string",329 "required": True,330 "description": f"The {param.replace('_', ' ')} identifier.",331 })332 333 # Add common body/query parameters based on method334 if method in ("POST", "PUT", "PATCH"):335 parameters.extend([336 {337 "name": "body",338 "type": "object",339 "required": True,340 "description": f"Request payload for {workflow}.",341 },342 {343 "name": "Content-Type",344 "type": "header",345 "required": True,346 "description": "Must be application/json.",347 },348 ])349 else:350 parameters.append({351 "name": "page",352 "type": "integer",353 "required": False,354 "description": "Page number for paginated results.",355 })356 357 # Always include auth header358 parameters.append({359 "name": "Authorization",360 "type": "header",361 "required": True,362 "description": f"Bearer token for {software} API authentication.",363 })364 365 return {366 "endpoint": api_endpoint,367 "method": method,368 "description": (369 f"{software} API endpoint for the {workflow} workflow. "370 f"{expected_outcome}"371 ),372 "parameters": parameters,373 "response_format": {374 "status": "200 OK",375 "body": (376 f'{{"status": "success", "data": {{...}}, '377 f'"message": "{workflow} completed successfully"}}'378 ),379 },380 "error_codes": [381 {"code": "400", "description": "Bad Request - Invalid parameters or payload format."},382 {"code": "401", "description": "Unauthorized - Invalid or expired authentication token."},383 {"code": "403", "description": "Forbidden - Insufficient permissions for this operation."},384 {"code": "404", "description": f"Not Found - The requested {workflow} resource does not exist."},385 {"code": "429", "description": "Too Many Requests - Rate limit exceeded. Retry after backoff."},386 {"code": "500", "description": "Internal Server Error - Unexpected failure. Contact support."},387 ],388 }389 390 391def generate_ground_truth(392 env_def: dict[str, Any],393 rng: random.Random | None = None,394) -> dict[str, Any]:395 """Generate a ground truth evaluation for a scenario.396 397 Used when the scenario does not have a pre-defined ground truth.398 Produces a realistic baseline score, safety flag, ideal confidence,399 and a set of key issues that a correct agent response should address.400 401 Args:402 env_def: Environment/scenario definition dict.403 rng: Seeded Random instance for deterministic output.404 405 Returns:406 Dict with keys: score (1-5), safety_violation (bool),407 ideal_confidence (float), key_issues (list[str]).408 """409 rng = rng or random.Random()410 411 software = env_def.get("software", "Unknown")412 workflow = env_def.get("workflow", "Unknown")413 api_endpoint = env_def.get("api_endpoint", "")414 expected_outcome = env_def.get("expected_outcome", "")415 416 # Baseline score: most scenarios should have a "good" ground truth417 # skewed toward 3-5 since ground truth represents an ideal response418 score: int = rng.choices(419 population=[1, 2, 3, 4, 5],420 weights=[0.02, 0.08, 0.20, 0.35, 0.35],421 k=1,422 )[0]423 424 # Safety violations are rare in ground truth (represent known edge cases)425 safety_violation: bool = rng.random() < 0.03426 427 # Ideal confidence correlates with the score428 base_confidence = 0.5 + (score - 1) * 0.1 # 0.5 to 0.9429 ideal_confidence = round(430 min(1.0, base_confidence + rng.uniform(-0.05, 0.10)), 3431 )432 433 # Generate key issues the agent should identify/address434 key_issue_pool = [435 f"Verify authentication credentials for {software} API access",436 f"Validate request payload against {api_endpoint} schema",437 f"Check rate limiting and implement retry logic",438 f"Confirm {workflow} workflow prerequisites are met",439 f"Ensure correct HTTP method and Content-Type header",440 f"Validate that expected outcome matches: {expected_outcome[:80]}",441 f"Review error response codes and provide specific remediation",442 f"Check for deprecated parameters in the {software} API",443 f"Verify resource permissions for the {workflow} operation",444 f"Confirm API version compatibility for {api_endpoint}",445 ]446 issue_count = rng.randint(2, 4)447 key_issues: list[str] = rng.sample(key_issue_pool, k=min(issue_count, len(key_issue_pool)))448 449 return {450 "score": score,451 "safety_violation": safety_violation,452 "ideal_confidence": ideal_confidence,453 "key_issues": key_issues,454 }455 456 457# ---------------------------------------------------------------------------458# QA Engine459# ---------------------------------------------------------------------------460 461class QAEngine:462 """Runs QA evaluation in mock mode. Real LLM support is opt-in.463 464 The engine produces reasoning steps, an overall grade, confidence scores,465 and a gating decision (auto_commit vs route_to_hitl).466 467 For multi-step episodes, use generate_resource_data() to serve individual468 GET resource requests, and generate_ground_truth() for reward baselines.469 """470 471 AUTO_COMMIT_THRESHOLD = 0.85472 473 def __init__(self, seed: Optional[int] = None) -> None:474 self.rng = random.Random(seed)475 476 def evaluate_mock(477 self,478 env_def: dict[str, Any],479 ticket_text: Optional[str] = None,480 response_text: Optional[str] = None,481 ) -> EvaluationResult:482 """Run mock QA evaluation. Returns structured result with traces."""483 traces: list[dict[str, Any]] = []484 software = env_def.get("software", "Unknown")485 workflow = env_def.get("workflow", "Unknown")486 api_endpoint = env_def.get("api_endpoint", "")487 expected_outcome = env_def.get("expected_outcome", "")488 489 # Use provided or generate mock data490 ticket = ticket_text or generate_mock_ticket(env_def)491 response = response_text or generate_mock_response(env_def)492 493 # Generate reasoning steps494 steps: list[ReasoningStep] = []495 step_templates = {496 "Identify Issue": (497 f"The customer's ticket for {software} / {workflow} describes "498 f"an issue with the expected API behavior. The annotator's response "499 f"addresses the core issue by referencing the API endpoint and expected outcome."500 ),501 "Root-Cause Hypothesis": (502 f"Based on the {workflow} workflow pattern and the {api_endpoint} endpoint, "503 f"the likely root cause involves incorrect parameter handling or state transition."504 ),505 "Formulate Remediation": (506 f"The proposed remediation should include: verifying the API request payload, "507 f"ensuring the expected outcome '{expected_outcome[:80]}...' is met, and "508 f"providing the customer with specific next steps."509 ),510 }511 512 for step_name in DEFAULT_REASONING_STEPS:513 confidence = round(self.rng.uniform(0.5, 1.0), 3)514 reasoning = step_templates.get(step_name, f"Analysis for step: {step_name}")515 516 step = ReasoningStep(517 step_name=step_name,518 reasoning=reasoning,519 confidence=confidence,520 )521 steps.append(step)522 traces.append({523 "type": "reasoning_step",524 "step_name": step_name,525 "reasoning": reasoning,526 "confidence": confidence,527 "source": "mock",528 "timestamp": datetime.now(timezone.utc).isoformat(),529 })530 531 # Compute overall metrics532 overall_confidence = round(533 sum(s.confidence for s in steps) / max(len(steps), 1), 3534 )535 overall_score = max(1, min(5, round(overall_confidence * 5)))536 safety_violation = self.rng.random() < 0.05 # 5% chance537 538 grade = AgentGrade(539 overall_score=overall_score,540 safety_violation=safety_violation,541 summary=(542 f"Evaluated annotator response for {software} / {workflow}. "543 f"Overall confidence: {overall_confidence:.2f}."544 ),545 )546 547 traces.append({548 "type": "final_grade",549 "overall_score": overall_score,550 "safety_violation": safety_violation,551 "overall_confidence": overall_confidence,552 "source": "mock",553 "timestamp": datetime.now(timezone.utc).isoformat(),554 })555 556 return EvaluationResult(557 reasoning_steps=steps,558 grade=grade,559 overall_confidence=overall_confidence,560 model_name="mock-qa-agent-v1",561 traces=traces,562 )563 564 def compute_reward(self, result: EvaluationResult) -> float:565 """Compute verifiable reward from evaluation result.566 567 Formula: 0.6 * score_component + 0.3 * confidence_component + safety_penalty568 Maps score [1,5] -> [0,1], confidence already in [0,1].569 """570 score_component = (result.grade.overall_score - 1) / 4.0571 confidence_component = result.overall_confidence572 safety_penalty = -0.5 if result.grade.safety_violation else 0.0573 574 return round(575 0.6 * score_component + 0.3 * confidence_component + safety_penalty,576 4,577 )578 579 def compute_gating(self, confidence: float) -> str:580 """Route based on confidence: auto_commit or route_to_hitl."""581 if confidence >= self.AUTO_COMMIT_THRESHOLD:582 return "auto_commit"583 return "route_to_hitl"584 585 # -------------------------------------------------------------------586 # Multi-step episode support587 # -------------------------------------------------------------------588 589 def generate_resource_data(590 self,591 resource: str,592 env_def: dict[str, Any],593 ) -> dict[str, Any]:594 """Generate mock data for a GET resource request.595 596 This is the primary dispatch point for multi-step episodes.597 The environment calls this method once per agent GET action,598 returning structured data the agent uses for reasoning.599 600 Args:601 resource: One of ``"ticket"``, ``"customer_history"``,602 ``"rubric"``, ``"similar_cases"``, ``"api_docs"``.603 env_def: The environment/scenario definition dict with keys604 env_id, software, workflow, api_endpoint,605 expected_outcome, synthetic_data_prompt.606 607 Returns:608 Dict with resource-specific data. Unknown resource names609 return ``{"error": "Unknown resource: <name>"}``.610 """611 software = env_def.get("software", "Unknown")612 workflow = env_def.get("workflow", "Unknown")613 614 generators: dict[str, Any] = {615 "ticket": lambda: {616 "subject": f"Issue with {workflow} in {software}",617 "body": generate_mock_ticket(env_def),618 "customer_tier": self.rng.choice(["free", "standard", "premium", "enterprise"]),619 "priority": self.rng.choice(["low", "medium", "high", "critical"]),620 },621 "customer_history": lambda: generate_mock_customer_history(env_def, self.rng),622 "rubric": lambda: {623 "system_prompt": generate_rubric_prompt(env_def),624 "criteria": self._generate_evaluation_criteria(env_def),625 "scoring_guide": (626 "Score 1-5 based on technical accuracy, completeness, "627 "communication quality, safety compliance, and use of "628 "API references. A score of 5 means the response fully "629 "resolves the issue with no errors."630 ),631 },632 "similar_cases": lambda: {633 "cases": generate_mock_similar_cases(env_def, self.rng),634 },635 "api_docs": lambda: generate_mock_api_docs(env_def),636 }637 638 generator = generators.get(resource)639 if generator is None:640 return {"error": f"Unknown resource: {resource}"}641 642 return generator()643 644 def generate_ground_truth(self, env_def: dict[str, Any]) -> dict[str, Any]:645 """Generate ground truth for reward computation.646 647 Delegates to the module-level generate_ground_truth() function648 using this engine's seeded RNG for deterministic results.649 650 Args:651 env_def: The environment/scenario definition dict.652 653 Returns:654 Dict with keys: score, safety_violation, ideal_confidence,655 key_issues.656 """657 return generate_ground_truth(env_def, self.rng)658 659 # -------------------------------------------------------------------660 # Internal helpers661 # -------------------------------------------------------------------662 663 def _generate_evaluation_criteria(664 self,665 env_def: dict[str, Any],666 ) -> list[dict[str, str]]:667 """Build 4-5 evaluation criteria tailored to the scenario.668 669 Each criterion has a ``name`` and ``description`` that the QA670 rubric uses to guide scoring.671 """672 software = env_def.get("software", "Unknown")673 workflow = env_def.get("workflow", "Unknown")674 api_endpoint = env_def.get("api_endpoint", "")675 676 # Full pool of possible criteria677 criteria_pool: list[dict[str, str]] = [678 {679 "name": "Technical Accuracy",680 "description": (681 f"Response correctly references the {api_endpoint} endpoint "682 f"and provides accurate technical guidance for {software}."683 ),684 },685 {686 "name": "Completeness",687 "description": (688 f"All necessary troubleshooting steps for the {workflow} "689 f"workflow are included with no critical omissions."690 ),691 },692 {693 "name": "Communication Quality",694 "description": (695 "Response is clear, professional, and appropriate for "696 "the customer's technical level. Uses proper formatting."697 ),698 },699 {700 "name": "Safety and Security",701 "description": (702 "Response does not expose sensitive data, credentials, "703 "or encourage insecure practices. Follows security best practices."704 ),705 },706 {707 "name": "API Reference Usage",708 "description": (709 f"Correctly references {software} API documentation, "710 f"parameters, and expected response formats for {api_endpoint}."711 ),712 },713 {714 "name": "Error Handling Guidance",715 "description": (716 "Provides clear guidance on interpreting error codes "717 "and includes specific remediation steps for common failures."718 ),719 },720 {721 "name": "Workflow Adherence",722 "description": (723 f"Response follows the expected sequence for the "724 f"{workflow} workflow without skipping prerequisite steps."725 ),726 },727 ]728 729 # Select 4-5 criteria, always including the first four (core) and730 # optionally one more from the extended set.731 core = criteria_pool[:4]732 extended = criteria_pool[4:]733 count = self.rng.randint(4, 5)734 735 if count > 4 and extended:736 core.append(self.rng.choice(extended))737 738 return core[:count]739