Field Log · Entry
Typed Agent Runtime: Tool Registry와 반드시 끝나는 Bounded Loop (6/10)
오늘의 결론
- Agent loop는
while True안에서 model과 tool을 번갈아 호출하는 코드가 아닙니다. Typed state와 허용된 transition, 모든 exit가 보이는 실행 protocol입니다.- Model의 tool call은 실행 명령이 아니라
ToolProposal입니다. Registry allowlist, JSON schema, semantic rule, capability·tenant authorization을 통과해야 실행할 수 있습니다.- Step·tool·deadline budget은 외부 호출 전에 예약합니다. 실행 후 차감하면 마지막 허용치를 넘겨 한 번 더 side effect가 발생할 수 있습니다.
- 정상 완료뿐 아니라 insufficient evidence·budget exhausted·cancelled·failed를 terminal result로 남깁니다. 종료 이유는 log 문자열이 아니라 API와 eval이 읽는 data입니다.
- 첫 production runtime은 tool을 순차 실행하는 편이 좋습니다. Parallel call은 예산·순서·취소·부분 성공 의미가 정의된 뒤 추가합니다.
앞 글에서는 evidence를 bounded context와 server-verified citation으로 변환했습니다. 이제 retrieval과 generation을 필요할 때 호출하되 무한히 반복하거나 권한 밖 tool을 실행하지 않는 runtime을 만듭니다.
이 글의 대상과 학습 시간
- 대상: function calling demo를 통제 가능한 service runtime으로 옮기려는 Python 개발자
- 선수 지식: dataclass·Enum·Pydantic, tool contract 개념
- 빠르게 읽기: 약 15분
- 코드와 test까지 따라 하기: 약 65분
- 산출물: typed
AgentState, allowlistToolRegistry, budget-first sequential loop와 reducer test
1. while True Agent가 멈추지 않는 이유
while True:
decision = await llm(messages)
if decision.final_answer:
return decision.final_answer
result = await tools[decision.tool_name](**decision.arguments)
messages.append(result)이 코드에는 다음 질문의 답이 없습니다.
- Model이 존재하지 않는 tool name을 만들면?
- Arguments가 schema에는 맞지만 다른 tenant resource를 가리키면?
- Tool이 계속 빈 결과를 반환하면?
- Model이 같은 query를 반복하면?
- Deadline이 tool 실행 중 끝나면?
- Tool call limit의 마지막 한 번은 언제 차감하는가?
- Client가 연결을 끊으면?
- Partial answer와 final answer는 어떻게 구분하는가?
Runtime은 model보다 바깥에서 이 질문에 답해야 합니다.
2. State는 대화문보다 넓다
from dataclasses import dataclass, field
from enum import StrEnum
class Phase(StrEnum):
OBSERVE = "observe"
DECIDE = "decide"
VALIDATE = "validate"
EXECUTE = "execute"
REDUCE = "reduce"
TERMINAL = "terminal"
class TerminationReason(StrEnum):
COMPLETED = "completed"
INSUFFICIENT_EVIDENCE = "insufficient_evidence"
NEEDS_CLARIFICATION = "needs_clarification"
STEP_BUDGET_EXHAUSTED = "step_budget_exhausted"
TOOL_BUDGET_EXHAUSTED = "tool_budget_exhausted"
DEADLINE_EXCEEDED = "deadline_exceeded"
CANCELLED = "cancelled"
INVALID_PROPOSAL = "invalid_proposal"
NO_PROGRESS = "no_progress"
POLICY_DENIED = "policy_denied"
APPROVAL_REQUIRED = "approval_required"
FAILED = "failed"
@dataclass(frozen=True)
class RuntimeBudget:
steps_remaining: int
tool_calls_remaining: int
def __post_init__(self) -> None:
if self.steps_remaining < 0 or self.tool_calls_remaining < 0:
raise ValueError("runtime budget must not be negative")
@dataclass(frozen=True)
class Observation:
kind: str
payload: dict
source: str
sequence: int
@dataclass(frozen=True)
class AgentState:
run_id: str
tenant_id: str
question: str
phase: Phase
budget: RuntimeBudget
observations: tuple[Observation, ...] = ()
final_answer: AnswerPayload | None = None
termination: TerminationReason | None = None
last_error_code: str | None = None
@property
def is_terminal(self) -> bool:
return self.phase is Phase.TERMINALRaw chain-of-thought를 state에 저장하지 않습니다. Runtime이 감사해야 할 것은 외부 evidence, tool proposal, validation decision, tool receipt, final answer처럼 관찰 가능한 artifact입니다.
3. Model output은 decision union으로 제한한다
from typing import Annotated, Literal, Union
from pydantic import BaseModel, ConfigDict, Field, ValidationError
class ToolProposal(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
kind: Literal["tool"]
call_id: str = Field(min_length=8, max_length=128)
tool_name: str = Field(pattern=r"^[a-z][a-z0-9_]{1,63}$")
arguments: dict
class FinalProposal(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
kind: Literal["final"]
answer: AnswerPayload
class AbstainProposal(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
kind: Literal["abstain"]
reason: Literal["insufficient_evidence", "needs_clarification"]
AgentDecision = Annotated[
Union[ToolProposal, FinalProposal, AbstainProposal],
Field(discriminator="kind"),
]Structured output은 union shape를 안정화하지만 tool 실행 허가를 의미하지 않습니다. tool_name과 arguments는 모두 untrusted proposal입니다.
4. ToolSpec에 실행 의미를 모은다
from collections.abc import Awaitable, Callable
from typing import Generic, TypeVar
from pydantic import BaseModel
ArgsT = TypeVar("ArgsT", bound=BaseModel)
ResultT = TypeVar("ResultT")
class Effect(StrEnum):
READ = "read"
WRITE = "write"
EXTERNAL = "external"
@dataclass(frozen=True)
class ToolContext:
run_id: str
tenant_id: str
principal_id: str
deadline: Deadline
@dataclass(frozen=True)
class ToolSpec(Generic[ArgsT, ResultT]):
name: str
description: str
args_model: type[ArgsT]
effect: Effect
required_capability: str
timeout_seconds: float
execute: Callable[[ArgsT, ToolContext], Awaitable[ResultT]]description은 model 선택을 돕지만 authorization policy가 아닙니다. required_capability, effect, timeout은 trusted code/config에서 옵니다.
읽기 tool 예시
class SearchKnowledgeArgs(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
query: str = Field(min_length=1, max_length=2_000)
top_k: int = Field(default=8, ge=1, le=20)
async def search_knowledge(
args: SearchKnowledgeArgs,
context: ToolContext,
) -> SearchObservation:
request = SearchRequest(
request_id=context.run_id,
query=args.query,
principal=principal_scope(context),
filters=SearchFilter(),
top_k=args.top_k,
)
evidence = await retrieval.search(request, deadline=context.deadline)
return SearchObservation.from_evidence(evidence)Tool이 tenant ID를 model argument로 받지 않는 점이 중요합니다. Trusted ToolContext에서 가져오면 model이 다른 tenant를 지정할 surface가 줄어듭니다.
5. Registry는 allowlist이며 dynamic loader가 아니다
class UnknownTool(Exception):
pass
class ToolPolicyDenied(Exception):
code = "tool_policy_denied"
class HumanApprovalRequired(Exception):
code = "human_approval_required"
class ToolSemanticError(ValueError):
code = "invalid_tool_semantics"
class NoProgressDetected(Exception):
code = "no_progress_detected"
class ToolRegistry:
def __init__(self, specs: tuple[ToolSpec, ...]) -> None:
by_name = {spec.name: spec for spec in specs}
if len(by_name) != len(specs):
raise ValueError("duplicate tool name")
self._by_name = by_name
def get(self, name: str) -> ToolSpec:
try:
return self._by_name[name]
except KeyError as exc:
raise UnknownTool(name) from exc
def model_catalog(self) -> list[dict]:
return [
{
"name": spec.name,
"description": spec.description,
"parameters": spec.args_model.model_json_schema(),
}
for spec in self._by_name.values()
]Model이 만든 name으로 importlib, getattr(global_object, name), eval()을 호출하지 않습니다. Registry에 등록된 callable만 실행됩니다.
6. Validation은 네 층으로 나눈다
1. Registry validation
Tool name이 현재 runtime version의 allowlist에 존재하는가?
2. Schema validation
spec = registry.get(proposal.tool_name)
args = spec.args_model.model_validate(proposal.arguments, strict=True)3. Semantic validation
Schema에 맞아도 query가 empty-like value인지, date range가 역전됐는지, 동일 call을 반복하는지 검사합니다.
4. Authorization·effect validation
def authorize_tool(
spec: ToolSpec,
*,
principal: Principal,
state: AgentState,
) -> None:
if not principal.has(spec.required_capability):
raise ToolPolicyDenied(spec.name)
if spec.effect is not Effect.READ:
raise HumanApprovalRequired(spec.name, state.run_id)이 글의 첫 runtime은 READ tool만 자동 실행합니다. WRITE·EXTERNAL effect는 29편의 approval·commit-time authorization 원칙에 따라 별도 flow로 보냅니다.
7. Budget은 실행 전에 예약한다
from dataclasses import replace
class BudgetExhausted(Exception):
def __init__(self, reason: TerminationReason) -> None:
self.reason = reason
def reserve_step(state: AgentState) -> AgentState:
if state.budget.steps_remaining <= 0:
raise BudgetExhausted(TerminationReason.STEP_BUDGET_EXHAUSTED)
return replace(
state,
budget=replace(
state.budget,
steps_remaining=state.budget.steps_remaining - 1,
),
)
def reserve_tool_call(state: AgentState) -> AgentState:
if state.budget.tool_calls_remaining <= 0:
raise BudgetExhausted(TerminationReason.TOOL_BUDGET_EXHAUSTED)
return replace(
state,
budget=replace(
state.budget,
tool_calls_remaining=state.budget.tool_calls_remaining - 1,
),
)다음처럼 실행 후 차감하면 안 됩니다.
if remaining >= 0:
await execute() # remaining=0인데 한 번 실행됨
remaining -= 1예약 후 tool이 실패해도 budget을 돌려주지 않는 것이 기본입니다. 실패한 외부 호출도 시간·quota·비용을 소비했기 때문입니다.
8. Transition을 reducer로 고정한다
@dataclass(frozen=True)
class ToolSucceeded:
call_id: str
tool_name: str
result: dict
@dataclass(frozen=True)
class ToolFailed:
call_id: str
tool_name: str
error_code: str
retryable: bool
RuntimeEvent = ToolSucceeded | ToolFailed
def reduce_tool_event(state: AgentState, event: RuntimeEvent) -> AgentState:
sequence = len(state.observations)
if isinstance(event, ToolSucceeded):
observation = Observation(
kind="tool_result",
payload={
"call_id": event.call_id,
"tool_name": event.tool_name,
"result": event.result,
},
source=event.tool_name,
sequence=sequence,
)
else:
observation = Observation(
kind="tool_error",
payload={
"call_id": event.call_id,
"tool_name": event.tool_name,
"error_code": event.error_code,
"retryable": event.retryable,
},
source=event.tool_name,
sequence=sequence,
)
return replace(
state,
phase=Phase.OBSERVE,
observations=state.observations + (observation,),
)Tool result도 untrusted data입니다. 다음 model input에서 evidence와 같은 data block으로 넣고, result 안의 instruction을 system policy로 승격하지 않습니다.
9. 순차 bounded loop를 구현한다
import asyncio
async def run_agent(
state: AgentState,
*,
decision_model: DecisionModel,
registry: ToolRegistry,
principal: Principal,
deadline: Deadline,
clock: Clock,
) -> AgentState:
try:
while not state.is_terminal:
deadline.require_remaining(clock=clock)
state = reserve_step(state)
state = replace(state, phase=Phase.DECIDE)
decision = await decision_model.decide(
build_decision_input(state, registry.model_catalog()),
deadline=deadline,
)
if isinstance(decision, FinalProposal):
answer = validate_final_answer(decision.answer, state)
return terminate(
state,
TerminationReason.COMPLETED,
answer=answer,
)
if isinstance(decision, AbstainProposal):
reason = (
TerminationReason.INSUFFICIENT_EVIDENCE
if decision.reason == "insufficient_evidence"
else TerminationReason.NEEDS_CLARIFICATION
)
return terminate(
state,
reason,
)
state = replace(state, phase=Phase.VALIDATE)
spec = registry.get(decision.tool_name)
args = spec.args_model.model_validate(
decision.arguments,
strict=True,
)
validate_semantics(spec, args, state)
validate_no_progress(decision, state)
authorize_tool(spec, principal=principal, state=state)
state = reserve_tool_call(state)
state = replace(state, phase=Phase.EXECUTE)
tool_deadline = stage_deadline(
deadline,
max_stage_seconds=spec.timeout_seconds,
clock=clock,
)
result = await spec.execute(
args,
ToolContext(
run_id=state.run_id,
tenant_id=state.tenant_id,
principal_id=principal.id,
deadline=tool_deadline,
),
)
state = replace(state, phase=Phase.REDUCE)
state = reduce_tool_event(
state,
ToolSucceeded(
call_id=decision.call_id,
tool_name=spec.name,
result=serialize_tool_result(result),
),
)
return state
except BudgetExhausted as exc:
return terminate(state, exc.reason)
except DeadlineExceeded:
return terminate(state, TerminationReason.DEADLINE_EXCEEDED)
except asyncio.CancelledError:
record_cancelled(state.run_id)
raise
except ToolPolicyDenied as exc:
return terminate(
state,
TerminationReason.POLICY_DENIED,
error_code=exc.code,
)
except HumanApprovalRequired as exc:
return terminate(
state,
TerminationReason.APPROVAL_REQUIRED,
error_code=exc.code,
)
except NoProgressDetected as exc:
return terminate(
state,
TerminationReason.NO_PROGRESS,
error_code=exc.code,
)
except (UnknownTool, ValidationError, ToolSemanticError) as exc:
record_invalid_proposal(state.run_id, exc)
return terminate(
state,
TerminationReason.INVALID_PROPOSAL,
error_code="invalid_tool_proposal",
)
except Exception as exc:
record_internal_error(state.run_id, exc)
return terminate(
state,
TerminationReason.FAILED,
error_code="internal_runtime_error",
)이 뼈대는 흐름을 보여 주기 위해 tool error mapping을 줄였습니다. Production에서는 spec.execute의 typed error를 ToolFailed event로 바꾸고 retryable observation을 model 또는 fixed policy가 처리하게 합니다. Raw exception text는 model context에 넣지 않습니다.
Cancellation의 terminal record
Task 취소 중에는 새 event를 안전하게 yield하거나 DB에 긴 write를 보장하기 어렵습니다. In-memory state에 CANCELLED를 넣는 것과 durable cancel receipt 저장은 분리합니다. 87편에서 transaction과 idempotency를 연결합니다.
10. 종료는 한 함수에서만 만든다
def terminate(
state: AgentState,
reason: TerminationReason,
*,
answer: AnswerPayload | None = None,
error_code: str | None = None,
) -> AgentState:
if state.is_terminal:
raise RuntimeError("state is already terminal")
return replace(
state,
phase=Phase.TERMINAL,
final_answer=answer,
termination=reason,
last_error_code=error_code,
)Terminal state 이후 transition을 거부하면 double completion과 완료 후 tool execution을 막을 수 있습니다.
11. 반복 행동도 budget 전에 감지한다
Max step만으로는 같은 실패를 여러 번 비용 내며 반복할 수 있습니다.
def proposal_fingerprint(proposal: ToolProposal) -> str:
canonical = canonical_json({
"tool": proposal.tool_name,
"arguments": proposal.arguments,
})
return sha256(canonical.encode()).hexdigest()
def validate_no_progress(
proposal: ToolProposal,
state: AgentState,
*,
repeat_limit: int = 2,
) -> None:
fingerprint = proposal_fingerprint(proposal)
repeats = count_previous_fingerprint(state, fingerprint)
if repeats >= repeat_limit:
raise NoProgressDetected(fingerprint)Fingerprint에 secret·raw text를 넣지 않고 normalized argument hash만 trace에 저장합니다. Query rewrite처럼 작은 변화가 의미가 있는 경우는 tool별 progress predicate를 정의합니다.
12. Runtime event를 trace와 eval에 연결한다
각 step은 다음 manifest를 남깁니다.
{
"run_id": "run-01J...",
"step": 3,
"phase": "execute",
"decision_kind": "tool",
"tool_name": "search_knowledge",
"effect": "read",
"validation": "allowed",
"steps_remaining": 4,
"tool_calls_remaining": 1,
"deadline_remaining_ms": 820,
"outcome": "success"
}Prompt·tool result 전문 없이도 잘못된 transition, budget 소진, 반복 call을 분석할 수 있습니다. Payload가 필요할 때는 redacted artifact store에 별도 접근 제어합니다.
13. Budget과 allowlist를 test한다
def test_tool_budget_is_reserved_before_execution():
state = initial_state(tool_calls=1)
reserved = reserve_tool_call(state)
assert reserved.budget.tool_calls_remaining == 0
with pytest.raises(BudgetExhausted):
reserve_tool_call(reserved)
def test_unknown_tool_never_executes():
called = False
async def dangerous(args, context):
nonlocal called
called = True
registry = ToolRegistry((safe_search_spec(),))
with pytest.raises(UnknownTool):
registry.get("delete_everything")
assert called is False
def test_terminal_state_cannot_terminate_twice():
done = terminate(initial_state(), TerminationReason.COMPLETED)
with pytest.raises(RuntimeError, match="already terminal"):
terminate(done, TerminationReason.FAILED)Integration test에서는 scripted decision model을 사용합니다.
case A: search → final
expected: 2 steps, 1 tool, completed
case B: invalid tool × 1
expected: policy/validation failure, tool not called
case C: search empty repeated
expected: no-progress or step-budget terminal
case D: deadline expired before decide
expected: no model/tool call, deadline terminal자주 실패하는 구현 8가지
1. Model output을 바로 실행한다
Tool call은 proposal입니다. Registry·schema·semantic·authorization·budget gate를 거칩니다.
2. Tool name으로 dynamic import한다
Allowlist 밖 code execution surface가 생깁니다. Trusted registry의 callable만 사용합니다.
3. Schema validation을 authorization으로 착각한다
형식이 맞아도 다른 tenant resource나 금지된 effect일 수 있습니다. Trusted principal로 별도 검사합니다.
4. 실행 후 budget을 차감한다
경계에서 한 번 더 실행됩니다. Side effect 전에 reserve합니다.
5. Exception만 종료로 본다
Insufficient evidence와 budget exhaustion은 정상적으로 예상 가능한 terminal outcome입니다.
6. Tool error 전문을 model에게 돌려준다
Stack trace·credential·provider body가 유출되고 prompt injection channel이 됩니다. Safe error code만 observation으로 만듭니다.
7. 처음부터 parallel tools를 허용한다
부분 성공·순서·budget reservation·cancel semantics가 복잡해집니다. 순차 기준선을 먼저 검증합니다.
8. Raw chain-of-thought를 state에 저장한다
감사에는 evidence·proposal·receipt·decision outcome이 필요합니다. 숨은 추론 전문을 운영 계약으로 삼지 않습니다.
직접 적용 체크리스트
- State에 phase·budget·terminal reason이 typed field로 있다.
- Model output은 final·abstain·tool proposal union이다.
- Tool은 trusted registry allowlist에서만 찾는다.
- Arguments를 strict schema와 semantic rule로 검증한다.
- Tenant·capability·effect authorization이 model input과 분리돼 있다.
- Step·tool budget을 외부 호출 전에 예약한다.
- Deadline이 model과 tool 하위 호출로 전파된다.
- 모든 경로가 명시적 terminal outcome 또는 cancellation으로 끝난다.
- Repeated proposal과 no-progress를 감지한다.
- Tool result·error를 untrusted observation으로 다룬다.
스스로 확인하기
- JSON Schema에 맞는 tool call도 실행하면 안 되는 경우는 무엇인가?
- Budget을 실행 전에 차감해야 하는 off-by-one 이유는 무엇인가?
UnknownTool을 model에게 raw exception으로 돌려주면 어떤 위험이 있는가?- Insufficient evidence를 exception이 아닌 terminal data로 두면 무엇이 좋아지는가?
- Parallel tool call을 추가하기 전에 정의해야 할 의미는 무엇인가?
자주 묻는 질문
Q1. LangGraph 같은 framework를 쓰면 이 runtime이 필요 없나요?
Framework가 state persistence와 graph execution을 도와도 state schema, tool authority, budget reservation, terminal reason은 application contract로 정의해야 합니다. 이 설계는 framework adapter 안팎에서 기준선이 됩니다.
Q2. Tool error를 model이 보고 복구해야 하지 않나요?
가능합니다. 단, provider stack trace가 아니라 rate_limited, not_found, invalid_argument 같은 safe typed observation과 허용된 다음 행동만 전달합니다.
Q3. Read tool은 모두 자동 승인해도 되나요?
아닙니다. Read도 민감정보 유출·고비용 query·다른 tenant 접근 위험이 있습니다. Capability·resource ownership·query limit을 검사합니다.
Q4. Step budget은 몇 회가 적당한가요?
고정 정답은 없습니다. Task 성공률·trajectory 길이·비용·latency 분포를 eval해 intent별로 정합니다. 값이 클수록 지능이 높아진다고 가정하지 않습니다.
Q5. Cancellation을 terminal state로 반환하지 않고 다시 올리는 이유는 무엇인가요?
Python task cancellation은 상위 server가 resource cleanup과 client disconnect를 처리해야 하는 control signal입니다. 삼키지 않고 전파하되, durable run store에는 별도 cancel receipt를 기록합니다.
마무리
Agent runtime의 핵심은 model이 더 자유롭게 행동하게 하는 것이 아니라 허용된 상태 전이만 실행하게 하는 것입니다.
- Model은 typed proposal을 만든다.
- Runtime은 allowlist·schema·semantic·authorization을 검증한다.
- Budget을 먼저 예약한 뒤 tool을 실행한다.
- Observation을 reducer로 state에 반영한다.
- 모든 경로가 명시적 reason으로 끝난다.
다음 글에서는 이 state와 event를 PostgreSQL에 durable하게 저장하고, Redis cache와 idempotency key를 결합한 session layer로 중복 request·동시 append·crash window를 다룹니다.
검증 정보
- 글의 성격: Python·JSON Schema·공식 function calling 문서 해설 + 작성자 runtime 설계 제안 + 가상 코드 예시
- 마지막 검증일: 2026-07-16
- 검증 환경: Python 3.12, Pydantic 2 계열 discriminated union·strict validation 개념
- 실행 주의: Tool provider schema와 structured output event는 고정한 SDK/API version에 맞춰 adapter에서 mapping 필요
- 수치 표기: step·tool·timeout 값은 설명용이며 task eval로 조정
- 경험 표기: tenant·tool·run ID와 장비 예시는 실제 운영 정보가 아닌 가상 값