AI 에이전트 롤백 전략 완전 가이드 — 에이전트가 망쳤을 때 복구하는 법

에이전트가 프로덕션 DB를 잘못 수정했습니다. 파일 200개를 잘못 덮어썼습니다. 되돌릴 방법이 없습니다. 이 상황을 구조적으로 막는 법을 정리했습니다.

[핵심 요약]
→ 문제: AI 에이전트는 실수를 확신에 차서 함 — 일반 버그보다 위험
→ 해결: 액션 실행 전 스냅샷, 실행 후 검증, 실패 시 자동 롤백
→ 핵심 패턴: Dry Run, 트랜잭션 래퍼, 스냅샷, 사람 확인 게이트
→ 도구: Git, DB 트랜잭션, 파일 백업, 샌드박스 실행
→ 원칙: 되돌릴 수 없는 액션은 항상 사람이 확인

---

왜 AI 에이전트의 실수가 더 위험한가

일반 버그 vs AI 에이전트 실수:

일반 버그:
→ 재현 가능 — 같은 입력에 같은 오류
→ 스택 트레이스 명확
→ 원인 파악 후 수정

AI 에이전트 실수:
→ 비결정적 — 재현이 어려움
→ 에이전트는 틀렸어도 확신에 차서 실행
→ 여러 파일/DB/API를 연쇄적으로 건드린 후 실패
→ 롤백 대상이 분산됨

실제 사고 시나리오:
에이전트: "레거시 코드 정리 시작"
→ 파일 47개 삭제 (사용 안 한다고 판단)
→ 실제로는 동적 임포트로 사용 중이었음
→ 프로덕션 배포 후 서비스 다운
→ 삭제된 파일 목록? 에이전트 세션 종료로 소멸

[위험도별 액션 분류]
읽기 전용 (안전):
→ 파일 읽기, DB 조회, API GET
→ 롤백 불필요

가역적 (주의):
→ 파일 수정, DB 업데이트, 이메일 초안
→ 스냅샷 후 실행, 롤백 가능

비가역적 (위험):
→ 파일 삭제, DB 삭제, 이메일 발송, 결제, 배포
→ 반드시 사람 확인 후 실행

---

실전 1 — Dry Run 패턴

실제 실행 전에 "무엇을 할 것인지"만 먼저 보여주고 확인을 받습니다.

from dataclasses import dataclass
from enum import Enum
from typing import Any

class RiskLevel(Enum):
    SAFE        = "safe"       # 읽기 전용
    REVERSIBLE  = "reversible" # 되돌릴 수 있음
    IRREVERSIBLE = "irreversible" # 되돌릴 수 없음

@dataclass
class PlannedAction:
    """실행 예정 액션 (Dry Run 결과)"""
    action_id:   str
    tool_name:   str
    description: str         # 사람이 읽을 수 있는 설명
    args:        dict
    risk_level:  RiskLevel
    estimated_impact: str    # "파일 3개 삭제", "DB 행 150개 수정"


class DryRunAgent:
    """Dry Run 지원 에이전트"""

    async def plan(self, task: str) -> list[PlannedAction]:
        """실제 실행 없이 계획만 반환"""
        # LLM에게 계획 수립 요청
        response = await self.llm.complete(f"""
        다음 작업의 실행 계획을 JSON으로 작성해주세요.
        실제 실행은 하지 말고, 무엇을 할지만 설명하세요.

        작업: {task}

        각 액션은 다음 형식으로:
        {{
            "tool_name": "툴 이름",
            "description": "무엇을 하는지 한국어 설명",
            "args": {{}},
            "risk_level": "safe|reversible|irreversible",
            "estimated_impact": "영향 범위 설명"
        }}
        """)

        return self._parse_plan(response)

    async def execute_with_confirmation(
        self,
        task: str,
        auto_approve_safe: bool = True
    ) -> str:
        """계획 확인 후 실행"""

        # 1단계: 계획 수립
        plan = await self.plan(task)

        print("\n=== 실행 계획 ===")
        for i, action in enumerate(plan, 1):
            risk_emoji = {"safe": "✅", "reversible": "⚠️", "irreversible": "🚨"}
            print(f"{i}. {risk_emoji[action.risk_level.value]} "
                  f"[{action.risk_level.value.upper()}] "
                  f"{action.description}")
            print(f"   영향: {action.estimated_impact}")

        # 2단계: 위험 액션 확인
        dangerous = [a for a in plan
                    if a.risk_level == RiskLevel.IRREVERSIBLE]

        if dangerous:
            print(f"\n🚨 비가역적 액션 {len(dangerous)}개 발견:")
            for action in dangerous:
                print(f"  - {action.description}")

            confirm = input("\n계속 진행하시겠습니까? (yes/no): ")
            if confirm.lower() != "yes":
                return "사용자가 취소했습니다."

        # 3단계: 실행
        return await self._execute_plan(plan, auto_approve_safe)

[Dry Run 언제 쓰나]
→ 처음 실행하는 새로운 작업
→ 대규모 파일/DB 조작
→ 프로덕션 환경에서 실행 전
→ 비가역적 액션이 포함된 경우
→ 비용이 비싼 작업 (실수 시 재실행 부담)

---

실전 2 — 트랜잭션 래퍼 (DB 롤백)

DB 관련 에이전트 액션은 항상 트랜잭션으로 감쌉니다.

import asyncpg
from contextlib import asynccontextmanager

class TransactionalAgentDB:
    """트랜잭션 기반 에이전트 DB 래퍼"""

    def __init__(self, conn_str: str):
        self.conn_str = conn_str

    @asynccontextmanager
    async def transaction(self, savepoint: str = None):
        """
        에이전트 DB 작업을 트랜잭션으로 래핑
        실패 시 자동 롤백
        """
        conn = await asyncpg.connect(self.conn_str)
        try:
            async with conn.transaction():
                if savepoint:
                    await conn.execute(f"SAVEPOINT {savepoint}")
                try:
                    yield conn
                    if savepoint:
                        await conn.execute(f"RELEASE SAVEPOINT {savepoint}")
                except Exception as e:
                    if savepoint:
                        await conn.execute(f"ROLLBACK TO SAVEPOINT {savepoint}")
                    raise
        finally:
            await conn.close()


# 에이전트에서 사용
class DatabaseAgent:

    def __init__(self, db: TransactionalAgentDB):
        self.db = db

    async def bulk_update_with_rollback(
        self,
        updates: list[dict],
        validation_fn = None
    ) -> dict:
        """
        대량 업데이트 — 검증 실패 시 전체 롤백
        """
        results = []

        async with self.db.transaction(savepoint="bulk_update") as conn:
            for update in updates:
                # 업데이트 실행
                await conn.execute("""
                    UPDATE products
                    SET price = $1, updated_at = NOW()
                    WHERE id = $2
                """, update["price"], update["id"])

                results.append(update)

            # 검증 실행 (실패 시 전체 롤백)
            if validation_fn:
                validation_result = await validation_fn(conn)
                if not validation_result["passed"]:
                    raise ValueError(
                        f"검증 실패: {validation_result['reason']}\n"
                        f"전체 업데이트 롤백됨"
                    )

        return {"updated": len(results), "status": "success"}


# 사용 예시
async def run_price_update_agent():
    db_agent = DatabaseAgent(db=TransactionalAgentDB(CONN_STR))

    async def validate_prices(conn):
        """가격이 0 이하인 항목 없는지 검증"""
        count = await conn.fetchval(
            "SELECT COUNT(*) FROM products WHERE price <= 0"
        )
        return {
            "passed": count == 0,
            "reason": f"가격이 0 이하인 항목 {count}개 발견" if count > 0 else ""
        }

    try:
        result = await db_agent.bulk_update_with_rollback(
            updates=[
                {"id": 1, "price": 15000},
                {"id": 2, "price": 25000},
                {"id": 3, "price": -999},  # 잘못된 가격 → 롤백 트리거
            ],
            validation_fn=validate_prices
        )
    except ValueError as e:
        print(f"롤백 완료: {e}")
        # 모든 가격이 원래대로 복원됨

---

실전 3 — 파일 시스템 스냅샷

파일 조작 전에 스냅샷을 찍고, 실패 시 복원합니다.

import shutil
import os
from pathlib import Path
from datetime import datetime
from contextlib import contextmanager

class FileSystemSnapshot:
    """파일 시스템 스냅샷 및 롤백"""

    def __init__(self, snapshot_dir: str = ".agent_snapshots"):
        self.snapshot_dir = Path(snapshot_dir)
        self.snapshot_dir.mkdir(exist_ok=True)

    def create(self, target_paths: list[str]) -> str:
        """지정된 경로들의 스냅샷 생성"""
        snapshot_id = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
        snapshot_path = self.snapshot_dir / snapshot_id
        snapshot_path.mkdir()

        manifest = []
        for path in target_paths:
            src = Path(path)
            if src.exists():
                if src.is_file():
                    dst = snapshot_path / src.name
                    shutil.copy2(src, dst)
                    manifest.append({"path": str(src), "type": "file"})
                elif src.is_dir():
                    dst = snapshot_path / src.name
                    shutil.copytree(src, dst)
                    manifest.append({"path": str(src), "type": "dir"})

        # 매니페스트 저장
        import json
        (snapshot_path / "manifest.json").write_text(
            json.dumps(manifest, ensure_ascii=False)
        )

        print(f"스냅샷 생성: {snapshot_id} ({len(manifest)}개 경로)")
        return snapshot_id

    def restore(self, snapshot_id: str) -> None:
        """스냅샷에서 복원"""
        import json
        snapshot_path = self.snapshot_dir / snapshot_id

        if not snapshot_path.exists():
            raise ValueError(f"스냅샷 없음: {snapshot_id}")

        manifest = json.loads((snapshot_path / "manifest.json").read_text())

        for item in manifest:
            src = snapshot_path / Path(item["path"]).name
            dst = Path(item["path"])

            # 현재 파일 삭제
            if dst.exists():
                if dst.is_file():
                    dst.unlink()
                elif dst.is_dir():
                    shutil.rmtree(dst)

            # 스냅샷에서 복원
            if item["type"] == "file":
                shutil.copy2(src, dst)
            elif item["type"] == "dir":
                shutil.copytree(src, dst)

        print(f"복원 완료: {snapshot_id}")

    def cleanup(self, snapshot_id: str) -> None:
        """성공 후 스냅샷 삭제"""
        shutil.rmtree(self.snapshot_dir / snapshot_id)


@contextmanager
def file_transaction(target_paths: list[str], snapshot_dir: str = ".agent_snapshots"):
    """
    파일 작업을 트랜잭션처럼 처리
    실패 시 자동 롤백
    """
    fs = FileSystemSnapshot(snapshot_dir)
    snapshot_id = fs.create(target_paths)

    try:
        yield
        # 성공 시 스냅샷 삭제
        fs.cleanup(snapshot_id)
        print("파일 작업 완료 — 스냅샷 삭제")
    except Exception as e:
        # 실패 시 롤백
        print(f"오류 발생: {e}")
        print("파일 롤백 중...")
        fs.restore(snapshot_id)
        fs.cleanup(snapshot_id)
        print("롤백 완료")
        raise


# 에이전트에서 사용
async def refactor_agent(target_dir: str):
    target_files = list(Path(target_dir).glob("**/*.py"))

    with file_transaction([str(f) for f in target_files]):
        for file in target_files:
            # 파일 수정 (실패 시 전체 롤백)
            content = file.read_text()
            modified = apply_refactoring(content)
            file.write_text(modified)

        # 수정 후 검증
        if not validate_all_files(target_files):
            raise ValueError("검증 실패 — 롤백 트리거")

---

실전 4 — Git 기반 롤백 (코드 수정 에이전트)

코드 수정 에이전트는 Git을 자동 롤백 시스템으로 활용합니다.

import subprocess
from pathlib import Path

class GitBackedAgent:
    """Git을 롤백 시스템으로 사용하는 에이전트"""

    def __init__(self, repo_path: str):
        self.repo = Path(repo_path)

    def _git(self, *args) -> str:
        """Git 명령 실행"""
        result = subprocess.run(
            ["git", *args],
            cwd=self.repo,
            capture_output=True,
            text=True
        )
        if result.returncode != 0:
            raise RuntimeError(f"Git 오류: {result.stderr}")
        return result.stdout.strip()

    def create_checkpoint(self, message: str) -> str:
        """현재 상태를 Git 커밋으로 체크포인트 저장"""
        self._git("add", "-A")

        # 변경사항이 있을 때만 커밋
        status = self._git("status", "--porcelain")
        if not status:
            return self._git("rev-parse", "HEAD")  # 현재 커밋 반환

        self._git("commit", "-m", f"[AGENT CHECKPOINT] {message}")
        commit_hash = self._git("rev-parse", "HEAD")
        print(f"체크포인트 생성: {commit_hash[:8]} — {message}")
        return commit_hash

    def rollback_to(self, commit_hash: str) -> None:
        """특정 커밋으로 롤백"""
        print(f"롤백 중: {commit_hash[:8]}")
        self._git("reset", "--hard", commit_hash)
        print("롤백 완료")

    def get_diff(self, commit_hash: str) -> str:
        """체크포인트 이후 변경사항 조회"""
        return self._git("diff", commit_hash, "HEAD")

    async def run_with_git_rollback(self, task: str) -> str:
        """Git 체크포인트로 보호된 에이전트 실행"""

        # 시작 전 체크포인트
        checkpoint = self.create_checkpoint(f"작업 시작: {task[:50]}")

        try:
            # 에이전트 실행
            result = await self._execute_coding_task(task)

            # 실행 후 검증
            if not self._validate_code():
                raise ValueError("코드 검증 실패")

            # 성공 시 최종 커밋
            self.create_checkpoint(f"작업 완료: {task[:50]}")
            return result

        except Exception as e:
            print(f"오류: {e}")
            # 체크포인트로 롤백
            self.rollback_to(checkpoint)
            return f"실패 및 롤백 완료: {str(e)}"

    def _validate_code(self) -> bool:
        """코드 검증 (문법 오류, 테스트 등)"""
        try:
            # Python 문법 검사
            result = subprocess.run(
                ["python", "-m", "py_compile", "."],
                cwd=self.repo,
                capture_output=True
            )
            return result.returncode == 0
        except Exception:
            return False

---

실전 5 — 사람 확인 게이트 (Human-in-the-Loop)

비가역적 액션 직전에 반드시 사람이 확인하는 게이트를 삽입합니다.

from enum import Enum

class ApprovalResult(Enum):
    APPROVED = "approved"
    REJECTED = "rejected"
    MODIFIED = "modified"

class HumanApprovalGate:
    """사람 확인 게이트"""

    def __init__(self, mode: str = "cli"):
        self.mode = mode  # "cli", "slack", "email"

    async def request_approval(
        self,
        action_description: str,
        impact: str,
        preview: str = None,
        timeout_seconds: int = 300
    ) -> tuple[ApprovalResult, str]:
        """
        사람에게 확인 요청

        Returns:
            (결과, 코멘트)
        """
        if self.mode == "cli":
            return await self._cli_approval(
                action_description, impact, preview
            )
        elif self.mode == "slack":
            return await self._slack_approval(
                action_description, impact, preview, timeout_seconds
            )

    async def _cli_approval(
        self, description, impact, preview
    ) -> tuple[ApprovalResult, str]:
        print("\n" + "="*50)
        print("🚨 비가역적 액션 — 사람 확인 필요")
        print("="*50)
        print(f"액션: {description}")
        print(f"영향: {impact}")
        if preview:
            print(f"\n미리보기:\n{preview}")
        print("="*50)

        while True:
            choice = input("\n[a]승인 / [r]거부 / [m]수정 / [?]상세: ").lower()

            if choice == "a":
                comment = input("코멘트 (선택): ")
                return ApprovalResult.APPROVED, comment

            elif choice == "r":
                reason = input("거부 이유: ")
                return ApprovalResult.REJECTED, reason

            elif choice == "m":
                modification = input("수정 내용: ")
                return ApprovalResult.MODIFIED, modification

            elif choice == "?":
                print(f"\n상세 정보:\n{preview or '없음'}")

    async def _slack_approval(
        self, description, impact, preview, timeout
    ) -> tuple[ApprovalResult, str]:
        """Slack으로 승인 요청 (Slack Bolt 앱 필요)"""
        # Slack Block Kit으로 승인 버튼 전송
        # 사용자 클릭 대기
        # 응답 반환
        pass  # 실제 구현은 Slack Bolt SDK 참고


# 에이전트에서 사용
class SafeDeploymentAgent:

    def __init__(self, approval_gate: HumanApprovalGate):
        self.gate = approval_gate

    async def deploy(self, service: str, version: str) -> str:
        # 배포 계획 수립
        plan = await self._create_deploy_plan(service, version)

        # 비가역적 액션(배포)에 사람 확인 요청
        result, comment = await self.gate.request_approval(
            action_description=f"{service} v{version} 프로덕션 배포",
            impact="서비스 다운타임 30초 예상, 이전 버전 롤백 불가",
            preview=f"변경사항:\n{plan['changelog']}"
        )

        if result == ApprovalResult.APPROVED:
            return await self._execute_deploy(plan)
        elif result == ApprovalResult.REJECTED:
            return f"배포 취소됨: {comment}"
        elif result == ApprovalResult.MODIFIED:
            # 수정사항 반영 후 재계획
            return await self.deploy(service, version)

---

실전 6 — 통합: 전체 롤백 파이프라인

class ResilientAgent:
    """모든 롤백 전략이 통합된 에이전트"""

    def __init__(
        self,
        db: TransactionalAgentDB,
        fs_snapshot: FileSystemSnapshot,
        git: GitBackedAgent,
        approval_gate: HumanApprovalGate,
        state_store: StateStore
    ):
        self.db       = db
        self.fs       = fs_snapshot
        self.git      = git
        self.gate     = approval_gate
        self.state    = state_store

    async def execute_safely(self, task: str, session_id: str) -> str:
        """안전한 에이전트 실행 전체 파이프라인"""

        # 1. 상태 체크포인트
        state = AgentState(session_id=session_id, task=task,
                          status=AgentStatus.RUNNING)
        await self.state.save(state)

        # 2. Git 체크포인트
        git_checkpoint = self.git.create_checkpoint(f"작업 시작: {task[:50]}")

        # 3. Dry Run으로 계획 수립
        plan = await self._plan(task)
        dangerous_actions = [a for a in plan
                            if a.risk_level == RiskLevel.IRREVERSIBLE]

        # 4. 위험 액션 사람 확인
        if dangerous_actions:
            result, _ = await self.gate.request_approval(
                action_description=f"{len(dangerous_actions)}개 비가역적 액션 포함",
                impact="\n".join(a.estimated_impact for a in dangerous_actions)
            )
            if result != ApprovalResult.APPROVED:
                return "사용자 취소"

        # 5. DB 트랜잭션으로 실행
        try:
            async with self.db.transaction(savepoint="agent_work") as conn:
                result = await self._execute_plan(plan, conn)

                # 6. 실행 후 검증
                if not await self._validate(conn):
                    raise ValueError("검증 실패")

            # 7. 성공 상태 저장
            state.status = AgentStatus.COMPLETED
            await self.state.save(state)
            return result

        except Exception as e:
            # 8. 실패 시 모든 레이어 롤백
            self.git.rollback_to(git_checkpoint)  # 파일 롤백
            # DB는 트랜잭션으로 자동 롤백됨

            state.status = AgentStatus.FAILED
            state.context["error"] = str(e)
            await self.state.save(state)

            return f"실패 및 전체 롤백 완료: {str(e)}"

---

마무리

✅ 롤백 전략 필수인 상황
→ 파일 삭제/수정이 포함된 에이전트
→ 프로덕션 DB를 건드리는 에이전트
→ 배포, 이메일 발송, 결제 등 비가역적 액션
→ 여러 시스템을 연쇄적으로 수정하는 멀티 에이전트

❌ 과한 경우
→ 읽기 전용 에이전트 (조회만 함)
→ 샌드박스 환경에서만 실행
→ 실패해도 재실행 비용이 낮은 프로토타입

[구현 우선순위]
1순위: Dry Run + 사람 확인 게이트 (즉각 효과)
2순위: DB 트랜잭션 래퍼 (DB 안전망)
3순위: Git 체크포인트 (코드 수정 안전망)
4순위: 파일 스냅샷 (파일 시스템 안전망)
5순위: 통합 파이프라인 (모든 레이어 결합)

 

---

관련 글:

 

AI 에이전트 상태 관리 완전 가이드 — 장기 실행 에이전트에서 상태를 잃지 않는 법

AI 에이전트 테스트 전략 완전 가이드 — 단위 테스트부터 통합 테스트, E2E까지

AI 에이전트에 Shell Access 주면 안 되는 이유 — 실제 해킹 사례와 방어법