Langchain 코드 패턴 사전

Langchain으로 LLM을 활용한 코딩 작업을 할 때 종종 코드 패턴이 기억나지 않아 막히는 경험을 합니다. 이 포스팅에서는 langchain에서 사용하는 다양한 코드 패턴을 정리하여 두고두고 써먹을 수 있도록 langchain 코드 패턴 사전을 만들어 보겠습니다. 이 포스팅은 지속적으로 업데이트 되니 추가되었으면 하는 패턴이 있다면 댓글을 남겨주세요 🤗

Langchain 기본

기본 체인 실행 (chain.invoke)

Langchain의 가장 기본적인 chain 구성 및 실행은 아래와 같이 하면 됩니다. 3가지 요소, prompt, model 그리고 parser를 준비한 다음 pipe (|) 연산자로 순서대로 연결시키게 되면, chain이 하나 완성됩니다. 이 chain을 실행하는 방법은 여러 가지가 있는데, 그 중 가장 기본적이고 간단한 방법은 invoke 메서드를 이용하는 것입니다. invoke 메서드는 "동기적 체인 실행" 이라고 이해하면 되며, 체인 실행이 모두 끝날 때 까지 다음 작업으로 넘어가지 않습니다. 바로 다음에 알아볼 stream 메서드와의 차이점은, LLM이 생성하는 출력 토큰들을 그때그때마다 토큰 단위로 반환하는 것이 아니고, LLM의 결과 생성이 모두 완료되는 순간 토큰 전체를 반환한다는 특징이 있습니다.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser

# Define a prompt template
prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")

# Define the language model
model = ChatOpenAI(model="gpt-4o-mini")

# Create the chain
chain = prompt | model | StrOutputParser()

# Invoke the chain
result = chain.invoke({"topic": "programming"})
print(result)

기본 스트리밍 (chain.stream)

chain.invoke 와 달리 chain.stream 메서드는 LLM이 반환하는 토큰을 그때그때 반환하는 메서드입니다. 일종의 파이썬 generator를 만드는 것이라 이해하면 됩니다.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser

# Define a prompt template
prompt = ChatPromptTemplate.from_template("Tell me a story about {topic}")

# Define the language model
model = ChatOpenAI()

# Create the chain
chain = prompt | model | StrOutputParser()

# Stream the output
for chunk in chain.stream({"topic": "dragons"}):
    print(chunk, end="", flush=True)

비동기 체인 실행 (chain.ainvoke)

비동기 체인 실행은 앞서 알아본 동기적 체인 실행(chain.invoke)와는 달리 LLM 결과를 기다리는 동안 다른 작업을 수행할 수 있다는 차이가 있습니다. 만약 LLM chain을 복잡하게 구성하여, 최종 결과를 얻을 때까지 시간이 오래 걸린다면, ainvoke 메서드를 활용하여 그동안 다른 작업을 처리하는 식으로 효율적인 코드 구성이 가능합니다.

import asyncio
from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser

# Define a prompt template
prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")

# Define the language model
model = ChatOpenAI()

# Create the chain
chain = prompt | model | StrOutputParser()

# Define an asynchronous function
async def main():
    # Execute the chain asynchronously
    result = await chain.ainvoke({"topic": "programming"})
    print(result)

# Run the asynchronous function
asyncio.run(main())

# Response:
# Why do programmers prefer dark mode?
# Because light attracts bugs!

비동기 체인 스트리밍 (chain.astream)

chain.astream 메서드는 chain.stream 의 비동기 버전입니다. chain.stream 이 for문과 함께 사용되었다면, chain.astream의 경우 async for 문과 함께 사용되는 것에 주목해주세요.

import asyncio
from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser

# Define a prompt template
prompt = ChatPromptTemplate.from_template("Tell me a story about {topic}")

# Define the language model with streaming enabled
model = ChatOpenAI(streaming=True)

# Create the chain
chain = prompt | model | StrOutputParser()

# Define an asynchronous function to stream the output
async def main():
    async for chunk in chain.astream({"topic": "dragons"}):
        print(chunk, end="", flush=True)

# Run the asynchronous function
asyncio.run(main())

LangChain Expression Language (LCEL) 응용

RunnablePassthrough

RunnablePassthrough는 pipe로 연결할 수 있는 Runnable 객체이면서, 별다른 작업을 하지 않고 invoke, stream 등의 실행 메서드의 인자로 넘겨 받은 값을 그대로 넘겨주는(passthrough) 역할을 수행합니다.

이를 응용하면 아래와 같이 invoke에 3을 넘겨주고, 이를 chain 내부에서 x 라는 값에 바인딩하여 사용하는 식으로 사용이 가능해집니다. chain을 실행할 때, 단순히 값만 필요하게 되는 것이죠.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# Create a prompt template with multiple placeholders
prompt = ChatPromptTemplate.from_template("What is {x} * 3?")

# Define the language model with streaming enabled
model = ChatOpenAI(model="gpt-4o-mini")

# Create the chain using RunnablePassthrough
chain = {"x": RunnablePassthrough()} | prompt | model | StrOutputParser()

# Run the chain
print(chain.invoke(3))

# Response:
# 3 * 3 equals 9.

RunnableParallel

RunnableParallel을 활용하면 동시에 실행되는 chain 분기를 만드는 방법입니다. 여러 Runnable 객체가 동시에 실행되기 때문에 처리 시간이 최적화되는 장점이 있습니다. 개별 chain의 결과를 딕셔너리 형태로 결합하여 최종 결과를 내놓게 됩니다. 아래 예는 chain1 의 결과를 key x에, chain2의 결과를 key y에 바인딩하여 결과를 만드는 방법을 보여줍니다.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel

# Create a prompt template with multiple placeholders
prompt1 = ChatPromptTemplate.from_template("What is {x} * 3?")
prompt2 = ChatPromptTemplate.from_template("What is {y} * 4?")

# Define the language model with streaming enabled
model = ChatOpenAI(model="gpt-4o-mini")

# Create the chain using RunnablePassthrough
chain1 = {"x": RunnablePassthrough()} | prompt1 | model | StrOutputParser()
chain2 = {"y": RunnablePassthrough()} | prompt2 | model | StrOutputParser()

chain = RunnableParallel(x=chain1, y=chain2)

# Run the chain
print(chain.invoke(3))

# Response:
# {'x': '3 * 3 equals 9.', 'y': '3 * 4 equals 12.'}

RunnableLambda

RunnableLambda는 사용자가 정의한 함수를 Runnable 객체로 변환하여 chain의 요소로 포함할 수 있게 해줍니다. LLM 모델을 어떻게 보면 입력과 출력이 있는 함수라고 생각해본다면, RunnableLambda가 어떻게 정의되고 활용될 수 있을지 더 쉽게 감을 잡을 수 있을 겁니다.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser
from langchain_core.runnables import RunnableLambda

# Create a prompt template with multiple placeholders
prompt = ChatPromptTemplate.from_template("What is {x} * 3?")

# Define the language model with streaming enabled
model = ChatOpenAI(model="gpt-4o-mini")

# Create the chain using RunnablePassthrough
chain = {"x": RunnableLambda(lambda x: x**2)} | prompt | model | StrOutputParser()
# or
# chain = {"x": lambda x: x**2} | prompt | model | StrOutputParser()

# Run the chain
print(chain.invoke(3))

# Resposne:
# 9 * 3 equals 27.

with_retry() 사용하여 실패 시 재시도

가끔 알 수 없는 이유로 LLM의 실행은 언제든 실패할 수 있습니다. Langchain을 활용하여 어떤 서비스를 구성하고 싶다면, 안정성을 위해 LLM call의 예기치 않은 실패에 대처하는 방법이 필요합니다. Runnable 객체의 with_retry() 메서드를 이용하면 LLM call 실패 시 후속 조치를 정의해둘 수 있습니다. 보다 안정적인 체인 실행이 가능해지겠죠? 다만 한 번 오류가 발생한 상황이라면, 재시도 시에도 오류가 발생할 확률이 높아지다보니 비용 절감을 위해서는 retry 반복 수를 작게 유지하는 것이 좋습니다.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser

# Create a prompt template with multiple placeholders
prompt = ChatPromptTemplate.from_template("What is {x} * 3?")

# Define the language model with streaming enabled
model = ChatOpenAI(model="gpt-4o-mini")

# Create the chain using RunnablePassthrough
chain = prompt | model.with_retry(stop_after_attempt=3) | StrOutputParser()

# Run the chain
print(chain.invoke({"x": 3}))

Langchain prompt templates

ChatPromptTemplate.from_template

ChatPromptTemplate.from_template은 단일 프롬프트 템플릿을 생성하는 데 사용됩니다. 템플릿 내에서 {genre}, {character} 와 같이 {변수명} 형식으로 변수를 지정할 수 있고, prompt.format(genre='A', character='B') 처럼 변수에 값을 할당할 수 있습니다.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser

# Create a prompt template with multiple placeholders
prompt = ChatPromptTemplate.from_template("Write a {genre} story about a {character} who {action}.")

# Define the language model with streaming enabled
model = ChatOpenAI(model="gpt-4o-mini")

# Create the chain
chain = prompt | model | StrOutputParser()

# Run the chain
chain.invoke({"genre": "fantasy", "character": "wizard", "action": "saves the world"})

ChatPromptTemplate.from_messages

ChatPromptTemplate.from_template의경우 기본적으로 HumanMessage 타입의 프롬프트를 생성하게 됩니다. "system", "human", "ai" 등의 역할과 메시지를 명시적으로 지정하기 위해서는 ChatPromptTemplate.from_messages를 이용해야 합니다.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser

# Create a prompt template system and user messages
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("user", "Tell me a joke about {topic}."),
    ]
)

# Define the language model with streaming enabled
model = ChatOpenAI(model="gpt-4o-mini")

# Create the chain
chain = prompt | model | StrOutputParser()

# Run the chain
print(chain.invoke({"topic": "dragons"}))

Langchain output parsers

StrOutputParser

StrOutputParser는 LLM 혹은 ChatModel의 출력을 문자열로 변환하는 도구입니다. 크게 아래 두 가지 기능을 제공합니다.

1. LLM 출력 처리: LLM이 이미 문자열을 반환하는 경우, 추가 변환 없이 그대로 전달합니다.
2. ChatModel 출력 처리: ChatModel 이 반환하는 메시지 객체에서 .content 속성을 추출하여 문자여롤 변환합니다.

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


# Define a prompt template
prompt = ChatPromptTemplate.from_template("{question}")

# Define the language model
model = ChatOpenAI()

# Create the chain with JsonOutputParser
chain = prompt | model | StrOutputParser()

print(chain.invoke({"question": "Who is the first human to walk on the moon?"}))

# Response:
# Neil Armstrong

JsonOutputParser

StrOutputParser가 모델의 출력을 단순 문자열로 변환하는 객체였다면, JsonOutputParser는 모델의 출력을 JSON 형식으로 변환하는 객체입니다. 아래의 특징을 가집니다.

1. JSON 파싱: 모델 출력을 자동으로 JSON 객체로 파싱합니다.
2. 스트리밍 지원: 실시간 데이터 처리를 위해 스트리밍을 지원합니다. 부분적인 JSON 객체를 순차적으로 생성할 수 있습니다.
3. 유연한 입력 처리: 문자열과 메시지 형태의 입력을 모두 처리할 수 있습니다.
4. Pydantic 통합: Pydantic과 함께 사용하여 예상되는 스키마를 선언할 수 있습니다.
5. get_format_instructions(): 잘 정의된 field description을 가지는 pydantic 데이터 스키마를 전달한 상황에서는, parser.get_format_instructions() 메서드를 이용하여 JSON output을 유도하는 프롬프트를 쉽게 얻어낼 수 있습니다.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import JsonOutputParser

from pydantic import BaseModel, Field


class Answer(BaseModel):
    answer: str = Field(description="The answer to the question")
    reasoning: str = Field(description="The reasoning for the answer")


# Define a prompt template
prompt = ChatPromptTemplate.from_template("{question} {format_instructions}")

# Define the language model
model = ChatOpenAI()

# Create the chain with JsonOutputParser
parser = JsonOutputParser(pydantic_object=Answer)
chain = prompt | model | parser

print(
    chain.invoke(
        {
            "question": "Who is the first human to walk on the moon?",
            "format_instructions": parser.get_format_instructions(),
        }
    )
)

# Response:
# {'answer': 'Neil Armstrong', 
# 'reasoning': 'Neil Armstrong was the first human to walk on the moon during the Apollo 11 mission in 1969.'}

기타

사용량 확인 (get_openai_callback)

get_openai_callback()은 OpenAI API 호출과 관련된 메트릭을 수집하고 추적하는 데 사용됩니다. with 문과 함께 context manager로서 사용되어 블록 내에서 이루어지는 모든 OpenAI APi 호출을 모두 추적합니다. 아래의 특징을 가집니다.

1. 토큰 사용량 계산: API 호출 동안 사용된 총 토큰 수, input / output 토큰 수를 계산합니다.
2. 비용 계산: API 호출에 따른 총 비용을 계산합니다.
3. 성능 모니터링: 개발자가 OpenAI API 호출의 성능을 효율적으로 모니터링할 수 있게 합니다.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.callbacks.manager import get_openai_callback
from langchain.schema import StrOutputParser

# Define a prompt template
prompt = ChatPromptTemplate.from_template("Tell me a story about {topic}")

# Define the language model with streaming enabled
model = ChatOpenAI(model="gpt-4o-mini")

# Create the chain
chain = prompt | model | StrOutputParser()

# Run the chain
with get_openai_callback() as cb:
    chain.invoke({"topic": "dragons"})
    print(cb)

# Response:
# Tokens Used: 891  # cb.total_tokens
#        Prompt Tokens: 13  # cb.prompt_tokens
#                Prompt Tokens Cached: 0
#        Completion Tokens: 878  # cb.completion_tokens
#                Reasoning Tokens: 0
# Successful Requests: 1
# Total Cost (USD): $0.00052874 # cb.total_cost

마치며

이번 포스팅에서는 반복적으로 사용되는 langchain 코드 패턴들을 정리해 보았습니다. 자주 사용하다보면, 이 포스팅을 다시 찾아오지 않고도 자연스럽게 langchain 코딩을 해낼 수 있겠죠? 😁