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 ) ``` ---