1. LCEL 소개

LCEL (LangChain 표현 언어)은 복잡한 체인을 구축하기 위한 간단하고 사용하기 쉬운 프레임워크입니다. 통합된 인터페이스와 복합 기본 요소를 제공하여 체인을 쉽게 구성할 수 있습니다. 각 LCEL 객체는 Runnable 인터페이스를 구현하며 invoke, batch, stream, ainvoke 등의 일반적으로 사용되는 호출 메서드 집합을 정의합니다. 따라서 LCEL 객체의 체인도 이 호출 방법을 자동으로 지원하여 각 LCEL 객체의 체인 자체가 LCEL 객체가 될 수 있습니다.

2. 호출

2.1. LCEL을 사용하지 않는 경우

LCEL을 사용하지 않으면 다음 코드 조각을 사용하여 특정 주제 문자열을 전달하고 농담 문자열을 검색할 수 있습니다.

from typing import List
import openai

prompt_template = "Tell me a short joke about {topic}"
client = openai.OpenAI()

def call_chat_model(messages: List[dict]) -> str:
    response = client.chat.completions.create(
        model="gpt-3.5-turbo", 
        messages=messages,
    )
    return response.choices[0].message.content

def invoke_chain(topic: str) -> str:
    prompt_value = prompt_template.format(topic=topic)
    messages = [{"role": "user", "content": prompt_value}]
    return call_chat_model(messages)

invoke_chain("ice cream")

2.2. LCEL을 사용하는 경우

반면에, LCEL을 사용하면 동일한 기능을 더 간결하게 구현할 수 있습니다. 아래 코드 조각은 LCEL 객체를 사용하여 체인을 쉽게 구축하는 방법을 보여줍니다.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

prompt = ChatPromptTemplate.from_template(
    "Tell me a short joke about {topic}"
)
output_parser = StrOutputParser()
model = ChatOpenAI(model="gpt-3.5-turbo")
chain = (
    {"topic": RunnablePassthrough()} 
    | prompt
    | model
    | output_parser
)

chain.invoke("ice cream")

3. 스트리밍

3.1. LCEL을 사용하지 않는 경우

다음 코드 조각은 LCEL을 사용하지 않고 결과를 스트리밍하는 방법을 보여줍니다.

from typing import Iterator

def stream_chat_model(messages: List[dict]) -> Iterator[str]:
    stream = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages,
        stream=True,
    )
    for response in stream:
        content = response.choices[0].delta.content
        if content is not None:
            yield content

def stream_chain(topic: str) -> Iterator[str]:
    prompt_value = prompt.format(topic=topic)
    stream = stream_chat_model([{"role": "user", "content": prompt_value}])
    for chunk in stream:
        print(chunk, end="", flush=True)

stream_chain("ice cream")

3.2. LCEL을 사용하는 경우

LCEL을 사용하여 결과를 스트리밍하는 것이 더 간편합니다. 아래 코드 조각은 LCEL을 사용하여 스트리밍 하는 방법을 보여줍니다.

for chunk in chain.stream("ice cream"):
    print(chunk, end="", flush=True)

4. 일괄 처리

4.1. LCEL을 사용하지 않는 경우

다음 코드 조각은 LCEL을 사용하지 않고 입력의 일괄 처리를 병렬로 수행하는 방법을 보여줍니다.

from concurrent.futures import ThreadPoolExecutor

def batch_chain(topics: list) -> list:
    with ThreadPoolExecutor(max_workers=5) as executor:
        return list(executor.map(invoke_chain, topics))

batch_chain(["ice cream", "spaghetti", "dumplings"])

4.2. LCEL을 사용하는 경우

LCEL을 사용하여 일괄 처리하는 것은 매우 간단합니다. 아래 코드 조각은 LCEL을 사용하여 일괄 처리하는 방법을 보여줍니다.

chain.batch(["ice cream", "spaghetti", "dumplings"])