LCEL은 LangChain Expression Language의 약어인데 기본적인 방식으로 Chain을 사용해도 문제가 없지만 코드를 좀 더 간략하게 사용하고 병렬처리, 비동기, 스트리밍 기능을 제공하기 위해 LCEL을 사용한다.
LCEL은 LangChain 라이브러리에서 복잡한 LLM(Large Language Model) 애플리케이션을 구축하기 위한 선언적 인터페이스로, 코드의 간결성과 유지보수성을 극대화하도록 설계되었습니다. 파이썬의 “|” 연산자를 활용해 컴포넌트를 직관적으로 연결하는 방식이 특징입니다
사용자 정의 체인을 가능한 쉽게 만들 수 있도록 구현된 Runnable 프로토콜은 대부분의 컴포넌트에 구현되어 있습니다. 이는 표준 인터페이스로, 사용자 정의 체인을 정의하고 표준 방식으로 호출하는 것을 쉽게 만듭니다. 표준 인터페이스에는 다음이 포함됩니다.
- stream: 응답의 청크를 스트리밍합니다.
- invoke: 입력에 대해 체인을 호출합니다.
- batch: 입력 목록에 대해 체인을 호출합니다.
stream: 실시간 출력
이 함수는 chain.stream 메서드를 사용하여 주어진 토픽에 대한 데이터 스트림을 생성하고, 이 스트림을 반복하여 각 데이터의 내용(content)을 즉시 출력합니다. end=”” 인자는 출력 후 줄바꿈을 하지 않도록 설정하며, flush=True 인자는 출력 버퍼를 즉시 비우도록 합니다.
from langchain.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
llm_model = ChatOllama(model="exaone3.5", streaming=True, temperature=0)
prompt = ChatPromptTemplate.from_template("{topic}를(을) 1문장으로 간락하게 설명해줘")
# 1. LCEL 체인 구성
chain = prompt | llm_model | StrOutputParser()
# 2. 스트리밍 실행
input_topic = {"topic": "인공지능"}
for chunk in chain.stream(input_topic):
print(chunk, end="", flush=True) # 실시간 출력 유지
print()
인공지능은 컴퓨터 시스템이 인간의 지능을 모방하여 학습, 문제 해결, 의사결정 등을 수행할 수 있도록 설계된 기술입니다.
invoke: 호출
chain 객체의 invoke 메서드는 주제를 인자로 받아 해당 주제에 대한 처리를 수행합니다.
# 3. invoke 실행
result = chain.invoke({"topic": "인공지능"})
print(result)
인공지능은 컴퓨터 시스템이 인간의 지능을 모방하여 학습, 문제 해결, 의사결정 등을 수행할 수 있도록 설계된 기술입니다.
# 4. batch 실행
inputs = [
{"topic": "인공지능"},
{"topic": "IoT"},
{"topic": "블록체인"}
]
results = chain.batch(inputs) # 병렬 처리
print(results)
[‘인공지능은 컴퓨터 시스템이 인간의 지능을 모방하여 학습, 문제 해결, 의사결정 등을 수행할 수 있도록 설계된 기술입니다.’, ‘IoT는 인터넷을 통해 다양한 기기들이 서로 연결되어 데이터를 주고받으며 효율적으로 작동하도록 하는 기술입니다.’, ‘블록체인은 분산된 데이터베이스로, 거래 기록을 안전하고 투명하게 저장하여 중앙 관리 기관 없이도 신뢰성을 유지합니다.’]
max_concurrency 매개변수를 사용하여 동시 요청 수를 설정할 수 있습니다
config 딕셔너리는 max_concurrency 키를 통해 동시에 처리할 수 있는 최대 작업 수를 설정합니다. 여기서는 최대 3개의 작업을 동시에 처리하도록 설정되어 있습니다.
result = chain.batch (
[
{"topic": "지구의 자전 속도는?"},
{"topic": "달의 자전 속도는?"},
{"topic": "말의 달리기 속도는?"},
{"topic": "지구를 탈출 하기 위한 최소 속도는?"},
],
config={"max_concurrency": 2},
)
print(result)ㅇㄹ
[‘지구는 약 24시간에 걸쳐 자전하여 하루를 만듭니다.’, ‘달의 자전 속도는 약 1달(27.3일)에 걸쳐 지구를 한 바퀴 돌면서 동일한 면을 보이는 것으로, 이는 자전 주기와 공전 주기가 일치하는 상태입니다.’, ‘말의 달리기 속도는 종에 따라 다르지만, 일반적으로 경주마는 시속 55km에서 최고 시속 70km까지 달릴 수 있습니다.’, ‘지구 탈출을 위한 최소 속도는 탈출 속도로 약 11.2 km/s입니다.’]
Parallel: 병렬성
LangChain Expression Language가 병렬 요청을 지원하는 방법을 살펴봅시다. 예를 들어, RunnableParallel을 사용할 때(자주 사전 형태로 작성됨), 각 요소를 병렬로 실행합니다.
langchain_core.runnables 모듈의 RunnableParallel 클래스를 사용하여 두 가지 작업을 병렬로 실행하는 예시를 보여줍니다.
ChatPromptTemplate.from_template 메서드를 사용하여 주어진 country에 대한 수도 와 면적 을 구하는 두 개의 체인(chain1, chain2)을 만듭니다. 이 체인들은 각각 model과 파이프(|) 연산자를 통해 연결됩니다. 마지막으로, RunnableParallel 클래스를 사용하여 이 두 체인을 capital와 area이라는 키로 결합하여 동시에 실행할 수 있는 combined 객체를 생성합니다.
chain1.invoke() 함수는 chain1 객체의 invoke 메서드를 호출합니다.
이때, country이라는 키에 대한민국라는 값을 가진 딕셔너리를 인자로 전달합니다.
from langchain_ollama import ChatOllama
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel
llm_model = ChatOllama(model="exaone3.5", temperature=0)
# {country} 의 수도를 물어보는 체인을 생성합니다.
chain1 = (
PromptTemplate.from_template("{country} 의 수도는 어디야?")
| llm_model
| StrOutputParser()
)
# {country} 의 면적을 물어보는 체인을 생성합니다.
chain2 = (
PromptTemplate.from_template("{country} 의 면적은 얼마야?")
| llm_model
| StrOutputParser()
)
combined = RunnableParallel(capital=chain1, area=chain2)
# chain1 를 실행합니다.
result = chain1.invoke({"country": "대한민국"})
print(result)
대한민국의 수도는 서울입니다.
이번에는 chain2.invoke() 를 호출합니다. country 키에 다른 국가인 미국 을 전달합니다.
# chain2 를 실행합니다.
result = chain2.invoke({"country": "한국"})
print(result)
한국의 면적은 약 **100,363 제곱킬로미터**입니다.
이는 대략적으로 **남한**의 면적을 의미하며, **북한**을 포함하면 약 **220,847 제곱킬로미터** 정도입니다.
더 자세한 정보가 필요하시면 알려주세요! 😊
combined 객체의 invoke 메서드는 주어진 country에 대한 처리를 수행합니다. 이 예제에서는 대한민국라는 주제를 invoke 메서드에 전달하여 실행합니다.
# combined 를 실행합니다.
result = combined.invoke({"country": "한국"})
print(result)
{‘capital’: ‘한국의 수도는 서울입니다.’, ‘area’: ‘한국의 면적은 약 **100,363 제곱킬로미터**입니다. \n\n이는 대략적으로 **남한**의 면적을 의미하며, **북한**을 포함하면 약 **220,847 제곱킬로미터** 정도입니다. \n\n더 자세한 정보가 필요하시면 알려주세요! 😊’}
배치에서의 병렬 처리
병렬 처리는 다른 실행 가능한 코드와 결합될 수 있습니다. 배치와 병렬 처리를 사용해 보도록 합시다.
chain1.batch 함수는 여러 개의 딕셔너리를 포함하는 리스트를 인자로 받아, 각 딕셔너리에 있는 “topic” 키에 해당하는 값을 처리합니다. 이 예시에서는 “대한민국”와 “미국”라는 두 개의 토픽을 배치 처리하고 있습니다.
# chain1 를 실행합니다.
result = chain1.batch([{"country": "대한민국"}, {"country": "미국"}])
print(result)
[‘대한민국의 수도는 서울입니다.’, ‘미국의 수도는 워싱턴 D.C.입니다.’]
chain2.batch 함수는 여러 개의 딕셔너리를 리스트 형태로 받아, 일괄 처리(batch)를 수행합니다. 이 예시에서는 대한민국와 미국라는 두 가지 국가에 대한 처리를 요청합니다.
# chain2 를 실행합니다.
result = chain2.batch([{"country": "대한민국"}, {"country": "미국"}])
print(result)
[‘대한민국의 면적은 약 **100,363 제곱킬로미터**입니다.’, ‘미국의 면적은 약 **9,833,520 제곱 킬로미터 (km2)** 입니다. 이는 세계에서 **세 번째로 큰 나라**입니다. \n\n혹시 다른 단위로 알고 싶으신가요? 예를 들어, 제곱 마일로 알고 싶으시다면 약 **3.7 백만 제곱 마일**입니다.’]
combined.batch 함수는 주어진 데이터를 배치로 처리하는 데 사용됩니다. 이 예시에서는 두 개의 딕셔너리 객체를 포함하는 리스트를 인자로 받아 각각 대한민국와 미국 두 나라에 대한 데이터를 배치 처리합니다.
# combined 를 실행합니다.
result = combined.batch([{"country": "대한민국"}, {"country": "미국"}])
print(result)
[{‘capital’: ‘대한민국의 수도는 서울입니다.’, ‘area’: ‘대한민국의 면적은 약 **100,363 제곱킬로미터**입니다.’}, {‘capital’: ‘미국의 수도는 워싱턴 D.C.입니다.’, ‘area’: ‘미국의 면적은 약 **9,833,520 제곱 킬로미터 (km2)** 입니다. 이는 세계에서 **세 번째로 큰 나라**입니다. \n\n혹시 다른 단위로 알고 싶으신가요? 예를 들어, 제곱 마일로 알고 싶으시다면 약 **3.7 백만 제곱 마일**입니다.’}]