[Langchain] LCEL (LangChain Expression Language)

Hunie_07·2025년 3월 25일

langChain

Langchain

목록 보기

2/35

📌 LCEL(LangChain Expression Language)

LCEL은 | 연산자를 사용해 컴포넌트들을 순차적으로 연결하는 선언적 체이닝을 지원
재사용성이 높아 정의된 체인을 다른 체인의 컴포넌트로 활용 가능
다양한 실행 방식(.invoke(), .batch(), .stream(), .astream())으로 동기/비동기 처리가 가능
배치 처리시 자동 최적화를 통해 효율적인 작업 수행이 가능

1️⃣ LCEL

1. Prompt + LLM

기본 구조: Prompt | LLM 형태로, 파이프(|) 연산자를 사용해 프롬프트와 LLM을 순차적으로 연결
데이터 흐름: 사용자 입력이 Prompt 템플릿을 통해 처리된 후, LLM에 전달되어 최종 응답을 생성
실행 순서: 파이프라인은 왼쪽에서 오른쪽으로 순차적으로 실행되며, 각 컴포넌트의 출력이 다음 컴포넌트의 입력으로 전달됨

from langchain_openai import ChatOpenAI

# LLM model
llm = ChatOpenAI(
    model="gpt-4o-mini", 
    temperature=0.3, 
    max_tokens=100,
    )

# 모델에 프롬프트를 입력
response = llm.invoke("탄소의 원자 번호는 무엇인가요?")

# 응답 객체 확인
response

- 출력

AIMessage(content='탄소의 원자 번호는 6입니다.', additional_kwargs={'refusal': None}, response_metadata= ...

# 응답 객체의 텍스트를 확인
response.content

- 출력

'탄소의 원자 번호는 6입니다.'

from langchain_core.prompts import PromptTemplate

# 템플릿 문자열 정의
template = """
당신은 {topic} 분야의 전문가입니다. {topic}에 관한 다음 질문에 답변해주세요.
질문: {question}
답변: """

# PromptTemplate 객체 생성
prompt = PromptTemplate.from_template(template)

# 템플릿 사용 예시
formatted_prompt = prompt.format(
    topic="인공지능",
    question="딥러닝과 머신러닝의 주요 차이점은 무엇인가요?"
)

print("템플릿 변수:")
print(f"- 필수 변수: {prompt.input_variables}")
print("\n생성된 프롬프트:")
print(formatted_prompt)

- 출력

템플릿 변수:
- 필수 변수: ['question', 'topic']

생성된 프롬프트:

당신은 인공지능 분야의 전문가입니다. 인공지능에 관한 다음 질문에 답변해주세요.
질문: 딥러닝과 머신러닝의 주요 차이점은 무엇인가요?
답변:

# prompt 객체의 템플릿을 확인
print(prompt.template)

- 출력

당신은 {topic} 분야의 전문가입니다. {topic}에 관한 다음 질문에 답변해주세요.
질문: {question}
답변:

# chain을 구성
chain = prompt | llm

# chain 객체 확인
chain

- 출력

PromptTemplate(input_variables=['question', 'topic'], input_types={}, partial_variables={}, template='\n당신은 {topic} 분야의 전문가입니다. {topic}에 관한 다음 질문에 답변해주세요.\n질문: {question}\n답변: ')
| ChatOpenAI(client=<openai.resources. ...)

# chain 객체의 입력 스키마를 확인
chain.input_schema.model_json_schema()

- 출력

{'properties': {'question': {'title': 'Question', 'type': 'string'},
  'topic': {'title': 'Topic', 'type': 'string'}},
 'required': ['question', 'topic'],
 'title': 'PromptInput',
 'type': 'object'}

# chain 실행
response = chain.invoke( 
    {
        "topic": "화학(Chemistry)", 
        "question": "탄소의 원자 번호는 무엇인가요?"
    }
)

# 응답 객체를 출력
response

- 출력

AIMessage(content='탄소의 원자 번호는 6입니다.', additional_kwargs={'refusal': None}, response_metadata= ...)

2. Prompt + LLM + Output Parser

데이터 파이프라인: Prompt | LLM | OutputParser 형태로 구성되어 LLM의 출력을 구조화된 형식으로 변환
Parser 종류: JSON, XML 등 다양한 형식의 파서를 지원하여 LLM 출력을 원하는 데이터 구조로 변환 가능
유효성 검증: Parser가 출력 형식을 검증하여 잘못된 형식의 응답을 필터링하고 안정적인 데이터 처리를 보장

# 문자열 출력 파서 
from langchain_core.output_parsers import StrOutputParser

# 출력 파서를 생성
output_parser = StrOutputParser()

# 출력 파서를 실행
output_parser.invoke(response)

- 출력

'탄소의 원자 번호는 6입니다.'

# 출력 파서를 chain에 추가
chain = prompt | llm | output_parser

# chain을 실행
response = chain.invoke(
    {
        "topic": "화학(Chemistry)", 
        "question": "탄소의 원자 번호는 무엇인가요?"
    }
)

# 응답 객체를 출력
response

- 출력

'탄소의 원자 번호는 6입니다.'

from langchain_groq import ChatGroq

llm = ChatGroq(model="llama3-8b-8192")

from pydantic import BaseModel, Field

# Pydantic
class Atom(BaseModel):
    """Information about Atom"""

    number: int = Field(description="The number of element")
    name : str = Field(description='The name of element')
    symbol : str = Field(description='The symbol of element')
    summary : str = Field(description="The summary of user's input")

structured_llm = llm.with_structured_output(Atom)

structured_llm.invoke("Tell me the number of 붕소")

- 출력

Atom(number=5, name='Boron', symbol='B', summary='Information about Boron')

2️⃣ Runnable

실행 인터페이스: 모든 LangChain 컴포넌트는 Runnable 인터페이스를 구현하여 일관된 방식으로 실행
실행 메서드: .invoke(), .batch(), .stream(), .astream() 등 다양한 실행 방식을 제공
호환성: 모든 Runnable 컴포넌트는 파이프(|) 연산자를 통해 연결 가능하며, 재사용이 용이
Runnable의 주요 유형:
- RunnableSequence: 여러 Runnable을 순차적으로 실행
- RunnablePassthrough: 입력을 그대로 다음 단계로 전달
- RunnableParallel: 여러 Runnable을 병렬로 실행
- RunnableLambda: 파이썬 함수를 Runnable로 래핑하여 체인에서 사용

1. RunnableSequence

RunnableSequence는 컴포넌트들을 연결하여 순차적으로 데이터를 처리하는 체인
| 연산자로 연결된 각 단계의 출력이 다음 단계의 입력으로 전달됨
다양한 실행 방식(동기/비동기, 배치/스트리밍)을 지원
LLM 체인, 데이터 파이프라인, 자동화된 작업 등 다단계 처리에 활용됨

from langchain_core.runnables import RunnableSequence
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 컴포넌트 정의
prompt = PromptTemplate.from_template(
"'{text}'를 영어로 번역해주세요. 번역된 문장만을 출력해주세요.")
translator = ChatOpenAI(model="gpt-4o-mini", temperature=0.3, max_tokens=100)
output_parser = StrOutputParser()

# RunnableSequence 생성 - 1. 함수 사용 
translation_chain = RunnableSequence(
    first=prompt,
    middle=[translator],   # 리스트로 전달하는 점에 주의
    last=output_parser
)

# RunnableSequence 생성 - 2. 연산자 사용
translation_chain = prompt | translator | output_parser

# 동기 실행
result = translation_chain.invoke({"text": "안녕하세요"})
print(result)

- 출력

Hello

비동기 처리의 주요 이점

응답성: I/O 작업 중 다른 작업 수행이 가능하여 시스템 전체 응답성 향상
확장성: 여러 요청을 동시에 처리할 수 있어 처리량 증가
리소스 활용: CPU와 메모리를 효율적으로 활용하여 성능 최적화

import asyncio
import nest_asyncio

# Jupyter에서 비동기 실행을 위한 설정
nest_asyncio.apply()

# 비동기 실행 함수를 정의
async def run():
    result = await translation_chain.ainvoke({"text": "감사합니다"})
    print(result)

#  비동기 실행 함수를 호출
await run()

- 출력

Thank you.

2. RunnableParallel

RunnableParallel은 여러 컴포넌트를 딕셔너리 형태로 구성해 동시 실행
동일한 입력이 모든 병렬 컴포넌트에 전달되며, 결과는 키-값 쌍으로 반환됨
데이터 변환과 파이프라인 구성에 특화되어 있으며, 출력 형식을 다음 단계에 맞게 조정 가능

(1) 질문 분석 체인

from langchain_openai import ChatOpenAI

# LLM model
llm = ChatOpenAI(
    model="gpt-4o-mini", 
    temperature=0.3, 
    max_tokens=100,
    )

# 질문과 관련된 분야를 찾는 프롬프트
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 출력 파서 정의
output_parser = StrOutputParser()

# 질문 템플릿 정의
question_template = """
다음 카테고리 중 하나로 입력을 분류하세요:
- 화학(Chemistry)
- 물리(Physics)
- 생물(Biology)

# 예시:
입력: 사람의 염색체는 모두 몇개가 있나요?
답변: 생물(Biology)

입력: {question}
답변: """

# 프롬프트 생성
question_prompt = ChatPromptTemplate.from_template(question_template)

# 체인 구성
question_chain = question_prompt | llm | output_parser

# 체인 실행
result = question_chain.invoke({"question": "탄소의 원자 번호는 무엇인가요?"})
print(f"분류 결과: {result}")

- 출력

분류 결과: 화학(Chemistry)

(2) 언어 감지 체인

# 질문에 사용된 언어를 구분하는 프롬프트
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 출력 파서 정의
output_parser = StrOutputParser()

# 언어 분류 템플릿 정의
language_template = """
입력된 텍스트의 언어를 다음 카테고리 중 하나로 분류하세요:
- 영어(English)
- 한국어(Korean)
- 기타(Others)

# 예시:
입력: How many protons are in a carbon atom?
답변: English

입력: {question}
답변: """

# 프롬프트 생성
language_prompt = ChatPromptTemplate.from_template(language_template)

# 체인 구성
language_chain = language_prompt | llm | output_parser

# 체인 실행 예시
examples = [
    "What is the atomic number of carbon?",
    "탄소의 원자 번호는 무엇인가요?",
    "¿Cuál es el número atómico del carbono?"
]

for example in examples:
    result = language_chain.invoke({"question": example})
    print(f"입력: {example}")
    print(f"분류 결과: {result}\n")

- 출력

입력: What is the atomic number of carbon?
분류 결과: English

입력: 탄소의 원자 번호는 무엇인가요?
분류 결과: Korean

입력: ¿Cuál es el número atómico del carbono?
분류 결과: Others

(3) RunnableParallel을 사용한 병렬 실행 체인

# 질문과 관련된 분야를 찾아서 질문에 대한 답변을 생성하는 프롬프트
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
from langchain_core.output_parsers import StrOutputParser
from operator import itemgetter

# 답변 템플릿 정의
answer_template = """
당신은 {topic} 분야의 전문가입니다. {topic}에 관한 질문에 {language}로 답변해주세요.
질문: {question}
답변: """

# 프롬프트 및 체인 구성
answer_prompt = ChatPromptTemplate.from_template(answer_template)
output_parser = StrOutputParser()

# 병렬 처리 체인 구성
answer_chain = RunnableParallel({
    "topic": question_chain,            # 주제 분류 체인
    "language": language_chain,         # 언어 감지 체인
    "question": itemgetter("question")  # 원본 질문 추출
}) | answer_prompt | llm | output_parser

# 체인 실행 예시
result = answer_chain.invoke({
    "question": "탄소의 원자 번호는 무엇인가요?"
})

print("처리 결과:")
print(f"답변: {result}")

- 출력

처리 결과:
답변: 탄소의 원자 번호는 6입니다. 이는 탄소 원자가 6개의 양성자를 가지고 있음을 의미합니다.

3. RunnablePassthrough

RunnablePassthrough는 입력값을 그대로 전달하여 원본 데이터를 보존
RunnableParallel과 함께 사용되어 입력 데이터를 새로운 키로 매핑 가능
투명한 데이터 흐름으로 파이프라인 디버깅과 구성이 용이

from langchain_core.runnables import RunnablePassthrough
import re

runnable = RunnableParallel(
    passed=RunnablePassthrough(),   # 사용자 입력값 그대로 출력
    modified=lambda x: int(re.search(r'\d+', x["query"]).group()), # 숫자만 추출
)

runnable.invoke({"query": '탄소의 원자 번호는 6입니다.'})

- 출력

{'passed': {'query': '탄소의 원자 번호는 6입니다.'}, 'modified': 6}

runnable = RunnableParallel(
    passed=RunnablePassthrough(),
    modified=lambda x: int(re.search(r'\d+', x).group()),
)

runnable.invoke('탄소의 원자 번호는 6입니다.')

- 출력

{'passed': '탄소의 원자 번호는 6입니다.', 'modified': 6}

4. RunnableLambda

RunnableLambda는 일반 함수를 Runnable 객체로 변환하는 래퍼 컴포넌트
체인에 커스텀 로직을 쉽게 통합할 수 있어 데이터 전처리, 후처리에 유용
| 연산자로 다른 컴포넌트들과 연결해 복잡한 처리 흐름을 구성 가능

import re
from langchain_core.runnables import RunnableLambda, RunnablePassthrough

# 텍스트에서 숫자를 추출하는 함수
def extract_number(query):
    return int(re.search(r'\d+', query).group())

# RunnablePassthrough로 입력을 그대로 전달하고, RunnableLambda로 숫자 추출 함수 실행
runnable = RunnablePassthrough() | RunnableLambda(extract_number)

# 입력 텍스트에서 6을 추출
result = runnable.invoke('탄소의 원자 번호는 6입니다.')
print(result)  # 출력: 6

- 출력

from langchain_core.runnables import RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import AIMessage

# 데이터 전처리 함수 정의
def preprocess_text(text: str) -> str:
    """ 입력 텍스트를 소문자로 변환하고 양쪽 공백을 제거합니다. """
    return text.strip().lower()

# 후처리 함수 정의
def postprocess_response(response: AIMessage) -> dict:
    """ 응답 텍스트를 대문자로 변환하고 길이를 계산합니다. """
    response_text = response.content
    return {
        "processed_response": response_text.upper(),
        "length": len(response_text)
    }

# 프롬프트 템플릿 생성
prompt = ChatPromptTemplate.from_template(
"다음 주제에 대해 영어 한 문장으로 설명해주세요: {topic}")

# 처리 파이프라인 구성
chain = (
    RunnableLambda(preprocess_text) |  # 입력 전처리
    prompt |                           # 프롬프트 포맷팅
    llm |                              # LLM 추론
    RunnableLambda(postprocess_response)  # 출력 후처리
)

# 체인 실행
result = chain.invoke("Artificial Intelligence")
print(f"처리된 응답: {result['processed_response']}")
print(f"응답 길이: {result['length']}")

- 출력

처리된 응답: ARTIFICIAL INTELLIGENCE (AI) REFERS TO THE SIMULATION OF HUMAN INTELLIGENCE IN MACHINES THAT ARE PROGRAMMED TO THINK, LEARN, AND SOLVE PROBLEMS AUTONOMOUSLY.
응답 길이: 157

Hunie_07

이전 포스트

[Langchain] 랭체인 컴포넌트

다음 포스트

[Langchain] LCEL (LangChain Expression Language)

Langchain

📌 LCEL(LangChain Expression Language)

1️⃣ LCEL

1. Prompt + LLM

2. Prompt + LLM + Output Parser

2️⃣ Runnable

1. RunnableSequence

2. RunnableParallel

3. RunnablePassthrough

4. RunnableLambda

[Langchain] 랭체인 컴포넌트

[Gradio] 챗봇 어플리케이션

0개의 댓글