지난 8회차 동안 에이전트는 매 턴마다 동일한 방식으로 종료되었습니다. 바로 문장을 출력하고 우리가 그것을 읽는 방식입니다. 이는 데모용으로는 괜찮지만, 제품으로서는 막다른 길입니다. 에이전트를 기반으로 무언가를 _구축(build)_하려는 순간 — 즉, UI에 렌더링하거나, 테스트에서 점수를 매기거나, API로부터 반환하려는 순간 — 산문(prose)은 당신을 방해합니다. 결국 정규 표현식(regexes)으로 텍스트를 스크래핑하며 문구가 바뀌지 않기만을 기도하는 처지로 전락하게 됩니다.

이 포스트는 한 가지 기능으로 이 문제를 해결합니다. 최종 답변이 고정된 형태를 가진 **검증된 JSON 객체 (validated JSON object)**가 되는 것입니다. 동일한 에이전트, 동일한 도구(tools), 동일한 메모리(memory)를 사용하지만, 이제는 문단 대신 {"status": "...", "answer": "...", "items": [...], ...}를 반환합니다. 이 계약(contract)은 이 시리즈의 나머지 부분이 전개되는 중심축입니다. 트레이싱(tracing) 장에서는 이를 로그로 남기고, 평가(evaluation) 장에서는 필드에 대해 단언(assert)하며, 배포(deployment) 장에서는 이를 HTTP 응답 본문(response body)으로 반환합니다.

솔직한 주의 사항이 하나 있으며, 이것이 이 포스트의 진짜 교훈입니다. 깔끔한 답변은 "JSON 스키마(schema)와 함께 response_format을 사용하여 API가 이를 강제하도록 하라"는 것입니다. 하지만 llama-3.3-70b와 같은 오픈 모델이 있는 호스팅된 API 카탈로그(API Catalog)에서는 이것이 신뢰할 수 없습니다. 엄격한 스키마 모드(strict schema mode)는 서버 측 기능인데, 오픈 모델 엔드포인트들은 이를 일관성 없게 구현하며, 도구 호출(tool calling)과도 상호작용이 좋지 않습니다. 따라서 우리는 여기에 의존하지 않습니다. 프롬프트(prompt)에서 JSON을 요청한 다음, 성숙한 방식으로 처리합니다: 직접 파싱(parse)하고, 검증(validate)하며, 틀렸을 경우 한 번은 복구(repair)하는 것입니다. 프레임워크는 사용하지 않습니다.

(만약 NIM을 자체 호스팅한다면 — 4부 — NVIDIA는 서버 측 구조화된 생성(structured generation)에 대해 문서화하고 있습니다: NIM 1.x는 제약된 디코딩(constrained decoding)을 위해 extra_body={"nvext": {"guided_json": schema}}를 허용하며, 최신 릴리스는 OpenAI 호환 response_format JSON 모드로 전환하고 있습니다. 어느 쪽이든, NVIDIA의 자체 문서에서도 클라이언트 측에서 응답을 검증하라고 안내합니다. 이는 정확히 이 포스트가 구축하고자 하는 단계입니다. 문서: https://docs.nvidia.com/nim/large-language-models/latest/structured-generation.html)

저는 USC의 NVIDIA Developer Champion인 B Torkian입니다. 시리즈의 9부입니다.

추가되는 사항

Workshop 8: 최종 답변 -> 출력할 문자열 (string)
Workshop 9: 최종 답변 -> JSON 텍스트 -> 파싱 (parse) -> 검증 (validate) -> (1회 수리 (repair)) -> 사용할 수 있는 딕셔너리 (dict)

Workshop 8의 동작(behavior)은 변경되지 않습니다 — 도구(tools), 멀티턴 메모리(multi-turn memory), 턴 단위 트리밍(trim-by-turns), 스트리밍 누적기(streaming accumulator)는 그대로 유지됩니다. 유일한 동작 변화는 **최종 답변 경계(final-answer boundary)**에서 발생합니다. message.content를 반환하는 대신, 이를 검증된 dict로 변환합니다. (또한 약간의 정리 작업도 수행합니다. Workshop 8에서는 chat()과 stream()에 거의 동일한 루프가 두 개 있었는데, 이를 하나의 공유 루프로 통합하여 JSON 최종화(finalization)가 정확히 한 곳에서 이루어지도록 했습니다. 이에 대한 자세한 내용은 단계 4에서 다룹니다. 한 가지 솔직한 트레이드오프(trade-off)는, 최종 답변에 대해 토큰 단위의 실시간 디스플레이(display)가 중단된다는 점입니다. 형태가 완성되지 않은 JSON은 보여주는 것이 무의미하기 때문입니다. 반면, 8부에서 다룬 와이어 레벨(wire-level) 스트리밍과 도구 호출(tool-call) 파편 재조립 기능은 그대로 유지됩니다.)

단계 1 — 계약(contract) 결정하기

코드를 작성하기 전에, 에이전트가 무엇을 반환해야 하는지 결정해야 합니다. 캠퍼스 어시스턴트의 경우, 다음 여섯 가지 키(key)가 모든 케이스를 커버합니다:

{
  "status": "answered",
  "answer": "USC AI Club은 매주 목요일 오후 5시에 공학관 204호에서 모입니다.",
...

status와 category는 열거형 (enums) (허용된 값의 고정된 집합)이므로, 다운스트림(downstream) 코드에서 이를 기준으로 분기 처리를 할 수 있습니다. items는 기계가 읽을 수 있는 페이로드(payload)입니다. missing은 사용자가 요청했으나 찾지 못한 항목의 이름을 지정합니다. sources는 근거(grounding)로, 가드레일(guardrails) 장에서 다룬 정신을 이어받아 사용된 정확한 지식 베이스(knowledge-base) 라인을 나타냅니다.

단계 2 — 파싱, 검증, 수리 (가장 중요한 부분)

우리는 모델의 출력을 신뢰할 수 없는 것으로 취급합니다. 일반적인 Python으로 작성된 세 개의 작은 함수를 사용합니다:

STATUSES = {"answered", "not_found", "needs_clarification"}
CATEGORIES = {"campus_event", "campus_hours", "campus_resource", "comparison", "refusal"}
REQUIRED_KEYS = ("status", "answer", "category", "items", "missing", "sources")
...

{% endraw %}
json fences. Take the {...} span.
    start, end = text.find("{“"), text.rfind("}")
    if start == -1 or end == -1 or end < start:
        raise ValueError("no JSON object found")
    return json.loads(text[start:end + 1])

def validate_answer(data) -> list:
    if not isinstance(data, dict):
        return ["response is not a JSON object"]
    errors = []
    for key in REQUIRED_KEYS:
        if key not in data:
            errors.append(f"missing required key: {key}")
    if data.get("status") not in STATUSES:
        errors.append(f"status must be one of {sorted(STATUSES)}")
    if data.get("category") not in CATEGORIES:
        errors.append(f"category must be one of {sorted(CATEGORIES)}")
    if "answer" in data and not isinstance(data["answer"], str):
        errors.append("answer must be a string")
    for key in ("items", "missing", "sources"):
        if key in data and not isinstance(data[key], list):
            errors.append(f"{key} must be a list")
    if isinstance(data.get("items"), list) and not all(isinstance(it, dict) for it in data["items"]):
        errors.append("each entry in items must be an object")
    return errors
{% raw %}

만약 유효성 검사에 실패하면, 정확히 한 번 수정 시도를 합니다. 즉, 손상된 출력을 temperature=0으로 모델에 다시 전달하고 문제 목록과 함께 깨끗한 객체를 요청합니다:

```python
def repair_answer_json(raw_text: str, errors: list) -> dict | None:
    # (프롬프트 생략 — 리포지토리 버전은 필요한 키(keys)와 열거형(enums)을 다시 명시합니다)
    try:
        fix = client.chat.completions.create(
            model=MODEL, temperature=0, max_tokens=800,
            messages=[
                {"role": "system", "content": "당신은 잘못된 형식의 JSON을 수정합니다. 오직 유효한 JSON 객체만 반환하세요."},
                {"role": "user", "content": f"문제점: {errors}. 모든 사실을 보존하세요. 원문:\n{raw_text}\n\n수정된 JSON만 반환하세요."},
            ],
        )
        data = parse_json_object(fix.choices[0].message.content or "")
    except Exception:          # API 오류 또는 파싱 불가능 — 결정론적(deterministic)으로 폴백(fallback) 수행
        return None
    return data if not validate_answer(data) else None

그리고 수리(repair)조차 실패할 경우, 호출자가 예상치 못한 상황에서 충돌하지 않도록 **결정론적(deterministic)**인 에러 객체를 반환합니다:

def format_error(missing: str = "valid_json") -> dict:
    return {"status": "needs_clarification",
            "answer": "유효한 구조화된 응답을 생성할 수 없었습니다. 다시 질문해 주세요.",
            "category": "refusal", "items": [], "missing": [missing], "sources": []}

파싱(Parse) → 검증(validate) → 1회 수리(repair) → 결정론적 폴백(deterministic fallback). 이 4단계 사다리는 스키마 강제 프레임워크(schema-enforcement framework) 없이도 프로덕션 환경에서 구조화된 출력(structured output)을 안전하게 만드는 핵심입니다. (Part 8의 단계 제한 폴백 또한 구조화됩니다. 리포지토리에서 format_error는 선택적으로 원인별 answer를 인자로 받으므로, 포기하는 경로조차 계약(contract)을 준수합니다.)

...

SYSTEM_PROMPT = (
    "...Workshop 7-8의 모든 도구(tool) + 메모리(memory) 가이드라인...\n\n"
    "최종 답변 형식. 도구 사용을 마쳤을 때, 당신의 최종 답변은 반드시 "
    "단일 JSON 객체여야 하며 그 외의 것은 포함해서는 안 됩니다 — 산문(prose)이나 코드 펜스(code fences) 없이 작성하세요. "
    "정확히 다음 키(keys)를 사용하세요: status (answered|not_found|needs_clarification), answer (string), category "
    "(campus_event|campus_hours|campus_resource|comparison|refusal), items (list), "
    "missing (list), sources (list)."
)

도구 호출(Tool-calling) 단계는 영향을 받지 않습니다. 모델은 사실을 수집하는 동안 여전히 tool_calls를 생성합니다. 오직 최종 답변만이 JSON이어야 합니다.

...

def chat(self, user_message: str) -> dict:
    return self._run_turn(user_message, stream=False)

def stream(self, user_message: str) -> dict:
    return self._run_turn(user_message, stream=True)

_run_turn은 Workshop 8의 루프(loop)입니다 — 더 이상 남은 도구가 없을 때까지 도구를 실행합니다. 유일하게 새로 추가된 단계는 최종 분기(branch)입니다. 텍스트를 반환하는 대신, _finalize_json을 통해 검증된 딕셔너리(dict)로 최종화합니다.

def _finalize_json(self, raw_text: str) -> dict:
    try:
        data = parse_json_object(raw_text)
        errors = validate_answer(data)
    except (ValueError, json.JSONDecodeError):
        data, errors = None, ["response was not valid JSON"]
    if errors:
        repaired = repair_answer_json(raw_text, errors)
        data = repaired if repaired is not None else format_error()
    return data

# 모델이 도구 호출을 중단한 후, _run_turn의 최종 분기:
#     data = self._finalize_json(text)
#     self.messages.append({"role": "assistant", "content": json.dumps(data)})
#     self._trim()
#     return data

우리는 모델의 가공되지 않은(raw) 텍스트가 아니라, 표준(canonical) 형태인 json.dumps(data)를 히스토리(history)에 저장한다는 점에 주목하세요. 따라서 다음 턴(turn)의 메모리(memory) 또한 깨끗하게 검증된 JSON이 됩니다. 이제 chat()과 stream() 모두 딕셔너리(dict)를 반환합니다.

...

네트워크 연결이 끊겼습니다. 이제 chat()과 stream() 모두 딕셔너리(dict)를 반환합니다.

session = ChatSession(verbose=True)
for q in [
    "When does the USC AI Club meet?",                          # answered, campus_event
    "How many days until that?",                                # memory + tool
    "Which is sooner, that meeting or the AI/ML office hours?",  # comparison
    "What is the campus wifi password?",                        # not_found / refusal
]:
    result = session.chat(q)
    print(json.dumps(result, indent=2))

네 개의 잘 구성된 객체(well-formed objects)를 받게 됩니다. 와이파이 질문은 만족스러운 부분입니다. 거절하는 문장 대신, 코드가 처리할 수 있는 형식화된 거절 응답을 얻기 때문입니다:

{"status": "not_found", "answer": "I don't have that information — check with the USC AI Club.",
 "category": "refusal", "items": [], "missing": ["USC campus wifi password"], "sources": []}

메모리(Memory)와 다단계 추론(multi-step reasoning)은 그대로 유지됩니다. _"How many days until that?"_는 여전히 _"that"_을 해결하고, 비교 질문은 여전히 날짜당 한 번씩 도구를 호출합니다.

6단계 — 실제로 구축한 것들

• **워크숍 1(Workshop 1)**은 에이전트에게 두뇌를 주었습니다. 2개의 사실 메모리(memory of facts). 3의 판단력(judgment). 4의 이식성(portability). 5의 손(hands). 6의 계획(a plan). 7의 대화 기억(memory of the conversation). 8의 실시간 음성(real-time voice).
• **워크숍 9(Workshop 9)**는 에이전트에게 **계약서(contract)**를 주었습니다. 다른 소프트웨어가 소비할 수 있는 출력 형식입니다.

마지막 것이 이 시리즈의 조용한 전환점입니다. 그 이전의 모든 것은 에이전트를 더 똑똑하게 만들었지만, 이것은 에이전트를 _통합 가능(integratable)_하게 만듭니다. 그리고 이 시리즈의 후반부를 구성합니다:

• 다음 — 트레이스(Traces): 각 턴(turn)의 도구 사용 기록, 지연 시간(latency), 그리고 이 최종 객체를 JSONL 형식으로 기록하여 에이전트가 나중에 무엇을 했는지 볼 수 있게 합니다.
• 그 다음 — 평가(Evals): 디스크에 트레이스를 저장하고 고정된 계약서(fixed contract)를 사용하면, 마침내 소프트웨어처럼 에이전트를 테스트할 수 있습니다. 질문들을 재생하고 와이파이 질문에서 status, category, 그리고 missing 필드가 작동하는지 단언(assert)할 수 있습니다.
• 그 다음 — 영속적인 세션(Durable sessions) 및 배포(Deploy): 대화 기록을 유지하고 이 정확한 JSON 본문(JSON body)을 HTTP API 뒤에서 제공합니다.

에이전트는 여전히 동일한 while 루프입니다. 우리는 단지 그 마지막 단어가 데이터가 되도록 가르쳤을 뿐입니다.

코드 가져오기

저장소 (Repo): github.com/torkian/nvidia-nim-workshop
원클릭 Colab: part9_structured_output.ipynb 열기
로컬 Python: 저장소 내의 part9_structured_output.py (pip install -r requirements.txt 실행 후 python3 part9_structured_output.py 실행).

MIT 라이선스입니다. 저는 USC에서 이 코드를 실행하고 있습니다 — 포크(fork)하여 지식 베이스(knowledge base), 도구(tools), 그리고 계약(contract)을 여러분의 학교, 클럽, 또는 프로젝트에 맞게 교체하여 사용하세요.

전체 시리즈