[AIOps Agent 7] 하네스·도구·거버넌스
이 글에서 다룰 주제
- 하네스와 도구: 모델이 요청한 함수를 누가 검증하고 실행할까?
- 권한과 승인: 승인 이후 계획이나 실제 서비스 상태가 바뀌면 어떻게 막을까?
- 거버넌스: 조직의 운영 원칙을 코드·배포·감사 기록에 어떻게 연결할까?
주요 단어 · Harness · Tool Registry · Trusted Principal · Policy · Approval · Idempotency · Governance
읽기 안내 · Python 함수와 JSON을 아는 독자를 위한 중급 구현 편이다. 읽고 나면 도구를 모델에 보여 주는 단계와 실제 실행 권한을 확인하는 단계를 구별하고, 승인 재사용을 막는 작은 하네스를 실행할 수 있다. 메모리와 지식 저장소는 5편, 에이전트 종류는 6편에서 연결한다.
가상의 tenant-a에 속한 운영팀 team-a가 checkout 지연 장애 inc-42를 실행 run-101에서 조사한다고 하자. tenant는 조직 간 격리 경계이고 team은 그 안의 운영팀이다. 에이전트가 “replica를 2개에서 3개로 늘리겠습니다”라고 썼다. 운영자가 승인했는데, 실행 직전에 에이전트가 인자를 5개로 바꾸거나 다른 자동화가 이미 배포를 바꿨다면 어떻게 될까?
프롬프트에 “승인된 작업만 실행하라”를 적는 것으로는 이 질문에 답할 수 없다. 어떤 요청을 어떤 신원으로 실행할지, 승인한 내용과 아직 일치하는지를 프로그램이 검사해야 한다. 이 실행 구조가 이번 글의 주제다.
1. 하네스의 역할
하네스(Harness) — 이 글에서는 모델의 입력·도구 호출·상태·실행 한도·권한·기록을 관리하는 실행 환경이라는 뜻으로 사용한다. 모든 프레임워크가 같은 범위로 쓰는 표준 제품명은 아니다.
하네스는 별도의 LLM을 하나 더 붙이는 개념이 아니다. 처음에는 Python 애플리케이션 안의 몇 개 모듈일 수 있다. 규모가 커지면 승인 서비스, 실행 워커, 정책 서비스로 나누더라도 책임은 이어진다.
그림 1. 모델은 다음 도구 호출을 제안한다. 하네스는 Registry에서 계약을 찾고 입력·정책·승인·예산을 검사한 뒤 실행기로 보낸다. 상태·근거·승인 원장과 Knowledge Base는 아래쪽에 배치했다. 운영 API의 읽기 계정과 변경 계정은 구분하며 우회 호출 경로를 두지 않는다.
| 역할 | 하는 일 | inc-42 예시 |
|---|---|---|
| 모델·에이전트 루프 | 근거를 읽고 다음 조사나 계획을 제안 | 현재 replica 확인 요청 |
| 하네스 | 상태·예산·도구 호출·권한·승인 검사 | checkout 변경 권한과 승인 확인 |
| 도구 실행기 | 허용된 실제 API 작업 수행 | 조건부 replica 변경 |
| 감사·근거 저장 | 요청·판정·실행 결과를 연결 | 누구의 어떤 승인이 사용됐는지 기록 |
여기에 메모리와 Knowledge Base가 들어오는 위치도 중요하다. 메모리는 현재 조사 상태를 복구하고, Knowledge Base는 운영 절차의 근거를 제공한다. 둘 모두 인가 결과의 원본으로 사용하지 않는다. 어제 대화에 “승인됨”이 있다고 오늘 변경 권한이 생기지 않는다.
1.1 Markdown과 실행 통제
제공된 학습 슬라이드는 서브에이전트 역할과 Skill 절차를 Markdown으로 관리하는 방식을 보여 준다. 역할·절차를 버전 관리하고 팀이 공유한다는 점에서 유용하다. 다만 실제 실행을 제한하는 코드와 인프라 권한까지 Markdown 본문으로 대체된다는 뜻으로 읽으면 안 된다.
| 관리 대상 | 저장 예시 | 효력 |
|---|---|---|
| 역할·작업 순서·확인 질문 | agent 설정, SKILL.md, 프롬프트 | 모델의 판단을 안내 |
| 입력 형식·허용 범위 | Tool schema, 검증 코드 | 잘못된 호출 거부 |
| 사용자·tenant 권한 | 인증 문맥, 정책·ACL | 접근 가능한 대상 결정 |
| 실제 변경 권한 | 실행기 계정, DB 권한, Kubernetes RBAC | 대상 시스템이 작업 제한 |
| 변경 승인·실행 이력 | 승인 DB, 감사 저장소 | 특정 계획의 실행과 책임 추적 |
예를 들어 Skill에는 “변경 전에 영향과 복구 방법을 제시하라”를 적는다. 서버는 승인 ID가 없으면 scale_replicas를 거부한다. Kubernetes 쪽은 변경 실행기 계정이 허용된 namespace와 리소스 밖으로 나가지 못하게 한다. 각각 다른 실패를 막는 장치다.
2. 도구 구현
도구를 구현할 때는 실제 함수, 입력·출력 계약, 서버 등록, 모델 노출을 구분한다.
Tool Registry — 도구 이름을 실제 구현과 입력 계약, 권한, 실행 한도에 연결하는 서버 쪽 목록이다. 모델이 만든 문자열을 임의 Python 함수 이름이나 shell 명령으로 실행하지 않게 하는 경계이기도 하다.
get_service_snapshot 도구 하나를 만든다면 다음 네 가지를 준비한다.
- 구현: 현재 서비스 상태를 읽는 API adapter.
- 계약: 서비스 ID 등 허용 입력과 반환 형식.
- 등록: 이름 → 구현·권한·실행 계정의 연결.
- 노출: 해당 사용자와 작업에서 사용할 도구 정의를 모델 호출에 전달.
다음은 이 글의 실습에서 사용하는 입력 계약이다. 모델은 service만 제안할 수 있다. tenant, 사용자 ID, 자격증명은 입력 항목에 없다.
{
"type": "object",
"properties": {"service": {"type": "string"}},
"required": ["service"],
"additionalProperties": false
}additionalProperties: false를 선언했더라도 서버가 실제 검증해야 효과가 생긴다. 모델 API의 구조화 출력 기능도 애플리케이션의 권한 검증을 대신하지 않는다. 형식이 올바른 {"service":"payments"}라도 사용자가 checkout만 볼 수 있다면 거부해야 한다.
도구 설명은 선택 품질에 영향을 준다. “정보를 가져온다”보다 “서비스의 현재 replica와 리소스 버전을 읽는다. 과거 장애 원인은 이 결과만으로 확정할 수 없다”가 적합하다. 검색 도구와 현재 상태 조회 도구가 같은 이름·설명으로 겹치면 모델이 오래된 문서를 현재 상태로 오해하기 쉽다.
2.1 모델에 도구 제공
실습의 model_tools(ctx)는 사용자 scope에 맞는 도구 정의만 반환한다. 이 목록을 사용하는 모델 SDK의 tool 필드에 맞춰 변환해 제공하면 된다. 실습의 input_schema는 내부 표현이며, 모든 공급자의 요청 필드가 같다는 뜻은 아니다.
def model_tools(self, ctx):
return [
{"name": t.name, "description": t.description,
"input_schema": deepcopy(t.schema)}
for t in REGISTRY.values()
if t.scope in ctx.principal.scopes
]모델은 자격증명이나 adapter 객체를 받지 않는다. 도구 이름·설명·스키마를 보고 호출을 제안한다. 애플리케이션은 제안을 받아 invoke()를 호출하고, 결과를 해당 모델 응답의 tool-call ID와 연결해 돌려준다. 위 코드는 정의를 만드는 부분만 실행하며 실제 모델 API는 호출하지 않는다.
노출 필터는 오선택을 줄이지만 최종 권한 경계는 아니다. 오래된 대화, 잘못된 캐시, 변조된 클라이언트에서 숨겨진 도구 이름이 들어올 수 있으므로 invoke()에서도 매번 검사한다.
3. 실행 권한 경계
모델이 제안한 도구 인자와 서버가 인증한 사용자 문맥을 분리해 전달한다.
Principal — 인증을 거쳐 확인된 요청 주체다. 사용자 ID만이 아니라 tenant, 허용 범위, 위임 관계 등 접근 판단에 필요한 정보를 포함한다.
다음 두 데이터는 출처가 다르다.
모델 제안: {tool: get_service_snapshot, arguments: {service: checkout}}
서버 문맥: {subject: requester-1, tenant: tenant-a, scope: service:read}서버 문맥은 검증된 토큰·세션과 권한 저장소에서 만든다. 요청 JSON에 사용자가 적어 보낸 tenant를 그대로 복사해서 만드는 문맥은 신뢰할 수 없다. 실습에서는 인증 서버를 구현하지 않으므로 Principal을 테스트 코드에서 고정 생성한다. 운영에 가져갈 때 가장 먼저 바꿀 지점이다.
그림 2. 그림 1과 같은 배치에서 하네스의 책임 경계를 빨간 테두리로 강조했다. Registry에서 계약 찾기 → schema 검증 → 정책·승인·예산 검사 → 실행기 순서다. 모델이 제안한 값과 인증 문맥에서 확정한 값을 분리하고 결과는 Runtime으로 반환한다.
실행 전에는 이 질문들을 순서대로 확인한다.
- 실행해도 되는 상태인가? 취소됐거나 실행 시간이 끝났는가, 예산이 남았는가?
- 등록된 계약인가? 알 수 없는 도구나 추가 인자가 들어왔는가?
- 누구의 요청인가? tenant와 service가 인증 주체의 허용 범위인가?
- 어떤 작업인가? 조회인가, 외부 상태를 바꾸는 변경인가?
- 변경이라면 무엇을 승인했는가? 지금 계획·대상 상태와 여전히 일치하는가?
정책 거부는 모델이 극복할 장애가 아니다. policy_denied를 받으면 다른 도구로 같은 정보를 우회 조회하도록 반복하지 않고, 허용된 범위로 조사하거나 사람에게 권한 경계를 설명해야 한다.
3.1 조회·변경 계정 분리
하나의 관리자 토큰을 모든 adapter에서 공유하면 get_service_snapshot 코드의 실수 하나가 변경 권한으로 이어질 수 있다. 이 예제는 reader-identity와 changer-identity를 나눠 표시한다. 문자열 자체가 실제 보안 경계를 제공하는 것은 아니다. 운영에서는 별도 workload identity·서비스 계정과 대상 시스템의 권한 설정으로 구현한다.
에이전트 프로세스가 변경 계정의 비밀값을 자유롭게 읽을 수 있다면 코드상 함수 분리만으로 부족하다. 모델을 호출하는 런타임은 변경 실행기에 제한된 작업 요청을 보내고, 실행기는 자체 신원으로 대상 API를 호출하도록 분리할 수 있다. 이때 실행기에서도 정책·승인 검증을 건너뛰면 안 된다.
4. 변경 승인
승인 결합(approval binding) — 승인이 허용하는 계획·대상·사용자·유효기간을 고정하고 실제 실행 때 그 일치 여부를 검사하는 설계다.
우리의 변경 계획은 “checkout 확장”이라는 문장이 아니라 아래와 같은 구조다.
{
"tool": "scale_replicas",
"arguments": {"service": "checkout", "replicas": 3, "resource_version": 7},
"tenant": "tenant-a",
"subject": "requester-1",
"incident": "inc-42",
"run_id": "run-101"
}실습에서는 이 객체를 정렬된 JSON으로 직렬화한 뒤 SHA-256 해시를 계산한다. 승인 저장소에는 이 해시, 승인자, 만료 시각을 저장한다. 실행 때 다시 계산해 비교하므로 replicas: 3을 4로 바꾸면 plan_changed로 거부한다. 해시는 진위 확인용 서명이 아니다. 누구나 해시를 계산할 수 있으므로 승인 레코드를 만드는 API 자체가 인증·인가로 보호돼야 한다.
그림 3. 리소스·인자·대상 버전을 계획에 묶고 승인 원장에 해시·만료·승인자를 기록한다. 호출 직전에는 계획 일치와 현재 권한·예산을 재검사한다. 실행과 재사용 차단에는 멱등 키와 감사 결과를 연결하며, 모델의 approved=true는 승인 증거로 사용하지 않는다.
4.1 대상 버전 변경
운영자가 승인하는 동안 배포가 바뀌면 계획 해시는 그대로여도 전제가 틀릴 수 있다. 실습의 resource_version=7은 계획이 어떤 상태를 보고 만들어졌는지 나타낸다. 가짜 adapter는 버전 비교와 변경을 같은 lock 안에서 실행한다. 현재 버전이 8이면 변경을 거부한다.
Kubernetes에 적용할 때는 실제 리소스의 metadata.resourceVersion과 사용하는 update/patch 방식의 충돌·전제 조건을 확인해야 한다. 이 글의 정수 증가 규칙은 가짜 adapter의 규칙이다. Kubernetes의 resourceVersion을 직접 증가시키거나 숫자 크기로 상태 순서를 판정하는 예제가 아니다. Kubernetes API의 조건부 변경
4.2 재사용과 멱등성
이 실습의 승인 ID는 한 번의 변경 시도에만 쓴다. 실행 전에 사용 처리하므로 같은 ID를 다시 보내면 approval_replayed가 된다. 실행기에서 버전 충돌이 나더라도 소비된 승인을 재사용하지 않는다.
운영 시스템의 멱등성(idempotency)은 여기에 더해 “동일한 작업 요청이 다시 도착해도 부수효과를 반복하지 않음”을 다룬다. 보통 작업 키와 정규화된 요청 해시, 상태, 결과 참조를 durable 저장소에 함께 둔다. 같은 키·같은 요청이면 기존 결과를 돌려주고, 같은 키·다른 요청이면 거부한다. 승인 ID와 작업 키를 하나의 의미로 섞지 않는 편이 설명과 운영에 유리하다.
요청을 보낸 뒤 응답만 유실됐다면 실행 여부는 불확실하다. 이때 새 승인과 새 작업 키로 다시 변경하기보다 기존 job 상태를 먼저 조회한다. 분산 환경에서 승인 소비와 외부 API 변경을 하나의 DB 트랜잭션으로 묶을 수 없는 문제는 뒤의 운영 경계에서 다시 다룬다.
5. 하네스 실습
다음 실습은 Python 3.9.6에서 실행 확인했다. 표준 라이브러리만 사용하며 모델 API·네트워크·Kubernetes 접속은 없다. tenant-a / checkout의 메모리 안 딕셔너리만 바꾼다. LLM의 답변 품질을 측정하는 실습이 아니라 서버의 실행 판정을 검증하는 실습이다.
10절의 전체 코드를 harness_demo.py로 저장하고 실행한다. 프로젝트 자료에서는 revision-v2/lab/harness_demo.py에 같은 코드가 있다. now와 deadline은 재현을 위한 논리 시각이며 실제 벽시계의 초 단위와 혼동하지 않는다.
python3 harness_demo.py아래에서 먼저 실행 결과와 차단 이유를 읽어도 된다. 전체 구현을 확인할 때는 REGISTRY → model_tools → invoke → FakeAdapter 순서로 찾아보면 된다. approve()는 모델에 노출하지 않는다. 테스트의 별도 승인자가 승인 레코드를 만드는 동작을 모의한다.
실제 실행 출력은 다음과 같다.
HARNESS DEMO | standard-library | fake adapter only
tenant=tenant-a incident=inc-42 initial replicas=2 resource_version=7
read: ok replicas=2 resource_version=7
denied: rejected reason=policy_denied
tenant_injection: rejected reason=schema_rejected
approval_missing: rejected reason=approval_required
approval_expired: rejected reason=approval_expired
plan_changed: rejected reason=plan_changed
stale_resource: rejected reason=resource_version_changed
valid_approval: ok replicas=3 resource_version=8
replay: rejected reason=approval_replayed
cancelled: rejected reason=cancelled
deadline: rejected reason=deadline_exceeded
budget: rejected reason=budget_exhausted
final replicas=3 resource_version=8
audit_events=12 executed=2 rejected=10
assertions=passed읽기 성공의 replica는 2, 유효한 승인 뒤에는 3이다. 최종 버전이 8인 것은 실제로 변경된 작업이 한 번이라는 뜻이다. 처음 조회와 승인된 변경 두 번만 실행됐고, 나머지 열 번은 실행 경계에서 거부됐다.
tenant_injection에서는 모델 인자에 tenant-b를 추가해 보았다. 추가 필드를 허용하지 않아 schema 단계에서 거부된다. denied는 입력 형식은 올바르지만 허용되지 않은 서비스라 정책 단계에서 거부된다. 두 오류를 나눠야 구현 결함과 권한 문제를 구분할 수 있다.
approval_expired는 시각, plan_changed는 승인 내용, stale_resource는 대상 상태, replay는 재사용을 각각 검사한다. “승인 확인” 한 항목으로 뭉치지 않고 서로 다른 실패를 확인한 이유다. 출력과 코드의 assert가 함께 남아 있어 수정 뒤 회귀를 확인하기 쉽다.
6. 운영 실행 제어
위 실습이 구현한 것은 단일 프로세스의 판단 순서다. 운영 버전에서 아래 항목을 어디에 붙일지 정해야 한다.
| 항목 | 실습에서 확인한 범위 | 운영 구현 지점 |
|---|---|---|
| 인증 | 테스트가 만든 Principal | API 진입점의 토큰·세션 검증 |
| Schema | 두 도구에 필요한 작은 타입 검사 | 검증된 JSON Schema 라이브러리와 API 계약 |
| 예산 | 호출 횟수 감소와 거부 | run별 모델 토큰·도구 비용·동시 실행 한도 |
| 시간 제한 | 호출 전 deadline 검사 | HTTP connect/read timeout, 워커·job deadline |
| 취소 | 실행 전 취소 상태 확인 | 장기 job 취소와 실행 후 상태 재확인 |
| 승인·중복 | 메모리 저장소와 lock | durable DB, 원자적 상태 전이, 고유 키 |
| 대상 상태 | 가짜 버전 비교·변경 lock | 실제 API의 조건부 변경과 충돌 처리 |
| 감사 | 메모리 목록에 판정 기록 | 내구성·접근 제한·보존 정책을 갖춘 저장소 |
호출 전 deadline을 확인해도 API가 영원히 응답하지 않는 문제는 해결되지 않는다. adapter가 timeout을 실제 네트워크 클라이언트에 전달해야 한다. Python에서 기다리기를 중단했다고 원격 작업이 취소된 것도 아니다. “취소 요청됨”과 “실행 중단 확인됨”을 상태로 나눈다.
재시도 역시 오류별로 다르다. 일시적인 조회 실패는 남은 예산 안에서 지수 backoff와 jitter를 적용할 수 있다. 권한 거부·schema 오류는 같은 요청으로 재시도하지 않는다. 변경은 응답을 못 받았다고 무조건 재시도하지 말고, 멱등 키 또는 job 조회로 실행 여부를 확인한다.
프로세스가 “승인 사용 처리” 직후 죽으면 실습은 상태를 잃는다. 운영에서는 pending → reserved → dispatched → succeeded/failed/unknown 같은 durable 상태를 두고, outbox나 작업 큐와 대상 API의 중복 방지 기능을 연결한다. 모든 API에서 정확히 한 번 실행을 보장할 수 있다고 가정하지 않는다. unknown 상태의 재조정 절차가 필요하다.
7. MCP와 업무 권한
MCP(Model Context Protocol) — AI 애플리케이션이 외부 도구·데이터를 발견하고 사용하는 메시지 규약이다. 업무별 승인 정책이나 대상 시스템의 권한 설정을 자동으로 만들어 주지는 않는다.
MCP의 tools/list로 이름·설명·입력 schema를 발견하고, tools/call로 도구를 호출할 수 있다. 그래서 여러 클라이언트가 같은 도구 서버를 재사용하기 좋다. 하지만 readOnlyHint 같은 annotation만 믿고 권한 검사를 생략하면 안 된다. 이 글에서 확인한 2025-11-25 Tools 문서도 신뢰하지 않는 서버의 annotation을 신뢰하지 말라고 명시한다. MCP Tools 명세
HTTP 기반 연결의 인증·인가 흐름과, checkout replica 변경을 허용하는 업무 정책은 별개 층이다. MCP 서버는 자신을 대상으로 발급된 토큰인지 검증해야 한다. 클라이언트 토큰을 검증 없이 downstream API로 그대로 넘기는 token passthrough는 명세에서 금지한다. MCP Authorization, MCP Security Best Practices
따라서 “MCP 연결 완료” 뒤에도 호출별 tenant·service·operation 정책과 변경 승인을 적용한다. MCP 세션 ID도 사용자 신원을 대신하지 않는다. stdio로 띄우는 로컬 서버와 HTTP 원격 서버는 배포·인증 경계가 다르므로 HTTP OAuth 설명을 모든 transport에 그대로 적용하지 않는다.
우리 Registry를 MCP 서버의 핸들러에 연결할 수도 있고, 처음에는 같은 애플리케이션의 함수 호출로 시작할 수도 있다. 어느 쪽이든 검증된 도구 구현을 한 번 감싸는 경계를 유지하면 모델 공급자나 통신 방식이 바뀌어도 권한 코드가 흩어지지 않는다.
8. 에이전트 거버넌스
에이전트 거버넌스 — 누가 어떤 목적·위험·권한으로 에이전트를 운영하고, 변경을 승인하며, 실패를 검토하는지 정하는 책임 체계다. 정책 문서와 실행 가능한 통제, 기록을 함께 포함한다.
아래는 이 시리즈의 AIOps 설계 제안이다. 특정 규정의 인증 체크리스트는 아니다. OWASP는 불필요한 기능·권한·자율성이 과도하게 부여되는 문제를 Excessive Agency로 설명하며, 모델의 허용 판단에 기대지 않는 downstream 권한 검증을 권고한다. 우리는 이 원칙을 도구·승인·배포 책임으로 연결한다. OWASP LLM06:2025
| 결정할 정책 | 구체적인 실행 지점 | 남길 증거 |
|---|---|---|
| 에이전트의 목적·책임자 | Agent Registry와 배포 메타데이터 | owner, 용도, 모델·프롬프트 버전 |
| 허용 작업·위험 등급 | Tool Registry, 정책 평가기 | scope, 대상 범위, 정책 버전 |
| 사람의 변경 승인 | 별도 승인 API와 UI | 계획 해시, 승인자, 시각, 만료 |
| 배포 가능한 품질 | CI 평가와 릴리스 gate | 평가셋 버전, 실패 사례, 승인 기록 |
| 실행 중단 조건 | budget, 취소, kill switch | 중단 이유와 미완료 작업 목록 |
| 개인정보·보존 | 수집·마스킹·저장·삭제 경로 | 삭제 범위, 처리 시각, 접근 기록 |
8.1 위험 등급
위험 등급은 도구 이름뿐 아니라 실제 영향 범위, 가역성, 데이터 민감도와 비용을 기준으로 정한다.
| 제안 등급 | 예시 | 기본 실행 방식 |
|---|---|---|
| 조회 | 허용된 지표·배포 상태 확인 | 범위·시간·결과 크기 제한 후 실행 |
| 초안 | 장애 보고서, 변경 PR 초안 | 외부 효과가 생기는 지점에서 별도 판단 |
| 제한 변경 | 단일 서비스 replica 변경 | 계획·영향·복구 조건 확인 후 승인 |
| 광범위·파괴적 변경 | 여러 tenant 변경, 데이터 삭제 | 별도 업무 흐름·복수 검토·엄격한 계정 |
조회도 비용과 정보 노출 위험이 있다. 전 tenant 로그를 읽는 도구는 “read”라는 이유만으로 낮은 위험이라고 볼 수 없다. 변경의 가역성, 영향 범위, 민감 데이터, 비용을 함께 평가한다. replica 변경도 다른 controller와 충돌하거나 부하 전제를 잘못 판단할 수 있으므로 성공 기준은 API 응답 외에 후속 관측을 포함한다.
8.2 책임 분담
RACI는 실행 담당(R), 최종 책임(A), 협의(C), 통보(I)를 구분하는 방법이다. 작은 팀에서도 한 작업의 최종 책임자를 명확히 하는 데 쓸 수 있다.
| 작업 | R: 실행 담당 | A: 최종 책임 | C/I: 협의·통보 |
|---|---|---|---|
| 도구 구현·권한 테스트 | 플랫폼 개발자 | 플랫폼 책임자 | 보안·서비스 담당 |
| 운영 변경 승인 | 지정된 운영자 | 해당 서비스 책임자 | 장애 지휘·관련 팀 |
| 모델·프롬프트 배포 | 에이전트 개발자 | 업무 owner | 평가·운영 담당 |
| KB 최신성·삭제 | 지식 소유자 | 데이터 owner | 보안·에이전트 담당 |
| 사고 분석·중단 | 당직 운영자 | 서비스 책임자 | 플랫폼·보안 담당 |
모델의 자체 평가는 최종 승인자를 대체하지 않는다. 위험한 변경의 제안자와 승인자를 분리할지는 위험 등급별로 정한다. 실습은 다른 subject의 승인자만 허용하지만, 이것이 모든 조직에서 항상 필요한 유일한 정책은 아니다.
8.3 평가·배포·보존
도구 설명을 바꾸면 선택이 달라지고, KB 문서가 바뀌면 근거가 달라진다. 모델만 버전 관리하지 말고 tool schema·정책·프롬프트·retrieval 설정을 함께 기록한다. 새 버전은 고정된 정상·실패 사례로 평가하고 조회 전용 shadow 실행, 제한된 서비스 적용 등 영향 범위를 단계적으로 넓힌다.
품질 지표는 정답률 하나로 끝내지 않는다. 잘못된 변경 제안, 승인 누락, tenant 간 접근, 근거 없는 RCA 단정, 비용 한도 초과를 따로 센다. 정책 테스트를 통과했다는 사실과 실제 장애 진단이 유용하다는 사실은 서로 다른 검증이다. Langfuse의 모델·도구 관측과 평가 데이터는 4편에서 설명한 실행 ID로 연결할 수 있지만, 이 평가 점수가 권한 승인을 대신하지는 않는다.
감사 기록에는 run_id, incident_id, 사용자·위임 주체, tool·schema 버전, 정책 판정, 승인 참조, 대상 버전, 결과 참조를 남긴다. 프롬프트 전체·비밀 토큰·모든 원문 로그를 무조건 보존하지 않는다. 최소화한 감사 메타데이터와 민감할 수 있는 근거 본문을 다른 접근·보존 정책으로 관리한다.
예를 들어 원문 로그의 보존 기간이 끝나도 “어떤 범위의 데이터를 언제 조회했는가”라는 메타데이터는 별도 정책에 따라 남길 수 있다. 반대로 삭제 요청이 승인되면 원문만 지우고 메모리 요약·검색 chunk·캐시·평가 데이터에 내용이 남아 있지 않은지 추적해야 한다. 보존 기간은 이 글에서 임의의 일수로 고정하지 않고 조직의 데이터 분류와 업무 요건으로 결정한다.
9. Repository 구조
아래는 현재 실습을 서비스로 키울 때의 제안 구조다. 현재 폴더에 모든 모듈이 구현돼 있다는 뜻은 아니다.
aiops-agent/
├── src/aiops_agent/
│ ├── api/ # 인증, 사용자 요청, 승인 endpoint
│ ├── runtime/
│ │ ├── runner.py # 모델 호출, tool 결과 연결, 종료
│ │ ├── dispatcher.py # 검증·정책·승인·실행 조정
│ │ ├── context.py # 서버가 만든 Principal / run 문맥
│ │ └── jobs.py # durable 실행·중복 방지·재조정
│ ├── policy/
│ │ ├── authorize.py # tenant·service·operation 규칙
│ │ └── approvals.py # 승인 저장·조회·원자적 소비
│ ├── tools/
│ │ ├── registry.py # 이름·계약·권한·구현 등록
│ │ └── schemas/ # 입력·출력 계약
│ ├── adapters/ # 관측 조회 / 변경 API 구현
│ ├── memory/ # 조사 상태와 메모리 수명 관리
│ ├── knowledge/ # KB 수집·ACL·검색·원문 조회
│ ├── skills/ # 모델이 읽는 업무 절차
│ └── audit/ # 판정·근거 참조·보존
└── evals/ # 권한 실패 사례와 업무 품질 평가첫 구현에서 위 디렉터리를 모두 별도 마이크로서비스로 만들 필요는 없다. 코드 모듈로 책임을 나눈 뒤, 변경 계정 격리·부하·배포 주기가 실제 분리 이유가 될 때 프로세스를 나눈다. 테스트는 모듈 이름이 아니라 경계의 동작을 검증한다. “승인 후 인자가 바뀌면 거부하는가?”가 “approve 함수가 호출됐는가?”보다 유용한 기준이다.
직접 확장해 볼 과제는 세 가지다. 첫째, 조회만 가능한 Principal로 변경 요청을 보내 거부되는지 확인한다. 둘째, 승인 뒤 run ID나 tenant를 바꾸고 기존 승인이 통과하지 못하게 한다. 셋째, 가짜 adapter의 실행 후 응답 유실을 재현하고 unknown 상태에서 재조회하는 절차를 설계한다. 이 세 동작을 설명할 수 있으면 모델 호출과 운영 가능한 에이전트 사이의 차이가 훨씬 분명해진다.
10. 전체 실행 코드
설명과 결과를 읽은 뒤 직접 실행할 수 있도록 전체 파일을 모았다. 아래 블록 하나를 복사한다.
#!/usr/bin/env python3
"""Executable teaching harness. Standard library only; no network or real changes.
Python >= 3.9. This is deliberately NOT an OAuth server, Kubernetes client,
durable approval service, general JSON Schema validator, or production harness.
"""
from dataclasses import dataclass, field
from copy import deepcopy
from hashlib import sha256
import json
from threading import Lock
from typing import Dict, FrozenSet, Optional
class Rejected(Exception):
pass
@dataclass(frozen=True)
class Principal:
subject: str
tenant: str
services: FrozenSet[str]
scopes: FrozenSet[str]
@dataclass
class Context:
principal: Principal
incident: str = "inc-42"
run_id: str = "run-101"
remaining_calls: int = 20
deadline: int = 120
cancelled: bool = False
@dataclass(frozen=True)
class ToolSpec:
name: str
description: str
schema: dict
scope: str
mode: str
executor_identity: str
READ_SCHEMA = {
"type": "object",
"properties": {"service": {"type": "string"}},
"required": ["service"], "additionalProperties": False,
}
CHANGE_SCHEMA = {
"type": "object",
"properties": {
"service": {"type": "string"},
"replicas": {"type": "integer", "minimum": 1, "maximum": 5},
"resource_version": {"type": "integer", "minimum": 1},
},
"required": ["service", "replicas", "resource_version"],
"additionalProperties": False,
}
REGISTRY = {
"get_service_snapshot": ToolSpec(
"get_service_snapshot", "Read the current service snapshot.",
READ_SCHEMA, "service:read", "read", "reader-identity"),
"scale_replicas": ToolSpec(
"scale_replicas", "Apply an approved replica target, between 1 and 5.",
CHANGE_SCHEMA, "service:scale", "change", "changer-identity"),
}
def validate(schema: dict, args: dict) -> None:
"""Only the tiny schema subset used above; reject unknown fields."""
if type(args) is not dict or set(args) != set(schema["required"]):
raise Rejected("schema_rejected")
for key, rule in schema["properties"].items():
value = args[key]
expected = str if rule["type"] == "string" else int
# bool is a subclass of int: use exact types here.
if type(value) is not expected:
raise Rejected("schema_rejected")
if expected is int and not rule.get("minimum", value) <= value <= rule.get("maximum", value):
raise Rejected("schema_rejected")
def plan_hash(tool: str, args: dict, ctx: Context) -> str:
payload = {"tool": tool, "arguments": args,
"tenant": ctx.principal.tenant, "subject": ctx.principal.subject,
"incident": ctx.incident, "run_id": ctx.run_id}
raw = json.dumps(payload, sort_keys=True, separators=(",", ":"))
return sha256(raw.encode()).hexdigest()
@dataclass(frozen=True)
class Approval:
plan_hash: str
expires_at: int
approver: str
class FakeAdapter:
"""This object is in-memory only. Lock also protects version comparison."""
def __init__(self):
self.lock = Lock()
self.state = {("tenant-a", "checkout"): {"replicas": 2, "resource_version": 7}}
def read(self, identity: str, tenant: str, service: str) -> dict:
if identity != "reader-identity":
raise Rejected("adapter_identity_rejected")
with self.lock:
return deepcopy(self.state[(tenant, service)])
def change(self, identity: str, tenant: str, args: dict) -> dict:
if identity != "changer-identity":
raise Rejected("adapter_identity_rejected")
with self.lock:
current = self.state[(tenant, args["service"])]
if current["resource_version"] != args["resource_version"]:
raise Rejected("resource_version_changed")
current["replicas"] = args["replicas"]
current["resource_version"] += 1
return deepcopy(current)
class Harness:
def __init__(self):
self.adapter = FakeAdapter()
self.approvals: Dict[str, Approval] = {}
self.used = set()
self.approval_lock = Lock()
self.audit = []
self.policy_version = "policy-v1"
def model_tools(self, ctx: Context) -> list:
# Useful exposure filtering; invoke() still rechecks every call.
return [{"name": t.name, "description": t.description,
"input_schema": deepcopy(t.schema)}
for t in REGISTRY.values() if t.scope in ctx.principal.scopes]
def authorize(self, spec: ToolSpec, args: dict, ctx: Context) -> None:
if spec.scope not in ctx.principal.scopes or args["service"] not in ctx.principal.services:
raise Rejected("policy_denied")
if ctx.principal.tenant != "tenant-a" or ctx.incident != "inc-42":
raise Rejected("policy_denied")
def approve(self, approval_id: str, tool: str, args: dict, ctx: Context,
approver: Principal, expires_at: int) -> None:
# Production: only a separately authenticated approval endpoint calls this.
# Never expose this method to the model as an executable tool.
if ("approval:grant" not in approver.scopes or approver.tenant != ctx.principal.tenant
or args["service"] not in approver.services or approver.subject == ctx.principal.subject):
raise Rejected("approver_denied")
spec = REGISTRY[tool]
validate(spec.schema, args)
self.authorize(spec, args, ctx)
if spec.mode != "change":
raise Rejected("approval_not_required")
with self.approval_lock:
if approval_id in self.approvals:
raise Rejected("duplicate_approval_id")
self.approvals[approval_id] = Approval(plan_hash(tool, args, ctx), expires_at, approver.subject)
def invoke(self, tool: str, args: dict, ctx: Context, now: int,
approval_id: Optional[str] = None) -> dict:
args = deepcopy(args)
event = {"run_id": ctx.run_id, "incident": ctx.incident,
"tenant": ctx.principal.tenant, "subject": ctx.principal.subject,
"tool": tool, "policy_version": self.policy_version}
try:
if ctx.cancelled:
raise Rejected("cancelled")
if now >= ctx.deadline:
raise Rejected("deadline_exceeded")
if ctx.remaining_calls <= 0:
raise Rejected("budget_exhausted")
ctx.remaining_calls -= 1
if tool not in REGISTRY:
raise Rejected("unknown_tool")
spec = REGISTRY[tool]
validate(spec.schema, args)
self.authorize(spec, args, ctx)
event["plan_hash"] = plan_hash(tool, args, ctx)
event["executor_identity"] = spec.executor_identity
if spec.mode == "read":
data = self.adapter.read(spec.executor_identity, ctx.principal.tenant, args["service"])
else:
with self.approval_lock:
approval = self.approvals.get(approval_id)
if approval is None:
raise Rejected("approval_required")
if approval_id in self.used:
raise Rejected("approval_replayed")
if now >= approval.expires_at:
raise Rejected("approval_expired")
if approval.plan_hash != event["plan_hash"]:
raise Rejected("plan_changed")
# Reserve before side effect. A failed change consumes approval too.
self.used.add(approval_id)
event["approval_id"] = approval_id
event["approver"] = approval.approver
data = self.adapter.change(spec.executor_identity, ctx.principal.tenant, args)
result = {"status": "ok", "data": data,
"evidence": {"source": "fake-adapter", "observed_at": now,
"tenant": ctx.principal.tenant, "service": args["service"]}}
event["decision"] = "executed"
except Rejected as error:
result = {"status": "rejected", "reason": str(error)}
event["decision"] = "rejected"
event["reason"] = str(error)
self.audit.append(event)
return result
def main() -> None:
principal = Principal("requester-1", "tenant-a", frozenset({"checkout"}),
frozenset({"service:read", "service:scale"}))
approver = Principal("operator-2", "tenant-a", frozenset({"checkout"}),
frozenset({"approval:grant"}))
ctx = Context(principal)
harness = Harness()
change = {"service": "checkout", "replicas": 3, "resource_version": 7}
cases = []
def run(label, tool, args, expected, approval_id=None, at=10):
result = harness.invoke(tool, args, ctx, at, approval_id)
actual = result.get("reason", result["status"])
assert actual == expected, (label, expected, result)
cases.append((label, result))
run("read", "get_service_snapshot", {"service": "checkout"}, "ok")
run("denied", "get_service_snapshot", {"service": "payments"}, "policy_denied")
run("tenant_injection", "get_service_snapshot",
{"service": "checkout", "tenant": "tenant-b"}, "schema_rejected")
run("approval_missing", "scale_replicas", change, "approval_required")
harness.approve("a-expired", "scale_replicas", change, ctx, approver, expires_at=9)
run("approval_expired", "scale_replicas", change, "approval_expired", "a-expired")
harness.approve("a-plan", "scale_replicas", change, ctx, approver, expires_at=60)
run("plan_changed", "scale_replicas", dict(change, replicas=4), "plan_changed", "a-plan")
stale = dict(change, resource_version=6)
harness.approve("a-stale", "scale_replicas", stale, ctx, approver, expires_at=60)
run("stale_resource", "scale_replicas", stale, "resource_version_changed", "a-stale")
harness.approve("a-valid", "scale_replicas", change, ctx, approver, expires_at=60)
run("valid_approval", "scale_replicas", change, "ok", "a-valid")
run("replay", "scale_replicas", change, "approval_replayed", "a-valid")
ctx.cancelled = True
run("cancelled", "get_service_snapshot", {"service": "checkout"}, "cancelled")
ctx.cancelled = False
run("deadline", "get_service_snapshot", {"service": "checkout"}, "deadline_exceeded", at=120)
ctx.remaining_calls = 0
run("budget", "get_service_snapshot", {"service": "checkout"}, "budget_exhausted")
assert len(harness.audit) == 12
assert harness.adapter.state[("tenant-a", "checkout")] == {"replicas": 3, "resource_version": 8}
print("HARNESS DEMO | standard-library | fake adapter only")
print("tenant=tenant-a incident=inc-42 initial replicas=2 resource_version=7")
for label, result in cases:
if result["status"] == "ok":
data = result["data"]
print("{}: ok replicas={} resource_version={}".format(label, data["replicas"], data["resource_version"]))
else:
print("{}: rejected reason={}".format(label, result["reason"]))
print("final replicas=3 resource_version=8")
print("audit_events=12 executed=2 rejected=10")
print("assertions=passed")
if __name__ == "__main__":
main()자료와 검증 범위 · 2026-10-04에 로컬 학습 자료의 Tool·Runtime 역할을 다시 검토하고, 위에 연결한 MCP 2025-11-25 문서와 OWASP LLM06:2025의 해당 주장 범위를 확인했다. 실제 실행은 Python 표준 라이브러리의 가짜 adapter 실습이며 모델 API·MCP 서버·Kubernetes·분산 승인 저장소 연동은 수행하지 않았다. 문서의 위험 등급·RACI·디렉터리 구조는 이 AIOps 예제를 위한 설계 제안이다.
추가 그림 자산 출처
PostgreSQL 로고는 공식 자산의 색과 비율을 유지해 사용했다. PostgreSQL과 Slonik은 PostgreSQL Community Association of Canada의 상표 또는 등록 상표이며, 상표 정책에 따른 제품 식별이다. 공식 후원·제휴를 뜻하지 않는다.
Kubernetes 리소스 아이콘은 공식 아이콘 모음을 사용했다. 리소스별 모양과 색을 유지했으며 연결선과 설명은 자체 설계다.
AIOps 에이전트 개발 시리즈
