LlamaIndex Workflow 완벽 가이드

LlamaIndex의 워크플로 시스템을 활용하여 복잡한 RAG 파이프라인을 구축하는 방법을 알아봅니다. 이벤트 기반 워크플로부터 멀티 인덱스 쿼리까지 단계별로 학습합니다.

---

목차

1. LlamaIndex_워크플로_엔진
2. 쿼리_엔진_구축
3. 실습_인덱스와_워크플로_통합
4. 실습_멀티_인덱스_쿼리

---

1. LlamaIndex 워크플로 엔진

김개발 씨는 회사에서 RAG 시스템을 구축하라는 미션을 받았습니다. 여러 개의 문서를 검색하고, 결과를 종합해서 답변을 생성해야 하는 복잡한 요구사항이었습니다.

"이걸 어떻게 깔끔하게 구조화하지?" 고민하던 중, 선배가 LlamaIndex의 Workflow를 추천해주었습니다.

Workflow는 LlamaIndex에서 제공하는 이벤트 기반 오케스트레이션 시스템입니다. 마치 공장의 컨베이어 벨트처럼 각 단계가 순서대로 연결되어 데이터가 흘러갑니다.

복잡한 RAG 파이프라인을 깔끔한 스텝 단위로 분리할 수 있어 유지보수가 훨씬 쉬워집니다.

다음 코드를 살펴봅시다.

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

# Workflow 클래스를 상속하여 커스텀 워크플로 정의
class MyWorkflow(Workflow):

    # @step 데코레이터로 워크플로 단계 정의
    @step
    async def process_query(self, ev: StartEvent) -> StopEvent:
        # StartEvent에서 쿼리 추출
        query = ev.query

        # 쿼리 처리 로직 (여기서는 간단히 변환)
        result = f"처리된 쿼리: {query}"

        # StopEvent로 결과 반환하며 워크플로 종료
        return StopEvent(result=result)

# 워크플로 실행
workflow = MyWorkflow()
result = await workflow.run(query="LlamaIndex란 무엇인가요?")

김개발 씨는 입사 6개월 차 백엔드 개발자입니다. 최근 회사에서 고객 문의에 자동으로 답변하는 AI 시스템을 만들라는 프로젝트를 맡게 되었습니다.

수천 개의 문서에서 관련 정보를 찾아 답변을 생성해야 하는 꽤 복잡한 시스템이었습니다. 처음에 김개발 씨는 모든 로직을 하나의 함수에 작성했습니다.

문서 검색, 컨텍스트 구성, LLM 호출, 답변 생성까지 수백 줄의 코드가 뒤엉켜 있었습니다. 코드가 점점 복잡해지자 어디서 문제가 생겼는지 파악하기도 어려워졌습니다.

선배 개발자 박시니어 씨가 코드를 보더니 고개를 저었습니다. "이렇게 하면 나중에 유지보수가 힘들어요.

Workflow 패턴을 써보는 게 어때요?" 그렇다면 Workflow란 정확히 무엇일까요? 쉽게 비유하자면, Workflow는 마치 요리 레시피와 같습니다.

재료 손질, 볶기, 끓이기, 담기처럼 각 단계가 명확하게 구분되어 있습니다. 한 단계가 끝나면 다음 단계로 자연스럽게 넘어갑니다.

Workflow도 마찬가지로 각 스텝이 독립적으로 정의되고, 이벤트를 통해 데이터가 전달됩니다. LlamaIndex의 Workflow는 세 가지 핵심 개념으로 구성됩니다.

첫째는 Event입니다. 이벤트는 스텝 간에 데이터를 전달하는 메시지입니다.

StartEvent로 워크플로가 시작되고, StopEvent로 종료됩니다. 둘째는 Step입니다.

@step 데코레이터가 붙은 메서드가 하나의 처리 단계가 됩니다. 셋째는 Workflow 클래스 자체로, 모든 스텝을 담는 컨테이너 역할을 합니다.

위의 코드를 자세히 살펴보겠습니다. MyWorkflow 클래스는 Workflow를 상속받아 만들어졌습니다.

process_query 메서드에 @step 데코레이터가 붙어있는데, 이것이 이 메서드가 워크플로의 한 단계임을 나타냅니다. 메서드는 StartEvent를 받아서 처리한 후 StopEvent를 반환합니다.

주목할 점은 모든 스텝 메서드가 async로 정의된다는 것입니다. LlamaIndex Workflow는 비동기 기반으로 설계되어 있어 여러 작업을 동시에 처리할 수 있습니다.

이는 실무에서 성능 향상에 큰 도움이 됩니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 고객 문의 시스템을 만든다고 가정해봅시다. 첫 번째 스텝에서는 질문을 분석하고, 두 번째 스텝에서는 관련 문서를 검색하고, 세 번째 스텝에서는 답변을 생성합니다.

각 스텝이 독립적이므로 특정 스텝만 수정하거나 교체하기가 쉽습니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수는 모든 로직을 하나의 스텝에 몰아넣는 것입니다. 이러면 Workflow를 사용하는 의미가 없어집니다.

각 스텝은 하나의 명확한 책임만 가지도록 설계해야 합니다. 김개발 씨는 박시니어 씨의 조언대로 코드를 리팩토링했습니다.

복잡하게 얽혀있던 로직이 깔끔한 스텝들로 분리되니 코드 가독성이 훨씬 좋아졌습니다. 버그가 발생해도 어느 스텝에서 문제인지 바로 파악할 수 있게 되었습니다.

실전 팁

💡 - 각 스텝은 단일 책임 원칙을 따라 하나의 작업만 수행하도록 설계하세요

• 커스텀 이벤트를 정의하여 스텝 간에 복잡한 데이터를 전달할 수 있습니다
• 디버깅을 위해 각 스텝에서 로깅을 추가하면 데이터 흐름을 추적하기 쉽습니다

---

2. 쿼리 엔진 구축

Workflow의 기본 개념을 익힌 김개발 씨는 이제 본격적으로 문서 검색 기능을 구현해야 했습니다. 단순히 데이터를 전달하는 것이 아니라, 실제로 인덱스에서 검색하고 LLM으로 답변을 생성하는 QueryEngine을 Workflow 안에 통합해야 했습니다.

QueryEngine은 LlamaIndex의 핵심 컴포넌트로, 인덱스에서 관련 문서를 검색하고 LLM을 통해 답변을 생성합니다. Workflow와 결합하면 검색, 처리, 응답 생성을 체계적으로 관리할 수 있습니다.

마치 도서관 사서가 책을 찾아 요약해주는 것과 같습니다.

다음 코드를 살펴봅시다.

from llama_index.core.workflow import Workflow, StartEvent, StopEvent, step, Context
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

class RAGWorkflow(Workflow):

    @step
    async def setup_engine(self, ctx: Context, ev: StartEvent) -> StopEvent:
        # 문서 로드
        documents = SimpleDirectoryReader("./data").load_data()

        # 벡터 인덱스 생성
        index = VectorStoreIndex.from_documents(documents)

        # QueryEngine 생성
        query_engine = index.as_query_engine()

        # 쿼리 실행 및 결과 반환
        response = query_engine.query(ev.query)

        return StopEvent(result=str(response))

김개발 씨는 Workflow의 기본 구조를 이해했지만, 아직 갈 길이 멀었습니다. 실제로 문서에서 정보를 찾아 답변을 생성하는 기능이 필요했기 때문입니다.

박시니어 씨가 옆에서 말했습니다. "이제 QueryEngine을 연결할 차례야." QueryEngine이란 무엇일까요?

쉽게 비유하자면, 도서관의 사서와 같습니다. 손님이 "파이썬 입문서 있나요?"라고 물으면, 사서는 서가를 뒤져 관련 책들을 찾아옵니다.

그리고 책의 핵심 내용을 요약해서 알려주기도 합니다. QueryEngine도 마찬가지로 질문을 받으면 인덱스에서 관련 문서를 검색하고, LLM을 통해 답변을 생성합니다.

QueryEngine을 사용하려면 먼저 인덱스가 필요합니다. 인덱스는 문서들을 검색 가능한 형태로 저장한 것입니다.

LlamaIndex에서 가장 흔히 사용하는 것은 VectorStoreIndex입니다. 이 인덱스는 문서를 벡터로 변환하여 의미적으로 유사한 내용을 빠르게 찾을 수 있게 해줍니다.

위 코드에서는 SimpleDirectoryReader로 ./data 폴더의 문서들을 읽어옵니다. 이 Reader는 PDF, TXT, DOCX 등 다양한 형식의 파일을 자동으로 인식하여 로드합니다.

그 다음 VectorStoreIndex.from_documents()로 인덱스를 생성합니다. 이 과정에서 각 문서가 임베딩 벡터로 변환됩니다.

인덱스가 준비되면 as_query_engine() 메서드로 QueryEngine을 생성합니다. 이 엔진은 내부적으로 검색기(Retriever)와 응답 합성기(Response Synthesizer)를 포함하고 있습니다.

검색기는 관련 문서 조각을 찾고, 응답 합성기는 LLM을 호출하여 최종 답변을 만듭니다. 코드에서 주목할 또 다른 점은 Context입니다.

Workflow에서 Context는 스텝 간에 데이터를 공유하는 저장소 역할을 합니다. 현재 코드에서는 단일 스텝이라 직접 사용하지 않았지만, 여러 스텝으로 나눌 때 매우 유용합니다.

예를 들어 첫 번째 스텝에서 인덱스를 생성하고 Context에 저장하면, 이후 스텝에서 꺼내 사용할 수 있습니다. 실무에서는 인덱스 생성을 매번 하지 않습니다.

문서가 많으면 인덱스 생성에 시간이 오래 걸리기 때문입니다. 대신 인덱스를 한 번 생성한 후 디스크에 저장해두고, 필요할 때 불러와서 사용합니다.

LlamaIndex는 storage_context를 통해 인덱스 저장과 로드를 지원합니다. 흔히 하는 실수 중 하나는 QueryEngine의 기본 설정을 그대로 사용하는 것입니다.

실제 서비스에서는 검색할 문서 개수(similarity_top_k), 응답 모드(response_mode), 사용할 LLM 등을 상황에 맞게 조정해야 합니다. 기본값이 항상 최적은 아닙니다.

김개발 씨는 QueryEngine을 Workflow에 통합하고 테스트해보았습니다. 질문을 입력하니 문서에서 관련 내용을 찾아 깔끔한 답변이 생성되었습니다.

"오, 제대로 동작하네!" 작은 성공에 기뻐하는 김개발 씨였습니다.

실전 팁

💡 - 인덱스는 한 번 생성 후 저장해두고 재사용하면 성능이 크게 향상됩니다

• similarity_top_k 파라미터로 검색할 문서 개수를 조절할 수 있습니다
• streaming=True 옵션으로 실시간 스트리밍 응답을 받을 수 있습니다

---

3. 실습 인덱스와 워크플로 통합

이론은 충분히 배웠습니다. 이제 김개발 씨는 실제로 동작하는 RAG 워크플로를 처음부터 끝까지 구현해보기로 했습니다.

문서 로딩, 인덱스 생성, 검색, 응답 생성까지 각 단계를 별도의 스텝으로 분리하여 체계적인 파이프라인을 구축하는 것이 목표입니다.

실제 프로덕션 환경에서는 RAG 파이프라인의 각 단계를 명확히 분리해야 합니다. 문서 전처리, 인덱싱, 검색, 응답 생성을 각각의 스텝으로 나누면 디버깅이 쉬워지고 특정 단계만 교체하기도 용이합니다.

마치 레고 블록처럼 조립하고 분해할 수 있는 구조를 만드는 것입니다.

다음 코드를 살펴봅시다.

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

# 커스텀 이벤트 정의
class IndexReadyEvent(Event):
    index: VectorStoreIndex

class QueryResultEvent(Event):
    result: str

class ProductionRAGWorkflow(Workflow):

    @step
    async def create_index(self, ctx: Context, ev: StartEvent) -> IndexReadyEvent:
        # 문서 로드 및 인덱스 생성
        docs = SimpleDirectoryReader(ev.data_path).load_data()
        index = VectorStoreIndex.from_documents(docs)
        await ctx.set("query", ev.query)
        return IndexReadyEvent(index=index)

    @step
    async def execute_query(self, ctx: Context, ev: IndexReadyEvent) -> StopEvent:
        query = await ctx.get("query")
        engine = ev.index.as_query_engine(similarity_top_k=3)
        response = engine.query(query)
        return StopEvent(result=str(response))

박시니어 씨가 김개발 씨에게 새로운 과제를 주었습니다. "지금까지는 모든 걸 하나의 스텝에 넣었잖아.

이번에는 각 단계를 분리해서 더 체계적으로 만들어봐." 김개발 씨는 고개를 끄덕였습니다. 이전에 작성한 코드는 동작은 했지만, 모든 로직이 한 곳에 몰려 있어서 테스트하기도 어렵고 수정하기도 까다로웠습니다.

이제 각 단계를 독립적인 스텝으로 분리할 때가 되었습니다. 먼저 커스텀 이벤트를 정의해야 합니다.

기본 제공되는 StartEvent와 StopEvent만으로는 복잡한 데이터를 전달하기 어렵습니다. 위 코드에서 IndexReadyEvent는 생성된 인덱스를 담아 다음 스텝으로 전달합니다.

이렇게 하면 각 스텝이 어떤 데이터를 주고받는지 명확해집니다. 커스텀 이벤트를 만드는 방법은 간단합니다.

Event 클래스를 상속받고, 전달할 데이터를 타입 힌트와 함께 선언하면 됩니다. IndexReadyEvent의 경우 index라는 필드에 VectorStoreIndex를 담도록 정의했습니다.

이 타입 정보 덕분에 IDE에서 자동완성도 지원받을 수 있습니다. 이제 워크플로의 흐름을 살펴봅시다.

첫 번째 스텝인 create_index는 StartEvent를 받습니다. 이 이벤트에서 data_path와 query를 추출합니다.

문서를 로드하고 인덱스를 생성한 뒤, IndexReadyEvent를 반환합니다. 반환된 이벤트는 자동으로 다음 스텝에 전달됩니다.

여기서 Context의 역할이 중요합니다. query는 첫 번째 스텝에서 받지만, 실제로 사용하는 것은 두 번째 스텝입니다.

이럴 때 Context에 저장해두면 다른 스텝에서 꺼내 쓸 수 있습니다. await ctx.set("query", ev.query)로 저장하고, await ctx.get("query")로 가져옵니다.

두 번째 스텝 execute_query는 IndexReadyEvent를 받습니다. 이벤트에서 인덱스를 꺼내고, Context에서 쿼리를 가져와서 검색을 수행합니다.

as_query_engine에 similarity_top_k=3을 전달하여 가장 관련성 높은 3개의 문서 조각만 사용하도록 설정했습니다. LlamaIndex Workflow의 강력한 점 중 하나는 자동 라우팅입니다.

스텝이 반환하는 이벤트 타입을 보고 자동으로 해당 이벤트를 받을 수 있는 다음 스텝을 찾아 실행합니다. 개발자가 명시적으로 연결을 지정할 필요가 없습니다.

만약 하나의 이벤트를 여러 스텝이 받을 수 있다면 어떻게 될까요? 모든 해당 스텝이 동시에 실행됩니다.

이를 활용하면 병렬 처리도 자연스럽게 구현할 수 있습니다. 예를 들어 하나의 쿼리에 대해 여러 인덱스를 동시에 검색하는 것도 가능합니다.

김개발 씨는 분리된 스텝들을 보며 뿌듯함을 느꼈습니다. 이제 인덱스 생성 로직만 수정하고 싶다면 create_index 스텝만 건드리면 됩니다.

쿼리 엔진 설정을 바꾸고 싶다면 execute_query만 수정하면 됩니다. 마치 레고 블록을 조립하듯이 유연한 구조가 완성되었습니다.

실전 팁

💡 - 커스텀 이벤트에 타입 힌트를 명확히 지정하면 IDE 지원을 받을 수 있습니다

• Context는 워크플로 전체에서 공유되므로 키 이름이 충돌하지 않도록 주의하세요
• 복잡한 워크플로는 시각화하여 데이터 흐름을 확인하는 것이 좋습니다

---

4. 실습 멀티 인덱스 쿼리

김개발 씨의 프로젝트가 점점 복잡해졌습니다. 이제는 하나의 인덱스로는 부족했습니다.

제품 문서, FAQ, 기술 블로그 등 여러 종류의 문서를 각각 다른 인덱스로 관리하고, 질문에 따라 적절한 인덱스를 선택하거나 여러 인덱스를 동시에 검색해야 했습니다.

멀티 인덱스 쿼리는 여러 개의 인덱스를 동시에 또는 선택적으로 검색하는 기술입니다. 대규모 시스템에서는 문서 종류별로 인덱스를 분리하여 관리하는 경우가 많습니다.

Workflow를 활용하면 이런 복잡한 검색 로직도 깔끔하게 구현할 수 있습니다.

다음 코드를 살펴봅시다.

class MultiIndexEvent(Event):
    indices: dict  # {"product": index1, "faq": index2, "blog": index3}

class MultiIndexWorkflow(Workflow):

    @step
    async def load_indices(self, ev: StartEvent) -> MultiIndexEvent:
        # 각 카테고리별 인덱스 로드 (실제로는 저장된 인덱스 로드)
        indices = {
            "product": VectorStoreIndex.from_documents(
                SimpleDirectoryReader("./data/product").load_data()
            ),
            "faq": VectorStoreIndex.from_documents(
                SimpleDirectoryReader("./data/faq").load_data()
            ),
        }
        return MultiIndexEvent(indices=indices)

    @step
    async def parallel_search(self, ctx: Context, ev: MultiIndexEvent) -> StopEvent:
        query = await ctx.get("query")
        results = {}

        # 모든 인덱스에서 병렬 검색
        for name, index in ev.indices.items():
            engine = index.as_query_engine()
            results[name] = str(engine.query(query))

        # 결과 종합
        combined = "\n\n".join([f"[{k}] {v}" for k, v in results.items()])
        return StopEvent(result=combined)

프로젝트가 성장하면서 김개발 씨는 새로운 도전에 직면했습니다. 처음에는 모든 문서를 하나의 인덱스에 넣었지만, 문서가 수만 개로 늘어나자 문제가 생겼습니다.

제품 매뉴얼과 FAQ와 블로그 글이 뒤섞여 검색 품질이 떨어진 것입니다. 박시니어 씨가 해결책을 제시했습니다.

"문서 종류별로 인덱스를 나눠봐. 그리고 질문에 따라 적절한 인덱스를 검색하거나, 필요하면 여러 인덱스를 동시에 검색해." 이것이 바로 멀티 인덱스 쿼리 패턴입니다.

마치 큰 도서관이 주제별로 서가를 나누는 것과 같습니다. 소설 코너, 과학 코너, 역사 코너가 따로 있으면 원하는 책을 더 빨리 찾을 수 있습니다.

위 코드에서는 indices를 딕셔너리로 관리합니다. 키는 인덱스의 이름이고 값은 실제 VectorStoreIndex 객체입니다.

이렇게 하면 인덱스를 쉽게 추가하거나 제거할 수 있습니다. 새로운 문서 카테고리가 생기면 딕셔너리에 항목만 추가하면 됩니다.

load_indices 스텝에서는 각 카테고리의 문서를 읽어 별도의 인덱스를 생성합니다. 실제 프로덕션 환경에서는 매번 인덱스를 생성하지 않고, 미리 저장해둔 인덱스를 로드합니다.

인덱스 저장소로는 로컬 디스크, Redis, 또는 Pinecone 같은 벡터 데이터베이스를 사용할 수 있습니다. parallel_search 스텝이 핵심입니다.

모든 인덱스를 순회하며 동일한 쿼리로 검색합니다. 현재 코드는 순차적으로 실행되지만, asyncio.gather를 사용하면 실제로 병렬 실행할 수 있습니다.

대규모 시스템에서는 이런 병렬화가 응답 시간을 크게 단축시킵니다. 검색 결과를 종합하는 방법도 다양합니다.

가장 단순한 방법은 위 코드처럼 모든 결과를 이어붙이는 것입니다. 더 정교한 방법으로는 각 결과의 관련성 점수를 비교하여 가장 높은 것만 선택하거나, LLM을 한 번 더 호출하여 여러 결과를 하나의 답변으로 합성할 수 있습니다.

때로는 모든 인덱스를 검색할 필요가 없습니다. 질문의 종류에 따라 특정 인덱스만 선택하고 싶을 수 있습니다.

이럴 때는 라우터를 사용합니다. LlamaIndex는 RouterQueryEngine을 제공하는데, 질문을 분석하여 가장 적절한 인덱스를 자동으로 선택합니다.

김개발 씨는 멀티 인덱스 시스템을 구축한 후 검색 품질이 눈에 띄게 향상되었음을 확인했습니다. 제품 관련 질문에는 제품 인덱스에서 정확한 정보가, FAQ 질문에는 FAQ 인덱스에서 명쾌한 답변이 나왔습니다.

더 이상 엉뚱한 블로그 글이 검색 결과에 끼어들지 않았습니다. 마지막으로 성능 최적화 팁을 알려드립니다.

인덱스가 많아지면 메모리 사용량이 늘어납니다. 자주 사용하지 않는 인덱스는 필요할 때만 로드하는 지연 로딩(lazy loading) 전략을 고려하세요.

또한 캐싱을 적용하면 반복되는 쿼리에 대해 빠른 응답이 가능합니다.

실전 팁

💡 - asyncio.gather를 사용하면 여러 인덱스 검색을 진정한 병렬로 실행할 수 있습니다

• RouterQueryEngine으로 질문 유형에 따라 자동으로 적절한 인덱스를 선택할 수 있습니다
• 인덱스 수가 많아지면 지연 로딩과 캐싱 전략을 고려하세요

---

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!