플로우 — 조건 분기 (flow-route)

# Registry 에이전트 기반 (기본)
uv run llamon flow my-flow --template flow-route --yes

# 로컬 LLM 기반
uv run llamon flow my-flow --template flow-route --runtime-source local --yes

--runtime-source registry(기본): 노드가 create_registry_node() 기반으로 동작합니다. --runtime-source local: 노드가 Agent(model=..., provider=...) 기반 로컬 LLM으로 동작합니다.

이 페이지는 입력 조건에 따라 다른 에이전트나 다른 경로로 분기하고 싶을 때 보면 됩니다.

토폴로지

START → router → agent_a → business_logic → END
               → agent_b → business_logic → END

router에서 질의 키워드를 분석해 agent_a 또는 agent_b로 분기하고, 공통 business_logic을 거쳐 종료합니다.

파일 구조

순서	파일	역할
①	`app/config.py`	에이전트 UUID 상수
②	`app/nodes.py`	노드 함수 + 라우팅 함수 + 비즈니스 로직
③	`app/graph.py`	노드 연결 (GraphBuilder + conditional_edge)
④	`app/agent_card.py`	카드 정보·스킬
—	`main.py`	진입점 (수정 불필요)

① app/config.py

AGENT_A_ID = "<YOUR_FLOW_AGENT_ID>"  # [수정]
AGENT_B_ID = "<YOUR_FLOW_AGENT_ID>"  # [수정]

② app/nodes.py

router_node은 상태를 변경하지 않고 통과만 합니다. route_by_query()의 반환값이 conditional_edge 매핑 키와 일치해야 분기가 동작합니다.

agent_a(call_agent)와 agent_b(call_agent_auto)는 의도적으로 두 헬퍼를 모두 보여줍니다 — call_agent_auto는 state를 받아 스트리밍 여부를 자동 분기합니다. 최종 응답을 내보내는 노드라면 call_agent_auto(state=state)를 권장합니다.

from llamon_agent import AIMessage, runtime_output_text
from llamon_agent.graph import build_agent_context, call_agent_auto, extract_query, extract_latest_text
from llamon_agent.a2a import call_agent


async def router_node(_state) -> dict:
    return {}


# [수정] 라우팅 함수 — 반환값은 graph.py의 conditional_edge 키와 일치해야 합니다
def route_by_query(state) -> str:
    query = extract_query(state).lower()
    business_keywords = ("사업", "매출", "수익", "market", "business")
    return "agent_b" if any(kw in query for kw in business_keywords) else "agent_a"


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


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


# 비즈니스 로직 — 분기 결과 후처리
async def business_logic(state) -> dict:
    agent_result = extract_latest_text(state)
    return {"messages": [AIMessage(content=agent_result)], "output": agent_result}

라우팅 함수 작성 규칙

반환값은 문자열이며 conditional_edge 딕셔너리의 키와 정확히 일치해야 합니다
state에서 원하는 정보(query, metadata 등)를 읽고 분기 키를 반환합니다
분기 경로를 추가하려면 반환값 + conditional_edge 매핑 + 노드를 함께 추가하세요

③ app/graph.py

from llamon_agent.config import Settings
from llamon_agent.graph import START, END, GraphBuilder
from llamon_agent.a2a import LLaMONRegistryClient
from app.config import AGENT_A_ID, AGENT_B_ID
from app.nodes import router_node, route_by_query, agent_a_node, agent_b_node, business_logic


async def build_graph(settings: Settings):
    registry = LLaMONRegistryClient(host=settings.LLAMON_REGISTRY_HOST)
    url_a = await registry.resolve_url(AGENT_A_ID)
    url_b = await registry.resolve_url(AGENT_B_ID)

    async def _agent_a(state): return await agent_a_node(state, agent_a_url=url_a)
    async def _agent_b(state): return await agent_b_node(state, agent_b_url=url_b)

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

스트리밍 동작

message/stream 요청 시 router/분기 전 노드는 sync로 실행됩니다. Registry/LLM을 호출하는 exit 경로에서는 call_agent_auto / call_llm이 스트리밍 여부를 자동 분기합니다. business_logic처럼 Python 후처리만 하는 exit 경로는 완성된 output을 최종 artifact로 반환합니다.

동작 원리와 구현 코드는 플로우 공통 패턴 — 스트리밍을 참고하세요.

커스터마이징

분기 경로 추가

config.py에 AGENT_C_ID 추가
nodes.py에 agent_c_node 작성
route_by_query()에 "agent_c" 반환 조건 추가
graph.py에서:
- .node("agent_c", _agent_c, ...) 추가
- conditional_edge 매핑에 "agent_c": "agent_c" 추가
- .edge("agent_c", "business_logic") 추가