인간-인-더-루프 (HITL)

가 결제 버튼을 누르기 직전, 외부 메일을 보내기 직전, 데이터를 영구 삭제하기 직전 — 사람의 마지막 한 번이 필요한 순간이 있다. 은 그 순간에 멈추는 기술이다. 와 두 축으로 본다.

왜 사람이 필요한가

자율성을 높일수록 책임도 같이 올라간다. 모델이 대부분 정확해도 가끔의 비싼 실수가 시스템을 죽인다. 의 끝은 결국 사람이다.

모델 정확도가 아니라, 한 번의 실수 비용이 HITL 도입 기준이다.

세 종류 케이스에 보통 들어간다.

고비용 도구 호출 — 결제·송금·자원 프로비저닝.
외부 부수효과 — 메일 발송·SNS 게시·DB 변경.
법적 책임 — 의료·법률 자문, 계약서 발행.

의 자리는 늘 이 셋 중 하나다. 도입 결정은 결국 비용 곱하기 확률.

HITL 게이트 한 장

다이어그램 로딩…

흐름은 단순하다 — 위험 분류, 대기열, 사람 결정, 분기. 결정은 셋 중 하나 — 승인·거부·수정. 수정이 가장 까다롭다. 사람이 인자를 고친 다음 흘려보내는 모드. 가 단순 yes/no 가 아니어야 한다는 뜻이다. 자체가 인간-기계 협업의 최소 단위다. 모드를 셋으로 두는 게 표준.

구현 1 — LangGraph interrupt (개념)

는 interrupt_before 와 interrupt_after 로 노드 실행 직전/직후에 멈출 수 있다. 가 상태를 저장한 채 사람을 기다린다.

# 개념 코드 — 자세한 건 17장
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import InMemorySaver

graph = (
  StateGraph(MyState)
  .add_node("plan", plan_node)
  .add_node("dangerous_tool", dangerous_tool_node)
  .add_edge("plan", "dangerous_tool")
  .compile(checkpointer=InMemorySaver(),
           interrupt_before=["dangerous_tool"])
)

# 1차 실행 — interrupt_before 에서 멈춤
graph.invoke({"input": "환불 처리"}, config={"configurable":
                                        {"thread_id": "t1"}})
# 사람 검토 후 재개
graph.invoke(None, config={"configurable": {"thread_id": "t1"}})

// 개념 — 17장에서 자세히
import { StateGraph, MemorySaver } from '@langchain/langgraph'

const graph = new StateGraph(MyState)
.addNode('plan', planNode)
.addNode('dangerousTool', dangerousToolNode)
.addEdge('plan', 'dangerousTool')
.compile({
  checkpointer: new MemorySaver(),
  interruptBefore: ['dangerousTool'],
})

await graph.invoke({ input: '환불 처리' },
                 { configurable: { thread_id: 't1' } })
// 사람 결정 후
await graph.invoke(null, { configurable: { thread_id: 't1' } })

장점은 그래프 상태가 통째로 저장되니 재개가 깔끔하다는 점. 가 자료구조 수준으로 1급이다. 메커니즘은 17장에서 깊게 다룬다.

구현 2 — 자체 승인 큐

프레임워크 없이도 같은 효과를 낼 수 있다. 큐와 이벤트 두 개면 된다.

class ApprovalQueue:
  def __init__(self):
      self._items: dict[str, ApprovalRequest] = {}
      self._events: dict[str, asyncio.Event] = {}

  async def submit(self, action, payload):
      rid = str(uuid.uuid4())
      req = ApprovalRequest(id=rid, action=action, payload=payload)
      self._items[rid] = req
      self._events[rid] = asyncio.Event()
      return req

  async def wait(self, rid, timeout=None):
      await asyncio.wait_for(self._events[rid].wait(), timeout=timeout)
      return self._items[rid]

  def decide(self, rid, status, by, modified=None):
      req = self._items[rid]
      req.status = status; req.decided_by = by
      if modified: req.modified_payload = modified
      self._events[rid].set()

export class ApprovalQueue {
private items = new Map<string, ApprovalRequest>()
private waiters = new Map<string, (r: ApprovalRequest) => void>()

submit(action: string, payload: Record<string, unknown>) {
  const id = randomUUID()
  const req = { id, action, payload, status: 'pending' as const }
  this.items.set(id, req)
  return req
}
wait(id: string): Promise<ApprovalRequest> {
  return new Promise((res) => this.waiters.set(id, res))
}
decide(id: string, status: ApprovalStatus, by: string) {
  const req = this.items.get(id)!; req.status = status; req.decidedBy = by
  this.waiters.get(id)?.(req)
}
}

핵심은 Event(혹은 Promise resolver). 에이전트는 wait 에서 블록되고, 운영자가 HTTP API 로 decide 를 호출하면 깨어난다. 의 가장 단순한 구현. 이 자연스럽게 들어가는 자리다. 를 자료구조로만 풀어낸 셈.

에이전트에 끼우기

위험 도구만 게이트를 통과시키고, 나머지는 즉시 실행.

DANGEROUS = {"charge_card", "send_email", "delete_data"}

async def call_tool_with_gate(tool, args, corr_id):
  audit.log("tool_call_proposed", correlation_id=corr_id,
            tool=tool, args=args)
  if tool in DANGEROUS:
      req = await queue.submit(tool, args)
      decided = await queue.wait(req.id, timeout=600)
      audit.log("human_decision", correlation_id=corr_id,
                request_id=req.id, status=decided.status,
                by=decided.decided_by)
      if decided.status == "rejected":
          return {"error": "rejected"}
      if decided.status == "modified":
          args = decided.modified_payload or args
  return await _execute(tool, args)

const DANGEROUS = new Set(['charge_card', 'send_email', 'delete_data'])

export async function callToolWithGate(
tool: string, args: Record<string, unknown>, corrId: string,
) {
await audit.log('tool_call_proposed', { corrId, tool, args })
if (DANGEROUS.has(tool)) {
  const req = queue.submit(tool, args)
  const decided = await queue.wait(req.id)
  await audit.log('human_decision', { corrId, status: decided.status })
  if (decided.status === 'rejected') return { error: 'rejected' }
  if (decided.status === 'modified') args = decided.modifiedPayload ?? args
}
return execute(tool, args)
}

게이트는 도구 호출 레이어에 끼워 넣는다. 코드는 위험 여부를 모르고 그냥 도구를 부른다. 의 정책은 한 곳에만. 도구 호출 결과는 그대로 에 흘러간다.

감사 추적

다이어그램 로딩…

은 사후 분쟁에서 살아남기 위한 것이다. 어떤 데이터를 남겨야 충분한가.

필드	왜 필요한가
`correlation_id`	한 요청을 끝까지 추적
`agent`	어느 에이전트가 제안
`tool`, `args`	실제로 일어난 행동
`model`, `model_input`	어떤 입력이 그 행동을 유발
`human_decision`, `decided_by`	누가 책임지나
`ts` (UTC)	시간 순서

이 여섯이 있으면 사고 후 재현이 가능하다. 가 모든 줄을 잇는 척추. 은 의 기억 장치다.

감사 로그 구현

append-only JSONL 이면 충분하다. SQL 도 좋지만, 단순함이 신뢰의 출발점.

class AuditTrail:
  def __init__(self, path):
      self.path = Path(path)
      self.path.parent.mkdir(parents=True, exist_ok=True)

  def log(self, event_type, **fields):
      record = {"ts": datetime.now(timezone.utc).isoformat(),
                "type": event_type, **fields}
      with self.path.open("a", encoding="utf-8") as f:
          f.write(json.dumps(record, ensure_ascii=False) + "\n")

export class AuditTrail {
constructor(private path: string) {}
async log(eventType: string, fields: Record<string, unknown> = {}) {
  await mkdir(dirname(this.path), { recursive: true })
  const record = { ts: new Date().toISOString(), type: eventType, ...fields }
  await appendFile(this.path, JSON.stringify(record) + '\n', 'utf8')
}
}

write-only 보장이 핵심. 수정 가능하면 감사가 아니라 그냥 로그다. 이 곧 의 기반이다. 는 결국 의 한 부분 집합 — 23장 관측성과 자연스레 합쳐진다.

에스컬레이션 정책

언제 누구에게 올릴지는 코드 곳곳에 흩지 말고 한 파일에 모은다. 은 데이터다.

POLICY = {
  "charge_card":   {"threshold_amount": 100000, "reviewer": "finance"},
  "send_email":    {"recipients_max": 1,        "reviewer": "ops"},
  "delete_data":   {"always": True,              "reviewer": "ciso"},
}

def needs_human(tool, args):
  p = POLICY.get(tool)
  if not p: return False
  if p.get("always"): return True
  if "threshold_amount" in p:
      return args.get("amount", 0) >= p["threshold_amount"]
  if "recipients_max" in p:
      return len(args.get("to", [])) > p["recipients_max"]
  return False

export const POLICY: Record<string, any> = {
charge_card: { thresholdAmount: 100000, reviewer: 'finance' },
send_email:  { recipientsMax: 1,        reviewer: 'ops' },
delete_data: { always: true,            reviewer: 'ciso' },
}

export function needsHuman(tool: string, args: any): boolean {
const p = POLICY[tool]
if (!p) return false
if (p.always) return true
if (p.thresholdAmount) return (args.amount ?? 0) >= p.thresholdAmount
if (p.recipientsMax) return (args.to?.length ?? 0) > p.recipientsMax
return false
}

리뷰어 채널까지 정책에 박아두면, 거버넌스 변경이 코드 한 줄 수정이 된다. 는 정책 객체 뒤에 숨고, 는 정책 변경을 모른다.

HITL 의 함정

사람이 곧 병목 — 승인이 늦으면 에이전트가 멈춘다. SLA 정의가 필수.
승인 피로 — 매번 누르다 보면 사람이 자동 승인 모드가 된다. 를 좁게.
모디파이 모드 누락 — yes/no 만 받으면 사람이 인자를 직접 못 고친다.
감사 누락 — 없는 HITL 은 그냥 수동 자동화. 사후 분쟁에서 무력.

의 자리는 좁게, 정책은 명시적으로, 로그는 풍부하게.

다음 챕터

패턴은 충분히 봤다. ···토론·HITL 다섯 가지를 손에 잡았다. 이제 실전 프레임워크로 넘어간다. 가장 영향력 큰 1군 — 부터다. 다음 챕터에서 그래프·노드·엣지·상태·를 본다.