18. 출력파서와 체인용 LCEL

LLM 시대, LangChain(랭체인)으로 배우는 AI 소프트웨어 개발

Mar 14. 2025

(1) 출력파서

LLM의 출력은 일반적으로 문자열로 리턴됩니다. 그러나, LLM의 출력을 프로그래밍 방식으로 사용할 때, 문자열을 구문분석하여 데이터구조로 변환해야 합니다. 이 해석 처리를 실시하는 컴포넌트가 출력파서입니다. LangChain(랭체인)은 아래 표와 같이 다양한 출력파서를 제공합니다.

파서입력은 LLM의 출력을 가정하기 때문에 문자열 또는 메시지입니다. 문자열과 메시지의 내용은 파서가 기대하는 문법을 따라야 합니다. 예를 들어, JsonOutputParser가 기대하는 메시지내용은 JSON문법에 따른 데이터여야 합니다. 파서 출력은 파싱된 데이터구조로 파서마다 다릅니다. 예를 들어, StrOutputParser는 문자열을 그대로 반환합니다. 반면에 JsonOutputParser는 JSONㅌ객체를 반환합니다.

① StrOutputParser 사용

위 표의 파서에서 특히 자주 사용되는 것은 StrOutputParser입니다. StrOutputParser는 LCE에서 체인을 만들 떄 출력을 문자열로 취급하는 기본 파서입니다. 다른 파서는 LLM의 출력지정(구조화 출력)에 의해 불필요하게 되고 있습니다. LLM의 출력지정을 실시하면 LLM의 출력을 구조화시킬 수 있ㅆ기 때문에 파서가 불필요하게 되기 때문입니다.

StrOutputParser 사용예시(src/langchain/parser.py)

여기서 StrOutputParser를 사용하여 문자열과 메시지를 구문분석합니다. 먼저 StrOutputParser를 가져옵니다.

StrOutputParser는 langchain_core.parsers 모듈에 포함되어 있습니다. 그런 다음 StrOutputParser를 사용하여 다음 문자열과 메시지를 구문분석합니다.

여기서는 대화모델의 invoke메소드가 반환하는 메시지를 가정하고 AIMessage클래스의 인스턴스를 제공합니다. 또한, 문자열 변수s는 간단한 문자열을 대체합니다.

이것들을 파싱하기 위해서 StrOutputParser를 다음과 같이 사용합니다.

invoke메소드는 구문분석할 문자열이나 메시지를 전달합니다. 해석한 결과를 변수 result1, result2에 대입합니다.

StrOutputParser가 문자열과 메시지에서 문자열을 그대로 꺼내고 있는지 확인할 수 있습니다.

(2) 체인용 LCEL

LCEL(LangChain Expression Language)은 Python에 내장된 일종의 도메인 특화 언어(DSL, Domain-Specific Language)입니다. 도메인 특화 언어는 특정 용도를 위한 전용언어를 의미합니다. 여기서 사용은 체인 구축입니다. LCEL은 Python연산자를 오버로드하여 체인구축을 위한 구문을 제공합니다. 사용자는 LCEL이 제공하는 구문을 사용하여 짧은 설명량으로 체인을 구축할 수 있습니다. 일반적으로 이러한 구문을 당의 구문(Syntactic Sugar)라고 합니다.

당의 구문은 복잡한 코드와 같은 것을 보다 간단한 코드로 쓸 수 있도록 하기 위한 구문입니다. 따라서 LCEL을 사용하지 않고도 LCEL을 사용하는 것과 동일한 것을 실현할 수 있습니다. 그러나 LangChain(랭체인)의 유용한 샘플과 같은 많은 것들은 LCEL로 설명됩니다. 따라서 LangChain(랭체인)을 배우는데에도 LCEL의 이해가 유용합니다. 또한 LCEL을 마스터하면 더 낳은 LangChain(랭체인)같은 프로그램을 만들 수 있습니다. 그래서 이번에는 LCEL에 대해 자주 사용되는 패턴을 망라하는 형태로 설명합니다.

① 프롬프트 체인 및 LCEL

프롬프트 체인은 하나의 프롬프트에서 얻은 응답을 다음 프롬프트의 입력으로 사용하는 프롬프트 엔지니어링 기술이었습니다. LCEL을 사용하면 프롬프트 체인을 간결하게 작성할 수 있습니다.

이번에는 프롬프트 체인을 예로 들어 LCEL을 사용하는 방법을 설명합니다. 주어진 프롬프트를 영어로 번역하고 영어 프롬프트를 사용하여 LLM에 문의하는 프롬프트 체인을 수행합니다. 또한 응답을 원래 언어로 번역하여 원래 언어의 프롬프트에 대한 응답을 얻는데 도움이 됩니다. 입력 프롬프트의 설명언어는 한국어로 제한되지 않습니다. LLM을 이용하여 자동으로 언어를 판정하고, 판정한 언어를 이용하여 원래의 언어로 번역하는 곳까지를 실시합니다. 이 프롬프트 체인을 목표로 LCEL을 습득하면서 점차 프로그램을 구축해 나갑니다.

② 체인이란?

LangChain(랭체인)의 체인이란, LangChain(랭체인)의 컴포넌트끼리가 연결된 것입니다. 하나의 컴포넌트의 출력이 다른 컴포넌트의 입력이 되는 경우에 체인을 구축할 수 있습니다. 또한 체인 자체도 구성요소입니다. 따라서 체인을 연결하여 더 긴 체인을 구성할 수 있습니다.

체인을 구성하는 요소는 Runnable인터페이스를 구현해야 합니다. Runnable인터페이스에서는 주로 다음 3가지 종류의 메소드를 정의하고 있습니다.

invoke/ainvoke

stream/astream

batch/abatch

invoke는 입력에서 출력을 얻는 기본 메소드입니다. 한편, stream와 batch는 각각 입출력을 스트림으로서 취급하고 싶은 경우, 복수의 입출력을 정리해 처리하고 싶은 경우에 사용하기 위한 메소드입니다. LangChain(랭체인)에서 제공하는 대부분의 구성요소는 Runnable인터페이스를 구현합니다. 이 때문에 대부분의 구성요소를 체인요소로 만들 수 있습니다.

Runnable 인터페이스에서는 메소드의 입출력의 형태는 정해져 있지 않습니다. 컴포넌트마다 입출력의 형태가 정해져 있ㅅ습니다. 하나의 컴포넌트A의 출력유형과 다른 컴포넌트B의 입력유형이 일치하면 컴포넌트A 다음 B를 체인으로 연결할 수 있ㅆ습니다. 입출력의 형태가 다른 컴포넌트끼리를 연결하는 경우, 한쪽 출력의 형태를 다른쪽의 입력의 형태에 맞출 필요가 있습니다. LangChain(랭체인)은 이를 위한 메커니즘도 제공합니다.

그러면 컴포넌트의 출력이 다른 컴포넌트의 입력이 되는 경우를 확인하고 LCEL을 사용하여 체인에 다시 써 봅시다. 먼저 체인을 사용하기 전의 예제를 아래와 같이 보여줍니다. 이 프로그램은 콘솔에서 입력한 문자열을 영어로 번역하는 프로그램입니다.

생성된 프롬프트에 llm에 전달하여 번역된 텍스트가 포함된 메시지를 검색합니다. 마지막으로 검색된 메시지를 parser에 전달하여 번역된 텍스트를 검색합니다. 여기서 주목해야 할 점은 순서대로 행해지는 3개의 컴포넌트 호출이 모두 invoke로 행해지고 있는 것입니다. 또한, llm과 parser의 입력은 각각 prompt_template와 llm의 출력이 되어 있는 것에도 주의해야 합니다. 이 경우 LCEL에서 다음과 같이 간단하게 다시 작성할 수 있습니다.

여기에서는 각 컴포넌트를 파이프(|)로 연결하는 것으로 체인을 구축하고 있습니다. 체인은 translation변수에 저장되어 translation을 invoke로 호출할 수 있습니다. invoke로 전달된 인수는 체인 내부에서 선두 구성요소인 prompt_template에 전달됩니다. 마찬가지로 prompt_template의 출력은 llm으로 전달됩니다. llm출력에 대해서도 마찬가지이며 최종적으로 체인마지막에 있는 parser출력이 체인의 출력이 됩니다. 이전 코드를 다시 작성한 전체 프로그램은 아래와 같습니다.

'안녕하세요'라는 사용자입력이 영어로 번역되어 'Hello'로 표시됩니다.

LCEL을 사용하지 않는 번역프로그램(src/langchain/chain0.py)

LCEL을 이용한 번역 프로그램 (src/langchain/chain1.py)

③ 시퀀스와 병렬

LangChain(랭체인)은 체인을 구축하는 구문으로 시퀀스와 병렬의 2가지 구문을 제공합니다. 시퀀스는 구성요소를 순서대로 연결하는 구문입니다. 반면에 병렬은 여러 구성요소를 병렬로 연결하는 구문입니다. 앞의 예시에서는 시퀀스를 사용하여 구성요소를 연결했습니다. 이번에는 시퀀스와 병렬구문을 사용하여 체인을 만드는 방법을 설명합니다.

▣ 시퀀스

시퀀스는 Runnable구성요소를 순서대로 연결하는 구문입니다. 다음과 같이 여러 컴포넌트를 파이프(|)로 연결하여 시퀀스를 구축할 수 있습니다.

체인에 대해 invoke를 호출하면 시퀀스의 구성요소가 차례로 호출됩니다. 이 상황은 아래와 같습니다.

일반적인 체인을 사용하는 방법은 체인을 변수에 저장하고 invoke로 호출하는 것입니다. 예를 들어, 다음과 같이 체인을 변수에 저장한 다음 invoke에서 호출할 수 있습니다.

이 코드는 다음 코드와 동일합니다.

또한, 파이프로 연결하는 구성요소 중 하나가 Runnable이면 다른 하나는 Runnable로 강제로 변환됩니다.

위 예시에서 입력x에 대해 x.foo를 반환하는 Runnable은 lambda표현식의 함수로 정의됩니다. 따라서 Runnable로 변환할 수 있는 표현식을 사용하여 Runnable과 Runnable을 연결할 수 있ㅆ습니다. 이 외에 Runnable로 변환할 수 있는 식의 예시로서는 함수 외에 다음 항에서 소개하는 사전이 있스빈다.

위 코드는 아래 코드와 동일합니다.

▣ 병렬(패럴)

값으로 Runnable 또는 Runnable로 변환할 수 있는 식을 가진 사전을 파이프로 Runnable과 연결하여 사전의 Runnable을 병렬화할 수 있습니다.

여기서 <key_1>, <key_2>, ..., <key_n>은 문자열을 나타냅니다. 또한, <runnable_1>, <runnable_2>, ..., <runnable_n>은 Runnable 또는 함수와 같은 Runnable로 변환할 수 있는 표현식을 나타냅니다. 이러한 사전을 파이프로 Runnable객체와 연결하면 사전이 RunnableParallel로 변환되고 사전의 Runnable은 병렬화됩니다. RUnnableParallel은 이름에서 알수 있듯이 Runnable이 병렬로 연결된 Runnable입니다. 또한, RunnableParallel의 invoke메소드의 반환값의 형태는 <key_1>, <key_2>, ..., <key_n>를 키로 하는 사전형이 됩니다.

병렬 구문

전형적인 이용패턴을 병렬구문의 사전의 값 종류마다 나타냅니다.

▣ 사전값이 Runnable일 경우

사전값에는 Runnable을 지정할 수 있습니다. 일반적인 사용패턴은 아래와 같습니다.

이 사전 리터럴은 RunnableParallel로 변환됩니다. 이것은 runnable과 파이프로 연결되어 있기 때문입니다. 여기서 RunnablePassthrough()는 입력을 그대로 출력하는 Runnable입니다.

즉, 사전 리터럴의 각 키에 대해 Runnable이 invoke되고 그 결과가 사전에 저장됩니다. 병렬화된 각 Runnable에는 동일한 입력(여기서는 input)이 전달됩니다. 또한, RunnablePassthrough()가 지정되고 있는 경우, 입력이 그대로 출력됩니다. 그런 다음 value에 저장된 사전이 뒤 runnable에 전달되고 invoke가 호출됩니다.

▣ 사전값을 Runnable로 변환할 수 있는 식인 경우

사전값은 Runnable외에도 Runnable로 변환할 수 있는 표현식을 지정할 수 있습니다. 자주 사용된는 패턴은 다음과 같습니다.

여기에서는 'key_1'의 값에는 lambda식을 'key_2'의 값에는 itemgetter함수의 호출을 지정하고 있습니다. itemgetter는 사전키를 지정하여 값을 얻는 함수를 반환하는 함수입니다. 따라서 두 값은 모두 함수이며 Runnable로 변환할 수 있는 표현식입니다.

사전에서 병렬화된 각 함수에는 동일한 입력(input)이 전달됩니다. 그런 다음 value에 저장된 사전이 후속 runnable로 전달되고 invoke가 호출됩니다.

④ RunnableParallel의 사용예시

RunnableParallel에는 크게 2가지 역할이 있습니다.

사전형의 출력 만들기

여러 체인을 논리적으로 병렬화

체인으로 연결하는 컴포넌트는 이전 컴포넌트의 출력유형이 후속 컴포넌트의 입력유형과 일치해야 했습니다. 사전형을 입력으로 하는 컴포넌트에 다른 형태의 데이터를 입력하고 싶은 경우에 RunnableParallel를 사용하는 것으로 형태를 명시적으로 변환하는 것입니다.

RunnableParallel의 또 다른 역할은 이름에서 알 수 있듯이 여러 체인을 논리적으로 병렬화하는 것입니다. 동일한 입력을 가진 여러 체인을 병렬화하고 사전형식의 데이터로 결과를 집계할 수 있습니다.

▣ 사전형 출력을 만들기 위한 이용

이제 RunnableParallel의 사용법중 사전의 출력을 얻는 것을 목적으로 하는 이용법을 확인해 봅시다. 이전 코드를 RunnableParallel을 사용하여 업데이트합니다. 구체적으로 다음과 같이 사전형 데이터로 호출한 부분을 문자열형 데이터로 호출합니다.

이 체인의 호출을 다음과 같이 호출할 수 있도록 수정합니다.

invoke인수를 사전형에서 문자열 형으로 변경합니다. 이 때문에, 체인에 입력은 사전형으로부터 문자열형이 됩니다. 한편, 원래 체인이 기대하는 형식은 사전형이었습니다. 따라서 다음과 같이 RunnableParallel을 사용하여 입력을 사전형으로 변환합니다.

새로운 RunnableParallel과 원래의 translation체인을 파이프로 연결해 이 체인을 to_english에 대입하고 있습니다. RunnableParallel의 "input"키와 대응하는 값에는 RunnablePassthrough()를 지정하고 있습니다. 이것은 RunnableParallel의 입력을 그대로 'input'키와 대응하는 값으로 하는 것을 의미합니다. 한편 "language"키와 대응하는 값에는 lambdatlrdmfh "English"를 지정하고 있습니다. 값은 Runnable인터페이스를 구현하는 구성요소를 지정해야 하지만 lambda식은 자동으로 Runnable로 형변환됩니다. 이 lambda식에도 RunnablePassthrough()와 같은 값이 입력됩니다. 여기서 lambda식에서는 그 값을 사용하지 않고 "English"를 돌려주도록 하고 있습니다. 임시 인수의 밑줄(_)은 임시인수를 사용하지 않음을 나타냅니다.

지금까지 RunnableParallel을 사용하여 한국어 프롬프트를 영어프롬프트로 변환하는 to_english를 정의했습니다. 같은 요령으로 LLM에 의한 영어 출력을 한국어로 고치는 to_korean도 아래와 같이 정의할 수 있습니다.

to_english와의 차이점은 lambda 표현식의 "English"가 "Korean"으로 변경된 것입니다. 이제 새로 정의한 체인을 사용하여 한국어 프롬프트를 영어로, 영어 LLM출력을 한국어로 변환하는 전체 프로그램을 아래 코드와 같습니다. 이후 이 프로그램에 대해서 순서대로 설명합니다.

우선 다음 코드에서 RunnablePassthrough를 가져옵니다.

또한 다음 코드에서 to_english와 to_korean을 사용하여 새로운 체인을 만듭니다.

이 체인은 다음과 같이 사용할 수 있습니다.

이 프로그램을 실행한 결과의 예시는 다음과 같습니다.

입력 텍스트 '내가 지금 말하고 있는 언어는 무엇입니까?'가 영어로 번역된 후 다시 한국어로 번역됩니다. 번역된 텍스트는 'What language am I speaking right now?'와 같아야 합니다. 그리고 LLM은 이 입력에서 'You are using English right now.'와 같이 반환합니다. 이 문장이 한국어로 번역되어 결과가 나옵니다.

RunnableParallel을 이용한 사전형 출력 생성(src/langchain/chain2.py)

▣ 병렬화를 위한 이용

RunnableParallel의 또 다른 중요한 역할을 여러 체인을 논리적으로 병렬화하는 것입니다. 동일한 입력을 가진 여러 체인을 병렬화하면 각 체인이 독립적으로 처리하고 결과를 사전형 데이터로 집계할 수 있습니다. 또한 병렬화를 통해 코드가 단순해지고 가독성과 유지보수성이 향상됩니다.

아래 코드는 체인을 병렬화하는데 사용하는 예시를 보여줍니다. 이 코드는 입력된 텍스트 언어를 결정하는 체인과 텍스트를 영어로 번역하는 체인을 병렬화합니다.

RunnableParallel을 이용한 체인병렬 실행

먼저 언어모델의 출력을 받기 위한 데이터 모델을 정의하고 언어모델의 출력형식으로 지정합니다.

Language클래스는 언어명을 나타내는 language_name필드가 있는 데이터 모델입니다. 이 데이터 모델은 언어판정 체인의 출력을 숫신하기 위해 llm_with_language_output의 출력형식으로 사용됩니다. 다음은 언어를 결정하는 프롬프트 템플릿과 체인을 정의합니다.

여기에서는 입력 텍스트의 언어를 결정하는 프롬프트 템플릿 ask_language_prompt를 작성합니다. 이 템플릿은 입력 텍스트를 받고 해당 언어를 묻는 프롬프트를 생성합니다. 그리고 ask_language_prompt와 llm_with_language_output을 결합하여 언어결정 체인 get_language_chain을 정의합니다. llm_with language_output은 Language클래스의 형식으로 언어모델의 출력을 수신하도록 설정된 언어모델입니다.

마찬가지로 번역을 위한 프롬프트 템플릿과 체인도 다음과 같이 정의합니다.

translation_prompt는 입력텍스트와 목표언어를 받고 번역을 요청하는 프롬프트를 생성합니다. translation체인은 translation_prompt, llm, StrOutputParser()를 조합하여 정의됩니다. 이 체인은 입력 텍스트(input)를 목표언어(language)로 번역합니다. 그런 다음 translation체인과 RunnableParallel을 사용하여 영어로 번역할 to_english체인을 정의합니다.

to_english는 입력 텍스트를 영어로 번역하는 체인입니다. 'input'키에는 RunnablePassthrough()를 지정하는 것으로 입력텍스트를 그대로 번역체인에 건네주고 있습니다. 'language'키는 항상 'English'를 반환하는 lambda식을 지정합니다. 마지막으로 언어판정체인과 영어 번역 체인을 병렬화하고 있습니다.

여기에서 사전형을 사용해 2개의 체인을 병렬화하고 있습니다. 'input'키에는 영어로 번역하는 체인이 연관되어 있습니다. 이 체인은 다음 3가지 구성요소로 구성됩니다.

to_english: 입력텍스트를 영어로 번역하는 체인

llm: 영어로 번역하기 위한 LLM

StrOutputParser(): LLM출력을 문자열로 구문분석하는 파서

다음 'language'키에는 언어를 판정하는 체인과 그 출력으로부터 language_name을 꺼내는 lambda식이 대응지어져 있습니다. 이 체인은 다음 2가지 구성요소로 구성됩니다.

get_language_chain: 입력텍스트의 언어를 결정하는 체인

lambda x: x.language_name: get_langchain_chain의 출력에서 language_name을 검색하는 lambda식

이러한 병렬화된 체인의 출력을 사전형의 데이터로 집계되고 마지막으로 번역체인으로 전달됩니다. 번역체인은 사전형 데이터에서 'input'과 'language'의 값을 검색하고 입력텍스트를 지정된 언어로 번역합니다.

입력텍스트 "내가 지금 말하는 언어는 무엇입니까?"가 영어로 번역된 후 다시 한국어로 번역됩니다. 번역된 텍스트는 'What language am I speaking right now?'와 같아야 합니다. 이 문장이 한국어로 번역되어 위 결과가 됩니다. 이 결과로부터 언어판정 체인에 의해 원래의 입력 텍스트의 언어가 한국어로서 올바르게 판장되고 있는 것 그 정보를 이용해 LLM의 영어출력이 원래 한국어로 올바르게 번역이 행해지는 것을 알 수 있습니다.

이런 식으로 RunnableParallel을 사용하면 여러 체인을 병렬화하고 각 체인의 출력을 사전형의 데이터로 집계할 수 있습니다. 벙렬화를 통해 코드의 단순화와 가독성 향상을 도모할 수 있기 때문에 LangChain(랭체인)을 사용한 개발에 있어서 중요한 기법의 하나가 되고 있습니다.