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

플로우 작성 규칙과 상태

State 한눈에 보기

섹션 제목: “State 한눈에 보기”
필드의미갱신 방식
query이번 호출의 사용자 입력호출 동안 유지
messagesHuman·AI·System 메시지reducer로 누적
output직전 노드가 남긴 결과새 값으로 교체
metadataA2A Part와 요청별 정보필요한 값을 복사해 유지

다음 노드가 읽을 값은 output에 둡니다. 사람에게 보여 줄 문장은 output_text, 코드가 읽을 값은 output_data, 파일은 output_files로 나눕니다.

return {
    "messages": [AIMessage(content=summary)],
    "output": {
        "output_text": summary,
        "output_data": {"order_id": order_id, "status": "created"},
    },
}

output은 누적되지 않습니다. 텍스트만 바꾸면서 기존 DataPart를 보존하려면 이전 dict를 펼쳐 넣으세요.

previous = state.get("output") or {}
if not isinstance(previous, dict):
    previous = {}


return {
    "output": {
        **previous,
        "output_text": "처리했습니다.",
    }
}

output_text만 담은 새 dict를 반환하면 앞 노드의 output_data와 output_files도 사라집니다.

A2A 입력 읽기

섹션 제목: “A2A 입력 읽기”

SDK는 요청의 DataPart와 FilePart를 실행 전에 state["metadata"]로 옮깁니다.

입력헬퍼
사용자 원문extract_query(state)
요청 DataPartextract_a2a_data(state)
요청 FilePartextract_a2a_files(state)
직전 노드 텍스트extract_latest_text(state)
직전 노드 DataPartextract_latest_output_data(state)
직전 노드 FilePartextract_output_files(state)

여러 DataPart·FilePart 중 하나를 고를 때는 순서 대신 InputContracts를 선언하세요. Flow helper는 현재 호출을, 오케스트레이터 상태·기억·재개는 여러 턴을 다룹니다.

최종 응답 규칙

섹션 제목: “최종 응답 규칙”

END에 연결된 종료 노드의 output이 A2A artifact가 됩니다.

output최종 Part
문자열TextPart
{output_text, output_data}TextPart + DataPart
{output_text, output_files}TextPart + FilePart

고정 artifact 이름은 main.py의 ExtensionConfig에 둡니다. 요청마다 이름이 달라야 할 때만 종료 노드의 output에 artifact_name과 artifact_description을 넣어 덮어쓰세요.

출력 스키마 이름 (schema)

섹션 제목: “출력 스키마 이름 (schema)”

최종 Flow 스키마는 main.py의 서버 경계에서 설정합니다.

from app.config import FLOW_OUTPUT_SCHEMA


extension = ExtensionConfig(output_schema=FLOW_OUTPUT_SCHEMA)

노드 간 output_data에는 데이터만 담습니다. SDK는 최종 artifact의 첫 DataPart에 기본 schema를 붙이되 명시된 값은 유지합니다.

환경별 이름은 config.py에서 resolve_env_override("FLOW_OUTPUT_SCHEMA", "my-flow.output.v1", source_file=__file__)로 읽을 수 있습니다. 소비자와 공유하는 버전 계약이라면 코드에 고정하세요.

Agent Card의 .set_capability_metadata(output_schemas=[...])는 discovery용 선언이며 응답 payload를 바꾸지 않습니다.

노드 등록과 실행 규칙

섹션 제목: “노드 등록과 실행 규칙”

node_kind

섹션 제목: “node_kind”

GraphBuilder.node()에는 역할에 맞는 표준 node_kind를 지정합니다. 이 값은 Studio의 노드 역할과 SDK의 기본 관측 유형을 정합니다.

kind역할
registry_nodeRegistry 에이전트 A2A 호출
registry_llmRegistry LLM·Prompt·Guardrail 호출
businessPython 업무 로직과 외부 연동
merge병렬 결과 병합
guardrail입력·출력 검증
http외부 HTTP 호출
postgresPostgreSQL 직접 접근
transform구조화 데이터 변환
llmCode-first 모델 호출

기존 별칭(agent, default, fallback, model, normal, router, tool)도 실행되지만 새 코드에서는 표준 node_kind를 사용하세요.

Registry URL 주입과 클로저 패턴

섹션 제목: “Registry URL 주입과 클로저 패턴”

Registry Agent 호출 노드는 .node(..., agent_id=...)로 등록합니다. builder가 첫 호출 때 URL을 resolve(캐시)해 노드 함수의 *_url 파라미터에 주입하므로 클로저가 필요 없습니다. GraphBuilder(registry=...)가 전제입니다.

from llamon_agent.a2a import LLaMONRegistryClient


registry = LLaMONRegistryClient(host=settings.LLAMON_REGISTRY_HOST)
agent_id = resolve_env_override("AGENT_A_ID", AGENT_A_ID, source_file=__file__)


builder = GraphBuilder(registry=registry)
builder.node("agent", agent_node, node_kind="registry_node", agent_id=agent_id)

Registry URL이 아닌 다른 부팅 시점 값을 keyword-only 인자로 넘길 때는 클로저로 감쌉니다. keyword-only 인자가 있는 함수를 그대로 등록하면 실행할 때 인자가 빠집니다.

async def _report(state):
    return await report_node(state, api_base=api_base)


builder.node("report", _report, node_kind="business")

기존처럼 URL을 클로저로 직접 바인딩한 코드도 계속 동작합니다. Registry 노드와 로컬 Agent 노드는 한 그래프에 섞을 수 있습니다.

Registry LLM 주입 (ext=)

섹션 제목: “Registry LLM 주입 (ext=)”

노드 전용 LLM은 .node(..., node_kind="registry_llm", ext=ExtensionConfig(llm=...))로 등록합니다. builder가 Registry 모델로 노드 전용 에이전트를 만들어(첫 호출 때 생성 후 캐시) 노드 함수의 *_agent 파라미터에 주입합니다. 역시 GraphBuilder(registry=...)가 전제입니다.

async def generate_node(state, *, generate_agent) -> dict:
    text = await call_llm(generate_agent, prompt, state=state)
    ...




answer_ext = ExtensionConfig(llm=LLMConfig(id=model_id, temperature=0.0))
builder.node("generate", generate_node, node_kind="registry_llm", ext=answer_ext)

주입 대상은 _agent로 끝나는 keyword 인자입니다. 없으면 유일한 추가 인자를, 그것도 없으면 <노드이름>_agent를 찾습니다. 받을 인자가 없으면 첫 요청을 기다리지 않고 등록 시점에 ValueError로 실패합니다.

스트리밍 규칙

섹션 제목: “스트리밍 규칙”

중간 노드는 항상 완성된 결과를 내고 END에 연결된 종료 노드만 사용자에게 토큰을 스트리밍합니다. 외부 A2A 에이전트는 call_agent_auto(), Code-first 모델은 call_llm()을 쓰면 state.metadata.is_stream_request를 보고 실행 방식을 고릅니다.

Python merge 노드가 종료점이면 완성된 artifact를 한 번만 반환합니다. 토큰 스트리밍이 필요하다면 마지막 노드를 Agent·LLM 호출 노드로 두세요.

미등록 Agent·MCP ID

섹션 제목: “미등록 Agent·MCP ID”

Registry에 아직 없는 Agent ID는 PendingAgentRef로 남아 서버 기동을 막지 않습니다. 호출할 때 한 번 더 확인하고 여전히 없으면 미해결 상태를 담은 빈 결과를 반환합니다. 필수 자식이라면 call_agent_auto(..., raise_on_pending=True)로 UPSTREAM_UNAVAILABLE 오류를 내세요.

MCP도 일부 서버가 없어도 나머지를 로드합니다. mcp.failures()로 상태를 확인하고 필요하면 try_recheck_pending()을 직접 호출합니다. 자동 재조회는 TTL과 잠금으로 Registry 요청 폭주를 막습니다.

Business 노드에서 MCP 직접 호출

섹션 제목: “Business 노드에서 MCP 직접 호출”

LLM이 도구를 고르게 하지 않고 특정 MCP 도구를 호출하려면 MCPHandle을 한 번 연결한 뒤 노드에서 사용합니다.

from llamon_agent import MCPHandle


mcp = MCPHandle()
await mcp.bind_registry(settings, mcp_ids=["<YOUR_MCP_ID>"], mcp_id="<YOUR_MCP_ID>")


result = await mcp.call("lookup_customer", customer_id="12345")
if result.status != "ok":
    raise ValueError(result.summary)

MCP가 여러 개면 핸들도 나누세요. 연결은 부팅 경로에서 한 번만 하고 노드에서는 call()을 실행합니다.

직렬과 병렬

섹션 제목: “직렬과 병렬”

직렬에서는 바로 앞 output을 다음 노드가 읽습니다. 병렬 분기가 동시에 output을 쓰면 마지막 값만 남습니다. 각 분기는 messages에 source를 표시하고 병합 노드만 output을 한 번 작성합니다.

async def branch_a(state):
    return {
        "messages": [
            AIMessage(content="A 결과", additional_kwargs={"source": "a"})
        ]
    }




async def merge(state):
    results = {}
    for message in state.get("messages", []):
        source = getattr(message, "additional_kwargs", {}).get("source")
        if source:
            results[source] = str(message.content)
    return {
        "output": {
            "output_text": "병렬 결과를 합쳤습니다.",
            "output_data": results,
        }
    }

다음 문서

섹션 제목: “다음 문서”