LCEL-왜 LCEL을 사용하는가?

인공지능-기계학습/LangChain

LCEL-왜 LCEL을 사용하는가?

The Yellow Lion King 2024. 1. 21. 15:28

지난 시간에는 LCEL Get-started를 통해서 LCEL의 간단하면서 다양한 예시를 알아보았습니다.

이번 시간에는 LCEL을 사용하는 이유에 대해서 알아보도록 하겠습니다. 즉, 사용하면 무엇이 좋은지 어떤 가치가 있는지 알아보겠습니다. 참고로 LCEL Get-started를 먼저 보고 오시면 이해에 도움이 됩니다. 혹시 안보셨다면 먼저 읽어보시기를 추천 드립니다.

LCEL은 간단한 콤포넌트 부터 복잡한 체인을 쉽게 만들 수 있게 합니다. 바로 다음과 같은 것들을 이용해서 만듭니다.

1. 통일된 인터페이스:

모든 LCEL 객체는 Runnable 인터페이스를 구현합니다. Runnable은 호출하는 메소드들의 공통 집합을 정의합니다.(invoke, batch, stream, ainvoke,...등과 같은...) 이것은 LCEL객체가 자동적으로 이러한 호출들을 지원할 수 있게 만듭니다. 즉, LCEL 객체로 만든 모든 체인은 하나의 LCEL객체라는 것을 말합니다.

2. 원시도구들의 구성:

LCEL의 가치를 이해하기 위해서, LCEL 없이 비슷한 기능을 어떻게 개발하는 지 생각해보고 동작하는 지를 살펴보는 것은 도움이 됩니다. LCEL은 콤포넌트들을 병렬화하고, fallbacks을 추가하고, 체인 내부의 구성을 동적으로 바꾸는 등을 통해서 체인을 쉽게 구성하게 해줍니다. 이번 포스팅에서는 지난 번 LCEL Get-started의 내용에 있던 기본 예시를 다루어 보겠습니다. 간단한 프롬프트와 모델이 결합된 체인을 다루어 볼껀데요, 이것은 이미 많은 기능들이 정의되어 있습니다. 이것을 재생성하는 데 필요한 것이 무엇인지 살펴보겠습니다.

준비

먼저 공통적으로 필요한 체인 하나를 만들어 보겠습니다.

%pip install –upgrade –quiet langchain-core langchain-openai

관련된 패키지를 설치 합니다.

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


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

chain = prompt | model | output_parser

프롬프트, 모델, 아웃풋 파서를 연결한 체인을 만듭니다.

이제부터 다음 내용을 구현할 때 두가지 경우의 코드를 비교하겠습니다. 즉, LCEL 없이 구현할 때와 LCEL로 구현할 때의 코드를 보겠습니다.

invoke, stream, batch, async, changing model provider, runtime configuration, logging, Fallback

Invoke

제일 간단한 사용예를 보면, 토픽(주제) 문자열을 전달해 주 농담 문자열을 받는 것 입니다. 아래는 이러한 내용을 LCEL 없이 개발했을 때와 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")

LCEL로 만드는 경우

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")

Stream

만약에 결과로 스트림을 원한다면 위에서 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)
    return stream_chat_model([{"role": "user", "content": prompt_value}])


for chunk in stream_chain("ice cream"):
    print(chunk, end="", flush=True)

LCEL로 만드는 경우

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

LCEL로 만드는 경우는 그저 두 줄의 코드면 충분합니다.

Batch

병렬적으로 여러 입력을 배치로 실행하기 원한다면 또 다시 새로운 함수가 필요합니다.

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"])

LCEL로 만드는 경우

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

Async

비동기 버전을 원한다면,

LCEL 없이 만드는 경우

async_client = openai.AsyncOpenAI()

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

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

LCEL로 만드는 경우

chain.ainvoke("ice cream")

Chat Model 대신 LLM 사용

chat 엔드포인트 대신에 completion 엔드포인트를 사용하기 원한다면,

LCEL없이 만드는 경우

def call_llm(prompt_value: str) -> str:
    response = client.completions.create(
        model="gpt-3.5-turbo-instruct",
        prompt=prompt_value,
    )
    return response.choices[0].text

def invoke_llm_chain(topic: str) -> str:
    prompt_value = prompt_template.format(topic=topic)
    return call_llm(prompt_value)

invoke_llm_chain("ice cream")

LCEL로 만드는 경우

from langchain_openai import OpenAI

llm = OpenAI(model="gpt-3.5-turbo-instruct")
llm_chain = (
    {"topic": RunnablePassthrough()} 
    | prompt
    | llm
    | output_parser
)

llm_chain.invoke("ice cream")

모델 제공자가 다른 경우

OpenAI 대신에 Anthropic을 사용하기 원하면,

LCEL 없이 만드는 경우

import anthropic

anthropic_template = f"Human:\n\n{prompt_template}\n\nAssistant:"
anthropic_client = anthropic.Anthropic()

def call_anthropic(prompt_value: str) -> str:
    response = anthropic_client.completions.create(
        model="claude-2",
        prompt=prompt_value,
        max_tokens_to_sample=256,
    )
    return response.completion    

def invoke_anthropic_chain(topic: str) -> str:
    prompt_value = anthropic_template.format(topic=topic)
    return call_anthropic(prompt_value)

invoke_anthropic_chain("ice cream")

LCEL로 만드는 경우

from langchain_community.chat_models import ChatAnthropic

anthropic = ChatAnthropic(model="claude-2")
anthropic_chain = (
    {"topic": RunnablePassthrough()} 
    | prompt 
    | anthropic
    | output_parser
)

anthropic_chain.invoke("ice cream")

실행 환경 설정(Runtime configurability)

LCEL 없이 만드는 경우

def invoke_configurable_chain(
    topic: str, 
    *, 
    model: str = "chat_openai"
) -> str:
    if model == "chat_openai":
        return invoke_chain(topic)
    elif model == "openai":
        return invoke_llm_chain(topic)
    elif model == "anthropic":
        return invoke_anthropic_chain(topic)
    else:
        raise ValueError(
            f"Received invalid model '{model}'."
            " Expected one of chat_openai, openai, anthropic"
        )

def stream_configurable_chain(
    topic: str, 
    *, 
    model: str = "chat_openai"
) -> Iterator[str]:
    if model == "chat_openai":
        return stream_chain(topic)
    elif model == "openai":
        # Note we haven't implemented this yet.
        return stream_llm_chain(topic)
    elif model == "anthropic":
        # Note we haven't implemented this yet
        return stream_anthropic_chain(topic)
    else:
        raise ValueError(
            f"Received invalid model '{model}'."
            " Expected one of chat_openai, openai, anthropic"
        )

def batch_configurable_chain(
    topics: List[str], 
    *, 
    model: str = "chat_openai"
) -> List[str]:
    # You get the idea
    ...

async def abatch_configurable_chain(
    topics: List[str], 
    *, 
    model: str = "chat_openai"
) -> List[str]:
    ...

invoke_configurable_chain("ice cream", model="openai")
stream = stream_configurable_chain(
    "ice_cream", 
    model="anthropic"
)
for chunk in stream:
    print(chunk, end="", flush=True)

# batch_configurable_chain(["ice cream", "spaghetti", "dumplings"])
# await ainvoke_configurable_chain("ice cream")

LCEL로 만드는 경우

from langchain_core.runnables import ConfigurableField


configurable_model = model.configurable_alternatives(
    ConfigurableField(id="model"), 
    default_key="chat_openai", 
    openai=llm,
    anthropic=anthropic,
)
configurable_chain = (
    {"topic": RunnablePassthrough()} 
    | prompt 
    | configurable_model 
    | output_parser
)



configurable_chain.invoke(
    "ice cream", 
    config={"model": "openai"}
)
stream = configurable_chain.stream(
    "ice cream", 
    config={"model": "anthropic"}
)
for chunk in stream:
    print(chunk, end="", flush=True)

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

# await configurable_chain.ainvoke("ice cream")

로깅(Logging)

LCEL 없이 만드는 경우

def invoke_anthropic_chain_with_logging(topic: str) -> str:
    print(f"Input: {topic}")
    prompt_value = anthropic_template.format(topic=topic)
    print(f"Formatted prompt: {prompt_value}")
    output = call_anthropic(prompt_value)
    print(f"Output: {output}")
    return output

invoke_anthropic_chain_with_logging("ice cream")

LCEL 로 만드는 경우

import os

os.environ["LANGCHAIN_API_KEY"] = "..."
os.environ["LANGCHAIN_TRACING_V2"] = "true"

anthropic_chain.invoke("ice cream")

Fallbacks

LCEL 없이 만드는 경우

def invoke_chain_with_fallback(topic: str) -> str:
    try:
        return invoke_chain(topic)
    except Exception:
        return invoke_anthropic_chain(topic)

async def ainvoke_chain_with_fallback(topic: str) -> str:
    try:
        return await ainvoke_chain(topic)
    except Exception:
        # Note: we haven't actually implemented this.
        return ainvoke_anthropic_chain(topic)

async def batch_chain_with_fallback(topics: List[str]) -> str:
    try:
        return batch_chain(topics)
    except Exception:
        # Note: we haven't actually implemented this.
        return batch_anthropic_chain(topics)

invoke_chain_with_fallback("ice cream")
# await ainvoke_chain_with_fallback("ice cream")
batch_chain_with_fallback(["ice cream", "spaghetti", "dumplings"]))

LCEL 로 만드는 경우

fallback_chain = chain.with_fallbacks([anthropic_chain])

fallback_chain.invoke("ice cream")
# await fallback_chain.ainvoke("ice cream")
fallback_chain.batch(["ice cream", "spaghetti", "dumplings"])

전체 코드 비교

이렇게 단순한 경우에도 LCEL 체인이 많은 기능들을 잘 압축해서 제공합니다. 체인들이 더 복잡해 줄 수록 이런 간단함은 더욱 특별한 가치를 가지게 됩니다.

LCEL 없이 만드는 경우

from concurrent.futures import ThreadPoolExecutor
from typing import Iterator, List, Tuple

import anthropic
import openai


prompt_template = "Tell me a short joke about {topic}"
anthropic_template = f"Human:\n\n{prompt_template}\n\nAssistant:"
client = openai.OpenAI()
async_client = openai.AsyncOpenAI()
anthropic_client = anthropic.Anthropic()

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:
    print(f"Input: {topic}")
    prompt_value = prompt_template.format(topic=topic)
    print(f"Formatted prompt: {prompt_value}")
    messages = [{"role": "user", "content": prompt_value}]
    output = call_chat_model(messages)
    print(f"Output: {output}")
    return output

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]:
    print(f"Input: {topic}")
    prompt_value = prompt.format(topic=topic)
    print(f"Formatted prompt: {prompt_value}")
    stream = stream_chat_model([{"role": "user", "content": prompt_value}])
    for chunk in stream:
        print(f"Token: {chunk}", end="")
        yield chunk

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

def call_llm(prompt_value: str) -> str:
    response = client.completions.create(
        model="gpt-3.5-turbo-instruct",
        prompt=prompt_value,
    )
    return response.choices[0].text

def invoke_llm_chain(topic: str) -> str:
    print(f"Input: {topic}")
    prompt_value = promtp_template.format(topic=topic)
    print(f"Formatted prompt: {prompt_value}")
    output = call_llm(prompt_value)
    print(f"Output: {output}")
    return output

def call_anthropic(prompt_value: str) -> str:
    response = anthropic_client.completions.create(
        model="claude-2",
        prompt=prompt_value,
        max_tokens_to_sample=256,
    )
    return response.completion   

def invoke_anthropic_chain(topic: str) -> str:
    print(f"Input: {topic}")
    prompt_value = anthropic_template.format(topic=topic)
    print(f"Formatted prompt: {prompt_value}")
    output = call_anthropic(prompt_value)
    print(f"Output: {output}")
    return output

async def ainvoke_anthropic_chain(topic: str) -> str:
    ...

def stream_anthropic_chain(topic: str) -> Iterator[str]:
    ...

def batch_anthropic_chain(topics: List[str]) -> List[str]:
    ...

def invoke_configurable_chain(
    topic: str, 
    *, 
    model: str = "chat_openai"
) -> str:
    if model == "chat_openai":
        return invoke_chain(topic)
    elif model == "openai":
        return invoke_llm_chain(topic)
    elif model == "anthropic":
        return invoke_anthropic_chain(topic)
    else:
        raise ValueError(
            f"Received invalid model '{model}'."
            " Expected one of chat_openai, openai, anthropic"
        )

def stream_configurable_chain(
    topic: str, 
    *, 
    model: str = "chat_openai"
) -> Iterator[str]:
    if model == "chat_openai":
        return stream_chain(topic)
    elif model == "openai":
        # Note we haven't implemented this yet.
        return stream_llm_chain(topic)
    elif model == "anthropic":
        # Note we haven't implemented this yet
        return stream_anthropic_chain(topic)
    else:
        raise ValueError(
            f"Received invalid model '{model}'."
            " Expected one of chat_openai, openai, anthropic"
        )

def batch_configurable_chain(
    topics: List[str], 
    *, 
    model: str = "chat_openai"
) -> List[str]:
    ...

async def abatch_configurable_chain(
    topics: List[str], 
    *, 
    model: str = "chat_openai"
) -> List[str]:
    ...

def invoke_chain_with_fallback(topic: str) -> str:
    try:
        return invoke_chain(topic)
    except Exception:
        return invoke_anthropic_chain(topic)

async def ainvoke_chain_with_fallback(topic: str) -> str:
    try:
        return await ainvoke_chain(topic)
    except Exception:
        return ainvoke_anthropic_chain(topic)

async def batch_chain_with_fallback(topics: List[str]) -> str:
    try:
        return batch_chain(topics)
    except Exception:
        return batch_anthropic_chain(topics)

LCEL로 만드는 경우

import os

from langchain_community.chat_models import ChatAnthropic
from langchain_openai import ChatOpenAI
from langchain_openai import OpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, ConfigurableField

os.environ["LANGCHAIN_API_KEY"] = "..."
os.environ["LANGCHAIN_TRACING_V2"] = "true"

prompt = ChatPromptTemplate.from_template(
    "Tell me a short joke about {topic}"
)
chat_openai = ChatOpenAI(model="gpt-3.5-turbo")
openai = OpenAI(model="gpt-3.5-turbo-instruct")
anthropic = ChatAnthropic(model="claude-2")
model = (
    chat_openai
    .with_fallbacks([anthropic])
    .configurable_alternatives(
        ConfigurableField(id="model"),
        default_key="chat_openai",
        openai=openai,
        anthropic=anthropic,
    )
)

chain = (
    {"topic": RunnablePassthrough()} 
    | prompt 
    | model 
    | StrOutputParser()
)

요약

이번 시간에는 LCEL을 실제로 사용할 때와 안했을 때의 차이 비교를 통해서 LCEL이 제공하는 가치에 대해서 알아보았습니다. LCEL이 통일된 인터페이스를 제공하고 많은 원시 구성 요소들을 제공하기 때문에 간단한 메소드 호출 만으로 많은 기능을 대신할 수 있음을 사례로 살표 보았습니다. 이러한 사례를 통해서 단순하고 간결한 코드 작성으로 구현이 가능함을 확인 하였습니다.

Next Steps

LCEL에 대한 학습을 계속하려면 다음을 권장합니다:

여기서 부분적으로 다룬 LCEL 인터페이스에 대해 전체 내용을 자세히 읽어보세요.

LCEL이 제공하는 추가적인 구성 기본 요소에 대해 학습하기 위해서 How-to 섹션을 탐험하세요.

공통 사용 사례에 대한 LCEL 동작을 보려면 Cookbook 섹션을 확인하세요. 그 다음에 살펴볼 좋은 사용 사례는 검색 증강 생성입니다.

저작자표시 비영리 변경금지