콘텐츠로 이동
LLaMON Agent SDK v0.6.0

RuntimeAdapter 고급

고정된 Pydantic 스키마로 충분하면 SchemaValidatedRuntimeAdapter를 쓰세요. 이 페이지는 동적 payload, LLM 없는 실행, 서버 응답 정책처럼 기본 경로를 벗어나는 경우만 다룹니다.

선택 순서

섹션 제목: “선택 순서”
단계선택쓰는 경우
1SchemaValidatedRuntimeAdapter[PayloadT]고정 Pydantic 스키마로 분류·추출·판정
2StructuredOutputAgent키가 런타임마다 달라지는 dict
3SchemaValidated + InputContracts여러 DataPart 가운데 규칙으로 입력 선별
4RuntimeAdapter 직접 상속호출 전후 분기나 여러 모델 호출

단계가 내려갈수록 검증, 스트리밍, 오류 처리를 직접 책임져야 합니다. 필요한 지점까지만 확장하세요.

동적 dict — StructuredOutputAgent

섹션 제목: “동적 dict — StructuredOutputAgent”

Pydantic 모델을 고정하기 어렵다면 payload 추출과 요약만 재정의합니다.

app/runtime_adapter.py
from llamon_agent import StructuredOutputAgent




class MyAgent(StructuredOutputAgent):
    stream_filter_mode = "summary_only"


    def extract_payload(self, text: str, data: list[dict]) -> dict | None:
        payload = super().extract_payload(text, data)
        return payload or {"response": text.strip(), "source": "llm"}


    def build_summary(self, payload: dict | None) -> str:
        if payload and isinstance(payload.get("output_text"), str):
            return payload["output_text"]
        return super().build_summary(payload)

payload_schema는 지정하지 않습니다. 검증이 필요하면 extract_payload()에서 키와 타입을 확인하세요. 가공 전 JSON 토큰을 사용자에게 보여 주지 않으려면 stream_filter_mode="summary_only"를 유지합니다.

LLM 호출 우회

섹션 제목: “LLM 호출 우회”

가드레일 게이트웨이나 입력 전달처럼 주 LLM 출력이 필요 없는 에이전트는 skip_llm=True로 모델 로드와 호출을 건너뜁니다.

app/config.py
from llamon_agent.config import ExtensionConfig, LLMConfig


extension = ExtensionConfig(
    llm=LLMConfig(skip_llm=True),
    artifact_name="validated-input",
)

이 설정에서는 id, 프롬프트, temperature, reasoning, MCP의 자율 도구 선택을 무시합니다. 출력 가드레일은 그대로 실행됩니다. 다만 응답은 RuntimeAdapter.postprocess(), EchoAdapter 또는 별도 비즈니스 로직이 만들어야 합니다. 조립 경로가 없으면 빈 문자열이 반환됩니다.

입력을 검증한 뒤 그대로 돌려주는 서버라면 기본 어댑터로 충분합니다.

main.py
from llamon_agent import EchoAdapter


app = await create_server(
    card=card,
    extension=extension,
    settings=settings,
    agent=EchoAdapter(),
)

프롬프트 기반 가드레일은 별도 심사 모델이 필요합니다. 선택 순서는 GuardrailConfig.judge_llm → Registry Guardrail의 모델 → LLMConfig.id입니다. 셋 다 없으면 정규식 규칙만 실행되고 프롬프트 검사는 경고와 함께 생략됩니다.

Upstream Parts 정책

섹션 제목: “Upstream Parts 정책”

create_server(upstream_parts_policy=...)는 이전 A2A 응답의 DataPart·FilePart를 현재 응답에 남길지 정합니다.

정책현재 결과가 있음현재 결과가 텍스트뿐임
replace 기본값현재 Part만 반환upstream 제외
append현재 Part 뒤에 upstream 추가upstream 그대로 전달
app = await create_server(
    card=card,
    agent=graph,
    upstream_parts_policy="append",
)

현재 결과가 항상 배열의 앞에 오므로 data[0]의 의미는 바뀌지 않습니다. SDK는 중복을 제거하지 않습니다.

InputContracts는 에이전트가 사용할 입력을 고르고 upstream_parts_policy는 최종 응답에 남길 Part를 정합니다. append는 선별 결과가 아니라 원본 upstream 전체를 붙입니다. 워크플로우가 이미 이전 Part를 누적한다면 각 에이전트는 replace를 유지해야 중복이 생기지 않습니다.

RuntimeAdapter 직접 경로에는 이 서버 정책이 자동 적용되지 않으므로, 필요한 병합은 postprocess()에서 명시하세요. 다만 Registry 출력 가드레일이 에이전트를 바깥에서 감싸는 경우에는 원본 upstream이 검사를 우회하지 않도록 그 guard 경계 안에서 append를 처리합니다.

EchoAdapter는 입력 DataPart·FilePart를 이미 current output으로 반환합니다. 따라서 출력 가드레일과 append를 함께 설정해도 SDK가 같은 upstream을 다시 붙이지 않으며, 설정이 중복임을 시작 로그에 한 번 남깁니다. 일반 append는 기존과 같이 current와 upstream을 모두 보존하고 자동 dedupe는 수행하지 않습니다. HITL 재개 경로는 이 정책의 영향을 받지 않습니다.

RuntimeAdapter 직접 구현

섹션 제목: “RuntimeAdapter 직접 구현”

단순 출력 변환은 postprocess() 하나로 끝내는 편이 안전합니다. 호출 전에 입력을 보강하거나 모델을 여러 번 부를 때만 진입 메서드를 재정의하세요.

메서드호출 경로기본 응답
dispatchA2A message/send완성된 단일 응답
streamA2A message/stream청크 스트림
invoke코드에서 직접 agent.invoke()완성된 단일 응답
postprocess세 경로의 공통 후처리RuntimeOutput

dispatch를 재정의해도 stream에는 적용되지 않습니다. 기본 stream()은 내부 runtime을 호출한 뒤 postprocess() 결과를 한 번만 내보내므로 실시간 토큰이 필요하면 별도로 구현해야 합니다.

app/runtime_adapter.py
from llamon_agent import RuntimeAdapter, RuntimeOutput




class MyAgent(RuntimeAdapter):
    async def postprocess(
        self,
        output,
        *,
        query,
        a2a_files=None,
        a2a_data=None,
        context=None,
    ):
        payload = self.extract_json(output)
        return RuntimeOutput(
            text=payload.get("summary", "결과를 생성했습니다."),
            data=[payload],
            files=list(a2a_files or []),
        )

stream()을 직접 구현하면 각 청크에 _inject_artifact_metadata()를 호출해야 Adapter ClassVar metadata가 빠지지 않습니다. 신규 프로젝트의 고정 artifact 이름은 ClassVar보다 ExtensionConfig.artifact_name에 두는 편이 낫습니다.

경계 정리

섹션 제목: “경계 정리”