#
LlamaIndex에서 에이전트 워크플로우 만들기

LlamaIndex의 워크플로우는 코드를 순차적이고 관리 가능한 단계로 구성하는 구조화된 방법을 제공합니다.

이러한 워크플로우는 `Events`에 의해 트리거되는 `Steps`를 정의하여 생성되며, 이들 자체가 추가 단계를 트리거하기 위해 `Events`를 발생시킵니다. RAG 작업을 위한 LlamaIndex 워크플로우를 보여주는 알프레드를 살펴보겠습니다.

워크플로우는 몇 가지 주요 이점을 제공합니다:

- 코드를 개별 단계로 명확하게 구성
- 유연한 제어 흐름을 위한 이벤트 기반 아키텍처
- 단계 간 타입 안전 통신
- 내장 상태 관리
- 간단하고 복잡한 에이전트 상호작용 모두 지원

추측하셨을 수 있듯이, 워크플로우는 전체 워크플로우에 대한 제어를 유지하면서 에이전트의 자율성 사이에서 훌륭한 균형을 제공합니다.

그럼 우리만의 워크플로우를 만드는 방법을 배워보겠습니다!

##
워크플로우 만들기

> 이 노트북을 따라가며 Google Colab에서 코드를 실행해볼 수 있습니다.

###
기본 워크플로우 생성

워크플로우 패키지 설치
LlamaHub 섹션에서 소개된 대로, 다음 명령어로 워크플로우 패키지를 설치할 수 있습니다:

    pip install llama-index-utils-workflow

`Workflow`를 상속하는 클래스를 정의하고 함수에 `@step` 데코레이터를 사용하여 단일 단계 워크플로우를 만들 수 있습니다. 또한 워크플로우의 시작과 끝을 나타내는 특별한 이벤트인 `StartEvent`와 `StopEvent`를 추가해야 합니다.

    from llama_index.core.workflow import StartEvent, StopEvent, Workflow, step

    class MyWorkflow(Workflow):
        @step
        async def my_step(self, ev: StartEvent) -> StopEvent:
            # 여기서 무언가 수행
            return StopEvent(result="안녕하세요, 세계!")

    w = MyWorkflow(timeout=10, verbose=False)
    result = await w.run()

보시다시피, 이제 `w.run()`을 호출하여 워크플로우를 실행할 수 있습니다.

###
여러 단계 연결하기

여러 단계를 연결하기 위해 단계 간에 데이터를 전달하는 사용자 정의 이벤트를 만듭니다. 이를 위해 단계 간에 전달되고 첫 번째 단계의 출력을 두 번째 단계로 전송하는 `Event`를 추가해야 합니다.

    from llama_index.core.workflow import Event

    class ProcessingEvent(Event):
        intermediate_result: str

    class MultiStepWorkflow(Workflow):
        @step
        async def step_one(self, ev: StartEvent) -> ProcessingEvent:
            # 초기 데이터 처리
            return ProcessingEvent(intermediate_result="1단계 완료")

        @step
        async def step_two(self, ev: ProcessingEvent) -> StopEvent:
            # 중간 결과 사용
            final_result = f"처리 완료: {ev.intermediate_result}"
            return StopEvent(result=final_result)

    w = MultiStepWorkflow(timeout=10, verbose=False)
    result = await w.run()
    result

타입 힌팅은 워크플로우가 올바르게 실행되도록 보장하므로 여기서 중요합니다. 조금 더 복잡하게 만들어보겠습니다!

###
루프와 분기

타입 힌팅은 워크플로우의 가장 강력한 부분입니다. 더 복잡한 워크플로우를 용이하게 하기 위해 분기, 루프, 조인을 만들 수 있게 해주기 때문입니다.

합집합 연산자 `|`를 사용하여 루프를 만드는 예제를 보여드리겠습니다. 아래 예제에서 `LoopEvent`가 단계의 입력으로 사용되고 출력으로도 반환될 수 있음을 볼 수 있습니다.

    from llama_index.core.workflow import Event
    import random

    class ProcessingEvent(Event):
        intermediate_result: str

    class LoopEvent(Event):
        loop_output: str

    class MultiStepWorkflow(Workflow):
        @step
        async def step_one(self, ev: StartEvent | LoopEvent) -> ProcessingEvent | LoopEvent:
            if random.randint(0, 1) == 0:
                print("나쁜 일이 발생했습니다")
                return LoopEvent(loop_output="1단계로 돌아갑니다.")
            else:
                print("좋은 일이 발생했습니다")
                return ProcessingEvent(intermediate_result="첫 번째 단계 완료.")

        @step
        async def step_two(self, ev: ProcessingEvent) -> StopEvent:
            # 중간 결과 사용
            final_result = f"처리 완료: {ev.intermediate_result}"
            return StopEvent(result=final_result)

    w = MultiStepWorkflow(verbose=False)
    result = await w.run()
    result

###
워크플로우 그리기

워크플로우를 그릴 수도 있습니다. `draw_all_possible_flows` 함수를 사용하여 워크플로우를 그려보겠습니다. 이는 워크플로우를 HTML 파일에 저장합니다.

    from llama_index.utils.workflow import draw_all_possible_flows

    w = ... # 이전 섹션에서 정의된 대로
    draw_all_possible_flows(w, "flow.html")

강의에서 다룰 마지막 멋진 트릭이 있는데, 그것은 워크플로우에 상태를 추가하는 능력입니다.

###
상태 관리

상태 관리는 워크플로우의 상태를 추적하고 싶을 때 유용하므로, 모든 단계가 동일한 상태에 접근할 수 있습니다. 단계 함수의 매개변수 위에 `Context` 타입 힌팅을 사용하여 이를 수행할 수 있습니다.

    from llama_index.core.workflow import Context, StartEvent, StopEvent

    @step
    async def query(self, ctx: Context, ev: StartEvent) -> StopEvent:
        # 컨텍스트에 쿼리 저장
        await ctx.store.set("query", "프랑스의 수도는 무엇인가요?")

        # 컨텍스트와 이벤트로 무언가 수행
        val = ...

        # 컨텍스트에서 쿼리 검색
        query = await ctx.store.get("query")

        return StopEvent(result=val)

훌륭합니다! 이제 LlamaIndex에서 기본 워크플로우를 만드는 방법을 알게 되었습니다!

> 워크플로우에는 더 복잡한 뉘앙스가 있으며, LlamaIndex 문서에서 배울 수 있습니다.

하지만 `AgentWorkflow` 클래스에 의존하는 워크플로우를 만드는 다른 방법이 있습니다. 이를 사용하여 다중 에이전트 워크플로우를 만드는 방법을 살펴보겠습니다.

##
다중 에이전트 워크플로우로 워크플로우 자동화

수동 워크플로우 생성 대신 `AgentWorkflow` 클래스를 사용하여 다중 에이전트 워크플로우를 만들 수 있습니다. `AgentWorkflow`는 워크플로우 에이전트를 사용하여 전문화된 기능을 기반으로 협업하고 서로에게 작업을 넘길 수 있는 하나 이상의 에이전트 시스템을 만들 수 있게 해줍니다. 이를 통해 서로 다른 에이전트가 작업의 서로 다른 측면을 처리하는 복잡한 에이전트 시스템을 구축할 수 있습니다. `llama_index.core.agent`에서 클래스를 가져오는 대신, `llama_index.core.agent.workflow`에서 에이전트 클래스를 가져올 것입니다. 하나의 에이전트는 `AgentWorkflow` 생성자에서 루트 에이전트로 지정되어야 합니다. 사용자 메시지가 들어오면 먼저 루트 에이전트로 라우팅됩니다.

각 에이전트는 다음을 수행할 수 있습니다:

- 툴을 사용하여 요청을 직접 처리
- 작업에 더 적합한 다른 에이전트에게 작업 넘기기
- 사용자에게 응답 반환

다중 에이전트 워크플로우를 만드는 방법을 살펴보겠습니다.

    from llama_index.core.agent.workflow import AgentWorkflow, ReActAgent
    from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI

    # 몇 가지 툴 정의
    def add(a: int, b: int) -> int:
        """두 숫자를 더합니다."""
        return a + b

    def multiply(a: int, b: int) -> int:
        """두 숫자를 곱합니다."""
        return a * b

    llm = HuggingFaceInferenceAPI(model_name="Qwen/Qwen2.5-Coder-32B-Instruct")

    # FunctionTool 없이 함수를 직접 전달할 수 있습니다 -- 함수/독스트링이 이름/설명을 위해 파싱됩니다
    multiply_agent = ReActAgent(
        name="multiply_agent",
        description="두 정수를 곱할 수 있습니다",
        system_prompt="숫자를 곱하는 툴을 사용할 수 있는 도움이 되는 어시스턴트입니다.",
        tools=[multiply],
        llm=llm,
    )

    addition_agent = ReActAgent(
        name="add_agent",
        description="두 정수를 더할 수 있습니다",
        system_prompt="숫자를 더하는 툴을 사용할 수 있는 도움이 되는 어시스턴트입니다.",
        tools=[add],
        llm=llm,
    )

    # 워크플로우 생성
    workflow = AgentWorkflow(
        agents=[multiply_agent, addition_agent],
        root_agent="multiply_agent",
    )

    # 시스템 실행
    response = await workflow.run(user_msg="5와 3을 더할 수 있나요?")

에이전트 툴은 앞서 언급한 워크플로우 상태도 수정할 수 있습니다. 워크플로우를 시작하기 전에 모든 에이전트가 사용할 수 있는 초기 상태 딕셔너리를 제공할 수 있습니다. 상태는 워크플로우 컨텍스트의 상태 키에 저장됩니다. 각 새로운 사용자 메시지를 보강하는 state_prompt에 주입됩니다.

이전 예제를 수정하여 함수 호출을 세는 카운터를 주입해보겠습니다:

    from llama_index.core.workflow import Context

    # 몇 가지 툴 정의
    async def add(ctx: Context, a: int, b: int) -> int:
        """두 숫자를 더합니다."""
        # 카운트 업데이트
        cur_state = await ctx.store.get("state")
        cur_state["num_fn_calls"] += 1
        await ctx.store.set("state", cur_state)

        return a + b

    async def multiply(ctx: Context, a: int, b: int) -> int:
        """두 숫자를 곱합니다."""
        # 카운트 업데이트
        cur_state = await ctx.store.get("state")
        cur_state["num_fn_calls"] += 1
        await ctx.store.set("state", cur_state)

        return a * b

    ...

    workflow = AgentWorkflow(
        agents=[multiply_agent, addition_agent],
        root_agent="multiply_agent",
        initial_state={"num_fn_calls": 0},
        state_prompt="현재 상태: {state}. 사용자 메시지: {msg}",
    )

    # 컨텍스트와 함께 워크플로우 실행
    ctx = Context(workflow)
    response = await workflow.run(user_msg="5와 3을 더할 수 있나요?", ctx=ctx)

    # 상태를 꺼내서 검사
    state = await ctx.store.get("state")
    print(state["num_fn_calls"])

축하합니다! 이제 LlamaIndex에서 에이전트의 기본을 마스터했습니다! 🎉

지식을 확고히 하기 위해 마지막 퀴즈를 계속해보겠습니다! 🚀

Update on GitHub