The Yellow Lion King 데이터와 함께 살아가기

 안녕하세요

지난 주 AWS Partner Tech Grue Community Meet-up 에 다녀 왔습니다. 

TGC는 AWS 에서 Partner Engineer를 대상으로 기술 관련 공유 토의하는 모임 입니다.
올해 그 첫번째 모임으로 지난주 강남 센터필드에서 열렸어요. 오신 분들 모두 파트너사의 기술 관련 업무를 (한가닥?) 하시는 분들이었어요. 조건이 파트너사의 직원이고 AWS 자격증 요건 또는 AWS Ambassador 등 여러가지가 있었습니다. 어려울 것 같았는데 저도 운 좋게 참여하게 되었습니다.

이번은 올해 첫 모임으로 전반적인 계획을 공유하는 시간이었고요, 기술 모임인 만큼 Ambassador 분들의 발표도 있었습니다.

저녁 퇴근후에 모임이다보니 출출함을 달래줄 피자와 치킨, 맥주와 음료 등도 제공해 주셨습니다.

새로운 기술과 트랜드 그리고 이에 대한 AWS의 새로운 기능 들을 먼저 만나볼 수있을 것 같아서 기분이 설래입니다.
앞으로 따끈따끈한 소식 전하겠습니다.
감사합니당~~~

지난 시간에는 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 섹션을 확인하세요. 그 다음에 살펴볼 좋은 사용 사례는 검색 증강 생성입니다.

LangChain 표현 언어 또는 LCEL은 체인을 쉽게 조립하는 선언적인 방법입니다. LCEL은 상용 서비스부터 프로토 타입을 구현하는 데 코드 변경 없이 지원하도록 처음부터 설계되었습니다. 가장 간단한 "프롬프트 + LLM" 체인에서부터 가장 복잡한 체인까지(사람들이 100개 이상의 단계로 구성된 LCEL 체인을 성공적으로 운영했습니다) 모두 지원합니다. LCEL을 사용하려는 이유 중에서 몇 가지를 강조해보겠습니다:

1. 스트리밍 지원: LCEL로 체인을 빌드하면 첫 번째 출력이 나올 때까지의 경과 시간 최대한 줄일 수 있습니다. 일부 체인의 경우, LLM에서 스트리밍 출력 파서로 토큰을 직접 전송하고, LLM이 원시 토큰을 출력하는 속도와 동일한 속도로 파싱된 증분 출력 묶음(청크)를 얻을 수 있습니다.
2. 비동기 지원: LCEL로 작성된 모든 체인은 동기식 API(예: 프로토타입을 만들 때 Jupyter 노트북에서) 및 비동기식 API(예: LangServe 서버에서)로 호출할 수 있습니다. 이는 프로토타입 및 프로덕션에서 동일한 코드를 사용하여 훌륭한 성능을 내고, 동일한 서버에서 많은 동시 요청을 처리할 수 있게 해줍니다.
3. 최적화된 병렬 실행: LCEL 체인에 병렬로 실행할 수 있는 단계가 있는 경우(예: 여러 Retriver에서 문서를 가져오는 경우), 대기 시간을 최소화 하기 위해서 동기 및 비동기 인터페이스를 자동으로 실행됩니다.
4. 재시도 및 예비 기능: LCEL 체인의 어떤 부분이든 재시도 및 예비 기능을 구성할 수 있습니다. 이것은 규모가 큰 체인을 더 신뢰할 수 있게 만드는 좋은 방법입니다. 현재 재시도/예비 기능에 대한 스트리밍 지원을 추가 중이므로 나중에는 추가된 신뢰성을 얻을 수 있으면서도 동시에 추가 대기 시간이 필요 없습니다.
5. 중간 결과에 대한 액세스: 더 복잡한 체인의 경우 종종 최종 출력이 생성되기 전에 중간 단계의 결과에 액세스하는 것이 매우 유용합니다. 이는 최종 사용자에게 무언가가 발생하고 있음을 알리거나 체인을 디버그하는 데 사용될 수 있습니다. 중간 결과를 스트리밍할 수 있으며 모든 LangServe 서버에서 사용 가능합니다.
6. 입력 및 출력 스키마: 입력 및 출력 스키마는 각 LCEL 체인에 대한 Pydantic 및 JSONSchema 스키마를 제공하며 체인의 구조에서 유추됩니다. 이는 입력 및 출력의 유효성 검사에 사용될 수 있으며 LangServe의 중요한 부분입니다.
7. 원활한 LangSmith 추적 통합: 체인이 더 복잡해질수록 모든 단계를 정확히 이해하는 것이 점점 중요해집니다. LCEL을 사용하면 모든 단계가 최대의 가시성과 디버깅 가능성을 위해 자동으로 LangSmith에 로그됩니다.
8. 원활한 LangServe 배포 통합: LCEL로 생성된 모든 체인은 LangServe를 사용하여 쉽게 배포할 수 있습니다.

LCEL 시작하기 : Get started

LCEL은 기본 콤포넌트를 가지고 복잡한 체인을 쉽게 만들 수 있게 합니다.  그리고 로깅과 병렬처리, 스트리밍과 같은 외부 기능으로의 확대를 지원 합니다.

기본 예제 : 프롬프트 + 모델 + 출력 파서

가장 기본적이고 일반적인 사용 예제는 프롬프트 템플릭과 모델을 함께 체인으로 묶는 것 입니다. 이것이 어떻게 동작하는지 보기 위해서 하나의 토픽을 받아서 하나의 농담을 생성하는 체인을 만들어 보시지요.

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("tell me a short joke about {topic}")
model = ChatOpenAI()
output_parser = StrOutputParser()

chain = prompt | model | output_parser

chain.invoke({"topic": "ice cream"})

"Why did the ice cream go to therapy?\n\nBecause it had too many toppings and couldn't find its cone-fidence!"

아래의 코드를 통해서 우리는 다른 콤포넌트들을 LCEL을 이용해서 하나로 묶을 수 있습니다.

chain = prompt | model | output_parser

여기서 | 기호는 유닉스의 파이프 연산자 입니다. 이것은 하나의 콤포넌트에서 나온 아웃풋을 다음 콤포넌트의 입력으로 제공하여 체인을 만듭니다.

이 체인의 입력은 프롬프트 템플릿으로 전달 됩니다. 그리고 그 프롬프트 템플릿의 출력은 모델로 전달 됩니다. 그리고 모델의 출력은 아웃풋 파서에게 전달 됩니다. 무슨일이 벌어지는지 정확히 이해하기 위해서 콤포넌트들을 각각 알아보시죠.

1. Prompt​

프롬프트는 BasePromptTemplate입니다. 이것은 템플릿 변수로 사전(dictionary)을 받아서 PromptValue를 생성합니다. PromptValue는 (입력으로 문자열을 받는) LLM 이나 (입력으로 연속된 메시지를 받는) ChatModel에 전달될 수 있는 완성된 프롬프트로 만드는 랩퍼입니다. 이것은 BaseMessages나 문자열을 만드는 로직을 정의하기 때문에 두 언어모델과 함께 사용 할 수 있습니다.

prompt_value = prompt.invoke({"topic": "ice cream"})
prompt_value

# 출력 내용
ChatPromptValue(messages=[HumanMessage(content='tell me a short joke about ice cream')])

prompt_value.to_messages()

# 출력 내용 : HumanMessage 객체
[HumanMessage(content='tell me a short joke about ice cream')]

prompt_value.to_string()

# 출력 내용: 문자열
'Human: tell me a short joke about ice cream'

2. Model​

그리고나서 PromptValue 는 모델에게 전달 됩니다. 이번에 우리가 사용한 모델이 ChatModel이기 때문에 아웃풋은 BaseMessage일 껍니다.

message = model.invoke(prompt_value)
message

# 출력 내용
AIMessage(content="Why did the ice cream go to therapy? \n\nBecause it had too many toppings and couldn't find its cone-fidence!")

만약에 우리의 모델이 LLM이라면 이것의 출력은 문자열일 껍니다.

from langchain.llms import OpenAI

llm = OpenAI(model="gpt-3.5-turbo-instruct")
llm.invoke(prompt_value)

3. Output parser​

그리고 마지막으로 우리는 우리의 모델 아웃풋을 아웃풋 파서에 전달 합니다. 아웃풋 파써가 BaseOutputParser라는 것은 문자열이나 BaseMessage를 입력으로 받는 다는 것을 의미합니다.

output_parser.invoke(message)

# 출력 내용
"Why did the ice cream go to therapy? \n\nBecause it had too many toppings and couldn't find its cone-fidence!"

4. 전체 Pipeline​

다음의 단계를을 따라 갑니다.:

1. {"topic": "ice cream"}과 같이 원하는 토픽을 사용자 입력으로 전달합니다.
2. 프롬프트는 사용자 입력을 받는데 이것은 토픽을 이용해서 프롬프트를 만든 다음 PromptValue를 생성하는데 사용합니다.
3. 모델 콤포넌트는 생성된 프롬프트를 받아서 OpenAI LLM 모델에 전달합니다. 이 모델의 출력으로 생성된 것은 ChatMessage 오브젝트입니다. 
4. 마지막으로, output_parser 콤포넌트는 ChatMessage 오브젝트를 입력 받고 파이썬 문자열로 변환하여 invoke 메소드로 반환합니다. 

만약에 어떤 콤포넌트든지 출력에 대해서 궁금한 점이 있다면 이 체인의 작은 버전을 언제든지 테스트 할 수 있습니다. 예를 들면 중간 결과를 보기 위해서 prompt 또는 prompt | model 같은 것을 테스트 할 수 있습니다. 

input = {"topic": "ice cream"}

# 프롬프트만 실행
prompt.invoke(input)
# > ChatPromptValue(messages=[HumanMessage(content='tell me a short joke about ice cream')])

# 프롬프트와 모델만 실행
(prompt | model).invoke(input)
# > AIMessage(content="Why did the ice cream go to therapy?\nBecause it had too many toppings and couldn't cone-trol itself!")

RAG 검색 예시

다음 예시로 우리는 질문에 대답할 때 어떤 내용을 추가하는 retrieval-augmented generation 체인을 만들기 원합니다.

# Requires:
# pip install langchain docarray tiktoken

from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate
from langchain.vectorstores import DocArrayInMemorySearch
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

vectorstore = DocArrayInMemorySearch.from_texts(
    ["harrison worked at kensho", "bears like to eat honey"],
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()

template = """Answer the question based only on the following context:
{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()
output_parser = StrOutputParser()

setup_and_retrieval = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
)
chain = setup_and_retrieval | prompt | model | output_parser

chain.invoke("where did harrison work?")

이 경우에 구성된 체인은 다음과 같습니다.

chain = setup_and_retrieval | prompt | model | output_parser

제일 첫번째 단계로 우리는 메모리 저장소를 이용해서 retriever를 설정할 수 있습니다. 이 retriever는 질문에 기반해서 문서를 가져올 수 있습니다. 이것은 다른 콤포넌트들과 함께 묶여질 수 있을 뿐만 아니라 실행될 수 있는 콤포넌트입니다. 뿐만아니라 여러분은 retriever만 분리해서 실행 할 수도 있습니다. 

retriever.invoke("where did harrison work?")

setup_and_retrieval = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
)

setup_and_retrieval = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
)
chain = setup_and_retrieval | prompt | model | output_parser

흐름은 다음과 같습니다:

1. 첫 번째 단계에서는 두 항목을 포함하는 RunnableParallel 객체를 만듭니다. 첫 번째 항목인 "context"에는 retriever에 의해 가져온 문서 결과가 포함됩니다. 두 번째 항목인 "question"에는 사용자의 원래 질문이 포함됩니다. 질문을 전달하기 위해 RunnablePassthrough를 사용하여 복사합니다.
2. 위 단계에서 생성된 사전을 프롬프트 컴포넌트에 전달합니다. 그런 다음 사용자 입력인 "question" 과 검색된 문서인 "context"를 사용하여 프롬프트를 구성하고 PromptValue를 출력합니다.
3. 모델 컴포넌트는 생성된 프롬프트를 가져와 OpenAI LLM 모델에 전달하여 실행 평가합니다. 모델에서 생성된 출력은 ChatMessage 객체입니다.
4. 마지막으로 output_parser 구성 요소는 ChatMessage를 입력으로 받아서 이를 Python 문자열로 변환하고, invoke 메서드에 반환합니다.

Next steps​

동일한 기능을 LCEL로 구현할 때의 코드와 LCEL 없이 구현할 때의, 두 코드를 나란히 놓고 비교하는 내용이 있는 '왜 LCEL을 사용하는지'를 읽어 보시기 바랍니다. 

이번 시간을 통해서 LCEL이 무엇인지 알아 보았습니다. 그리고 prompt + model  + output_parsert 형태의 기본 예제를 만들고 이해했으며, 마지막에는 RAG 예제를 통해서 관련된 문서를 입력 프롬프트에서 함께 사용하도록 하는 방법에 대해서 알아 보았습니다.

LangServe Error

현상:

LangServe 파일을 실행시 서버가 올라오지 못하고 아래의 에러를 출력하면서 종료되는 현상 발생

ERROR:    [Errno 99] error while attempting to bind on address ('::1', 8000, 0, 0): cannot assign requested address

원인:

위와 같은 langserve 파일에서 localhost를 OS가 제대로 할당해주지 못해서 발생하는 에러입니다.

조치: 

따라서, 위의 코드에서 localhost 문자 대신에 0.0.0.0  으로 바꾸어 주면 잘 동작 합니다.