본 게시글은 패스트캠퍼스 테디노트님의 “테디노트의 RAG 비법노트” 를 수강하면서 개인적으로 정리한 내용입니다.
강의 링크

---

💡LCEL

• LangChain 구성 요소를 단일 체인으로 결합하는 선언적 방식
• 유닉스(Unix)의 파이프(|) 연산자처럼, 한 컴포넌트의 출력을 다음 컴포넌트의 입력으로 자연스럽게 전달
  기본구조 : Prompt + Model + Output Parser

promptTemplate의 활용: from_template() 메서드를 활용해 쉽게 객체화 가능

from langchain_core.prompts import PromptTemplate

# 1. 템플릿 정의 ({country}가 변수)
template = "{country}의 수도는 어디인가요?"

# 2. 객체 생성
prompt_template = PromptTemplate.from_template(template)

# 3. 프롬프트 생성 예시
print(prompt_template.format(country="대한민국"))

# 출력
대한민국의 수도는 어디인가요?

Chain 생성과 파이프(|) 연산자

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 1. 프롬프트 생성
prompt = PromptTemplate.from_template("{country}의 수도는 어디인가요?")

# 2. 모델 생성
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)

# 3. 출력 파서
output_parser = StrOutputParser()

# 4. Chain 연결 (LCEL)
chain = prompt | model | output_parser

# 실행
print(chain.invoke({"country": "대한민국"}))

# 출력
대한민국의 수도는 서울입니다.

---

💡Runnable 인터페이스

LangChain의 모든 체인은 Runnable 프로토콜을 따른다.

✍️ 기본 메서드

• invoke : 입력 후 응답이 완성될 때까지 대기, 한번에 반환
• stream : 응답을 Chunk 단위로 실시간으로 스트리밍
• batch : 리스트 형태의 입력을 받아 일괄 처리

stream 출력 예제 코드

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    temperature=0.1,  # 창의성 (0.0 ~ 2.0)
    model_name="gpt-4.1-nano",  # 모델명
)

answer = llm.stream("대한민국의 아름다운 관광지 10곳과 주소를 알려주세요!")

for token in answer:
    print(token.content, end="", flush=True)
    
# 출력
물론입니다! 대한민국의 아름다운 관광지 10곳과 그 주소를 소개해 드리겠습니다.

1. 경복궁 (Gyeongbokgung Palace)  
서울특별시 종로구 사직로 161

2. 제주도 한라산 국립공원 (Hallasan National Park, Jeju Island)  
제주특별자치도 제주시 특별자치도 제주시 조천읍 한라산로 2070

3. 부산 해운대 해수욕장 (Haeundae Beach, Busan)  
부산광역시 해운대구 해운대해변로 264

4. 강원도 설악산 국립공원 (Seoraksan National Park)  
강원도 속초시 설악산로 833

5. 경상남도 통영 한산도 (Tongyeong Hansando)  
경상남도 통영시 한산면 죽림리 한산도

6. 전라남도 여수 오동도 (Oedo Island, Yeosu)  
전라남도 여수시 오동도길 61

7. 충청남도 공주 무령왕릉 (Royal Tomb of King Muryeong)  
충청남도 공주시 송산리 1-1

8. 전라북도 전주 한옥마을 (Jeonju Hanok Village)  
전라북도 전주시 완산구 기린대로 99
...

10. 강원도 강릉 경포대 (Gyeongpodae Pavilion, Gangneung)  
강원도 강릉시 경포로 365

이 외에도 대한민국에는 아름다운 곳이 많지만, 위의 장소들은 대표적이고 인기 있는 관광지입니다. 방문 계획에 참고하시기 바랍니다!

Stream 의 출력은 우리가 흔히 사용하는 AI Model 의 응답과 같은 UI

🚀 비동기 처리

LangChain은 메서드앞에 a를 붙여 비동기로 처리할 수 있다.
ainvoke astream abatch

import asyncio

async def async_run():
    response = await chain.ainvoke({"country": "일본"}) 
    print(f"\n[LLM 답변 도착]: {response}")

async def main():
    task = asyncio.create_task(async_run())
    
    print("--- 숫자 세기 시작 ---")
    for i in range(10):
        print(i, end=" ")
        await asyncio.sleep(0.5) # 텀주기
    print("\n--- 숫자 세기 끝 ---")

    await task 

# 실행
await main()

# 출력
--- 숫자 세기 시작 ---
0 1 2 
[LLM 답변 도착]: 일본의 수도는 도쿄(東京)입니다. 도쿄는 일본의 정치, 경제, 문화의 중심지로, 많은 인구와 다양한 명소가 있는 대도시입니다.
3 4 5 6 7 8 9 
--- 숫자 세기 끝 ---

import asyncio

async def async_run():
    response = await chain.abatch([ 
        {"country": "대한민국"}, 
        {"country": "미국"}, 
        {"country": "프랑스"} 
    ])
    print(f"\n[LLM 답변 도착]: {response}")

async def main():
    task = asyncio.create_task(async_run())
    
    print("--- 숫자 세기 시작 ---")
    for i in range(10):
        print(i, end=" ")
        await asyncio.sleep(0.5) # 텀주기
    print("\n--- 숫자 세기 끝 ---")

    await task 

# 실행
await main()

# 출력
--- 숫자 세기 시작 ---
0 1 2 
[LLM 답변 도착]: ['대한민국의 수도는 서울입니다.', '미국의 수도는 워싱턴 D.C.입니다.', '프랑스의 수도는 파리입니다.']
3 4 5 6 7 8 9 
--- 숫자 세기 끝 ---

🎈 RunnableParallel (병렬처리)

독립적인 두 개의 체인을 동시에 실행 가능

from langchain_core.runnables import RunnableParallel

# Chain 1: 수도 질문
chain1 = (
    PromptTemplate.from_template("{country}의 수도는?")
    | model
    | StrOutputParser()
)

# Chain 2: 면적 질문
chain2 = (
    PromptTemplate.from_template("{country}의 면적은?")
    | model
    | StrOutputParser()
)

# 병렬 체인 결합
combined = RunnableParallel(capital=chain1, area=chain2)

# 실행
result = combined.invoke({"country": "대한민국"})
print(result)

# 출력
{'capital': '대한민국의 수도는 서울입니다.', 
'area': '대한민국의 면적은 약 100,210 평방킬로미터입니다. 이는 한반도의 남쪽 부분에 해당하며, 북한과 함께 한반도를 구성하고 있습니다.'}

✔️ RunnablePassthrough & RunnableLambda

• RunnablePassthrough : 입력받은 값을 그대로 변수 전달

from langchain_core.runnables import RunnablePassthrough

# {"num": 5} 대신 5만 넣어도 {"num": 5}로 변환되어 프롬프트로 전달됨
chain = {"num": RunnablePassthrough()} | PromptTemplate.from_template("{num}의 10배는?") | model

print(chain.invoke(5).content)
# 출력
5의 10배는 50입니다.

	- assign (값 추가) : 기존 데이터 흐름에 새로운 데이터를 추가

# 입력값 x에 대해 x["num"] * 3 한 값을 'mult'라는 키로 추가
(RunnablePassthrough.assign(mult=lambda x: x["num"] * 3)).invoke({"num": 1})

# 결과: {'num': 1, 'mult': 3}

• RunnableLambda : 사용자 정의 함수를 맵핑 가능

from langchain_core.runnables import RunnableLambda
from datetime import datetime

# 사용자 정의 함수를 RunnableLambda 에서 사용할 때 매개변수가 없으면 오류 반환
def get_today(x):
    # 매개변수와 상관없이 오늘 날짜 return
    return datetime.today().strftime("%b-%d")

# 체인 구성: 오늘 날짜를 가져와서 프롬프트에 주입
chain = (
    {"today": RunnableLambda(get_today), "n": RunnablePassthrough()}
    # today에 오늘 날짜, n은 invoke 시 전달받은 값
    | PromptTemplate.from_template("{today}가 생일인 대한민국의 유명인 {n}명을 알려줘")
    | model
    | StrOutputParser()
)

chain.invoke(2) 
# prompt: 오늘 날짜(Dec-23) 가 생일인 대한민국의 유명인 2명을 알려줘

# 출력 (🤔 결과값이 이상하다 ... 생일이신분이 아니신데)
12월 23일에 생일인 대한민국의 유명인으로는 다음 두 명이 있습니다:

1. **이상민** - 대한민국의 가수이자 방송인으로, 그룹 '젝스키스'의 멤버로 잘 알려져 있습니다.
2. **김범수** - 대한민국의 가수로, 뛰어난 가창력으로 많은 사랑을 받고 있는 아티스트입니다.

이 외에도 12월 23일에 태어난 유명인들이 있을 수 있습니다.

---