RuntimeAdapter 고급

이 페이지는 SchemaValidatedRuntimeAdapter 가 아닌 상위 베이스 클래스를 직접 상속해야 할 때의 선택지와 주의사항을 정리합니다.

레벨 한눈에 보기

레벨	베이스 클래스	언제
Level 1	`SchemaValidatedRuntimeAdapter[PayloadT]`	고정 Pydantic 스키마 (대부분의 분류·추출·판정)
Level 2	`StructuredOutputAgent`	스키마 고정이 어려운 동적 dict / Pydantic 의존성 제거
Level 3	`SchemaValidatedRuntimeAdapter` + `InputContracts`	다중 DataPart 중 shape 기반 선별 + priority fallback chain
Level 4	`RuntimeAdapter` 직접 상속	`dispatch`/`invoke`/`stream` 자체 재정의 필요

이 페이지는 위 표 중 Level 2 (StructuredOutputAgent) 와 Level 4 (RuntimeAdapter 직접 상속) 를 본문에서 다룹니다. Level 1 은 Registry 기반 에이전트, Level 3 은 InputContracts 가이드 참조.

Level 2: `StructuredOutputAgent`

Pydantic 없이 dict 를 직접 다루고 싶을 때 사용합니다.

언제 선택하나

상황	이유
LLM 응답 키가 런타임에 가변으로 변함	Pydantic 스키마로 고정하기 어려움
Pydantic 의존성을 피하고 싶음	임베디드·경량 배포 환경
기존 JSON 파이프라인과 자유 dict 형태 호환 필요	레거시 integration

최소 예시

from llamon_agent import StructuredOutputAgent


class MyAgent(StructuredOutputAgent):
    """자유 dict 형태의 응답을 표준화하는 어댑터."""

    # [선택] A2A artifact 메타 덮어쓰기 (어댑터 클래스에 이름이 붙박인 고급 케이스).
    # v0.2.0+ 에서는 `ExtensionConfig.artifact_name` 사용을 권장
    # — [agent-composition 가이드](/guide/agent-composition#a2a-artifact-메타데이터-artifact_name--artifact_description) 참조.
    artifact_name = "dynamic-response"
    artifact_description = "동적 키 기반 응답"

    def extract_payload(self, text: str, data: list[dict]) -> dict | None:
        """LLM 응답에서 dict payload 를 추출.

        기본 동작: fenced JSON 또는 data[0] → dict 반환.
        """
        payload = super().extract_payload(text, data)
        if payload is not None:
            return payload
        # 폴백 경로 — 추출 실패 시 텍스트 전체를 단순 응답으로 감쌉니다.
        return {"response": text.strip() or "응답 없음", "source": "llm"}

    def build_summary(self, payload: dict | None) -> str:
        """사용자 대면 요약 문구 생성 (RuntimeOutput.text)."""
        if payload and isinstance(payload.get("output_text"), str):
            return payload["output_text"]
        return super().build_summary(payload)

오버라이드 포인트

메서드	용도	오버라이드 빈도
`extract_payload(text, data)`	문자열/기존 data 에서 dict 추출	자주
`build_summary(payload)`	payload → 요약 텍스트	자주
`stream_filter_mode` (ClassVar)	streaming 토큰 필터 (`raw`/`summary_only`/`prefix_only`)	가끔

주의사항

payload_schema 는 지정하지 마세요 — Level 2 에서는 무시되며 경고도 없습니다.
Pydantic 없이 검증이 필요할 때는 extract_payload 안에서 직접 isinstance·키 체크 후 None 을 반환하면 SDK 기본 폴백 경로로 진입합니다.
streaming UX: stream_filter_mode 를 "summary_only" 로 두면 raw JSON 토큰이 클라이언트에 노출되지 않습니다 (자세한 내용은 SDK 소스 core/runtime/structured.py 참고).

`skip_llm` 최적화 — `SchemaValidatedRuntimeAdapter` + passthrough 시 LLM 호출 우회

적용 조건 — 다음 네 조건을 모두 만족할 때만 안전합니다:

SchemaValidatedRuntimeAdapter (또는 StructuredOutputAgent) 를 상속.
create_server(agent=MyAgent(runtime_agent), ...) 로 RuntimeAdapter 인스턴스를 agent= 인자로 주입하거나, registered guardrail-only 런타임처럼 primary LLM 없이 응답 정책이 명확해야 함.
apply_business_rules (혹은 extract_payload/build_summary)가 LLM 출력 텍스트에 의존하지 않음 — 빈 문자열을 받아도 정상 응답을 만들 수 있어야 함.
응답 조립을 a2a_data passthrough + 입력 텍스트 echo + 비즈니스 룰만으로 끝낼 수 있음.

동작:

LLMConfig(skip_llm=True) 가 True 면 SDK 가 ReAct 그래프 빌드를 건너뛰고 SkipLLMStub 을 사용합니다.
SkipLLMStub.dispatch/invoke 는 빈 문자열을 즉시 반환 → postprocess 파이프라인은 그대로 실행 → apply_business_rules 가 a2a_data 와 입력 텍스트로 응답을 조립.
출력 가드레일은 정상 동작 (최종 RuntimeOutput 에 적용).

설정 방법과 주의사항은 agent-composition 가이드의 LLM 호출 우회 섹션 을 참조하세요.

Level 4: `RuntimeAdapter` 직접 상속

dispatch/invoke/stream 자체를 재정의해야 할 때 사용합니다.

언제 선택하나

상황	예시
여러 LLM 호출 조합	router → executor → validator 체인
가드레일 차단 메시지 직접 분기	차단 텍스트 감지 시 별도 응답 구성
LLM 호출 전후 여러 단계 검증	pre-validation + post-enrichment
외부 시스템 호출을 adapter 내부에 포함	조건부 DB 조회 후 프롬프트 보강

베이스 클래스 구조

RuntimeAdapter (src/llamon_agent/core/runtime/adapter.py, ~227줄) 의 핵심 계약:

class RuntimeAdapter:
    # A2A artifact 메타 덮어쓰기 (ClassVar)
    artifact_name: ClassVar[str | None] = None
    artifact_description: ClassVar[str | None] = None

    def __init__(self, wrapped: Any) -> None:
        self._wrapped = wrapped
        self.name: str = getattr(wrapped, "name", "")

    # [오버라이드 대상]
    async def postprocess(self, output, *, query, a2a_files=None, a2a_data=None, context=None):
        """LLM 응답 → RuntimeOutput 변환. 기본: normalize + 입력 패스스루."""
        ...

    # [SDK 계약 — 어느 경로를 바꾸려는지 기준으로 오버라이드]
    async def dispatch(self, query, ...): ...   # A2A 일반 요청(message/send) — LLM 호출 전후 흐름을 직접 짤 때 (router/체인/분기)
    async def invoke(self, query, ...): ...     # 코드에서 .invoke() 직접 호출 — 드묾 (A2A 와 별개)
    async def stream(self, query, ...): ...     # A2A 스트리밍 요청(message/stream) — 토큰 단위 실시간 응답이 필요할 때 (기본은 single-yield)

    # [헬퍼 — 오버라이드 불필요]
    def _inject_artifact_metadata(self, result): ...  # ClassVar → RuntimeOutput 자동 주입
    def extract_json(self, output): ...
    def extract_text(self, output): ...

dispatch / invoke / stream — 셋의 차이

셋 다 어댑터의 진입 메서드지만 누가 호출하고, 응답을 어떻게 돌려주는지가 다릅니다.

메서드	진입 경로	응답 형태
`dispatch`	A2A `message/send` (외부 에이전트의 일반 요청)	완성된 단일 응답
`stream`	A2A `message/stream` (외부 에이전트의 실시간 요청)	토큰/청크 스트림
`invoke`	코드에서 `agent.invoke()` 직접 호출 (A2A 미경유)	완성된 단일 응답

dispatch↔stream 은 같은 A2A 입구가 send냐 stream이냐로 갈리는 것이고, invoke 는 서버를 거치지 않는 별도 진입점입니다.

기본 구현에서 셋이 호출하는 내부 메서드가 다릅니다 — 특히 stream 은 _wrapped.dispatch 를 호출합니다 (self.dispatch 가 아님). 그래서 dispatch 만 재정의하면 streaming 요청에는 반영되지 않습니다.

오버라이드	`message/send` (`dispatch`)	`message/stream` (`stream`)	`agent.invoke()` (`invoke`)
`dispatch`	✅	—	—
`invoke`	—	—	✅
`stream`	—	✅	—
`postprocess`	✅	✅	✅

→ 세 경로 공통 후처리는 postprocess 한 곳에. dispatch/invoke/stream 재정의는 경로별로 정말 다른 동작이 필요할 때만.

최소 예시 — LLM 호출 전 입력 보강 + 후 가드레일 분기

dispatch 오버라이드의 핵심은 LLM 호출을 사이에 두고 앞뒤로 무엇을 끼워넣을지입니다. 아래 예시는 세 위치를 모두 보여줍니다 — 호출 전 보강 / 호출 / 호출 후 분기.

from typing import Any, AsyncIterator
from llamon_agent import RuntimeAdapter, RuntimeOutput
from llamon_agent.exceptions import GuardrailBlockedError


class MyAgent(RuntimeAdapter):
    """LLM 호출 전 입력 보강, 후 가드레일 분기."""

    # v0.2.0+ 에서는 `ExtensionConfig.artifact_name` 을 권장. ClassVar 는 여전히 작동하나
    # 이 어댑터 클래스에 이름이 결합되어야 하는 고급 케이스용. 일반적으로는 config.py 사용.
    artifact_name = "multi-stage-response"
    artifact_description = "pre/post 분기 + 조건부 재호출"

    async def dispatch(self, query, *, skill_id=None, thread_id=None,
                       a2a_files=None, a2a_data=None, context=None):
        # ───────── ① LLM 호출 전 (pre) ─────────
        # 1-a. 입력 검증 — 짧은 쿼리는 LLM 호출 자체를 우회 (비용 0)
        if len(query.strip()) < 3:
            return self._inject_artifact_metadata(
                RuntimeOutput(text="더 자세한 질문을 입력해주세요.", data=[])
            )

        # 1-b. 컨텍스트 주입 — a2a_data 에 user_id 가 있으면 쿼리 보강 + skill 강제
        user_ctx = next((d for d in (a2a_data or []) if "user_id" in d), None)
        if user_ctx:
            query = f"[사용자 {user_ctx['user_id']}] {query}"
            skill_id = skill_id or "personalized"

        # ───────── ② LLM 호출 + 가드레일 분기 ─────────
        # 가드레일 차단은 GuardrailBlockedError 로 raise 됩니다 — try/except 로 분기.
        # 잡지 않으면 SDK 기본 동작 (state:"failed" + GUARDRAIL_BLOCKED 응답) 으로 그대로 위임.
        try:
            raw = await self._wrapped.dispatch(
                query, skill_id=skill_id, thread_id=thread_id,
                a2a_files=a2a_files, a2a_data=a2a_data, context=context,
            )
        except GuardrailBlockedError:
            # 차단 시 안내 응답으로 교체 (postprocess 우회)
            return self._inject_artifact_metadata(
                RuntimeOutput(text="요청 내용이 정책에 위반되어 처리할 수 없습니다.", data=[])
            )

        # ───────── ③ LLM 호출 후 (post) ─────────
        # 정상 응답 — postprocess 에 위임 (조건부 재호출은 거기서 처리)
        result = await self.postprocess(
            raw, query=query, a2a_files=a2a_files,
            a2a_data=a2a_data, context=context,
        )
        return self._inject_artifact_metadata(result)

    async def postprocess(self, output, *, query, a2a_files=None, a2a_data=None, context=None):
        """1차 응답 후처리 — 필요 시 2차 호출 주입."""
        payload = self.extract_json(output)
        if payload.get("needsEnrichment"):
            # 2차 호출로 데이터 보강
            enriched = await self._wrapped.dispatch(
                f"다음을 상세화: {payload}",
                a2a_files=a2a_files, a2a_data=a2a_data, context=context,
            )
            payload["detail"] = self.extract_text(enriched)

        return RuntimeOutput(
            text=payload.get("summary", "결과를 생성했습니다."),
            data=[payload],
            files=list(a2a_files or []),  # 첨부 패스스루
        )

각 단계에 어떤 로직을 두면 좋은지:

단계	흔한 use case
① Pre (LLM 호출 전)	입력 검증, 짧은 쿼리/특정 의도에 대한 LLM 우회, query/skill_id/context 보강, 라우팅 강제
② LLM 호출	`self._wrapped.dispatch(...)` — 보통 한 번. 라우팅이나 분기는 ①에서 결정.
③ Post (LLM 호출 후)	가드레일 차단 분기, 응답 검증, postprocess 위임

→ 단순 응답 변환은 postprocess 한 곳에. dispatch 오버라이드는 ①·③ 처럼 LLM 호출 자체와 엮이는 분기가 필요할 때.

stream() 를 직접 구현하는 경우

기본 RuntimeAdapter.stream() 은 _wrapped.dispatch() → postprocess() → 단일 청크 yield 로 강등됩니다. 실제 토큰 스트리밍이 필요하면 async for 로 직접 yield 하세요:

async def stream(self, query, *, skill_id=None, thread_id=None,
                 a2a_files=None, a2a_data=None, context=None) -> AsyncIterator[Any]:
    async for chunk in self._wrapped.stream(
        query, skill_id=skill_id, thread_id=thread_id,
        a2a_files=a2a_files, a2a_data=a2a_data, context=context,
    ):
        # 청크 가공 (가드레일, 토큰 필터 등)
        # 마지막 청크에만 artifact 메타 주입
        yield self._inject_artifact_metadata(chunk)

A2A 메시지 스트리밍 요청 매칭

message/send 요청은 dispatch() 로, message/stream 요청은 stream() 으로 라우팅됩니다. message/stream 요청인데 stream() 이 구현 안 되어 있거나 _wrapped.stream 이 없으면 SDK 가 자동으로 dispatch() 경로로 폴백합니다 (단일 최종 응답만 전달).

Flow 노드용 DX 헬퍼

런타임 어댑터가 아닌 Flow(멀티 에이전트) 노드에서 streaming/비스트리밍을 자동 분기하는 call_llm / call_agent_auto 헬퍼는 플로우 상태와 헬퍼 가이드에서 다룹니다 — state.metadata.is_stream_request 기반으로 .stream()/.invoke() (또는 call_agent / call_agent_stream_result) 를 자동 선택해 1줄 호출로 축약합니다 (v0.2.0+).

세 가지 경로 비교

축	Level 1 (`SchemaValidated`)	Level 2 (`StructuredOutputAgent`)	Level 4 (`RuntimeAdapter` 직접)
주 오버라이드	`payload_schema` + `format_summary`	`extract_payload` + `build_summary`	`dispatch` / `postprocess` / `stream`
검증	Pydantic 자동	수동 (dict isinstance)	직접 구현
스트리밍	`dispatch` 강등 (JSON 누설 방지)	`stream_filter_mode` 로 제어	완전 수동
한 어댑터당 LLM 호출 수	1	1	제한 없음 (2~N)
보일러플레이트	최소 (~20줄)	중간 (~40줄)	많음 (~100줄+)

참고 자료

Registry 기반 에이전트 (Level 1)
입력 선별 (Level 3 — InputContracts)
플로우 — 직렬 (flow-seq) — call_agent_auto 실사용 예시
SDK 소스:
- src/llamon_agent/core/runtime/adapter.py — RuntimeAdapter 베이스 (~227줄)
- src/llamon_agent/core/runtime/structured.py — StructuredOutputAgent (~600줄, stream filter 포함)
- src/llamon_agent/core/runtime/schema_validated.py — SchemaValidatedRuntimeAdapter (~430줄)

RuntimeAdapter 고급

레벨 한눈에 보기

Level 2: StructuredOutputAgent

언제 선택하나

최소 예시

오버라이드 포인트

주의사항

skip_llm 최적화 — SchemaValidatedRuntimeAdapter + passthrough 시 LLM 호출 우회

Level 4: RuntimeAdapter 직접 상속

언제 선택하나

베이스 클래스 구조

dispatch / invoke / stream — 셋의 차이

최소 예시 — LLM 호출 전 입력 보강 + 후 가드레일 분기

stream() 를 직접 구현하는 경우

A2A 메시지 스트리밍 요청 매칭

Flow 노드용 DX 헬퍼

세 가지 경로 비교

참고 자료

Level 2: `StructuredOutputAgent`

`skip_llm` 최적화 — `SchemaValidatedRuntimeAdapter` + passthrough 시 LLM 호출 우회

Level 4: `RuntimeAdapter` 직접 상속