Langchain - Model 컴포넌트와 호출 방법

Model은 Langchain 에서 ChatGPT나 PaLM 모델을 추상화한 객체이다.

보통 LLM모델들이 문장을 완성하는 LLM  기능 (질문에 대한 답변, 문서 요약, 아이디어 제공 등등)의 멀티모달 모델(__AzureOpenAI__)과,사람과 상호작용을 하는 채팅을 위한 챗봇 모델(__AzureChatOpenAI__), 이렇게 두 가지 모델을 제공한다.

 

Langchain도 마찬가지로 이 두 가지 모델에 대한 추상화 계층을 제공하고 있다.

LM은 입력된 프롬프트의 명령에 따라 답변을 내는 모델이다.

ChatGPT, PaLM API등의 LLM 모델에 대한 추상화 객체를 제공한다.

 

LLM 모델 객체를 생성하는 방법은 모델 제공자에 따라 다르며, 특히 모델 제공자에 따라서 지원하는 튜닝 가능한 패러미터들도 다르다.

 

예를 들어 chatgpt의 경우에는 temperature값을 설정할 수 있고, 구글의 PaLM Vertex AI의 경우 temperature, Top-K/P 등의 값을 추가로 설정할 수 있다.

 

Langchain에서 지원되는 LLM 모델에 대해서는 langchain 공식 문서 https://python.langchain.com/docs/integrations/llms/ 를 참고하기 바란다.

 

LCEL (LangChain Expression Language)

Chain은 개념적으로는 훌륭하지만, 코드양이 다소 많아지고, 병렬처리나 비동기 처리, 스트리밍 같은 고급 기능을 구현하기 어렵다.

이런 한계를 극복하기 위해서 2023년 8월에 LangChain Expression Language (이하 LCEL이 개발되었다.

Chain의 기능을 대처하는 컴포넌트로, 병렬, 비동기, 스트리밍 같은 고급 워크플로우 처리에서 부터 FallBack이나 Retry 와 같은 장애 처리 기능을 지원하며,

Langchain 모니터링/평가 솔루션인 LangSmith와 쉽게 연동이 된다.

 

Invoke

동기 방식으로 LLM에 질문을 하는 방법

from dotenv import load_dotenv
import os
from typing import List
import openai

# 환경 변수 로드
load_dotenv()

# 환경 변수 사용
azure_openai_api_key = os.getenv('azure_openai_api_key')
azure_openai_endpoint = os.getenv('azure_openai_endpoint')
azure_open_api_version = os.getenv('azure_open_api_version')
azure_deployment_gpt35turbo = os.getenv('azure_deployment_gpt35turbo')


client = openai.AzureOpenAI(
    api_version=azure_open_api_version,
    azure_deployment=azure_deployment_gpt35turbo,    
    azure_endpoint=azure_openai_endpoint,
    api_key=azure_openai_api_key,
)

prompt_template = "{topic}에 대한 짧은 농담을 들려줘."

def call_chat_model(messages: List[dict]) -> str:
    response = client.chat.completions.create(
        model="gpt-4",
        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("아이스크림")

"당신: 아이스크림이 결혼할 때 무슨 말을 할까요?\n챗봇: \"당신이 내 사랑스러운 콘입니다!\"

#LCEL
from langchain_openai import AzureChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

#LCEL
from langchain_openai import AzureChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

prompt = ChatPromptTemplate.from_template(
    "{topic}에 대한 짧은 농담을 들려줘."
)
output_parser = StrOutputParser()
model = AzureChatOpenAI(
    api_version=azure_open_api_version,
    azure_deployment=azure_deployment_gpt35turbo,    
    azure_endpoint=azure_openai_endpoint,
    api_key=azure_openai_api_key,
    )
chain = (
    {"topic": RunnablePassthrough()} 
    | prompt
    | model
    | output_parser
)

chain.invoke("아이스크림")

"당연하지, 여기 한 개 있어. \n\n\"아이스크림 만큼 달콤한 건 없어. 그럼에도 불구하고, 나는 아이스크림을 먹을 때마다 참을성이 녹아버린다!\"

 

 

앞의 예제는 LLM을 호출할때, 단순하게, 하나의 프롬프트만 동기형으로 호출하였는데,

여러개의 프롬프트를 동시에 호출하거나 또는 호출을 비동기로 하는 등의 다양한 호출 패턴이 있을 수 있다.

Langchain은 이러한 다양한 호출 패턴을 지원한다.

(참고. 지원하는 LLM에 따라서 지원되는 호출 패턴이 다를 수 있으니 https://python.langchain.com/docs/integrations/llms/ 에서 지원 되는 호출 패턴을 미리 확인하기 바란다.)

 

Stream

모델의 사이즈가 크거나 프롬프트가 클때는 응답 시간이 느릴 수 있다.

그래서 이 응답을 모두 기다렸다가 결과를 출력하는데 시간이 많이 소요될 수 있는데, 이런 문제를 해결 하기 위해서 Langchain은 스트리밍 패턴을 지원한다.

응답이 생성되는대로 스트리밍으로 응답을 실시간으로 리턴할 수 있다.

주로 챗봇과 같은 실시간성이 필요한 애플리케이션에서 자주 사용된다.

stream()을 사용하면, 계속해서 결과를 리턴하는 것을 확인할 수 있다.

(예제를 실행해보면 애니메이션 처럼, 문장이 순차적으로 출력되는 것을 확인할 수 있다.)

#without LCEL
from typing import Iterator
from typing import List
import openai

client = openai.AzureOpenAI(
    api_version=azure_open_api_version,
    azure_deployment=azure_deployment_gpt35turbo,    
    azure_endpoint=azure_openai_endpoint,
    api_key=azure_openai_api_key,
)

# prompt_template = "Tell me a short joke about {topic}"
prompt_template = "{topic}에 대하여 400자 이내로 설명해줘."

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

def stream_chain(topic: str) -> Iterator[str]:
    prompt_value = prompt_template.format(topic=topic)
    return stream_chat_model([{"role": "user", "content": prompt_value}])


for chunk in stream_chain("한국의 유명 아이돌"):
    print(chunk, end="", flush=True)

#LCEL
from langchain_openai import AzureChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

prompt = ChatPromptTemplate.from_template(
    "{topic}에 대하여 400자 이내로 설명해줘."
)
output_parser = StrOutputParser()
model = AzureChatOpenAI(
    api_version=azure_open_api_version,
    azure_deployment=azure_deployment_gpt35turbo,    
    azure_endpoint=azure_openai_endpoint,
    api_key=azure_openai_api_key,
    )
chain = (
    {"topic": RunnablePassthrough()} 
    | prompt
    | model
    | output_parser
)

for chunk in chain.stream("한국의 유명 아이돌"):
    print(chunk, end="", flush=True)

Batch

만약에 여러개의 질문이나 명령을 내려야 할 경우,

루프를 도는 것이 아니라, 파이썬 리스트에 프롬프트 목록을 저장한 후에 batch 호출을 이용하여 한번에 여러 질문을 호출할 수 있다.

#Without LCEL
from concurrent.futures import ThreadPoolExecutor
from typing import List
import openai

client = openai.AzureOpenAI(
    api_version=azure_open_api_version,
    azure_deployment=azure_deployment_gpt35turbo,    
    azure_endpoint=azure_openai_endpoint,
    api_key=azure_openai_api_key,
)

prompt_template = "{topic}에 대한 짧은 농담을 들려줘."

def call_chat_model(messages: List[dict]) -> str:
    response = client.chat.completions.create(
        model="gpt-4",
        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)

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

batch_chain(["아이스크림", "스파게티", "떡복이"])

["당신: 왜 아이스크림이 컴퓨터를 싫어할까요?\nAI: 모든 커넥션을 끊어버리기 때문에요!","왜 스파게티가 행복한가요?\n파스타 수준에서 살고 있기 때문이죠!","떡복이가 떡볶이 먹고 있다가 물어보면, \"떡복이 먹냐?\""]

#LCEL
from langchain_openai import AzureChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

prompt = ChatPromptTemplate.from_template(
    "{topic}에 대한 짧은 농담을 들려줘."
)
output_parser = StrOutputParser()
model = AzureChatOpenAI(
    api_version=azure_open_api_version,
    azure_deployment=azure_deployment_gpt35turbo,    
    azure_endpoint=azure_openai_endpoint,
    api_key=azure_openai_api_key,
    )
chain = (
    {"topic": RunnablePassthrough()} 
    | prompt
    | model
    | output_parser
)

chain.batch(["아이스크림", "스파게티", "떡복이"])

["당신: 왜 아이스크림은 학교에서 인기가 많을까요?\nAI: 그런데 왜요?\n당신: 아이스크림을 좋아하는 학생들이 항상 '아이스크림이 녹을 때까지 기다리면 한번에 먹을 수 있다'고 말하기 때문이죠!","어느 날 한 남자가 레스토랑에 들어가 스파게티를 주문했습니다. 그런데 남자가 스파게티를 먹으며 신기하게 웃기 시작했어요. 서둘러 그에게 다가가서 물어보았죠. \"왜 웃고 있어요?\" 그는 대답했습니다. \"스파게티를 먹으면서 마음이 풍요로워지니까요! 이게 바로 '파스타 풀'이라는 건가요?\"","떡복이가 자기 소개하면서 \"저는 떡복이예요. 떡과 복이 따로 있지만 합쳐서 먹으면 맛있는 놈이죠!\"라고 말했다."]

 

Async

비동기 방식으로도 호출이 가능하다.

#without LCEL
from typing import List
import openai

prompt_template = "{topic}에 대한 짧은 농담을 들려줘."

async_client = openai.AsyncAzureOpenAI(
    api_version=azure_open_api_version,
    azure_deployment=azure_deployment_gpt35turbo,    
    azure_endpoint=azure_openai_endpoint,
    api_key=azure_openai_api_key,
)

async def acall_chat_model(messages: List[dict]) -> str:
    response = await async_client.chat.completions.create(
        model="gpt-4", 
        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)

await ainvoke_chain("아이스크림")

#LCEL
from langchain_openai import AzureChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import asyncio
import nest_asyncio

prompt = ChatPromptTemplate.from_template(
    "{topic}에 대한 짧은 농담을 들려줘."
)
output_parser = StrOutputParser()
model = AzureChatOpenAI(
    api_version=azure_open_api_version,
    azure_deployment=azure_deployment_gpt35turbo,    
    azure_endpoint=azure_openai_endpoint,
    api_key=azure_openai_api_key,
    )
chain = (
    {"topic": RunnablePassthrough()} 
    | prompt
    | model
    | output_parser
)

await chain.ainvoke("아이스크림")