[👤LLM] Parser, Runnable Component 써보기

LCEL : LangChain Expression Language

체인을 아래와 같이 선언됩니다.

chain = prompt | model

 

 | 사이사이 있는 것들은 컴포넌트입니다. 컴포넌트들은 순서대로 자신의 결과를 다음 컴포넌트로 넘겨줍니다.

즉, 한 컴포넌트의 출력은 다음 컴포넌트의 입력이 됩니다.

이 모습이 마치 체인과 같기 때문에 langchain이라는 이름이 붙었습니다. 😄⛓️‍💥

체인은 어떻게 시작될까요?

아래와 같이 체인이름.invoke(전달) 로 체인을 깨우면

response = chain.invoke({"input" : "말티즈"})

 

정의한 체인의 첫 번째 컴포넌트에 (전달)을 전달하며 차례대로 실행하기 시작합니다.

차례대로 실행되고, 마지막 컴포넌트의 출력이 최종적으로 리턴되며 체인이 끝나게 됩니다.

 

 

그러면 아무거나 이  |  파이프 속에 들어갈 수 있을까요?

 

 |  파이프 사이에 연결될 수 있는 컴포넌트

1. Runnable 인터페이스
2. '조건'을 만족하는 함수

Runnable 인터페이스는 무엇일까요?

우리가 위에서 당연하게 실행했던 코드에 있던 요소들입니다.

# LLM model 생성
model = ChatOpenAI(model="gpt-4o-mini")

# prompt 템플릿 생성
prompt = ChatPromptTemplate.from_template("{dog_breeds_input}들의 고향은 어디야?")

# chain 생성
chain = prompt | model

 

ChatPromptTemplate로 만든 prompt, ChatOpenAI로 만든 model은

랭체인이 처음부터 "얘네는 chain에 들어갈 수 있는 요소들이야!"라고 정해둔 Runnable 인터페이스입니다.

→ 따라서 | 파이프 속에 들어갈 수 있는 있습니다.

이외에도 langchain에 자주쓰이는 output_parser(출력을 예쁘게 만들어줌), Retriever(RAG에서 검색결과를 물어오는 리트리버)도 만들어질 때부터 chain에 들어갈 수 있게 정해진 Runnable 인터페이스입니다.

'조건'을 만족하는 함수?

그럼 '조건'을 만족하는 함수는 무엇일까요?

• 조건 1 : 입력을 받고 출력을 반환하는 형태의 함수.
• 조건 2 : 이전 체인의 출력과 데이터 타입이 호환되어야 함, 다음 체인의 입력과 데이터 타입이 호환이 되어야 함

---

my_print로 출력한 결과에 엔터(\n)를 넣어서 출력하는 함수를 만들어 주세요.

# 직접 함수를 정의내려서 답변 깔끔하게 출력받기 : splited_string = string.split('.') 
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# LLM model 생성
model = ChatOpenAI(model="gpt-4o-mini")

# prompt 템플릿 생성
prompt = ChatPromptTemplate.from_template("나 영화 {preference_movie} 좋아하는데, 비슷한 영화 2개 추천해줘")

def my_print(string) :
  return string.content

def your_print(string) :
  splited_string = string.split('.') # . 을 기준으로 나눔. # 합치고 싶은 리스트
  # \n으로 구분
  return '\n'.join(splited_string)

# chain 생성
chain = prompt | model | my_print | your_print

# chain 실행
answer = chain.invoke({"preference_movie": "하이큐!! 쓰레기장의 결투"})
print(answer)

---

자주 쓰이는 StrOutputParser()
간단한 모델 출력에서 content만 그대로 출력할때

# 제공되는 StrOutputParser 함수로 답변 깔끔하게 출력받기
# 간단한 모델 출력에서 content만 그대로 출력할때
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser


# LLM model 생성
model = ChatOpenAI(model="gpt-4o-mini")

# prompt 템플릿 생성
prompt = ChatPromptTemplate.from_template("나 영화 {preference_movie} 좋아하는데, 비슷한 영화 2개 추천해줘")

# chain 생성
chain = prompt | model | StrOutputParser() # 깔끔하게 출력된다.

# chain 실행
answer = chain.invoke({"preference_movie": "하이큐!! 쓰레기장의 결투"})
print(answer)

---

자주 쓰이는 runnable component: RunnablePassthrough()
▶ 입력을 변환하지 않음: 입력 데이터를 그대로 반환합니다.
▶ 데이터 검증 또는 테스트용: 파이프라인에서 데이터 흐름을 확인하거나 테스트할 때 유용합니다.
▶ 복잡한 체인에서 간단한 연결 구성: 변환이 필요 없는 경우 체인의 일부로 사용됩니다.

import time
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough

# LLM model 생성
model = ChatOpenAI(model="gpt-4o-mini")

# prompt 템플릿 생성
prompt = ChatPromptTemplate.from_template("나 영화 {preference_movie} 좋아하는데, 비슷한 영화 2개 추천해줘")

# 처리 시간을 기록하는 Custom RunnablePassthrough
class TimingRunnablePassthrough(RunnablePassthrough):
    def invoke(self, input, *arg):
        start_time = time.time()  # 시작 시간 기록
        output = super().invoke(input)  # 기존 Passthrough 동작 수행
        end_time = time.time()  # 종료 시간 기록

        # 처리 시간 출력
        processing_time = end_time - start_time
        print(f"Processing time: {processing_time:.6f} seconds")

        return output

# Custom RunnablePassthrough 객체 생성
timed_passthrough = TimingRunnablePassthrough()

# chain 생성
chain = prompt | timed_passthrough | model | timed_passthrough | StrOutputParser() | timed_passthrough

# chain 실행
answer = chain.invoke({"preference_movie: 하이큐!! 쓰레기장의 결투"})
print(answer)