AI 에이전트 신뢰성 2편: 실패 감지와 자동 복구

Layer 4: End-to-End 시나리오 테스트
Layer 3: 툴 호출 정확도 테스트
Layer 2: 프롬프트 회귀 테스트
Layer 1: 출력 형식 / 스키마 검증

from pydantic import BaseModel, ValidationError

class AgentResponse(BaseModel):
    action: str
    parameters: dict
    confidence: float  # 0.0 ~ 1.0

def test_schema(raw_output: str):
    try:
        r = AgentResponse.model_validate_json(raw_output)
        assert 0.0 <= r.confidence <= 1.0
        return True
    except ValidationError as e:
        print(f"Schema error: {e}")
        return False

async def probabilistic_assert(agent, prompt, expected, n=20, threshold=0.8):
    results = await asyncio.gather(*[agent.run(prompt) for _ in range(n)])
    rate = Counter(r.action for r in results)[expected] / n
    assert rate >= threshold, f"통과율 {rate:.0%} < 목표 {threshold:.0%}"

class ToolCallRecorder:
    def __init__(self): self.calls = []
    def __call__(self, tool_name, **kwargs):
        self.calls.append({"tool": tool_name, "params": kwargs})
        return mock_response(tool_name, **kwargs)

def test_tool_selection():
    recorder = ToolCallRecorder()
    agent = build_agent(tools=recorder)
    agent.run("고객 ID 12345의 최근 주문을 조회해줘")
    assert recorder.calls[0]["tool"] == "get_orders"
    assert recorder.calls[0]["params"]["customer_id"] == "12345"

from langsmith import Client

client = Client()
def export_golden_tests(project: str, limit=100):
    # 긍정 피드백을 받은 트레이스를 골든 데이터셋으로
    runs = client.list_runs(
        project_name=project,
        filter="eq(feedback_score, 1)",
        limit=limit
    )
    return [{"input": r.inputs, "expected": r.outputs} for r in runs]

# .github/workflows/agent-test.yml
on: [pull_request]
jobs:
  test:
    steps:
      - run: pytest tests/test_schema.py       # 항상, 빠름
      - run: pytest tests/test_tool_calls.py   # PR마다
      - run: pytest tests/test_replay.py       # 야간 스케줄

# 의사코드: 스텝 가드레일
if state.steps >= MAX_STEPS or state.tool_errors >= 3:
    return escalate_to_human(state)  # 티켓/대기열

for i in $(seq 1 20); do pytest -q tests/agent/test_refund_flow.py >/dev/null && echo pass || echo fail; done | sort | uniq -c