为 Agentic AI 工作流加装断路器：控制自主代码审查的爆炸半径

Originally published on tamiz.pro. The rapid adoption of agentic AI in software engineering has introduced a new class of operational risks: autonomous agents that can interpret, execute, and commit code. While powerful, these systems lack the inherent safety rails of traditional deterministic software. When an LLM agent makes a cascading error in a code review pipeline, the "blast radius" can expand from a single file to a corrupted build, a failing deployment, or even a security vulnerability. This article dissects the architectural necessity of applying Circuit Breaker patterns to agentic workflows, specifically focusing on AI-driven code review. We will move beyond simple retry logic to explore state-machine-based fail-safes that isolate failures, prevent resource exhaustion, and enforce human-in-the-loop gating when confidence levels drop below operational thresholds. 1. The Problem: Non-Determinism and Cascade Failure 1.1 Agentic Autonomy vs. Deterministic Control Traditional code review tools are deterministic: regular expressions, static analyzers (like ESLint or SonarQube), and linters produce predictable outputs for predictable inputs. However, agentic workflows powered by Large Language Models (LLMs) are probabilistic. An agent might: Hallucinate Context: Misinterpret the architectural intent of a module. Loop Infinite: Get stuck in a reflection loop where it critiques its own correction, then critiques the critique. Execute Side Effects: Run git commit or trigger CI/CD pipelines that have cascading financial or security implications. In a standard microservices architecture, a circuit breaker opens when a service fails N times in a rolling window, stopping traffic to that service to allow it to recover. In an agentic AI workflow, the "service" is the agent's decision-making capability. The "traffic" is the sequence of actions it takes. If we don't break the circuit, a buggy prompt engineering update or a specific edge-case PR can trigger an autonomous agent to continuously generate bad code, exhaust API tokens, or introduce subtle security flaws into the main branch. 1.2 Defining Blast Radius in AI Context In traditional SRE, blast radius refers to the scope of impact when a component fails. In AI engineering, it is multidimensional: Cognitive Blast Radius: The agent's "mental model" of the codebase is corrupted by bad context, leading to sustained logical errors. Resource Blast Radius: Infinite loops consuming GPU tokens or API credits. Security Blast Radius: The agent generates code that bypasses security checks (e.g., SQL injection in generated database migrations). Operational Blast Radius: The agent triggers a deployment that breaks production, affecting users. 2. Architectural Design: The Agentic Circuit Breaker 2.1 Core State Machine We define our circuit breaker not as a simple flag, but as a finite state machine (FSM) that gates agent actions. The states are: CLOSED: Normal operation. The agent is allowed to execute actions (e.g., suggest code changes, run tests). OPEN: Failure threshold breached. All autonomous actions are blocked. The system falls back to a "Safe Mode" (human review only or static analysis only). HALF-OPEN: A trial period. A limited set of low-risk actions is allowed to test if the underlying issue (e.g., a context drift or model instability) has resolved. 2.2 Defining Failure Metrics What constitutes a "failure" in an agentic review? We must move beyond HTTP 500 errors. Key metrics for opening the circuit include: Token Exhaustion: The agent exceeds a predefined token limit for a single PR without reaching a conclusion. Negative Feedback Loop: The agent generates a correction, the tests fail, and it generates the same correction again (detected via hash similarity). Security Flag: A secondary "Guardrail" model (a smaller, faster LLM or static analyzer) flags the generated code as high-risk (e.g., eval(), hardcoded secrets). Latency Spike: The time to complete a review step exceeds the SLO, indicating potential infinite loops or API throttling. 3. Implementation Strategy 3.1 The Wrapper Pattern Rather than hardcoding checks into every agent prompt, we wrap the agent's execution loop in a CircuitBreakerMiddleware. This middleware intercepts every action the agent requests to perform. class AgenticCircuitBreaker: def __init__(self, failure_threshold=5, recovery_timeout=300, token_budget=10000, security_guardrail=None): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.token_budget = token_budget self.security_guardrail = security_guardrail self.state = "CLOSED" self.failure_count = 0 self.last_failure_time = None self.total_tokens_used = 0 self.consecutive_duplicate_actions = 0 async def execute_action(self, agent, action: AgentAction, context: ReviewContext): """ Guts the agent's action. If circuit is OPEN, blocks and triggers fallback. """ if self.state == "OPEN": if self._should_attempt_half_open(): self.state = "HALF-OPEN" logger.info("Circuit Breaker: Transitioning to HALF-OPEN for test") else: raise CircuitOpenError( f"Circuit is OPEN. Last failure: {self.last_failure_time}. " f"Falling back to human review queue." ) try: # 1. Pre-Flight Check: Security Guardrail if self.security_guardrail: risk_score = await self.security_guardrail.evaluate(action) if risk_score > 0.8: # High risk self._record_failure("Security Guardrail Flag") raise SecurityViolationError(action) # 2. Token Budget Check estimated_tokens = self._estimate_tokens(action) if self.total_tokens_used + estimated_tokens > self.token_budget: self._record_failure("Token Budget Exhausted") raise TokenExhaustionError() # 3. Execute the Action (e.g., LLM call, git commit, test run) result = await agent.execute(action) # 4. Post-Flight Check: Detect Infinite Loops if result.is_duplicate_of_previous: self.consecutive_duplicate_actions += 1 if self.consecutive_duplicate_actions >= 2: self._record_failure("Infinite Loop Detected") raise InfiniteLoopError() else: self.consecutive_duplicate_actions = 0 self._record_success() return result except Exception as e: self._record_failure(str(e)) raise e def _record_failure(self, reason): self.failure_count += 1 self.last_failure_time = time.time() if self.failure_count >= self.failure_threshold: self._open_circuit() logger.warning(f"Circuit Breaker Failure: {reason}. Count: {self.failure_count}") def _record_success(self): self.failure_count = 0 self.state = "CLOSED" def _should_attempt_half_open(self): if self.state == "OPEN": if time.time() - self.last_failure_time > self.recovery_timeout: return True return False def _open_circuit(self): self.state = "OPEN" logger.error(f"Circuit Breaker OPENED after {self.failure_count} failures.") Enter fullscreen mode Exit fullscreen mode 3.2 Contextual Isolation A major cause of "cognitive blast radius" is context pollution. If an agent reads a 5,000-line file, it may hallucinate dependencies. To mitigate this, the circuit breaker should also enforce Context Sanitization. Before the agent begins its review, a preprocessing step chunks the codebase. The breaker tracks which chunks have been injected into the agent's window and which chunks have triggered context overflow or semantic drift alerts. If the drift metric exceeds a predefined threshold—indicating the agent is losing focus on the actual code structure—the breaker trips the Context Limiter, truncating the prompt history and re-injecting a high-fidelity summary of the relevant architectural layers. This prevents the "lost in the middle" phenomenon where critical logic buried thousands of tokens away is ignored during the review. Implementation: The Adaptive Threshold Controller Static thresholds are insufficient for agentic systems. A monolithic review of a core banking service requires different safety margins than a pull request touching a static documentation generator. We implement an adaptive controller that adjusts breaker parameters based on the repository's historical volatility and the agent's confidence metrics. The core logic resides in a lightweight state machine. Here is a Python implementation using dataclasses and standard library components, designed to be drop-in compatible with LangChain or LlamaIndex agent frameworks. import time from enum import Enum from dataclasses import dataclass, field from typing import Optional, List, Callable import threading class BreakerState(Enum): CLOSED = "closed" OPEN = "open" HALF_OPEN = "half_open" @dataclass class BreakerConfig: failure_threshold: int = 5 recovery_timeout: float = 60.0 success_threshold: int = 3 context_drift_limit: float = 0.85 # 85% drift triggers trip confidence_floor: float = 0.60 # Below this, count as failure class AgenticCircuitBreaker: def __init__(self, config: BreakerConfig = None): self.config = config or BreakerConfig() self.state = BreakerState.CLOSED self.failure_count = 0 self.last_failure_time = 0.0 self.half_open_calls = 0 self._lock = threading.RLock() self.history: List[dict] = [] def _trip(self, reason: str): """Transitions breaker to OPEN state.""" with self._lock: self.state = BreakerState.OPEN self.last_failure_time = time.time() self.failure_count = 0 self.history.append({ "timestamp": self.last_failure_time, "state": "OPEN", "reason": reason }) # In a production system, emit a telemetry event here def _attempt_reset(self): """Transitions from OPEN to HALF_OPEN if timeout has elapsed.""" with self._lock: if self.state == BreakerState.OPEN: if time.time() - self.last_failure_time >= self.config.recovery_timeout: self.state = BreakerState.HALF_OPEN self.half_open_calls = 0 self.history.append({ "timestamp": time.time(), "state": "HALF_OPEN", "reason": "Recovery attempt initiated" }) return True return False def execute(self, agent_call: Callable, *args, **kwargs): """ Wraps the agent's execution step. Args: agent_call: The function representing the agent's next step. *args, **kwargs: Arguments passed to the agent. Returns: The result of the agent_call if successful. Raises CircuitBreakerOpenError if the breaker is tripped. """ # Check if in OPEN state and eligible for HALF_OPEN if self._attempt_reset() and self.state == BreakerState.HALF_OPEN: # In HALF_OPEN, we allow limited traffic if self.half_open_calls >= self.config.success_threshold: self.state = BreakerState.CLOSED self.half_open_calls = 0 else: self.half_open_calls += 1 if self.state == BreakerState.OPEN: raise CircuitBreakerOpenError( f"Breaker is OPEN. Last trip: {time.ctime(self.last_failure_time)}" ) try: # Execute the agent step result, metrics = agent_call(*args, **kwargs) # Evaluate metrics for soft failures if metrics.get("confidence", 1.0) self.config.context_drift_limit: self._handle_soft_failure("High context drift detected") else: self._handle_success() return result except Exception as e: self._handle_hard_failure(str(e)) raise def _handle_soft_failure(self, reason: str): with self._lock: self.failure_count += 1 self.history.append({ "timestamp": time.time(), "state": self.state.value, "reason": reason, "failure_count": self.failure_count }) if self.failure_count >= self.config.failure_threshold: self._trip(reason) def _handle_hard_failure(self, reason: str): with self._lock: # Hard failures (exceptions) trip the breaker immediately # or contribute more heavily to the count depending on strategy. # Here we use a 1-strike policy for critical errors. self.failure_count += 1 self.history.append({ "timestamp": time.time(), "state": self.state.value, "reason": f"Hard Failure: {reason}", "failure_count": self.failure_count }) if self.failure_count >= 1: # Aggressive for autonomous agents self._trip(reason) def _handle_success(self): with self._lock: self.failure_count = 0 if self.state == BreakerState.HALF_OPEN: # Successful call in half-open state progresses recovery pass # Logic handled in execute() for state transition check class CircuitBreakerOpenError(Exception): pass Enter fullscreen mode Exit fullscreen mode Integrating with Fallback Strategies When the breaker trips, the workflow cannot simply halt; it must degrade gracefully. In an agentic code review pipeline, degradation involves shifting from autonomous generation to Supervised Verification. Human-in-the-Loop (HITL) Escalation: When the breaker moves to OPEN, the orchestrator pauses the autonomous loop. It packages the last N interactions, the specific code diff, and the reason for tripping (e.g., "High context drift") into a structured handoff payload. This payload is sent to a human reviewer's queue. The agent is effectively "parked." Simpler Heuristic Fallback: If HITL is unavailable, the system falls back to a static, rule-based linter (like Pylint or ESLint) instead of the LLM. This ensures that critical security vulnerabilities are still caught, even if the nuanced architectural critique is lost. Retry with Reduced Scope: In the HALF_OPEN state, the orchestrator does not retry the original complex task. Instead, it slices the review into smaller, non-overlapping chunks. For example, instead of reviewing an entire UserSession class, it reviews only the validate_token method. This reduces the context load and allows the agent to "warm up" its performance metrics before the breaker potentially re-closes. Monitoring and Telemetry A circuit breaker without telemetry is a black box. Every state transition must be logged to a time-series database. Key metrics to track include: Trip Rate: The percentage of review sessions that resulted in a breaker trip. A high trip rate indicates that the agent's context window is too small for the typical codebase size, or the model is inherently unstable on this domain. Mean Time to Recovery (MTTR): The time between the breaker opening and the first successful HALF_OPEN transition. This helps tune the recovery_timeout. Drift Correlation: Plotting context_drift scores against token count. This often reveals a non-linear relationship where drift spikes exponentially past 40% of the model's context window, allowing you to set hard cap limits on chunk size. Concluding Thoughts Agentic AI systems are powerful, but they are inherently stochastic. They do not fail gracefully; they fail confusingly. By wrapping these autonomous loops in circuit breakers, we shift the safety model from "preventing errors" (which is impossible with LLMs) to "containing errors." The key insight is that the breaker is not just a failure detector; it is a context governor. It enforces hygiene by forcing context sanitization and preventing the accumulation of hallucinated state. As we build more complex multi-agent systems, where one agent reviews another's code, these breakers become the immune system of the workflow. They allow the system to be aggressive in its exploration while remaining robust in its failure modes. Start small. Implement the breaker around your most critical, least-automated agent. Monitor the trip reasons. You will likely discover that the "failures" are not actually code errors, but context management issues that can be solved by better chunking strategies, not better prompts. The breaker will teach you where the agent's cognitive limits lie, allowing you to design workflows that respect those boundaries rather than fighting them.