플로우 헬퍼 API

이 페이지는 함수 사전입니다. state와 병렬 병합이 낯설다면 플로우 작성 규칙을 먼저 읽으세요.

from llamon_agent.graph import (
    build_agent_context,
    call_agent_auto,
    call_llm,
    extract_a2a_data,
    extract_a2a_files,
    extract_latest_output_data,
    extract_latest_text,
    extract_output_files,
    extract_query,
    parse_raw_json,
    require_state_list,
    require_state_str,
    resolve_file_bytes,
)

`GraphBuilder.node()` 평가 옵션

builder.node(
    "answer",
    answer_node,
    evaluators=[groundedness],
    evaluator_timeout_seconds=120.0,
)

옵션	기본값	계약
`evaluators`	`None`	runtime evaluator 목록. node observation 자동 생성
`evaluator_timeout_seconds`	`90.0`	evaluator별 제한 시간. `None`은 무제한

평가는 노드 성공 후 observation 종료 전에 실행하며 오류를 기록하고 출력을 유지합니다. 함수·built-in 작성법과 score topology는 Runtime Evaluator Framework을 참고하세요.

State와 Part 추출

함수	반환	선택 기준
`extract_query(state)`	`str`	이번 호출의 사용자 원문
`extract_latest_text(state)`	`str`	직전 노드의 텍스트
`extract_latest_output_data(state)`	`list[dict]`	직전 노드의 구조화 결과
`extract_a2a_data(state)`	`list[dict]`	요청에 들어온 DataPart
`extract_a2a_files(state)`	`list[dict]`	요청에 첨부된 FilePart
`extract_output_files(state)`	`list[dict]`	직전 노드가 만든 FilePart

extract_query()는 마지막 HumanMessage를 먼저 보고 없으면 state["query"]를 읽습니다. extract_latest_text()는 output_text → 최근 message → query 순으로 찾습니다. extract_latest_output_data()도 현재 output을 먼저 보고 message를 거슬러 올라갑니다.

요청 Part와 노드 산출물을 구분하세요. extract_a2a_files()는 사용자가 보낸 파일, extract_output_files()는 앞 노드가 만든 파일입니다. 여러 후보 가운데 하나를 고를 때는 InputContracts를 사용합니다.

파일 바이트

resolve_file_bytes(file_dict)는 bytes_b64 → data: URI → file:// URI → 로컬 경로 순서로 읽습니다. 실패하면 ValueError나 FileNotFoundError를 올립니다.

HTTP·HTTPS URI는 자동으로 내려받지 않습니다. 원격 파일은 앱이 소유한 httpx.AsyncClient로 받되 시간 제한, 허용 호스트, 크기 제한을 적용한 뒤 바이트를 처리하세요.

Registry Agent 호출

build_agent_context(state)는 사용자 질문과 앞 노드의 텍스트·데이터를 하나의 프롬프트로 합칩니다. 외부 A2A 에이전트를 부를 때 call_agent_auto()와 함께 쓰면 message/send와 message/stream을 자동으로 고릅니다.

from llamon_agent import AIMessage, runtime_output_text
from llamon_agent.graph import build_agent_context, call_agent_auto


async def final_agent_node(state, *, agent_url: str) -> dict:
    output = await call_agent_auto(
        agent_url,
        build_agent_context(state),
        state=state,
    )
    return {
        "messages": [AIMessage(content=runtime_output_text(output))],
        "output": output,
    }

주요 옵션은 다음과 같습니다.

옵션	기본값	켤 때
`forward_inbound_data`	`False`	요청 DataPart를 자식이 그대로 읽어야 할 때
`forward_inbound_files`	`False`	요청 FilePart를 자식에게 넘길 때
`forward_inbound_metadata`	`False`	SDK 내부 키까지 필요한 특수한 경우
`forward_session`	`True`	기본 유지. 자식에게 관측 ID를 숨길 때만 `False`
`raise_on_pending`	`False`	필수 자식이 미등록이면 즉시 실패시킬 때
`stream`	`None`	`None` 자동, `True` 강제 stream, `False` 강제 send

forward_session=True는 sessionId, userId, workflow·agent·chatbot ID와 이름만 선별해 전달합니다. forward_inbound_metadata처럼 내부 metadata 전체를 보내지 않습니다. 호출자가 metadata=에 같은 키를 넣으면 그 값이 우선합니다.

내부 처리용 자식 응답을 사용자 스트림에 노출하고 싶지 않다면 stream=False를 지정합니다. raise_on_pending=True는 미등록 자식의 빈 sentinel 대신 UPSTREAM_UNAVAILABLE을 반환합니다.

Code-first 모델 호출

call_llm(agent, prompt, *, state=None, on_text_chunk=None, text_extractor=None) -> str은 프로세스 안의 LLM 객체를 호출합니다. state.metadata.is_stream_request가 참이면 stream을 순회하고 아니면 invoke 경로를 사용합니다.

외부 에이전트에는 call_agent_auto(), 로컬 모델에는 call_llm()을 쓰세요. 같은 프로세스의 LLM 토큰은 LangGraph 메시지 채널로 흐르므로 SDK가 별도 writer로 중복 전송하지 않습니다.

반환 텍스트는 기본 추출기 runtime_output_text가 뽑습니다. 이 추출기는 순수 JSON 응답을 구조화 data로 정규화하고 빈 문자열을 돌려주므로, JSON 원문이 그대로 필요하면(예: LLM judge 판정 파싱) text_extractor=를 바꾸세요.

raw = await call_llm(
    judge_agent,
    prompt,
    state=state,
    text_extractor=lambda value: value if isinstance(value, str) else str(value),
)
verdict = parse_raw_json(raw, fallback={})

parse_raw_json(raw, *, fallback=None)은 dict·list는 그대로 통과시키고 문자열만 JSON으로 파싱합니다. 실패하면 예외 대신 fallback(없으면 원본)을 반환합니다.

필수 State 값

필수 metadata를 직접 isinstance로 검사하는 대신 require_state_str()과 require_state_list()를 사용합니다.

metadata = state.get("metadata", {}) or {}
order_id = require_state_str(
    metadata,
    "orderId",
    reason="order_id_required",
    message="주문 ID가 누락되었습니다.",
)
items = require_state_list(metadata, "items")

reason을 생략하면 state_field_required 또는 state_field_invalid_type을 사용합니다. 오류에는 field, expectedType, receivedType이 함께 기록됩니다. 반환 타입이 좁혀지므로 TypedDict와 엄격한 타입 검사에도 그대로 쓸 수 있습니다.