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

Agents Course

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

워크플로우 만들기

기본 워크플로우 생성

여러 단계 연결하기

루프와 분기

워크플로우 그리기

상태 관리

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