[LLM & RAG] Langchain 기초 - Chain

1. Chain의 연결방식

(1) LCEL (LangChain Expression Language) : 

LangChain에서 제공하는 선언형(Declarative) 방식으로 체인을 구성하는 방법입니다. Python 코드로 직접 체인을 구성하지 않고, 직관적인 YAML 스타일의 표현식을 사용합니다. (비유하자면 R에서 파이프라인 연산자? 같이 쓰는 방식이라고 할까요?)

• 간결한 표현: 복잡한 체인도 짧은 코드로 정의 가능.
• 재사용성: 정의된 체인을 쉽게 재사용하거나 공유 가능.
• 시각적 구성 가능: LCEL 기반의 GUI 도구를 활용할 수도 있음.

from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

# LCEL 스타일 체인 구성
prompt = PromptTemplate.from_template("Tell me a short fact about {topic}")

llm = ChatOpenAI(model="gpt-4")

# LCEL 체인 연결
chain = prompt | llm

# 실행
print(chain.invoke({"topic": "space"}))

 

(2) Python 체인 방식 (Imperative 방식) : 

Python 체인 방식은 명령형(Imperative) 방식으로 체인을 구성하는 방법입니다. 즉, 개발자가 SequentialChain, LLMChain 등의 클래스를 직접 사용해 체인을 연결하고 실행해야 합니다. 보다 세밀한 제어가 가능하지만, 코드 작성이 상대적으로 복잡할 수 있습니다. 

from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

template = "{language}로 사랑한다는 말을 번역해줘."

prompt_template = PromptTemplate.from_template(template)
prompt = prompt_template.format(language="영어")
llm = ChatOpenAI()
result = llm.invoke(prompt)
print(result)

 

2. Chain에 input을 주는 방식

아래와 같은 Chain이 있다고 해봅시다. 

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

# 프롬프트 템플릿 정의
chat_prompt = ChatPromptTemplate.from_template("Tell me a fact about {topic}")

# LLM 설정
chat = ChatOpenAI(model="gpt-4")

# 출력 파싱
output_parser = StrOutputParser()

# 체인 구성
runnable = chat_prompt | chat | output_parser

 

1) invoke를 활용한 단일 입력 처리

invoke()는 하나의 입력값을 처리하는 동기(synchronous) 방식
결과가 완전히 생성된 후 반환

# 단일 입력 실행
result = runnable.invoke({"topic": "black holes"})
print(result)

 

2) batch를 이용한 다중 입력 처리

batch()는 여러 개의 입력을 병렬로 실행하여 속도 증가
invoke()를 여러 번 실행하는 것보다 효율적

# 여러 개의 입력을 리스트로 전달
batch_inputs = [
    {"topic": "black holes"},
    {"topic": "quantum mechanics"},
    {"topic": "neural networks"}
]

# 배치 실행
batch_results = runnable.batch(batch_inputs)

# 결과 출력
for topic, result in zip(batch_inputs, batch_results):
    print(f"Topic: {topic['topic']} → Result: {result}")

 

3) stream을 이용한 스트리밍 방식 처리

stream()은 모델이 생성하는 출력(토큰)을 실시간으로 반환
flush=True를 사용하면 터미널에서 즉시 출력
네트워크 응답을 기다리지 않고 빠르게 출력

# 스트리밍 방식으로 실행
for chunk in runnable.stream({"topic": "black holes"}):
    print(chunk, end="", flush=True)

 

---

참고) 여러개의 Runnable을 병렬적으로 실행하기

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.runnable import RunnableParallel
from langchain.output_parsers import StrOutputParser

# 첫 번째 프롬프트: 기본 정보 제공
prompt1 = ChatPromptTemplate.from_template("Tell me a fact about {topic}")
chat1 = ChatOpenAI(model="gpt-4")
output_parser = StrOutputParser()
chain1 = prompt1 | chat1 | output_parser

# 두 번째 프롬프트: 주어진 주제에 대한 장점 설명
prompt2 = ChatPromptTemplate.from_template("What are the benefits of {topic}?")
chat2 = ChatOpenAI(model="gpt-4")
chain2 = prompt2 | chat2 | output_parser

# RunnableParallel을 이용해 두 개의 체인을 병렬로 실행
parallel_chain = RunnableParallel(
    fact_chain=chain1,  # 첫 번째 체인
    benefit_chain=chain2  # 두 번째 체인
)

# 실행 (병렬로 두 개의 체인이 동시에 실행됨)
result = parallel_chain.invoke({"topic": "quantum computing"})

# 결과 출력
print(result)

{
    'fact_chain': 'Quantum computing leverages quantum mechanics to perform computations exponentially faster than classical computers in certain tasks.',
    'benefit_chain': 'Quantum computing can solve complex problems in cryptography, drug discovery, and optimization much faster than classical computers.'
}