플로우 템플릿

flow-seq·flow-parallel·flow-route는 기본 구조이고, flow-http는 외부 API를 연결한 실용 예제입니다. 네 템플릿의 파일 구조와 노드 계약은 같습니다.

구분	템플릿	선택할 때	핵심 모양
기본 구조	`flow-seq`	앞 결과를 다음 단계가 차례로 처리	직렬
기본 구조	`flow-parallel`	같은 입력을 여러 작업이 동시에 처리	병렬 분기·병합
기본 구조	`flow-route`	앞 결과에 따라 다음 경로가 달라짐	조건 분기
실용 예제	`flow-http`	외부 API 결과를 가공한 뒤 Agent가 정리	HTTP + 병렬 변환

# Registry 에이전트 기반
uv run llamon flow my-flow --template flow-seq --yes

# Code-first 모델 기반
uv run llamon flow my-flow --template flow-seq --runtime-source local --yes

flow-seq, flow-parallel, flow-route는 --runtime-source local을 지원합니다. Registry 모드는 A2A URL을 확인하고 로컬 모드는 프로세스 안의 Agent를 호출합니다.

관측·평가 예제 추가

네 템플릿 모두 기존 노드 함수를 유지한 채 관측·평가 예제를 덧붙일 수 있습니다.

uv run llamon flow my-flow \
  --template flow-seq \
  --example observability \
  --yes

이 옵션은 관측 설정을 활성화하고 종료 노드에 evaluators=[output_completeness, output_ready]를 연결합니다. 또한 app/evaluation.py·비교 스크립트·계약 테스트를 추가하되 노드 로직과 배포 설정은 유지합니다. Runtime evaluator는 노드 반환값을 유지하면서 child evaluation observation과 대상 노드의 Numeric score를 기록합니다. 검증 방법은 계약 테스트와 운영에서 확인하세요.

공통 파일

파일	역할
`app/config.py`	여러 노드가 공유하는 에이전트 ID와 기본값
`app/nodes.py`	에이전트 호출, 업무 로직, 병합 함수
`app/graph.py`	`GraphBuilder`로 노드와 edge 연결
`app/agent_card.py`	외부에 공개할 카드와 스킬
`main.py`	서버 조립. 보통 수정하지 않음

생성된 nodes.py에는 핵심 계약만 남깁니다. 전체 예제는 문서에 있습니다.

A2A 데이터 선별: 입력 선별
state 검증·타입 좁히기: 필수 State 값
환경 변수 배치: 프로젝트 설정 구조

Registry 노드는 URL을 미리 확인한 뒤 클로저로 감싸 등록합니다. 노드가 읽고 쓰는 state, output, messages 규칙은 플로우 작성 규칙을 먼저 확인하세요.

직렬 `flow-seq`

START → agent_a → business_logic → agent_b → END

앞 단계의 output을 바로 다음 단계가 읽습니다. 입력 정리 → 조회 → 최종 답변처럼 순서가 고정된 작업에 맞습니다.

return (
    GraphBuilder()
    .node("agent_a", _agent_a, node_kind="registry_node")
    .node("business_logic", business_logic, node_kind="business")
    .node("agent_b", _agent_b, node_kind="registry_node")
    .edge(START, "agent_a")
    .edge("agent_a", "business_logic")
    .edge("business_logic", "agent_b")
    .edge("agent_b", END)
    .build()
)

중간 노드는 extract_latest_text()나 extract_latest_output_data()로 직전 결과를 읽습니다. 종료 노드인 agent_b에서 call_agent_auto()를 사용하면 message/send와 message/stream을 자동으로 구분합니다.

병렬 `flow-parallel`

                   ┌→ agent_a ─┐
START → prepare ───┤           ├→ merge → END
                   └→ agent_b ─┘

같은 입력을 여러 관점에서 동시에 처리할 때 씁니다. 분기 노드는 output을 함께 갱신하지 말고 messages에 source를 붙이세요. merge가 출처별 결과를 모아 최종 output을 한 번만 작성합니다.

return (
    GraphBuilder()
    .node("prepare", prepare, node_kind="business")
    .node("agent_a", _agent_a, node_kind="registry_node")
    .node("agent_b", _agent_b, node_kind="registry_node")
    .node("merge", merge, node_kind="merge")
    .edge(START, "prepare")
    .edge("prepare", "agent_a").edge("prepare", "agent_b")
    .edge("agent_a", "merge").edge("agent_b", "merge")
    .edge("merge", END)
    .build()
)

Python merge가 종료점이면 완성된 artifact만 반환합니다. 합친 결과를 LLM이 설명해야 한다면 merge → final_agent → END를 추가하세요.

조건 분기 `flow-route`

START → agent_a → route ─┬→ agent_b → END
                         └→ finalize → END

입력이나 앞 노드의 결과에 따라 다음 경로가 달라질 때 선택합니다. 라우팅 함수의 반환값은 conditional_edge()의 키와 정확히 같아야 합니다.

builder = GraphBuilder()
builder.node("agent_a", _agent_a, node_kind="registry_node")
builder.node("agent_b", _agent_b, node_kind="registry_node")
builder.node("finalize", finalize, node_kind="business")
builder.edge(START, "agent_a")
builder.conditional_edge(
    "agent_a",
    route_after_a,
    {"agent_b": "agent_b", "finalize": "finalize"},
)
builder.edge("agent_b", END)
builder.edge("finalize", END)
return builder.build()

경로를 늘릴 때는 노드 등록, 라우팅 반환값, 키와 노드의 대응표, END edge를 함께 추가하세요. 하나라도 빠지면 특정 입력에서만 끊기는 그래프가 됩니다.

HTTP 파이프라인 `flow-http`

START → http_fetch ─┬→ transform_summary ─┐
                    └→ transform_flags ───┴→ merge → final_agent → END

외부 API 응답을 구조화 데이터로 저장하고 여러 변환을 병렬로 수행한 뒤 Agent가 최종 문장을 만들 때 사용합니다.

return (
    GraphBuilder()
    .node("http_fetch", http_fetch, node_kind="http")
    .node("transform_summary", transform_summary, node_kind="transform")
    .node("transform_flags", transform_flags, node_kind="transform")
    .node("merge", merge, node_kind="merge")
    .node("final_agent", _final_agent, node_kind="registry_node")
    .edge(START, "http_fetch")
    .edge("http_fetch", "transform_summary")
    .edge("http_fetch", "transform_flags")
    .edge("transform_summary", "merge")
    .edge("transform_flags", "merge")
    .edge("merge", "final_agent")
    .edge("final_agent", END)
    .build()
)

HTTP 노드는 시간 제한을 지정하고 raise_for_status()로 실패를 드러내세요. 응답 원문은 output_data에 넣고 변환 노드는 helper로 읽습니다. 병렬 변환의 결과는 다른 병렬 플로우처럼 source를 붙여 합칩니다.

공통으로 확인할 것

종료 노드만 사용자에게 토큰을 스트리밍합니다.
미등록 에이전트가 필수라면 raise_on_pending=True로 조용한 빈 결과를 막습니다.
DataPart·FilePart를 자식에게 넘겨야 하는 조합 노드는 forward_inbound_data=True, forward_inbound_files=True를 명시합니다.
DB 연결을 여러 번 사용하면 호출마다 연결하지 말고 공유 풀 또는 SQLAlchemy를 사용합니다.
헬퍼의 검색 순서와 옵션은 플로우 헬퍼 API에서 확인합니다.

플로우 템플릿

관측·평가 예제 추가

공통 파일

직렬 flow-seq

병렬 flow-parallel

조건 분기 flow-route

HTTP 파이프라인 flow-http

공통으로 확인할 것

직렬 `flow-seq`

병렬 `flow-parallel`

조건 분기 `flow-route`

HTTP 파이프라인 `flow-http`