4. LCEL 설명

RAG + AI 에이전트 개발 by LangChain, LangGraph

⑴ Runnable과 RunnableSequence : LCEL의 가장기본 컴포넌트

이번에는 LangChain을 잘 다루고 싶은 분들을 위해 LangChain Expression Language(LCEL)에 대해 자세히 설명합니다.

 LCEL의 가장 기본적인 구현은 Prompt template, Chat model, Output parser의 세 가지를 연결하는 것입니다. 앞 장에서 설명했듯이 Prompt template, Chat model, Output parser는 모두 'invoke' 메서드로 실행할 수 있는데, LCEL을 잘 이해할 수 있도록 먼저 이들을 순서대로 invoke해 봅시다. 먼저 prompt(ChatPromptTemplate), model(ChatOpenAI), output_parser(StrOutputParser)를 준비합니다.

prompt, model, output_parser를 순서대로 invoke하는 코드는 다음과 같습니다.

이런 코드에서도 Prompt template, Chat model, Output parser를 순서대로 실행할 수 있지만, LCEL에서는 이를 '|'로 연결한 후 실행합니다. 사실 ChatPromptTemplate, ChatOpenAI, StrOutputParser는 모두 LangChain의 'Runnable'이라는 추상 베이스 클래스를 상속받으며, Runnable을 '|'로 연결하면 'RunnableSequence'가 되고, RunnableSequence도 RunnableSequence가 됩니다. RunnableSequence도 Runnable의 일종입니다.

RunnableSequence를 호출하면 연결된 Runnable이 순차적으로 호출됩니다.

이렇게 Runnable을 '|'로 연결하여 새로운 Runnable을 만들고, 이를 호출하면 내부 Runnable이 차례로 호출되는 것이 LCEL의 기본입니다.

① Runnable 실행방법 : invoke, stream, batch

우선 Runnable의 실행방법으로는 invoke외에 stream, batch가 있습니다. 각각 알아봅시다. 먼저 Runnable을 스트리밍으로 실행하려면 다음과 같이 stream 메서드를 사용합니다.

또한, batch 메소드를 사용하면 여러 개의 입력을 한꺼번에 처리할 수 있습니다.

chain(RunnableSequence)의 stream 메서드를 호출하면 내부 Runnable의 stream 메서드가 순차적으로 호출됩니다. 또한, chain(RunnableSequence)의 batch 메서드를 호출하면 내부 Runnable의 batch 메서드가 차례로 호출됩니다. 이처럼 Runnable 클래스를 상속받은 클래스는 invoke, stream, batch라는 통일된 인터페이스로 호출할 수 있습니다. 또한, 이들을 비동기 처리한 ainvoke, astream, abatch라는 메소드도 제공됩니다.

② LCEL의 '|'로 다양한 Runnable을 연쇄적으로 연결

이제 '|'를 이용한 처리 체인에 대해 좀 더 자세히 알아보겠습니다. '|'를 사용하면 Runnable과 Runnable을 연결할 수 있고, Runnable을 연결한 chain도 Runnable이므로 chain과 chain도 '|'으로 연결할 수 있습니다. 예를 들어, Zero-shot CoT에서 단계별로 생각하게 하고, 그 결과에서 결론만 추출해 보겠습니다. 먼저 model(ChatOpenAI)과 output_parser(StrOutputParser)를 준비합니다.

첫 번째 Chain으로 Zero-shot CoT에서 단계별로 생각하게 하는 Chain을 만듭니다.

두 번째 Chain으로 단계별로 생각한 답변에서 결론을 추출하는 Chain을 생성합니다.

두 개의 Chain을 연결한 Chain을 만들어 실행해 보겠습니다.

최종적으로 요약된 간단한 답변을 얻을 수 있었습니다. 이때 cot_summarize_chain 내부에서는 먼저 cot_chain이 실행되어 단계별로 생각한 중복된 답을 얻습니다. 그 답변을 입력으로 summarize_chain을 실행하여 요약된 간단한 답변을 얻게 되는데, LLM을 두 번 호출함으로써 Zero-shot CoT를 사용하여 답변의 정확도를 높이면서 최종적으로 간단한 출력을 얻을 수 있었다는 것을 알 수 있습니다.

LLM 애플리케이션에서 복잡한 작업을 한 번의 LLM 호출로 해결하려고 하면 프롬프트 작성이나 개선이 어려워지는 경우가 많습니다. 여러 프롬프트에서 LLM을 여러 번 호출하는 정책을 사용하면, 작업을 더 쉽게 해결할 수 있는 경우가 많습니다. 참고로 cot_chain과 summarize_chain을 '|'로 연결하여 실행할 수 있는 것은 cot_chain의 출력 타입과 summarize_chain의 입력 타입이 일치하기 때문입니다.

단, Runnable을 '|'로 연결할 때는 출력타입과 입력타입의 정합성에 주의해야 합니다.

③ LangSmith에서 Chain의 내부 동작 확인

LCEL로 작성한 Chain 내부에서는 앞의 예시처럼 LLM을 여러 번 호출하거나, 뒤에서 소개할 것처럼 검색 처리를 수행하는 경우가 많습니다. 처리 연쇄가 복잡해지면, 처리 중 상황을 확인하고 싶을 때가 있습니다. 앞 설명에서 'LangSmith 설정'에서 설명한 절차에 따라 LangSmith를 설정했다면, LangSmith 화면에서 다음과 같이 Chain의 내부 동작을 확인할 수 있습니다.

⑵ RunnableLambda : 임의의 함수를 Runnable로 만들기

LCEL에서는 Runnable끼리 '|'로 연결하는 것 외에도 다양한 커스터마이징 방법을 제공하고 있습니다. 여기서는 임의의 함수를 Runnable로 만드는 'RunnableLambda'에 대해 설명합니다. LLM 애플리케이션에서는 LLM의 응답에 대해 규칙 기반으로 추가 처리를 추가하거나 어떤 변환을 가하고 싶은 경우가 많은데, RunnableLambda를 사용하면 LCEL의 Chain에 임의의 처리(함수)를 연결할 수 있습니다. 예를 들어, LLM이 생성한 텍스트에 대해 소문자를 대문자로 변환하는 처리를 체인으로 연결하는 Chain을 구현해 보겠습니다. 먼저 prompt(ChatPromptTemplate), model(ChatOpenAI), output_parser(StrOutputParser)를 준비합니다.

이어서 소문자를 대문자로 변환하는 함수를 구현하고, 이를 Chain으로 연결하여 실행합니다.

RunnableLambda를 사용하면 임의의 함수를 Runnable로 변환할 수 있으며, LLM 호출과 자체 처리를 연결하고 싶은 경우가 많기 때문에 RunnableLambda를 자주 사용하게 됩니다.

① chain 데코레이터를 이용한 RunnableLambda의 구현

RunnableLambda를 생성하기 위해 chain 데코레이터(@chain)를 사용할 수도 있는데, chain 데코레이터를 사용하는 샘플 코드는 다음과 같습니다.

chain 데코레이터(@chain)에 의해 상위 함수가 RunnableLambda로 변환되었다는 뜻입니다.

② RunnableLambda로 자동변환

지금까지 설명에서는 RunnableLambda를 명시적으로 생성하여 Runnable과 연결하는 예제를 살펴보았다. 사실 RunnableLambda를 명시적으로 생성하지 않아도 Runnable과 임의의 함수를 '|'로 연결할 수 있습니다. 우선 다음 샘플 코드를 살펴봅시다.

이 코드에서는 Runnable과 upper라는 함수가 '|'로 연결되어 있습니다. 이때 upper는 자동으로 RunnableLambda로 변환됩니다. 사실 '|'의 왼쪽과 오른쪽 중 하나가 Runnable인 경우, 다른 쪽이 함수라면 자동으로 RunnableLambda로 변환되도록 되어 있으며, LangChain 공식 문서나 쿡북에서는 RunnableLambda로 자동 변환을 사용하는 경우가 많습니다. 이후 예제에서도 RunnableLambda로 자동 변환을 적절히 사용합니다.

③ Runnable의 입력유형과 출력유형에 주의하기

지금까지 '|'를 사용하여 다양한 Runnable을 연결해 보았지만, 어떤 Runnable끼리 연결해도 잘 동작하지 않을 수 있습니다. 예를 들어, 다음 코드는 에러가 발생합니다.

이 코드에서는 model과 upper 함수를 직접 연결하고 있습니다. 실행하면 다음과 같은 오류가 발생합니다.

이 오류는 upper 함수 내 text.upper() 부분에서 발생하는데, model이 AIMessage를 출력하는 반면, 자체 제작한 upper 함수는 str을 입력으로 기대하기 때문에 발생합니다.

다시 한 번 강조하지만, Runnable을 '|'로 연결할 때는 출력의 타입과 입력의 타입의 정합성에 주의해야 합니다. 위의 오류 예제는 다음과 같이 StrOutputParser를 사용하여 model의 출력을 str로 변환한 후 upper 함수에 전달하면 해결됩니다.

⑶ RunnableParallel : 여러 개의 Runnable을 병렬로 연결

LCEL을 구현하다 보면 Runnable을 병렬로 연결하고 싶을 때가 있습니다. 예를 들어, 사용자가 입력한 주제에 대해 LLM이 낙관적인 의견과 비관적인 의견을 생성하도록 해봅시다.

먼저 model(ChatOpenAI)과 output_parser(StrOutputParser)를 준비합니다.

이어서 낙관적인 의견을 생성하는 Chain을 구현합니다.

또한, 비관적인 의견을 생성하는 Chain을 구현합니다.

'RunnableParallel'을 사용하여 낙관적인 의견을 생성하는 체인(optimistic_chain)과 비관적인 의견을 생성하는 체인(pessimistic_chain)을 병렬로 연결한 체인을 생성하고 실행합니다.

낙관적인 의견과 비관적인 의견이 dict로 출력되었습니다. 이처럼 여러 개의 Runnable을 병렬로 연결하여 실행할 수 있는 것이 'RunnableParallel'입니다. 실제로 위의 예시 optimistic_chain과 pessimistic_chain은 동시에 실행되기 때문에 순서대로 실행하는 것보다 짧은 시간에 전체 처리가 완료됩니다. RunnableParallel은 사실상 키가 str이고 값이 Runnable(또는 Runnable로 자동 변환할 수 있는 함수 등)인 dict입니다.

① RunnableParallel의 출력을 Runnable의 입력으로 연결

RunnableParallel도 Runnable의 일종이므로 Runnable과 '|'로 연결할 수 있는데, RunnableParallel과 Runnable을 '|'로 연결하는 예시로 낙관적인 의견과 비관적인 의견을 제시한 후, 객관적으로 정리하는 Chain을 연결해 봅시다.

먼저 낙관적인 의견과 비관적인 의견을 종합하는 프롬프트(synthesize_prompt)를 준비합니다.

낙관적인 의견을 생성하는 체인(optimistic_chain)과 비관적인 의견을 생성하는 체인(pessimistic_chain)을 병렬로 연결한 RunnableParallel을 synthesize_prompt, model, output_parse_parse에 연결하여 실행합니다. parser에 연결하여 실행합니다.

RunnableParallel의 두 체인에서 생성된 의견을 정리할 수 있습니다.

② RunnableParallel로 자동 변환

Runnable과 함수를 '|'로 연결하면 함수가 자동으로 RunnableLambda로 변환되었습니다. 마찬가지로 키가 str이고 값이 Runnable(또는 Runnable로 자동 변환할 수 있는 함수 등)인 dict는 RunnableParallel로 자동 변환됩니다. 예를 들어, 앞의 코드는 다음과 같이 작성할 수도 있습니다.

RunnableParalell을 사용할 때는 이 형식으로 코드를 작성하는 경우가 많습니다.

참고로 RunnableParallel에서는 병렬로 연결된 Runnable이 모두 실행되지만, 상황에 따라 두 체인 중 하나만 선택하여 실행하고 싶은 경우도 있을 수 있으며, LCEL에서는 이러한 체인의 '라우팅'도 가능합니다. 자세한 내용은 공식 문서 다음 페이지를 참고하시기 바랍니다.

관련 URL: https://python.langchain.com/docs/how_to/routing/

How to route between sub-chains | �️� LangChain

③ RunnableLambda와의 조합 : itemgetter 사용 예시

RunnableParalell 설명의 마지막에 'itemgetter'를 사용하는 예제를 소개합니다. itemgetter는 파이썬 표준 라이브러리에서 제공하는 함수로, itemgetter를 사용하면 dict 등으로부터 값을 추출하는 함수를 쉽게 만들 수 있습니다. 를 쉽게 만들 수 있습니다. 예를 들어, {“topic”: “생성 AI의 진화에 대하여”}라는 dict에서 itemgetter(“topic”)를 사용하여 topic을 가져오는 예제는 다음과 같습니다.

{“topic”: “생성형 AI의 진화에 대하여”}에서 “생성형 AI의 진화에 대하여”라는 값을 가져온 것입니다. LCEL에서는 이 itemgetter를 사용하면 유용한 경우가 많습니다. 예를 들어 다음 코드는 {“topic”:“ 생성 AI의 진화에 대하여”}에서 itemgetter(“topic”}로 값을 가져와서 ChatPromptTemplate의 {topic} 부분에 채워넣고 있습니다.

itemgetter를 이해하려면 조금 익숙해져야 할 수도 있지만, LCEL에서는 자주 등장하기 때문에 꼭 기억해 둡시다.

⑷ RunnablePassthrough : 입력 그대로 출력하기

LCEL의 구성 요소 중 마지막으로 소개할 것은 'RunnablePassthrough'입니다. 앞에서 설명한 RunnableParalell을 사용할 때, 해당 요소의 일부에서 입력 값을 그대로 출력하고 싶은 경우가 있습니다. 입력을 그대로 출력하기 위해 사용할 수 있는 것이 RunnablePassthrough입니다.

RunnablePassthrough를 사용하는 예로 간단한 RAG의 Chain을 구현해 보겠습니다. RAG의 일반적인 구현은 문서를 벡터 데이터베이스에 저장하고 벡터 검색을 하는 것이지만, 여기서는 좀 더 쉽게 구현할 수 있는 웹 검색으로 구현해 보겠습니다, LLM과 RAG에 최적화된 검색 엔진인 'Tavily'를 사용하기로 합니다. Tavily 웹사이트에서 회원가입을 하고 API 키를 발급받으세요. 획득한 API 키는 Google Colab의 시크릿에 'TAVILY_API_KEY'라는 이름으로 저장합니다. 이후 다음 코드를 실행하여 TAVILY_API_KEY를 환경 변수로 설정합니다.

LangChain에서 Tavily를 사용하기 위해 tavily-python이라는 패키지를 설치합니다. 다음 명령어를 실행합니다.

이제 Tavily에서 웹 검색을 이용한 RAG 처리를 구현해 보겠습니다. 먼저 prompt(ChatPromptTemplate), model(ChatOpenAI)을 준비합니다.

다음으로 Tavily를 LangChain의 Retriver로 사용하기 위한 TavilySearchAPIRetriever를 준비합니다.

TavilySearchAPIRetriever의 k라는 파라미터는 검색할 건수를 지정할 수 있습니다. 지금까지 준비한 prompt, model, retriever를 사용하여 RAG의 Chain을 구현합니다.

이 chain 중 prompt입력까지를 보여주면 아래와 같습니다.

RunnablePassthrough는 입력한 내용을 그대로 출력하는데, prompt의 입력 중 “question”에 대해서는 “서울의 오늘 날씨는?”이라는 문자열이 그대로 입력됩니다. 반면, prompt의 입력 중 'context'에 대해서는 retriever의 실행 결과인 검색 결과 목록이 입력됩니다. 이처럼 RunnableParallel의 출력 중 하나로 입력 값을 그대로 사용하고 싶다면 RunnablePassthrough를 사용할 수 있습니다.

① assign : RunnableParallel의 출력에 값 추가

앞에서 구현한 RAG의 Chain에서는 LLM이 생성한 최종 답변만이 Chain 전체의 출력으로 나옵니다. 하지만 retriever의 검색 결과도 전체 Chain의 출력에 포함시키고 싶은 경우가 종종 있습니다. 이때 사용할 수 있는 것이 RunnablePassthrough의 assign라는 클래스 메서드입니다.

RunnablePassthrough.assign 부분에서는 RunnableParallel의 실행 결과를 유지한 채 'answer'를 추가한 dict를 출력한 것입니다.

참고로 assign는 RunnablePassthrough의 클래스 메서드 외에 Runnable의 인스턴스 메서드로도 제공

됩니다. 따라서 이전 예제와 유사한 처리를 다음과 같이 작성할 수도 있습니다.

이렇게 assign을 사용하면 context와 같은 Chain의 중간 값을 Chain의 최종 출력에 포함시킬 수 있습니다. 따라서 프롬프트를 채운 결과를 화면에 표시하고 싶을 때와 같이 Chain의 중간 값을 UI에 표시하고 싶을 때에도 assign이 유용하게 쓰일 수 있습니다.

이번에는 LCEL(LangChain Expression Language)에 대해 설명했습니다. LangChain은 각 컴포넌트를 라이브러리처럼 사용하는 것만으로도 유용하게 사용할 수 있습니다. 하지만, LCEL을 잘 다룰 수 있어야 LangSmith의 트레이스와의 연동등을 보다 효과적으로 활용할 수 있습니다. LCEL은 부분적으로만 배우면 자신만의 방식으로 커스터마이징하여 사용하려고 할 때 어려움을 겪을 수 있습니다. 필자 생각으로는 라이브러리가 아닌 프레임워크를 따라잡을 수 있는 생각을 가지고 체계적으로 학습하는 것이 중요합니다.

물론, LCEL은 익숙하지 않으면 어려운 것도 사실이고 LCEL을 배우는 리스크를 가지고 싶지 않은 경우도 있습니다. 또한, LangChain을 사용한다고 해서 모든 처리를 LCEL로 작성해야 하는 것은 아니고 LCEL로 구현하기 어렵다고 생각되는 부분은 일반 프로그래밍으로 구현할 수 있습니다.

© 2024 ZeR0, Hand-crafted & made with Damon JW Kim.

Profile: https://gaebal.site

ZeR0 Knowledge Base | Jaewoo Kim resume

개발문의: https://naver.me/GalVgGKH

개발 및 강좌 문의 접수

블로그: https://blog.naver.com/beyond-zero

ZeR0 기술 블로그 : 네이버 블로그