AI 에이전트와 복잡한 워크플로우에 대한 수요가 급증하면서, 개발자들은 더 정교하고 제어 가능한 AI 애플리케이션을 구축하고자 합니다. 그러나 기존의 도구들은 복잡한 다단계 프로세스를 관리하기에는 한계가 있었죠. 이런 배경에서 LlamaIndex가 Workflows 1.0을 정식 출시했습니다. 이는 AI 에이전트 시스템을 위한 경량 프레임워크로, 개발자가 복잡한 AI 애플리케이션의 실행 흐름을 완전히 제어할 수 있게 해줍니다.
이벤트 기반 아키텍처로 구현하는 AI 워크플로우
LlamaIndex Workflows는 이벤트 기반(event-driven) 아키텍처를 통해 복잡한 AI 애플리케이션을 단계별로 구성할 수 있는 프레임워크입니다. 각 단계(step)는 특정 이벤트 타입을 처리하고 새로운 이벤트를 발생시키는 방식으로 동작합니다.
가장 큰 특징은 비동기 우선(async-first) 설계입니다. 이를 통해 병렬 처리, 복잡한 루프, 그리고 여러 단계에 걸친 상태 관리를 자연스럽게 구현할 수 있습니다. 개발자는 @step 데코레이터를 사용해 함수를 워크플로우 단계로 정의하고, 타입 추론을 통해 입력과 출력을 자동으로 검증받을 수 있습니다.
from llama_index.core.workflow import (
Event, StartEvent, StopEvent, Workflow, step
)
class JokeEvent(Event):
joke: str
class JokeFlow(Workflow):
@step
async def generate_joke(self, ev: StartEvent) -> JokeEvent:
topic = ev.topic
# AI 모델을 사용해 농담 생성
return JokeEvent(joke=generated_joke)
@step
async def critique_joke(self, ev: JokeEvent) -> StopEvent:
# 농담을 분석하고 평가
return StopEvent(result=critique_result)다양한 AI 애플리케이션 구축 가능
Workflows로 구축할 수 있는 애플리케이션의 범위는 상당히 넓습니다.
AI 에이전트와 추론 시스템에서는 여러 단계에 걸쳐 추론하고 결정을 내리며 행동을 취하는 지능형 시스템을 만들 수 있습니다. 문서 처리 파이프라인에서는 문서를 수집하고 분석하며 요약한 후 적절한 경로로 라우팅하는 복합적인 처리 과정을 구현할 수 있습니다.
멀티모달 AI 애플리케이션도 가능합니다. LLM, 비전 모델 등 서로 다른 AI 모델들과 다양한 데이터 형식을 조정하여 복잡한 작업을 해결할 수 있죠. 연구 어시스턴트의 경우 정보를 검색하고 분석하며 종합하여 포괄적인 답변을 제공하는 워크플로우를 구성할 수 있습니다.
특히 역할 기반 멀티 에이전트 시스템에서는 여러 에이전트가 필요에 따라 서로 제어권을 넘겨가며 협업하는 시스템을 구축할 수 있습니다. 콘텐츠 생성 시스템에서는 콘텐츠를 생성하고 검토하며 편집한 후 사람의 승인을 거쳐 발행하는 전체 파이프라인을 만들 수 있습니다.
출처: Unsplash
독립 프레임워크로의 전환과 그 의미
Workflows 1.0의 가장 중요한 변화는 독립적인 패키지로 분리되었다는 점입니다. 이전까지는 llama_index와 LlamaIndexTS의 일부였지만, 이제 전용 패키지로 제공됩니다.
# Python 버전
pip install llama-index-workflows
# TypeScript 버전
npm i @llamaindex/workflow-core이러한 분리는 몇 가지 중요한 의미를 갖습니다. 먼저 LlamaIndex 생태계와의 독립성이 강조됩니다. 이제 Workflows는 Python이나 TypeScript 애플리케이션의 오케스트레이션 로직을 작성하는 범용 프레임워크로 사용할 수 있습니다.
기여와 실험의 장벽도 낮아졌습니다. 이전에 LlamaIndex를 사용하지 않던 환경에서도 Workflows를 쉽게 도입할 수 있게 되었죠. 또한 더 넓은 사용 사례 지원이 가능해졌습니다. 에이전트 시스템부터 작업 오케스트레이션까지, 제한된 종속성으로 다양한 용도에 활용할 수 있습니다.
기존 사용자들은 걱정할 필요가 없습니다. llama_index와 LlamaIndexTS는 계속해서 기존 임포트 경로를 통해 Workflows 기능을 제공하므로, 기존 코드는 그대로 작동하면서 새로운 기능의 이점도 누릴 수 있습니다.
버전 1.0의 새로운 기능들
타입 안전성 강화
Python과 TypeScript 모두에서 타입화된 워크플로우 상태(Typed Workflow State)를 지원합니다. 이를 통해 개발자는 더 안전하고 예측 가능한 코드를 작성할 수 있습니다.
from pydantic import BaseModel, Field
class MyState(BaseModel):
user_query: str = Field(default="")
processed_results: list = Field(default_factory=list)
class MyWorkflow(Workflow):
@step
async def process_query(self, ctx: Context[MyState], ev: StartEvent) -> StopEvent:
state = await ctx.store.get_state()
state.user_query = ev.query
await ctx.store.set_state(state)
return StopEvent(result="완료")동적 리소스 주입
Python 워크플로우에서는 리소스 주입(Resource Injection)을 지원합니다. 데이터베이스 클라이언트나 외부 서비스 등을 동적으로 주입할 수 있어 더 유연한 아키텍처를 구성할 수 있습니다.
from llama_index.core.workflow.resource import Resource
from typing import Annotated
def get_database():
return DatabaseClient()
class MyWorkflow(Workflow):
@step
async def process_data(
self,
ev: StartEvent,
db: Annotated[DatabaseClient, Resource(get_database)]
) -> StopEvent:
# 주입된 데이터베이스 클라이언트 사용
result = await db.query("SELECT * FROM users")
return StopEvent(result=result)관찰 가능성(Observability) 지원
llama-index-instrumentation을 설치하면 OpenTelemetry, Arize Phoenix 등을 통해 워크플로우를 모니터링할 수 있습니다. 이는 프로덕션 환경에서 AI 애플리케이션을 운영할 때 필수적인 기능입니다.
출처: Unsplash
개발자를 위한 실용적 가이드
시작하기
Workflows를 시작하는 것은 간단합니다. 먼저 패키지를 설치하고 기본적인 워크플로우를 작성해보세요.
# 설치
pip install llama-index-workflows
# 기본 워크플로우 작성
from llama_index.core.workflow import Workflow, StartEvent, StopEvent, step
class SimpleWorkflow(Workflow):
@step
async def process(self, ev: StartEvent) -> StopEvent:
result = f"처리된 입력: {ev.input_data}"
return StopEvent(result=result)
# 실행
workflow = SimpleWorkflow()
result = await workflow.run(input_data="테스트 데이터")
print(result)고급 패턴
여러 이벤트 대기하기는 복잡한 워크플로우에서 자주 사용되는 패턴입니다.
@step
async def synthesize(self, ctx: Context, ev: QueryEvent | RetrieveEvent) -> StopEvent | None:
# 두 이벤트가 모두 도착할 때까지 대기
data = ctx.collect_events(ev, [QueryEvent, RetrieveEvent])
if data is None:
return None
query_event, retrieve_event = data
# 두 이벤트의 데이터를 조합하여 처리
result = process_combined_data(query_event, retrieve_event)
return StopEvent(result=result)스트리밍과 진행 상황 알림도 구현할 수 있습니다.
class ProgressEvent(Event):
message: str
progress: float
@step
async def long_running_task(self, ctx: Context, ev: StartEvent) -> StopEvent:
for i in range(10):
# 진행 상황을 스트림에 전송
ctx.write_event_to_stream(
ProgressEvent(message=f"단계 {i+1}/10 완료", progress=(i+1)/10)
)
await asyncio.sleep(1) # 실제 작업 시뮬레이션
return StopEvent(result="모든 작업 완료")
# 사용
workflow = MyWorkflow()
handler = workflow.run()
async for event in handler.stream_events():
if isinstance(event, ProgressEvent):
print(f"{event.message} ({event.progress*100:.1f}%)")
result = await handler다른 프레임워크와의 차별점
AI 에이전트 프레임워크 시장에서 LlamaIndex Workflows는 독특한 위치를 차지합니다. LangChain이나 LangGraph와 비교할 때, Workflows는 이벤트 기반 아키텍처에 더 집중합니다. 이는 동시에 많은 프로세스나 이벤트를 처리해야 하는 에이전트에 특히 적합합니다.
CrewAI나 AutoGen 같은 멀티 에이전트 프레임워크와 비교하면, Workflows는 개발자의 제어권을 더 강조합니다. 에이전트의 동작을 세밀하게 제어하면서도 복잡한 상호작용을 구현할 수 있죠.
특히 학습 곡선이 상대적으로 완만하다는 점도 장점입니다. 기존 Python이나 TypeScript 개발 경험이 있다면 빠르게 익힐 수 있으며, 타입 안전성과 비동기 프로그래밍의 장점을 자연스럽게 활용할 수 있습니다.
AI 개발 생태계에서의 의미
LlamaIndex Workflows 1.0의 출시는 AI 개발 생태계에서 중요한 의미를 갖습니다.
첫째, 프로덕션급 AI 애플리케이션 개발이 더욱 현실적이 되었습니다. 타입 안전성, 관찰 가능성, 리소스 관리 등 엔터프라이즈 환경에서 필요한 기능들이 기본으로 제공됩니다.
둘째, AI 에이전트의 민주화가 가속화될 것으로 예상됩니다. 복잡한 멀티 에이전트 시스템을 구축하는 데 필요한 진입 장벽이 크게 낮아졌기 때문입니다.
셋째, 오픈소스 AI 도구의 성숙도를 보여주는 사례입니다. 독립적인 프레임워크로 분리된 것은 커뮤니티 기반 개발과 혁신을 더욱 촉진할 것입니다.
LlamaIndex Workflows 1.0은 AI 애플리케이션 개발의 새로운 가능성을 열어주는 도구입니다. 이벤트 기반 아키텍처와 타입 안전성을 바탕으로, 개발자들은 더 정교하고 안정적인 AI 시스템을 구축할 수 있게 되었습니다. 앞으로 이 프레임워크가 AI 개발 생태계에 어떤 변화를 가져올지 기대가 됩니다.
참고자료:
Comments