πŸͺ΄ VaultByNr

Search

SearchSearch

πŸ“ Basic_LCEL ν™œμš©ν•˜κΈ°

May 17, 202410 min read

Summary

  • 상황별 Method ν™œμš©λ²•
  • Runnable ν™œμš©λ²•

LCEL(LangChain Expression Language)은 ν”„λ‘¬ν”„νŠΈ ꡬ성, λͺ¨λΈ μΈμŠ€ν„΄μŠ€ 생성, 좜λ ₯ μƒμ„±μ˜ 과정을 ==Chain==으둜 λ¬Άμ–΄ λ³΅μž‘ν•œ μ›Œν¬ν”Œλ‘œμš°λ₯Ό 쉽고 μ§κ΄€μ μœΌλ‘œ ꡬ좕할 수 μžˆλ„λ‘ λ•λŠ” μΈν„°νŽ˜μ΄μŠ€μ΄λ‹€.

특수문자(|)λ₯Ό ν™œμš©ν•˜μ—¬ 본인만의 Chain을 ꡬ좕할 수 μžˆλ‹€.

1️⃣ Methods

Sync/AsyncDescription
invoke/ainvokeμž…λ ₯에 λŒ€ν•œ κ²°κ³Όλ₯Ό 좜λ ₯ν•œλ‹€.
batch/abatchλ°˜λ³΅λ˜λŠ” μž…λ ₯을 리슀트둜 μž…λ ₯ν•˜μ—¬ μ²˜λ¦¬ν•œλ‹€.
stream/astreamchunkλ§ˆλ‹€ 좜λ ₯되게 ν•œλ‹€.
astream_log쀑간 단계λ₯Ό μŠ€νŠΈλ¦¬λ°ν•œλ‹€.

동기(Synchronous)와 비동기(Asynchronous)

동기(Synchronous)비동기(Asynchronous)
μ„€λͺ…μž‘μ—…μ„ 순차적으둜 μ§„ν–‰μ—¬λŸ¬ μž‘μ—…μ„ λ™μ‹œμ— 진행
μž₯μ μ½”λ“œκ°€ λ‹¨μˆœν•˜κ³  μ΄ν•΄ν•˜κΈ° 쉽닀
디버깅이 쉽닀		νš¨μœ¨μ μ΄λ‹€
λ°˜μ‘μ„±μ΄ μ’‹λ‹€
λ‹¨μ μžμ› ν™œμš©μ΄ λΉ„νš¨μœ¨μ μ΄λ‹€
ν•œ μž‘μ—…μ΄ 잘λͺ»λ˜λ©΄ 전체 ν”„λ‘œκ·Έλž¨μ΄ 멈좜 수 μžˆλ‹€		디버깅이 λ³΅μž‘ν•˜λ‹€
# κΈ°λ³Έ μ½”λ“œ
from langchain_openai import ChatOpenAI 
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = PromptTemplate.from_template("{input}에 λŒ€ν•΄ ν•œκ΅­μ–΄λ‘œ ν•œ μ€„λ‘œ μ„€λͺ…ν•΄μ€˜")
model = ChatOpenAI(model_name = "gpt-3.5-turbo") 
output_parser = StrOutputParser()
chain = prompt | model | output_parser

πŸ“‹ invoke / ainvoke

import time 
import asyncio
 
def run_sync(input_list):
    """invoke μ‹€ν–‰ ν•¨μˆ˜"""
    start_time = time.time()
    for input in input_list:
        result = chain.invoke(input)
        print(result)
    end_time = time.time()
    print("="*100)
    print(f"Sync execution time: {end_time - start_time:.2f} seconds")
 
async def run_async(input_list):
    """ainvoke μ‹€ν–‰ ν•¨μˆ˜"""
    start_time = time.time()
    tasks = [chain.ainvoke(input) for input in input_list]
    results = await asyncio.gather(*tasks)
    end_time = time.time()
    print(f"Async execution time: {end_time - start_time:.2f} seconds")
    print("="*100)
 
    for result in results:
        print(result)
 
run_sync(input_list)
await run_async(input_list)
# Sync execution time: 7.13 seconds
# Async execution time: 1.58 seconds

πŸ“‹ batch / abatch

batch와 abatch의 속도 차이가 크게 λ‚˜μ§€ μ•ŠλŠ” κ²ƒμ²˜λŸΌ λ³΄μ΄μ§€λ§Œ, 보닀 더 λ³΅μž‘ν•œ μ½”λ“œμ—μ„œλŠ” 차이가 λ‚  것이닀.

import time
import asyncio
 
def run_sync(input_list):
    """batch μ‹€ν–‰ ν•¨μˆ˜"""
    start_time = time.time()
    result = chain.batch(input_list)
    end_time = time.time()
    print(f"Sync execution time: {end_time - start_time:.2f} seconds")
    print("="*100)
    print("\n".join(result))
 
async def run_async():
    """abatch μ‹€ν–‰ ν•¨μˆ˜"""
    start_time = time.time()
    tasks = chain.abatch(input_list)
    result = await tasks
    end_time = time.time()
    print(f"Async execution time: {end_time - start_time:.2f} seconds")
    print("="*100)
    print("\n".join(result))
 
run_sync(input_list)
await run_async(input_list)
# Sync execution time: 1.78 seconds
# Async execution time: 1.65 seconds

πŸ“‹ stream / astream

generator둜 좜λ ₯λ˜μ–΄ for loop으둜 print ν•˜λ©΄ chunkλ³„λ‘œ streaming λœλ‹€.

# generator둜 좜λ ₯λ˜λŠ” 것을 확인
chain.stream({"input":"파이썬"}) 
 
# Output
# <generator object RunnableSequence.stream at 0x0000014E37FB6650>
# stream
for chunk in chain.stream({"input":"파이썬"}):
    print(chunk, end="", flush=True)
 
# astream
for chunk in chain.stream({"input":"파이썬"}):
    print(chunk, end="", flush=True)

πŸ“‹ stream_log

chain μ‹€ν–‰ 과정을 λ‘œκΉ…ν•˜λŠ” ν•¨μˆ˜λ‘œ 디버깅할 λ•Œ μš©μ΄ν•˜λ‹€.

stream = chain.astream_log({"input":"파이썬"})
async for chunk in stream:
    print(chunk)
    print("="*100)

πŸ“‹ callbacks

Streaming은 ChatOpenAIμ—μ„œ callbacks μ˜΅μ…˜μœΌλ‘œλ„ κ°€λŠ₯ν•˜λ‹€.

BaseCallbackHandlerν΄λž˜μŠ€κ°€ μžˆλŠ”λ° 이 클래슀λ₯Ό μ»€μŠ€ν…€ν•˜μ—¬ .stream()보닀 더 λ‹€μ–‘ν•œ 상황에 μ μš©ν•  수 μžˆλ‹€.

ν΄λž˜μŠ€μ™€ κ΄€λ ¨λœ λ‚΄μš©μ€ Langchain κ³΅μ‹λ¬Έμ„œμ—μ„œ μ°Έκ³ ν•˜μž.

# κΈ°λ³Έ μ½”λ“œ
from typing import Any
from langchain_openai import ChatOpenAI 
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.callbacks import BaseCallbackHandler
 
class CustomHandler(BaseCallbackHandler):
    def on_llm_new_token(self, token: str, **kwargs: Any) -> Any:
        """Run on new LLM token. Only available when streaming is enabled."""
        print(token, end="", flush=True)
 
prompt = PromptTemplate.from_template("{input}에 λŒ€ν•΄ ν•œκ΅­μ–΄λ‘œ ν•œ μ€„λ‘œ μ„€λͺ…ν•΄μ€˜")
output_parser = StrOutputParser()

방법 1

model = ChatOpenAI(
    model_name = "gpt-3.5-turbo", 
    streaming=True, 
    callbacks=[CustomHandler()]
) 
chain = prompt | model | output_parser
 
response = chain.invoke({"input": "파이썬"})

방법 2

model = ChatOpenAI(
    model_name = "gpt-3.5-turbo", 
    streaming=True
)
chain = prompt | model | output_parser
 
model.callbacks = [CustomHandler()]
response = chain.invoke({"input": "파이썬"})

방법 3

model = ChatOpenAI(
    model_name = "gpt-3.5-turbo", 
    streaming=True
)
chain = prompt | model | output_parser
 
response = chain.invoke(
    {"input": "파이썬"}, 
    {"callbacks": [CustomHandler()]}
)

2️⃣ Runnable

μž…λ ₯값듀에 λŒ€ν•œ λ³€ν˜•μ΄ ν•„μš”ν•œ 경우 μœ μ—°ν•˜κ²Œ μ»€μŠ€ν…€ν•  수 μžˆλŠ” 도ꡬ듀이닀.

FunctionDescription
RunnablePassthrough()μž…λ ₯된 값을 κ·ΈλŒ€λ‘œ μ „λ‹¬ν•œλ‹€.
RunnablePassthrough.assign()μž…λ ₯된 값을 λ³€ν™˜ν•˜κ±°λ‚˜ μƒˆλ‘œμš΄ λ³€μˆ˜λ₯Ό λ§Œλ“ λ‹€.
RunnableLambda()μž…λ ₯된 값을 μ΄μš©ν•΄μ„œ ν•¨μˆ˜λ‘œ μƒˆλ‘œμš΄ λ³€μˆ˜λ₯Ό λ§Œλ“ λ‹€.
RunnableParallel()λ™μΌν•œ μž…λ ₯을 가진 chain을 λ³‘λ ¬λ‘œ μ²˜λ¦¬ν•œλ‹€.
# κΈ°λ³Έ μ½”λ“œ
from langchain_openai import ChatOpenAI 
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = PromptTemplate.from_template("'{input}'을 μ˜μ–΄λ‘œ λ²ˆμ—­ν•΄μ£Όμ„Έμš”")
model = ChatOpenAI(model_name = "gpt-3.5-turbo") 
output_parser = StrOutputParser()

πŸ“‹ RunnablePassthrough()

μž…λ ₯된 값을 κ·ΈλŒ€λ‘œ μ „λ‹¬ν•œλ‹€.

λ”°λΌμ„œ μ›λž˜λŠ” λ”•μ…”λ„ˆλ¦¬λ‘œ μž…λ ₯ν•΄μ•Ό ν–ˆμ§€λ§Œ, λ¬Έμžμ—΄λ‘œ μž…λ ₯ λ°›μ•„ runnableμ—μ„œ λ”•μ…”λ„ˆλ¦¬λ₯Ό λ§Œλ“  ν›„ ν”„λ‘¬ν”„νŠΈμ— 전달할 수 μžˆλ‹€.

runnable = {"input": RunnablePassthrough()}
chain = runnable | prompt | model | output_parser 
chain.invoke("λ‹€λžŒμ₯")
 
# Output
# 'Squirrel'

πŸ“‹ RunnablePassthrough.assign()

μž…λ ₯값을 λ³€ν˜•ν•˜κ±°λ‚˜ μƒˆλ‘œμš΄ μž…λ ₯값을 λ§Œλ“€ 수 μžˆλ‹€.

add_runnable = RunnablePassthrough.assign(input = lambda x: x["input"] + "λ₯Ό λ³΄μ•˜μŠ΅λ‹ˆλ‹€")
add_runnable.invoke({"input": "λ‹€λžŒμ₯"})
 
# Output
# {'input': 'λ‹€λžŒμ₯λ₯Ό λ³΄μ•˜μŠ΅λ‹ˆλ‹€'}

πŸ“‹ RunnableLambda()

ν•¨μˆ˜λ₯Ό λ§Œλ“€μ–΄ μž…λ ₯값을 λ³€ν˜•ν•  수 μžˆλ‹€.

from langchain_core.runnables import RunnableLambda
 
def add_text(input):
    return "μ„Έμƒμ—μ„œ κ°€μž₯ μž‘μ€ " + input
 
runnable = {"input": RunnableLambda(add_text)}
chain = runnable | prompt | model | output_parser 
chain.invoke("λ‹€λžŒμ₯")
 
# Output
# 'The smallest squirrel in the world'

πŸ“‹ RunnableParallel()

μž…λ ₯값이 λ™μΌν•œ μ—¬λŸ¬ 개의 체인을 λ³‘λ ¬μ μœΌλ‘œ 관리할 수 μžˆλ‹€.

prompt1 = PromptTemplate.from_template("{country}의 μ£Όμš” μ–Έμ–΄λ₯Ό μ•Œλ €μ€˜")
prompt2 = PromptTemplate.from_template("{country}의 λŒ€ν‘œμ μΈ λžœλ“œλ§ˆν¬ 3개λ₯Ό μ•Œλ €μ€˜")
 
chain1 = prompt1 | model | output_parser
chain2 = prompt2 | model | output_parser
 
combined = RunnableParallel(
    language = chain1,
    landmarks = chain2
)
 
combined.invoke({"country":"ν•œκ΅­"})
 
# Output
# {'language': 'ν•œκ΅­μ˜ μ£Όμš” μ–Έμ–΄λŠ” ν•œκ΅­μ–΄μž…λ‹ˆλ‹€. ν•œκ΅­μ–΄λŠ” λŒ€λΆ€λΆ„μ˜ ν•œκ΅­ μ‚¬λžŒλ“€μ΄ μ‚¬μš©ν•˜λŠ” μ–Έμ–΄λ‘œ, κ΅­λ‚΄μ—μ„œλŠ” 곡식 μ–Έμ–΄λ‘œ μ‚¬μš©λ˜κ³  μžˆμŠ΅λ‹ˆλ‹€. λ˜ν•œ, μ˜μ–΄λ„ λ§Žμ€ μ‚¬λžŒλ“€μ΄ ν•™μŠ΅ν•˜κ³  μ‚¬μš©ν•˜κ³  있으며, 쀑ꡭ어와 일본어도 일뢀 μ§€μ—­μ—μ„œ μ‚¬μš©λ˜κ³  μžˆμŠ΅λ‹ˆλ‹€.',
# Β 'landmarks': '1. λ‚¨μ‚°νƒ€μ›Œ - μ„œμšΈμ˜ λŒ€ν‘œμ μΈ λžœλ“œλ§ˆν¬λ‘œμ„œ, μ„œμšΈ μ‹œλ‚΄μ™€ ν•œκ°•μ„ ν•œλˆˆμ— λ³Ό 수 μžˆλŠ” μ „λ§λŒ€κ°€ 유λͺ…ν•˜λ‹€.\n2. 경볡ꢁ - μ„œμšΈμ— μœ„μΉ˜ν•œ μ‘°μ„  μ‹œλŒ€μ˜ κΆκΆλ‘œμ„œ, μ•„λ¦„λ‹€μš΄ 전톡 ν•œμ˜₯ 건물과 κ·Όμ •μ „, 경회루 등을 λ³Ό 수 μžˆλ‹€.\n3. λΆ€μ‚° νƒ€μ›Œ - λΆ€μ‚°μ˜ λžœλ“œλ§ˆν¬λ‘œμ„œ, λΆ€μ‚° μ‹œλ‚΄μ™€ ν•΄μ•ˆλ„λ‘œλ₯Ό ν•œλˆˆμ— λ³Ό 수 μžˆλŠ” μ „λ§λŒ€μ™€ 야경이 유λͺ…ν•˜λ‹€.'}

πŸ“‹ itemgetter의 ν™œμš©

itemgetterλŠ” λ”•μ…”λ„ˆλ¦¬μ—μ„œ νŠΉμ • ν‚€μ˜ valueλ₯Ό μΆ”μΆœν•˜λŠ” κΈ°λŠ₯을 ν•œλ‹€.

from operator import itemgetter
 
prompt = PromptTemplate.from_template("'{input}'을 {language}λ²ˆμ—­ν•΄μ£Όμ„Έμš”")
runnable = {
    "input" : itemgetter("input") | RunnableLambda(lambda x: x + "λŠ” λ§›μžˆμ–΄"),
    "language": itemgetter("language")
}
chain = runnable | prompt | model | output_parser
chain.invoke({"input": "μ˜€λ Œμ§€", "language": "μ˜μ–΄"})
 
# Output
# '"Orange is delicious."'

References

Graph View

Backlinks

  • No backlinks found