# Reasoning Engine Internals > Implements dual-provider reasoning architecture (Gemini + Claude orchestration), cross-model reasoning pipelines, token budget management, and AI-assisted development velocity tracking for production AI systems. - Skill: `paulpas/reasoning-engine-internals` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/reasoning-engine-internals` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/reasoning-engine-internals/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/reasoning-engine-internals --- # Reasoning Engine Internals Implements the architectural backbone of multi-model AI reasoning systems — orchestrating dual-provider pipelines (Gemini for synthesis, Claude for analysis), managing token budgets across reasoning steps, delegating code generation across specialized agents, and tracking AI-assisted development velocity at scale. This skill governs HOW to architect the reasoning engine that powers agents, not individual reasoning techniques like CoT or ReAct. ## TL;DR Checklist - [ ] Map each incoming task to model capability profiles (synthesis vs analysis vs creative) before dispatching - [ ] Implement dual-provider pipeline with explicit result merging and conflict resolution logic - [ ] Allocate token budgets per reasoning step with early termination for simple queries and escalation for complex ones - [ ] Deploy multi-agent code generation delegation with architect → implementer → reviewer handoff patterns - [ ] Track AI development velocity metrics (code generation share, accuracy rate, latency) against industry benchmarks - [ ] Measure reasoning quality via per-step confidence scoring, answer consistency checks, and latency-vs-accuracy tradeoff analysis - [ ] Apply `code-philosophy` laws throughout — especially Law 1 (Early Exit) on budget exhaustion and Law 4 (Fail Fast) on model failures --- ## When to Use Use this skill when: - Architecting a production reasoning system that requires multiple LLM providers for optimal quality-to-cost ratio - Designing cross-model pipelines where different models handle distinct reasoning phases (e.g., Gemini for creative synthesis, Claude for logical verification) - Building multi-agent code generation workflows requiring role-based handoffs between architect, implementer, and reviewer agents - Measuring AI-assisted development velocity across a team — tracking what percentage of code is AI-generated, at what quality threshold, and with what latency impact - Managing token budgets in high-volume agent systems where cost control requires per-step allocation, early termination, and relevance scoring - Operating in accuracy-critical domains (financial analysis, medical diagnostics, legal research) where single-model answers are insufficient and cross-validation across models is required --- ## When NOT to Use Avoid this skill for: - **Simple tasks solvable by a single model** — A straightforward code fix, documentation update, or basic query does not justify dual-provider overhead. Use `reasoning-techniques` (CoT/ReAct with a single model) instead. - **Basic chatbot or prompt engineering projects** — If you are only designing conversational flows without multi-model reasoning depth, this architecture is over-engineering. Stick to simple prompt templates. - **Single-model deployments with no scalability concerns** — If cost and latency are not constraints and a single model (e.g., GPT-4 or Claude Sonnet) handles all tasks adequately, dual-provider orchestration adds complexity without value. - **Exploring reasoning techniques themselves** — If you need to learn about Chain-of-Thought, Tree of Thoughts, or ReAct patterns, use the `reasoning-techniques` skill. This skill assumes you already know those methods and focuses on how to compose them across models. --- ## Core Workflow 1. **Model Capability Mapping** — Classify each task by reasoning type (creative synthesis, logical analysis, factual retrieval, code generation) and map to the optimal model based on its strengths. Gemini 2.5 Pro excels at creative synthesis and multi-modal reasoning; Claude Opus 4 excels at analytical reasoning, logical verification, and structured output. Build a capability matrix that maps task categories to primary and fallback models. **Checkpoint:** Every task category must have a primary model, a fallback model, and an estimated token budget range before any dispatch occurs. 2. **Dual-Provider Task Assignment** — Route tasks through the dual-provider pipeline. For analytical tasks: send to Claude first for structured analysis, then pass the result to Gemini for synthesis and creative enrichment. For creative tasks: send to Gemini first for ideation, then to Claude for fact-checking and logical consistency verification. Merge results using weighted scoring based on confidence and domain expertise. **Checkpoint:** Both model responses must be received (or timeout after budget limit) before merging. If one provider fails, fall back to single-provider execution with a degraded-quality flag. 3. **Reasoning Execution with Token Budget Management** — Execute the reasoning pipeline while monitoring token consumption against per-step budgets. For simple queries (confidence ≥ 0.9), terminate reasoning early and return results immediately. For complex queries (confidence < 0.7), escalate to deeper reasoning with additional verification steps. Track relevance scoring for each message in the conversation turn to prioritize context window allocation for high-signal content. **Checkpoint:** Cumulative token usage must never exceed the total budget for the task. If approaching the limit (>85%), trigger early termination and return best-effort results with a budget-warning flag. 4. **Quality Verification via Cross-Model Consensus** — Compare outputs from different models for consistency on factual claims, logical conclusions, and numerical results. Use a confidence score per reasoning step (0.0–1.0) to weight each model's contribution during merging. For high-stakes decisions (financial, medical, legal), require unanimous agreement between models or escalate to a third-party verification model. **Checkpoint:** If cross-model disagreement exceeds the configured threshold (>15% on factual claims, >20% on recommendations), flag the output for human review before delivery. 5. **AI Development Velocity Tracking** — Measure and report AI-assisted development metrics: percentage of code lines generated by AI, first-pass acceptance rate, revision cycles per task, and average latency from request to deployed code. Compare against industry benchmarks (Google/Microsoft report 30%+ AI-generated code at scale). Track accuracy rates for AI-generated code (does it compile, pass tests, meet requirements?). **Checkpoint:** Velocity metrics must be computed at regular intervals (daily for teams, hourly for production CI pipelines) and surfaced in dashboards with trend analysis over 7-day and 30-day windows. 6. **Budget Monitoring and Adaptive Allocation** — Continuously monitor token usage patterns across all active tasks. Identify tasks that consistently exceed their budget and adjust allocation rules dynamically. Track cost-per-task, latency-per-reasoning-step, and accuracy-vs-cost tradeoffs to optimize the reasoning engine over time. Implement adaptive token reallocation: if one task completes early with surplus budget, redistribute tokens to a competing complex task. **Checkpoint:** Budget reports must be generated every hour showing per-task spend, remaining budget, predicted overrun risk, and recommended reallocation actions. --- ## Implementation Patterns / Reference Guide ### Pattern 1: Dual-Provider Reasoning Pipeline Orchestrate two LLM providers where each handles the task phase it excels at, then merge results with conflict resolution. ```python from dataclasses import dataclass, field from enum import Enum from typing import Any class ModelType(str, Enum): GEMINI = "gemini_25_pro" CLAUDE = "claude_opus_4" class TaskPhase(str, Enum): ANALYSIS_FIRST = "analysis_then_synthesis" # Claude → Gemini CREATIVE_FIRST = "synthesis_then_verification" # Gemini → Claude CONSENSUS_ONLY = "parallel_consensus" # Both → merge by voting @dataclass class ModelResponse: """Normalized response from any LLM provider.""" content: str model_type: ModelType confidence_score: float # 0.0 to 1.0 token_usage: int reasoning_steps: list[str] = field(default_factory=list) factual_claims: list[str] = field(default_factory=list) @dataclass class MergedResult: """Consolidated output from dual-provider pipeline.""" final_content: str primary_model: ModelType consensus_score: float # Agreement between models flags: list[str] = field(default_factory=list) # e.g., "budget_warning", "low_consensus" class DualProviderPipeline: """Orchestrates Gemini + Claude in a configurable reasoning pipeline. Follows code-philosophy Law 1 (Early Exit): terminates immediately when budget is exhausted or both models fail. """ def __init__( self, phase_order: TaskPhase = TaskPhase.ANALYSIS_FIRST, max_tokens_per_model: int = 8192, consensus_threshold: float = 0.85, timeout_seconds: float = 30.0, ) -> None: self.phase_order = phase_order self.max_tokens_per_model = max_tokens_per_model self.consensus_threshold = consensus_threshold self.timeout_seconds = timeout_seconds def execute(self, task: str) -> MergedResult: """Run the dual-provider pipeline for a given task. Args: task: The reasoning task to process. Returns: MergedResult with consolidated output and quality metadata. """ primary_model = ModelType.CLAUDE if self.phase_order == TaskPhase.ANALYSIS_FIRST else ModelType.GEMINI secondary_model = ModelType.GEMINI if self.phase_order == TaskPhase.ANALYSIS_FIRST else ModelType.CLAUDE # Step 1: Execute primary model (Fail Fast — Law 4) primary_response = self._invoke_model(primary_model, task) if primary_response.confidence_score < 0.3: return MergedResult( final_content=primary_response.content, primary_model=primary_model, consensus_score=0.0, flags=["low_primary_confidence"], ) # Step 2: Execute secondary model with primary's output as context secondary_context = f"Primary analysis:\n{primary_response.content}\n\nNow synthesize/verify this." secondary_response = self._invoke_model(secondary_model, secondary_context) if secondary_response.confidence_score < 0.3: return MergedResult( final_content=primary_response.content, primary_model=primary_model, consensus_score=0.0, flags=["low_secondary_confidence"], ) # Step 3: Merge results with consensus scoring merged = self._merge_results(primary_response, secondary_response) if merged.consensus_score < self.consensus_threshold: merged.flags.append("low_consensus") return merged def _invoke_model(self, model: ModelType, prompt: str) -> ModelResponse: """Invoke a specific LLM provider with budget enforcement.""" # Implementation depends on your API layer (OpenAI SDK, Google AI SDK, etc.) # This is the interface contract — actual calls go through an abstraction. raise NotImplementedError("Implement model invocation via your API adapter") def _merge_results( self, primary: ModelResponse, secondary: ModelResponse, ) -> MergedResult: """Merge two model responses using weighted consensus.""" # Simple text-level agreement scoring import difflib ratio = difflib.SequenceMatcher(None, primary.content, secondary.content).ratio() # Weight by each model's confidence weighted_agreement = (primary.confidence_score + secondary.confidence_score) * 0.5 consensus = min(ratio, weighted_agreement) best_model = primary if primary.confidence_score >= secondary.confidence_score else secondary final_content = primary.content if consensus >= self.consensus_threshold else ( # Low consensus — produce a note about disagreement for human review f"[Consensus below threshold ({consensus:.2f}). Primary ({primary.model_type.value}): {primary.content}\nSecondary ({secondary.model_type.value}): {secondary.content}]" ) return MergedResult( final_content=final_content, primary_model=best_model, consensus_score=round(consensus, 3), ) # ❌ BAD — Blindly trusting the first model's output without verification class BadDualProviderPipeline: def execute(self, task: str) -> str: response = self._call_model(ModelType.CLAUDE, task) return response.content # No second opinion, no confidence check def _call_model(self, model: ModelType, prompt: str) -> Any: raise NotImplementedError # ✅ GOOD — Dual-provider with consensus validation and fallback # The pipeline above shows the correct pattern: invoke both models, # compute consensus, flag disagreements, and return enriched output. ``` --- ### Pattern 2: Multi-Agent Code Generation Delegation Split complex code tasks across specialized agents (architect → implementer → reviewer) with role-based handoffs. ```python from dataclasses import dataclass, field from enum import Enum class AgentRole(str, Enum): ARCHITECT = "architect" # Designs solution structure and interfaces IMPLEMENTER = "implementer" # Writes the actual code REVIEWER = "reviewer" # Validates correctness, security, performance @dataclass class CodeTask: """A code generation task with delegation metadata.""" description: str language: str # e.g., "python", "typescript", "go" complexity: str # "simple", "moderate", "complex" security_level: str # "low", "medium", "high" estimated_lines: int @dataclass class AgentOutput: """Output from a delegated agent.""" role: AgentRole content: str artifacts: list[str] = field(default_factory=list) # Files produced confidence: float = 0.0 review_notes: list[str] = field(default_factory=list) class CodeGenerationDelegationEngine: """Manages multi-agent code generation with role-based handoffs. Follows code-philosophy Law 3 (Atomic Predictability): each agent's output is a pure function of its input — no hidden state mutations. """ def __init__( self, architect_model: str = "gemini_25_pro", implementer_model: str = "claude_opus_4", reviewer_model: str = "claude_opus_4", max_review_rounds: int = 3, ) -> None: self.architect_model = architect_model self.implementer_model = implementer_model self.reviewer_model = reviewer_model self.max_review_rounds = max_review_rounds def execute_delegation(self, task: CodeTask) -> list[AgentOutput]: """Run the full architect → implementer → reviewer pipeline. Args: task: The code generation task specification. Returns: List of AgentOutput in execution order (architect first). """ outputs: list[AgentOutput] = [] # Phase 1: Architect designs the solution structure architecture_prompt = self._build_architecture_prompt(task) architect_output = self._invoke_agent( AgentRole.ARCHITECT, architecture_prompt, self.architect_model, ) outputs.append(architect_output) # Phase 2: Implementer writes code based on architecture implementation_prompt = self._build_implementation_prompt( task, architect_output.content ) implementer_output = self._invoke_agent( AgentRole.IMPLEMENTER, implementation_prompt, self.implementer_model, ) outputs.append(implementer_output) # Phase 3: Reviewer validates — with iterative fix loop for round_num in range(1, self.max_review_rounds + 1): review_prompt = self._build_review_prompt( task, implementer_output.content ) reviewer_output = self._invoke_agent( AgentRole.REVIEWER, review_prompt, self.reviewer_model, ) outputs.append(reviewer_output) # Check if review passes — if not, feed fixes back to implementer if self._review_passes(reviewer_output): break # Law 1 (Early Exit) on success fix_prompt = self._build_fix_prompt( task, implementer_output.content, reviewer_output.review_notes, ) implementer_output = self._invoke_agent( AgentRole.IMPLEMENTER, fix_prompt, self.implementer_model, ) outputs.append(implementer_output) return outputs def _build_architecture_prompt(self, task: CodeTask) -> str: return ( f"Design an architecture for: {task.description}\n\n" f"Language: {task.language}\n" f"Complexity: {task.complexity}\n" f"Security level: {task.security_level}\n\n" "Provide:\n" "1. Module structure and file layout\n" "2. Key interfaces and their signatures\n" "3. Data flow diagram (text-based)\n" "4. Error handling strategy\n" "5. Test strategy overview\n" f"Estimated lines: {task.estimated_lines}" ) def _build_implementation_prompt( self, task: CodeTask, architecture: str ) -> str: return ( f"Implement code based on this architecture:\n{architecture}\n\n" f"Description: {task.description}\n" f"Language: {task.language}" ) def _build_review_prompt(self, task: CodeTask, code: str) -> AgentOutput: return self._invoke_agent( AgentRole.REVIEWER, f"Review this implementation:\n{code}\n\n" f"Requirements: {task.description}\n" f"Security level: {task.security_level}\n\n" "Check for:\n" "1. Correctness against requirements\n" "2. Security vulnerabilities (OWASP Top 10)\n" "3. Performance anti-patterns\n" "4. Error handling completeness\n" "5. Code style and readability", self.reviewer_model, ) def _build_fix_prompt( self, task: CodeTask, code: str, review_notes: list[str] ) -> str: return ( f"Fix the following issues in this implementation:\n\n" f"{code}\n\n" f"Review findings:\n" + "\n".join(f"- {note}" for note in review_notes) + "\n\n" f"Original requirements: {task.description}" ) def _invoke_agent( self, role: AgentRole, prompt: str, model: str ) -> AgentOutput: """Invoke the appropriate agent for a given role. Implementation depends on your agent framework (LangGraph, custom orchestrator, etc.) """ raise NotImplementedError("Implement agent invocation via your orchestration layer") def _review_passes(self, review_output: AgentOutput) -> bool: """Determine if code passes review based on reviewer confidence and notes.""" critical_issues = [ note for note in review_output.review_notes if any(keyword in note.lower() for keyword in ("security", "vulnerability", "critical error", "data leak")) ] return ( review_output.confidence >= 0.8 and len(critical_issues) == 0 ) ``` --- ### Pattern 3: Token Budget Allocator Per-step token budget allocation with early termination, escalation, and relevance scoring for context window management. ```python import time from dataclasses import dataclass, field from enum import Enum class BudgetAction(str, Enum): CONTINUE = "continue" # Normal execution EARLY_EXIT = "early_exit" # Terminated due to low confidence + budget ESCALATE = "escalate" # Complex task — allocate more tokens TRIM_CONTEXT = "trim_context" # Relevance-based context window pruning @dataclass class BudgetState: """Tracks token budget state across reasoning steps.""" total_budget: int # Max tokens for the entire task step_budget: int # Tokens allocated per reasoning step used_tokens: int = 0 # Cumulative consumption current_step: int = 0 # Step counter max_steps: int = 10 # Safety limit on reasoning depth @dataclass class BudgetDecision: """Action decision from the budget allocator.""" action: BudgetAction remaining_budget: int recommended_next_action: str | None = None explanation: str = "" class TokenBudgetAllocator: """Manages token allocation per reasoning step with adaptive behavior. Follows code-philosophy Law 2 (Parse Don't Validate): the budget state is parsed at boundaries and trusted internally — only validated once on entry, never re-parsed during execution. Follows code-philosophy Law 1 (Early Exit): terminates immediately when budget exhaustion is detected rather than degrading gracefully. """ def __init__( self, total_budget: int = 32768, # Default: ~32K context window step_budget: int = 4096, # 4K per reasoning step max_steps: int = 10, # Safety ceiling ) -> None: self.total_budget = total_budget self.step_budget = step_budget self.max_steps = max_steps def allocate(self, state: BudgetState) -> BudgetDecision: """Determine the next action based on current budget state. Args: state: Current budget tracking state. Returns: BudgetDecision with the recommended action and explanation. """ # Early exit: budget exhausted if state.used_tokens >= self.total_budget: return BudgetDecision( action=BudgetAction.EARLY_EXIT, remaining_budget=0, explanation="Budget exhausted — terminating reasoning", ) # Step limit reached if state.current_step >= self.max_steps: return BudgetDecision( action=BudgetAction.EARLY_EXIT, remaining_budget=self.total_budget - state.used_tokens, explanation="Max reasoning depth reached — returning best-effort result", ) remaining = self.total_budget - state.used_tokens # Early exit for simple queries with high confidence if remaining > (self.step_budget * 4) and state.current_step == 0: return BudgetDecision( action=BudgetAction.EARLY_EXIT, remaining_budget=remaining, recommended_next_action="Return immediate answer without further reasoning", explanation="Sufficient budget for single-step answer — no need for deep reasoning", ) # Escalation: complex task needs more tokens in next step if remaining < (self.step_budget * 2) and state.current_step > 0: return BudgetDecision( action=BudgetAction.ESCALATE, remaining_budget=remaining, recommended_next_action="Request budget increase for deeper analysis", explanation="Budget running low but task may need more depth — consider escalation", ) # Normal operation: continue with allocated step budget return BudgetDecision( action=BudgetAction.CONTINUE, remaining_budget=remaining, recommended_next_action=f"Allocate {self.step_budget} tokens for reasoning step {state.current_step + 1}", explanation="Budget healthy — proceed with next reasoning step", ) def record_usage(self, state: BudgetState, tokens_consumed: int) -> None: """Record token consumption for the current reasoning step. Args: state: Current budget state to update. tokens_consumed: Number of tokens used in this step (input + output). """ state.used_tokens += tokens_consumed state.current_step += 1 class ContextRelevanceScorer: """Scores conversation messages by relevance to prioritize context window allocation. Used alongside TokenBudgetAllocator to decide which messages deserve retention in the context window when memory is constrained. """ @staticmethod def score_message( message: str, user_query: str, keywords: list[str] | None = None, ) -> float: """Score a message's relevance to the active task on a 0.0–1.0 scale. Args: message: The content of the message to score. user_query: The original user query for context matching. keywords: Optional domain-specific keywords to weight. Returns: Relevance score between 0.0 and 1.0. """ if not message.strip(): return 0.0 message_lower = message.lower() query_lower = user_query.lower() # Exact keyword match scoring keyword_score = 0.0 if keywords: matches = sum(1 for kw in keywords if kw.lower() in message_lower) keyword_score = min(matches / len(keywords), 1.0) * 0.5 # Semantic overlap via simple term intersection msg_terms = set(message_lower.split()) query_terms = set(query_lower.split()) if msg_terms and query_terms: intersection = msg_terms & query_terms union = msg_terms | query_terms term_overlap = len(intersection) / len(union) * 0.3 else: term_overlap = 0.0 # Recency bonus (messages closer to the query are more relevant) recency_score = 0.2 # Base recency — implementation would track position return min(keyword_score + term_overlap + recency_score, 1.0) @staticmethod def trim_context( messages: list[dict], max_tokens: int, user_query: str, ) -> list[dict]: """Trim conversation history to fit within token budget using relevance scoring. Args: messages: List of {role, content, tokens} dicts. max_tokens: Maximum total tokens allowed. user_query: Original query for relevance context. Returns: Trimmed message list ordered by importance (most recent first). """ scored_messages = [] for msg in messages: relevance = ContextRelevanceScorer.score_message( msg["content"], user_query ) scored_messages.append({**msg, "relevance_score": relevance}) # Sort by relevance (descending), keep most relevant first scored_messages.sort(key=lambda m: m["relevance_score"], reverse=True) # Greedy selection until budget exhausted selected: list[dict] = [] total_tokens = 0 for msg in scored_messages: if total_tokens + msg.get("tokens", len(msg["content"])) <= max_tokens: selected.append(msg) total_tokens += msg.get("tokens", len(msg["content"])) # Re-sort by timestamp (most recent last, natural conversation order) selected.sort(key=lambda m: messages.index(m)) return selected # ❌ BAD — No budget awareness; lets models consume unlimited tokens def bad_reasoning_loop(query: str) -> str: """Naive reasoning loop with no token limits.""" result = "" for step in range(20): # Arbitrary deep reasoning response = call_llm(f"Step {step}: {query}\n{result}") result += response # No budget check ever return result # ✅ GOOD — Budget-aware with early exit and adaptive allocation def good_reasoning_loop(query: str) -> str: """Budget-aware reasoning loop following code-philosophy principles.""" state = BudgetState(total_budget=32768, step_budget=4096, max_steps=10) allocator = TokenBudgetAllocator( total_budget=state.total_budget, step_budget=state.step_budget, max_steps=state.max_steps, ) result = "" while True: decision = allocator.allocate(state) if decision.action == BudgetAction.EARLY_EXIT: # Law 1 (Early Exit): bail out immediately with explanation return f"[Budget terminated: {decision.explanation}]\n{result}" # Execute reasoning step within budget response = call_llm_with_budget(f"{query}\nCurrent analysis:\n{result}", max_tokens=state.step_budget) tokens_used = estimate_token_count(response) allocator.record_usage(state, tokens_used) result += response return result # Unreachable — early exit always fires ``` --- ### Pattern 4: AI Development Velocity Tracker Measure AI-assisted development velocity — code generation share, accuracy tracking, and latency metrics against industry benchmarks. ```python from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum class TaskOutcome(str, Enum): ACCEPTED = "accepted" # Code accepted on first pass REVISIONS_NEEDED = "revisions_needed" # Required iterations before acceptance REJECTED = "rejected" # AI output rejected entirely PARTIALLY_USED = "partially_used" # Some AI code kept, rest rewritten @dataclass class CodeGenerationEvent: """A single AI-assisted code generation event.""" timestamp: datetime task_type: str # e.g., "feature", "bugfix", "refactor", "test" language: str estimated_lines_ai: int # Lines generated by AI total_lines_changed: int # Total lines changed in the diff revision_cycles: int # Number of iterations needed time_to_first_output_sec: float # Latency until first code from AI final_acceptance_time_sec: float # Total time from request to acceptance outcome: TaskOutcome reviewer: str | None = None @dataclass class VelocityMetrics: """Aggregated AI development velocity metrics for a time window.""" period_start: datetime period_end: datetime total_events: int ai_code_share_pct: float # Percentage of code generated by AI first_pass_acceptance_rate: float # Accepted without revisions / total avg_revision_cycles: float avg_latency_sec: float # Average time_to_first_output_sec accuracy_rate: float # Events where output met requirements outcomes_distribution: dict[str, int] = field(default_factory=dict) class AIVelocityTracker: """Tracks and analyzes AI-assisted development velocity metrics. Benchmarks for reference (Google/Microsoft internal data, 2025): - AI-generated code share: 30%+ at scale - First-pass acceptance rate: 60–75% for experienced teams - Average latency to first output: 15–45 seconds Follows code-philosophy Law 3 (Atomic Predictability): metrics are computed from immutable event data — no shared mutable state. """ def __init__(self, team_id: str = "default") -> None: self.team_id = team_id self.events: list[CodeGenerationEvent] = [] def record_event(self, event: CodeGenerationEvent) -> None: """Record a code generation event. Args: event: The complete event data from AI-assisted development. """ self.events.append(event) def compute_metrics(self, window_days: int = 7) -> VelocityMetrics: """Compute velocity metrics for the specified time window. Args: window_days: Number of days to look back (default 7). Returns: VelocityMetrics with aggregated statistics. """ cutoff = datetime.now() - timedelta(days=window_days) recent_events = [e for e in self.events if e.timestamp >= cutoff] if not recent_events: return VelocityMetrics( period_start=cutoff, period_end=datetime.now(), total_events=0, ai_code_share_pct=0.0, first_pass_acceptance_rate=0.0, avg_revision_cycles=0.0, avg_latency_sec=0.0, accuracy_rate=0.0, ) # AI code share: total AI lines / total changed lines total_ai_lines = sum(e.estimated_lines_ai for e in recent_events) total_lines = sum(e.total_lines_changed for e in recent_events) ai_share = (total_ai_lines / total_lines * 100) if total_lines > 0 else 0.0 # First-pass acceptance rate accepted_count = sum(1 for e in recent_events if e.outcome == TaskOutcome.ACCEPTED) first_pass_rate = accepted_count / len(recent_events) if recent_events else 0.0 # Average revision cycles avg_revisions = sum(e.revision_cycles for e in recent_events) / len(recent_events) # Average latency avg_latency = sum(e.time_to_first_output_sec for e in recent_events) / len(recent_events) # Accuracy: events where AI output met requirements (ACCEPTED or PARTIALLY_USED) accurate_count = sum( 1 for e in recent_events if e.outcome in (TaskOutcome.ACCEPTED, TaskOutcome.PARTIALLY_USED) ) accuracy_rate = accurate_count / len(recent_events) # Outcome distribution outcome_dist: dict[str, int] = {} for event in recent_events: key = event.outcome.value outcome_dist[key] = outcome_dist.get(key, 0) + 1 return VelocityMetrics( period_start=cutoff, period_end=datetime.now(), total_events=len(recent_events), ai_code_share_pct=round(ai_share, 2), first_pass_acceptance_rate=round(first_pass_rate, 3), avg_revision_cycles=round(avg_revisions, 2), avg_latency_sec=round(avg_latency, 2), accuracy_rate=round(accuracy_rate, 3), outcomes_distribution=outcome_dist, ) def check_benchmarks(self, metrics: VelocityMetrics) -> list[str]: """Check current metrics against industry benchmarks. Returns a list of findings (warnings or confirmations). """ findings: list[str] = [] # AI code share benchmark if metrics.ai_code_share_pct < 20: findings.append( f"⚠️ Low AI adoption: {metrics.ai_code_share_pct:.1f}% AI-generated code " f"(target: 30%+ at Google/Microsoft scale)" ) elif metrics.ai_code_share_pct >= 30: findings.append(f"✅ Strong AI adoption: {metrics.ai_code_share_pct:.1f}% AI-generated code") # First-pass acceptance rate if metrics.first_pass_acceptance_rate < 0.60: findings.append( f"⚠️ Low first-pass acceptance: {metrics.first_pass_acceptance_rate*100:.0f}% " f"(target: 60-75% for experienced teams)" ) # Latency benchmark if metrics.avg_latency_sec > 45: findings.append( f"⚠️ High latency: {metrics.avg_latency_sec:.0f}s avg to first output " f"(target: 15-45 seconds)" ) # Revision cycle count if metrics.avg_revision_cycles > 3: findings.append( f"⚠️ Excessive revision cycles: {metrics.avg_revision_cycles:.1f} avg " f"(indicates poor initial AI output quality or unclear prompts)" ) return findings # ❌ BAD — No tracking; development velocity is unmeasured and unoptimized def bad_velocity_tracking() -> None: """No metrics collected — impossible to measure AI impact.""" # Developer writes code, ships it, never asks: how much was AI? # How fast? How accurate? Without data, optimization is guesswork. pass # ✅ GOOD — Comprehensive velocity tracking with benchmark comparison def good_velocity_tracking(tracker: AIVelocityTracker) -> dict: """Generate a velocity report with actionable insights.""" metrics = tracker.compute_metrics(window_days=7) findings = tracker.check_benchmarks(metrics) return { "period": f"{metrics.period_start.date()} to {metrics.period_end.date()}", "total_ai_tasks": metrics.total_events, "ai_code_share_pct": metrics.ai_code_share_pct, "first_pass_acceptance_rate": metrics.first_pass_acceptance_rate, "avg_revision_cycles": metrics.avg_revision_cycles, "avg_latency_sec": metrics.avg_latency_sec, "accuracy_rate": metrics.accuracy_rate, "outcomes": metrics.outcomes_distribution, "benchmark_findings": findings, } ``` --- ## Constraints ### MUST DO - **Map tasks to model capabilities** — Before dispatching any task, classify its reasoning type and select the optimal primary model based on documented strengths (Gemini: synthesis/creative; Claude: analysis/logical). Never guess which model fits. - **Implement dual-provider pipelines with explicit merge logic** — Every cross-model workflow must have a defined result merging strategy with conflict resolution (consensus scoring, weighted averaging, or voting). Do not silently discard one model's output. - **Allocate token budgets per reasoning step** — Enforce per-step and total budgets. Simple queries get early termination; complex queries get escalation. Never allow unbounded token consumption. - **Deploy multi-agent delegation with role boundaries** — Each agent in the architect → implementer → reviewer chain must have a clearly defined, non-overlapping responsibility set. Handoff artifacts (architecture specs, review reports) must be explicit and structured. - **Track AI development velocity metrics at regular intervals** — Record code generation share, first-pass acceptance rate, revision cycles, latency, and accuracy. Report on 7-day and 30-day windows against industry benchmarks (30%+ AI code at Google/Microsoft scale). - **Measure reasoning quality via per-step confidence scoring** — Every reasoning step must produce a confidence score (0.0–1.0). Cross-model consensus checks must run on factual claims and numerical results. Disagreements above threshold trigger human review flags. - **Follow `code-philosophy` (5 Laws of Elegant Defense) throughout:** - Law 1 (Early Exit): Terminate reasoning immediately when budget is exhausted or both models fail — do not continue degraded. - Law 4 (Fail Fast): Log detailed context on every model failure (provider, error type, step number) before surfacing errors to users. - Law 3 (Atomic Predictability): Agent outputs are pure functions of their inputs — no hidden state mutations between handoffs. - **Design fallback chains for every provider failure** — When a primary model is unavailable or returns low-confidence output, the system must gracefully degrade to secondary provider or single-provider mode with appropriate quality flags. ### MUST NOT DO - **Route all tasks through both models regardless of complexity** — A simple "fix this typo" does not need dual-provider orchestration. Match pipeline depth to task complexity (Law 1: early exit for simple queries). - **Merge conflicting model outputs by averaging text** — Text content cannot be meaningfully averaged. Use consensus scoring, voting on recommendations, or flag disagreements for human review instead. - **Allow any single model's latency to block the entire pipeline** — If one provider times out, return the other model's result with a degradation flag rather than waiting indefinitely or returning nothing. - **Track velocity metrics without accuracy measurements** …(truncated)