입력 선별 (InputContracts)

이 문서에서 할 일

인바운드 A2A 데이터 중 원하는 스키마·형태에 정확히 맞는 페이로드만 결정적으로 선별합니다. 용어가 낯설면 용어집 을 참고하세요.

언제 필요한가

외부 에이전트가 DataPart 를 2개 이상 보내는 환경에서 다음 패턴이 필요할 때 사용합니다.

• 단일 선별 — 여러 DataPart 중 특정 shape 한 개만 골라쓰기 (예: intent, profile, history 가 함께 와도 profile 만 사용)
• 복수 선택 — A, B, C 중 원하는 것만 동시에 추출 (아래 “여러 계약 동시 선별” 참조)
• 폴백 체인 (우선순위 기반) — A 있으면 A, 없으면 B, 그것도 없으면 원본으로 폴백 (아래 “Fallback chain (priority)” 참조)

세 패턴 모두 같은 InputContracts 선언으로 표현 가능하며, 사용 API 만 다릅니다. 텍스트 질의만 받는 에이전트라면 이 페이지를 건너뛰어도 됩니다.

이 페이지는 A2A DataPart / FilePart를 **결정적(deterministic)**으로 선별하는 InputContracts의 사용법을 다룹니다.

기본 원칙

• InputContracts는 선택 사항 — 선언하지 않아도 extract_a2a_data(state)가 모든 DataPart 전체 리스트를 그대로 반환 (SDK가 자동으로 잘라내지 않음)
• 계약을 선언하면 SDK가 shape scoring으로 들어온 DataPart 중 가장 적합한 것을 선별
• 매치가 없거나 모호하면 None 반환 → 사용자 코드에서 extract_a2a_data() 로 폴백 처리
• 파일 선별: Flow 노드는 select_input_file(state, contracts, name), Structured 어댑터는 result.matched_payload(name, kind="file")
• 매칭 실패/모호 원인 진단: explain_input_selection(state) (아래 매칭 디버깅 참고)

접근 패턴

extract_a2a_data(state)는 리스트를 그대로 돌려줍니다. 관행적으로 raw[0]으로 첫 번째만 꺼내 쓰는 코드가 많지만, SDK가 첫 번째만 노출하는 것이 아닙니다. 여러 개를 모두 쓰려면 리스트 전체를 순회하세요.

---

동작 방식

1. InputContracts에 기대하는 데이터 shape(필드명, 타입, 배열 아이템 필수키)와 파일 필터(mime_types, name_suffixes)를 선언
2. select_input_contracts()가 들어온 DataPart와 FilePart를 각각 shape scoring으로 비교
3. 유일한 최고점 후보가 있으면 해당 payload 반환
4. 같은 점수의 후보가 여러 개면 모호 → None
5. match가 없으면 → None
6. contracts가 선언된 경우 결과는 state["metadata"]["input_matches"]에 캐시 → explain_input_selection(state)로 조회

---

계약 선언

계약 선언 예시

from llamon_agent import InputContracts, DataContract, DataFieldContract

INPUT_CONTRACTS = InputContracts(
    data=[
        DataContract(
            name="validation_result",
            description="문서 검증 결과 payload",  # 문서화용 (런타임 선별에 미사용)
            fields=[
                DataFieldContract(
                    key="documents",
                    value_type="array",
                    item_required_keys=["documentType", "parsedData", "isValid"],
                )
            ],
        )
    ]
)

DataFieldContract 옵션

필드	타입	기본값	의미
key	str	(필수)	payload에서 찾을 필드명
value_type	str	"any"	any, string, number, integer, boolean, object, array
required	bool	True	이 필드가 반드시 있어야 하는지
item_required_keys	list[str]	[]	value_type="array"일 때 각 아이템에 반드시 있어야 하는 키

필드 매칭 동작

payload 상태	required=True	required=False
키 부재 OR 값이 null	계약 전체 실패	skip (다른 필드 평가 계속)
값 있음 + 타입 일치	+20	+5
값 있음 + 타입 불일치	계약 전체 실패	계약 전체 실패
array 검증 (item_required_keys) 통과	위에 +10	위에 +10

• 점수: schema +100 > 필수 +20 > array item +10 > 옵셔널 +5. 동점은 ambiguous=True 로 자동 선택 포기.
• 모든 필드가 optional + payload 에 한 개도 없으면 계약 자체 실패 (안전장치).
• null ≡ 키 부재: SDK 가 payload.get(key) is None 으로 합쳐 처리. “값이 의도적으로 비어있음” 을 표현하려면 빈 문자열 같은 sentinel 권장.

DataContract 옵션

필드	타입	기본값	의미
name	str	(필수)	계약 이름 — matched_payload(name)으로 결과 조회
schema	str	""	exact schema 이름 매칭 (있으면 fields보다 우선)
description	str	""	문서화용 설명 (런타임 선별에 미사용)
required	bool	False	계약 수준 예약 필드 (직렬화만 보존, 매칭 로직 미사용 — 필드 수준 필수 여부는 DataFieldContract.required 가 담당)
priority	int	0	폴백 체인 우선순위 — select_input_data_priority() 가 내림차순 순회 (v0.2.1+)
fields	list	[]	shape 매칭용 필드 규칙

FileContract 옵션

필드	타입	기본값	의미
name	str	(필수)	계약 이름 — matched_payload(name, kind="file") 으로 결과 조회
description	str	""	문서화용 설명 (런타임 선별에 미사용)
required	bool	False	계약 수준 예약 필드 (DataContract 와 동일)
priority	int	0	폴백 체인 우선순위 — select_input_file_priority() 가 내림차순 순회 (v0.2.1+)
mime_types	list[str]	[]	정확 매칭할 mime-type 화이트리스트
name_suffixes	list[str]	[]	파일명 접미사 화이트리스트 (예: .pdf)

---

템플릿별 사용법

Structured 에이전트 (runtime_adapter.py)

StructuredOutputAgent는 self.a2a_data로 A2A 구조화 데이터에 접근합니다.

app/runtime_adapter.py

from llamon_agent import StructuredOutputAgent, InputContracts, DataContract, DataFieldContract
from llamon_agent.input_contracts import select_input_contracts

INPUT_CONTRACTS = InputContracts(
    data=[
        DataContract(
            name="validation_result",
            fields=[
                DataFieldContract(key="documents", value_type="array",
                    item_required_keys=["documentType", "parsedData", "isValid"]),
            ],
        )
    ]
)

class MyAgent(StructuredOutputAgent):
    def _select_input(self) -> dict | None:
        """self.a2a_data에서 contract에 맞는 payload를 선별합니다."""
        result = select_input_contracts(
            data_parts=self.a2a_data or [],
            contracts=INPUT_CONTRACTS,
        )
        return result.matched_payload("validation_result", kind="data")

    def extract_payload(self, text, data):
        # ① 계약 기반 선별
        selected = self._select_input()
        if selected is not None:
            docs = selected["documents"]
            return {"documents": docs, "count": len(docs)}
        # ② 폴백: LLM 응답에서 추출
        return super().extract_payload(text, data)

    def build_summary(self, payload):
        if payload and "count" in payload:
            return f"{payload['count']}건의 문서를 처리했습니다."
        return super().build_summary(payload)

호출 측 payload 예시 — 위 계약이 어떤 DataPart 를 선택하는지:

외부 에이전트가 보낸 message

{
  "parts": [
    {"kind": "text", "text": "홍길동의 서류 검증 결과를 요약해줘"},
    {"kind": "data", "data": {
      "documents": [
        {"documentType": "INCOME_CERT", "isValid": true, "parsedData": {"name": "홍길동"}},
        {"documentType": "FAMILY_CERT", "isValid": false, "parsedData": {"reason": "expired"}}
      ]
    }},
    {"kind": "data", "data": {
      "sessionMetadata": {"traceId": "abc123", "user": "hong"}
    }}
  ]
}

결과:

• DataPart 1 선택 — documents 필드 (value_type="array") 가 계약과 일치하고, 배열 원소가 item_required_keys=["documentType","parsedData","isValid"] 를 모두 만족합니다.
• DataPart 2 무시 — documents 키 자체가 없어 계약과 무관합니다.
• result.matched_payload("validation_result", kind="data") 는 DataPart 1 의 dict 를 반환합니다.

계약이 매치되지 않거나 동점이면 None 을 반환합니다. 호출 측은 extract_payload 등에서 폴백 처리하면 됩니다.

Flow 노드 (nodes.py)

Flow에서는 select_input_data(state, contracts, name)를 사용합니다. graph state에서 직접 꺼내줍니다.

app/nodes.py

from llamon_agent import InputContracts, DataContract, DataFieldContract
from llamon_agent.graph import select_input_data, extract_a2a_data

INPUT_CONTRACTS = InputContracts(
    data=[
        DataContract(
            name="validation_result",
            fields=[
                DataFieldContract(key="documents", value_type="array",
                    item_required_keys=["documentType", "parsedData", "isValid"]),
            ],
        )
    ]
)

async def business_logic(state) -> dict:
    # ① 계약 기반 선별
    selected = select_input_data(state, INPUT_CONTRACTS, "validation_result")
    if selected is not None:
        docs = selected["documents"]
        # ... 비즈니스 로직
    else:
        # ② 폴백 — 원본은 전체 DataPart 리스트. 여기서는 첫 번째만 관행적으로 사용.
        raw = extract_a2a_data(state)
        docs = raw[0].get("documents", []) if raw else []
        # 여러 DataPart를 모두 쓰려면: [item.get("documents", []) for item in raw]
    # ...

General 에이전트

General 템플릿은 사용자 코드가 A2A 입력에 접근하는 접점이 없습니다. 여러 DataPart가 오면 SDK가 텍스트로 직렬화하여 LLM에 전달하고, LLM이 자율 판단합니다.

결정적 선별이 필요하면 Structured 또는 Flow 템플릿을 사용하세요.

---

여러 계약 동시 선별

계약을 여러 개 선언하면 각각 독립적으로 매칭합니다.

여러 계약 선언

INPUT_CONTRACTS = InputContracts(
    data=[
        DataContract(
            name="validation_result",
            fields=[
                DataFieldContract(key="documents", value_type="array",
                    item_required_keys=["documentType", "parsedData", "isValid"]),
            ],
        ),
        DataContract(
            name="applicant_info",
            fields=[
                DataFieldContract(key="applicantInfo", value_type="object"),
            ],
        ),
    ]
)

각각 독립 조회

result = select_input_contracts(data_parts=a2a_data, contracts=INPUT_CONTRACTS)

docs    = result.matched_payload("validation_result", kind="data")  # DataPart 1
profile = result.matched_payload("applicant_info", kind="data")     # DataPart 2

호출 측 payload 예시:

{
  "parts": [
    {"kind": "text", "text": "홍길동의 서류를 검증하고 자격을 판정해줘"},
    {"kind": "data", "data": {
      "documents": [
        {"documentType": "INCOME_CERT", "isValid": true, "parsedData": {"name": "홍길동"}}
      ]
    }},
    {"kind": "data", "data": {
      "applicantInfo": {"name": "홍길동", "age": 35}
    }},
    {"kind": "data", "data": {
      "sessionMetadata": {"traceId": "abc123"}
    }}
  ]
}

결과: validation_result → DataPart 1 선택, applicant_info → DataPart 2 선택, sessionMetadata는 어떤 contract에도 안 걸려 무시.

---

Fallback chain (priority)

여러 버전의 스키마를 동시에 받을 때 “v2 있으면 v2, 없으면 v1, 그것도 없으면 원본” 식으로 우선순위 폴백이 필요하면 priority 를 부여합니다 (v0.2.1+).

from llamon_agent import InputContracts, DataContract, DataFieldContract

INPUT_CONTRACTS = InputContracts(
    data=[
        DataContract(name="modern", schema="user.v2", priority=10, fields=[
            DataFieldContract(key="permissions", value_type="object"),
            DataFieldContract(key="preferences", value_type="object"),
        ]),
        DataContract(name="legacy", schema="user.v1", priority=5, fields=[
            DataFieldContract(key="is_admin", value_type="boolean"),
        ]),
        DataContract(name="raw_id", priority=0, fields=[
            DataFieldContract(key="id", value_type="string"),  # schema 없이 id 필드만으로 매칭 (기본 폴백)
        ]),
    ],
)

매칭 규칙

schema= 가 지정된 계약은 payload data 안에 "schema" 키가 있고 그 값이 정확히 일치해야 매칭됩니다. 그 위에서 fields 의 required 필드(기본 required=True)가 모두 존재하고 타입이 맞아야 합니다. value_type 은 string·number·integer·boolean·object·array·any 중 하나입니다.

Flow 노드

from llamon_agent.graph import select_input_data_priority

payload = select_input_data_priority(state, INPUT_CONTRACTS)
# priority 내림차순으로 시도 → 첫 매치 반환, 모두 실패하면 None

• 동순위는 선언 순서대로 시도 (stable sort).
• 파일 버전: select_input_file_priority(state, contracts).
• 매칭 평가 자체는 모든 계약에 대해 수행되어 explain_input_selection() 진단 캐시에 그대로 남습니다.
• priority=0 은 기본값 — 기존 계약 선언은 변경 없이 동작.

Fallback 시나리오 매트릭스

스키마 진화 (v2 → v1 → 원본) 시점별로 외부 에이전트가 보내는 DataPart 가 달라질 때, 하나의 InputContracts 선언으로 모든 케이스 대응:

Case A — v2 클라이언트 (modern 선택)

{
  "parts": [
    {"kind": "text", "text": "..."},
    {"kind": "data", "data": {
      "schema": "user.v2", "name": "홍길동",
      "permissions": {"admin": true},
      "preferences": {"theme": "dark"}
    }}
  ]
}

→ modern (priority=10) 매치 → payload 반환, legacy/raw_id 평가 안 됨.

Case B — v1 클라이언트 (legacy 폴백)

{
  "parts": [
    {"kind": "text", "text": "..."},
    {"kind": "data", "data": {
      "schema": "user.v1", "name": "홍길동",
      "is_admin": true
    }}
  ]
}

→ modern 매치 실패 (v2 schema 불만족) → legacy (priority=5) 매치 → payload 반환.

Case C — 구버전 클라이언트 (raw_id 최종 폴백)

{
  "parts": [
    {"kind": "text", "text": "..."},
    {"kind": "data", "data": {
      "id": "user-12345"
    }}
  ]
}

→ modern/legacy 모두 매치 실패 → raw_id (priority=0) 매치 → payload 반환.

Case D — 어떤 계약도 매치 안 됨

{
  "parts": [
    {"kind": "data", "data": {"sessionId": "xyz"}}
  ]
}

→ 모든 계약 평가 실패 → select_input_data_priority() 가 None 반환 → 호출 측에서 폴백 처리.

호출 코드는 4 케이스 모두 동일:

payload = select_input_data_priority(state, INPUT_CONTRACTS)
if payload is None:
    # Case D — 모든 contract 실패
    ...
else:
    # Case A/B/C — 가장 우선순위 높은 매치
    process(payload)

---

엣지 케이스

상황	결과
DataPart가 비어있음	모든 계약이 None
한쪽만 매칭됨	매칭된 것만 payload, 나머지 None
같은 shape의 DataPart가 2개	ambiguous=True → None (자동 선택 안 함)
하나의 DataPart에 두 계약 필드 모두 있음	양쪽 계약 모두 같은 DataPart를 선택
계약 선언 안 함	InputContracts() → 기존 동작 그대로

---

매칭 디버깅

계약이 왜 미매칭/모호로 떨어졌는지 확인하려면 explain_input_selection(state) 로 진단 정보를 꺼냅니다.

Flow 노드 안

from llamon_agent.graph import explain_input_selection, select_input_data

async def business_logic(state) -> dict:
    selected = select_input_data(state, INPUT_CONTRACTS, "validation_result")
    if selected is None:
        diagnostic = explain_input_selection(state)
        # {
        #   "data:validation_result": {
        #     "matched": False, "ambiguous": True,
        #     "reason": "ambiguous_multiple_candidates", ...
        #   },
        #   "file:income_cert": {"matched": False, "reason": "no_exact_file_match", ...}
        # }
        ...

키	의미
matched	매칭 성공 여부
ambiguous	동점 후보가 2개 이상이어서 자동 선택 포기했는지
reason	schema, shape, no_exact_schema_or_shape_match, ambiguous_multiple_candidates, no_filter_declared 등
score/index/payload	매칭된 경우에만 채워짐

캐시 조건

진단 캐시(state["metadata"]["input_matches"])는 contracts가 선언된 경우에만 기록됩니다. 선언하지 않은 상태에서 explain_input_selection()을 호출하면 {}가 돌아옵니다.

---

select_input_data vs select_input_contracts

	select_input_data	select_input_contracts
사용 위치	Flow 노드 (nodes.py)	Structured 어댑터 (runtime_adapter.py)
입력	state dict	data_parts / file_parts list
import	from llamon_agent.graph import select_input_data	from llamon_agent.input_contracts import select_input_contracts
내부 동작	state에서 a2a_data + a2a_files 추출 → select_input_contracts() 호출 → 결과를 state["metadata"]["input_matches"]에 캐시	직접 매칭 (캐시 없음)

select_input_data()는 select_input_contracts()의 graph-level 래퍼입니다. 매칭 로직(shape scoring, ambiguity 판정)은 동일합니다. 파일용은 select_input_file()을 사용합니다.

---

파일 선별 (FileContract)

DataContract 외에 FileContract로 A2A FilePart도 선별할 수 있습니다.

from llamon_agent import InputContracts, FileContract

INPUT_CONTRACTS = InputContracts(
    files=[
        FileContract(
            name="income_cert",
            description="소득증명서 PDF",
            mime_types=["application/pdf"],
            name_suffixes=[".pdf"],
        )
    ]
)

필드	의미
mime_types	허용할 MIME 타입 목록
name_suffixes	파일명 접미사 필터 (.pdf, .xlsx 등)

mime_types와 name_suffixes 모두 비어있으면 매칭 실패합니다. 최소 하나는 선언해야 합니다.

---

Type narrowing — select_input_data_as

select_input_data() 반환 타입은 dict[str, Any] | None — pyright/mypy strict 에서 키 접근이 Any 로 처리되어 자동 narrowing 안 됩니다. 도메인 TypedDict 로 type 안전성을 원하면 select_input_data_as() 를 옵트인합니다.

from typing import TypedDict
from llamon_agent.graph import select_input_data_as

class OrderSummary(TypedDict):
    line_items: list[dict]
    currency: str

order = select_input_data_as(state, CONTRACTS, "order_summary", OrderSummary)
if order is not None:
    items = order["line_items"]   # type: list[dict] (narrowed)

Runtime 동작은 select_input_data 와 완전 동일 (typing.cast 만 추가, 필드 검증 X — caller 책임). 기존 caller 는 무영향 — 신규 함수만 추가됐습니다.

---

관련 문서

• 에이전트 구성: Registry 기반 에이전트
• 플로우 공통 패턴: 플로우 공통 패턴
• 확정적 MCP 호출: deterministic_tool — 입력 분기 후 특정 tool을 코드로 확정 호출하고 싶을 때