메트릭 수집 메커니즘 & 개선 구성code-level

A1. events 수집 — 이벤트 → 레코드 디스패치

async for로 스트림을 소비하며 이벤트 타입별로 분기한다. seq는 INSERT마다 증가(전순서 보장). 토큰·턴·지연이 여기서 누적된다.

def _extract_tokens(event):
    usage = getattr(event, "models_usage", None)
    if usage is None: return 0, 0
    return int(usage.prompt_tokens or 0), int(usage.completion_tokens or 0)

# log_stream 누적:
total_input_tokens += inp;  total_output_tokens += out
total_turns += 1                       # agent_message 1건당
latency_ms = (last_ts - first_ts) * 1000   # 첫~마지막 이벤트

도구 에러 — 실행 이벤트의 is_error로 tool_result/tool_error 분기. mv_tool_summary.errors의 원천:

is_error = getattr(result, "is_error", False)
if is_error:
    payload = {"error_type": "ToolExecutionError", "message": raw_content, ...}
    # message_type = 'tool_error' 로 INSERT

A2. 비밀 마스킹 · 절단 · 로깅 안전성

마스킹(절대규칙 5): tool_args 저장 전 단일 지점에서 재귀 마스킹. 절단: content 8192자 / tool_result 2048자(+sha256). 로깅이 평가를 죽이지 않음(절대규칙 2): INSERT 실패해도 카운트만 하고 계속.

_SECRET_KEY_RE = re.compile(r"(password|token|secret|api_key|authorization)", re.I)
def _mask_secrets(obj):   # dict/list 재귀; 키가 패턴이면 값 → "***"
    if isinstance(obj, dict):
        return {k: ("***" if _SECRET_KEY_RE.search(k) else _mask_secrets(v)) ...}

async def _safe_insert(self, run_id, record):
    try: await _insert_event(...); return 0
    except Exception as exc: logger.warning(...); return 1  # 누락 카운트만

A3. success / format_valid — GAIA exact-match 채점

format_valid = FINAL ANSWER 패턴 추출 성공 여부. success = format_valid AND 정규화 일치. 정규화는 숫자/리스트/문자열을 구분하고 리스트는 순서 보존한다.

def score(predicted, gold):
    format_valid = predicted is not None          # FINAL ANSWER 패턴 있었나
    if not format_valid:
        return ScoreResult(success=False, format_valid=False)
    success = _compare_normalized(normalize(predicted), normalize(gold))
    return ScoreResult(success=success, format_valid=True)

# extract_final_answer: 정규식 FINAL ANSWER\s*:\s*(.+?)(?:\n|$), 마지막 매치 사용
# normalize: 통화·콤마·% 제거→숫자 / 세미콜론·콤마2개+→리스트(순서보존) / 그외 문자열
# _compare_normalized: 둘다 float이면 abs(a-b)<1e-6, 아니면 문자열 동등

template_literal 방어 — 로거가 planner 답을 잡을 때 플레이스홀더를 거른다:

_TEMPLATE_PLACEHOLDER_RE = re.compile(r"^[\`<*]")   # `<answer>` 시작이면 무시

A4. terminated_by → loop_rate · runs UPDATE

stop_reason을 3값으로 매핑. loop_rate = max_messages로 끝난 비율. run 1회는 단일 UPDATE로 확정된다.

def _terminated_by(stop_reason):
    lowered = (stop_reason or "").lower()
    if "approve" in lowered: return "approve"
    if "max" in lowered: return "max_messages"
    return None

# 종료 조건(팀 빌드): APPROVE(critic) OR MaxMessageTermination(max_messages)
await conn.execute("""UPDATE runs SET status='completed', terminated_by=$2,
  success=$4, format_valid=$5, total_turns=$6, total_tokens=$7, ... WHERE run_id=$1""", ...)

A5. 집계 — materialized view

모든 패널 수치의 최종 출처. 배치 종료 시 db.refresh_views()로 REFRESH.

CREATE MATERIALIZED VIEW mv_batch_summary AS SELECT r.batch_id, r.team_id, t.level,
  AVG(r.success::int)                        AS pass_at_1,
  AVG((NOT r.format_valid)::int) FILTER (WHERE status='completed') AS format_invalid_rate,
  AVG((r.terminated_by='max_messages')::int) AS loop_rate,
  SUM(r.total_tokens)::float / NULLIF(SUM(r.success::int),0) AS tokens_per_success,
  PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY r.latency_ms) AS p95_latency
FROM runs r JOIN tasks t USING (task_id) WHERE status='completed' GROUP BY 1,2,3;

B0. 공통 원칙 — 1변수 + 새 team_id + 재현성 가드

개입 1건 = 새 팀 모듈 + 새 team_id + 새 prompt_bundle_hash. 1변수 격리를 위해 안 바꾸는 부분은 베이스라인에서 import한다. 해시가 바뀌면 기존 team_id 재사용을 에러로 중단(AC9).

# executor·selector는 v1 그대로, planner·critic만 교체 → 1변수 격리
from agent_eval.teams.v1_baseline import EXECUTOR_PROMPT, SELECTOR_PROMPT, python_calc, read_attached_file

B1. v2-attach-path EXP-40 promoted

특이점: read_attached_file 100% 에러 (executor가 경로 추측 실패). 변경: 프롬프트·도구 불변, 질문 메시지에만 절대경로 1줄 주입(1변수).

def build_run_message(task, settings):
    if not task.file_name: return task.question
    abs_path = Path(settings.gaia_dir).resolve() / task.file_name
    return (f"{task.question}\n\n[첨부파일 경로: {abs_path}] "
            f"이 파일이 필요하면 read_attached_file 도구에 이 경로를 그대로 전달하라.")

B2. v3-prompt-fix EXP-41 promoted

특이점: final_answer = "<answer>" (template_literal). 변경: PLANNER/CRITIC 프롬프트 — 플레이스홀더 예시를 실제 값으로.

- 최종 답변은 반드시 `FINAL ANSWER: <answer>` 한 줄이어야 한다.
+   숫자:    FINAL ANSWER: 42
+   문자열:  FINAL ANSWER: Paris
+   리스트:  FINAL ANSWER: cat, dog, fish
+ - <answer> 같은 플레이스홀더는 절대 출력하지 않는다. 반드시 실제 값을 적는다.
# CRITIC: `FINAL ANSWER: <answer>` 꺾쇠 플레이스홀더가 남으면 반드시 반려한다.

B3. v4-prompt-fix EXP-46 promoted

특이점: v3에서 format_invalid 0.21로 악화(한 단어 답일 때 형식 생략). 변경: PLANNER/CRITIC에 규칙 한 줄씩 추가.

# PLANNER:
+ - 답이 한 단어 또는 한 글자여도 반드시 `FINAL ANSWER:` 형식을 사용한다. 절대 생략하지 않는다.
# CRITIC:
+ - `FINAL ANSWER:` 행이 없으면 답변 길이에 관계없이 반드시 반려한다.

B4. v5-multitool EXP-42 promoted

특이점: read 100% 에러 — read_text(encoding="utf-8")가 바이너리(xlsx/pdf/mp3)에서 UnicodeDecodeError. python_calc는 AST 산술 전용이라 파일 파싱 코드 거부. 변경: 도구 2개 교체 — 확장자별 디스패치 + 격리 실행.

def read_attached_file(path):
    p = Path(path).resolve()
    if not p.is_file(): raise FileNotFoundError(str(p))  # 경로 교정 유도
    ext = p.suffix.lower()
    if   ext in (".xlsx",".xls"): content = _read_xlsx(p)   # openpyxl
    elif ext == ".docx":        content = _read_docx(p)   # python-docx
    elif ext == ".pdf":         content = _read_pdf(p)    # pdfplumber
    elif ext in _IMAGE_EXTS:     return "[이미지: 텍스트 추출 불가 안내]"  # 크래시 대신
    elif ext in _AUDIO_EXTS:     return "[오디오: 전사 불가 안내]"
    else:                      content = _read_text(p)   # errors="replace"
    return _truncate(content, 20_000)

def python_exec(code):  # python_calc(AST) 대체 — 격리 서브프로세스
    proc = subprocess.run([sys.executable, "-c", code], timeout=20, cwd=workdir, ...)

개입 계보

v1-baseline ─B1(예시)─▶ v3 ─B3(형식강제)─▶ v4 ─B4(파서교체)─▶ v5-multitool
   └─B2(경로주입)─▶ v2-attach-path
# 각 화살표 = experiments 원장 1행 (가설 → 1변수 개입 → batch 비교 → verdict)