LangChain 기초 (2) - Runnable and LCEL: 부품 규격과 파이프 합성

모든 부품이 공유하는 Runnable 인터페이스를 하나씩 호출해 확인하고, 파이프 연산자가 만드는 RunnableSequence와 값을 흘려보내는 Passthrough, 갈래를 나누는 Parallel, 함수를 끼우는 Lambda를 다룹니다.

Posted 2026/08/07

By 강호승

5 min read

LangChain 기초 시리즈의 2편입니다. 전체 목차는 0편에 있습니다.

1편에서 프레임워크의 제안은 “모든 부품에 같은 인터페이스를 씌우는 것”이라고 했습니다. 이번 편은 그 인터페이스를 직접 만져 봅니다. 여기서 만든 체인을 7편까지 이어 씁니다.

부품이 되기 위한 조건

LangChain에서 부품이라고 부를 수 있는 것은 Runnable을 구현한 객체입니다. 조건은 메서드 몇 개를 갖추는 것뿐입니다.

메서드	하는 일
`invoke(input)`	입력 하나를 받아 출력 하나를 돌려준다
`batch(inputs)`	입력 리스트를 동시에 처리한다
`stream(input)`	출력을 조각으로 흘려보낸다
`ainvoke`, `abatch`, `astream`	위 셋의 async 버전

프롬프트도 모델도 파서도 전부 이 규격을 따릅니다.

  
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

SYSTEM = """너는 온라인 쇼핑몰의 고객 지원 담당자다.

<규칙>
- <자료>에 적힌 내용만 근거로 답한다. 자료에 없는 내용은 답에 쓰지 않는다.
- 자료로 답할 수 없으면 "자료에 없습니다"라고만 답하고 추측하지 않는다.
</규칙>"""

model = init_chat_model("anthropic:claude-sonnet-5", max_tokens=1024)
prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM),
    ("user", "<자료>\n{context}\n</자료>\n\n<질문>\n{question}\n</질문>"),
])
parser = StrOutputParser()

for part in (prompt, model, parser):
    print(type(part).__name__, [m for m in ("invoke", "batch", "stream") if hasattr(part, m)])

# ChatPromptTemplate ['invoke', 'batch', 'stream']
# ChatAnthropic ['invoke', 'batch', 'stream']
# StrOutputParser ['invoke', 'batch', 'stream']

하나씩 호출해 본다

부품을 잇기 전에 따로따로 불러 보면 데이터가 어떤 모양으로 흐르는지 보입니다. 이 습관이 나중에 체인이 깨졌을 때 어느 부품에서 틀어졌는지 찾는 방법이 됩니다.

  
payload = {
    "context": "골드 등급부터는 주문 금액과 관계없이 배송비가 무료입니다.",
    "question": "골드 등급이면 배송비가 무료인가요?",
}

a = prompt.invoke(payload)
print(type(a).__name__)        # ChatPromptValue
print(a.to_messages())         # [SystemMessage(...), HumanMessage(...)]

b = model.invoke(a)
print(type(b).__name__)        # AIMessage
print(b.text)                  # 네, 골드 등급부터는 배송비가 무료입니다.

c = parser.invoke(b)
print(type(c).__name__)        # str

dict 하나가 들어가서 ChatPromptValue가 되고, 그것이 AIMessage가 되고, 마지막에 str이 됩니다. 부품이 서로 이어질 수 있는 이유는 앞 부품의 출력 타입이 뒤 부품이 받을 수 있는 입력 타입이기 때문입니다.

AIMessage에서 텍스트를 꺼낼 때는 .text를 씁니다. .content는 문자열일 수도 있고 블록 리스트일 수도 있어서, 텍스트만 필요할 때는 .text가 안전합니다.

파이프가 만드는 것

이제 잇습니다.

  
chain = prompt | model | parser

print(type(chain).__name__)          # RunnableSequence
print([type(s).__name__ for s in chain.steps])
# ['ChatPromptTemplate', 'ChatAnthropic', 'StrOutputParser']

print(chain.invoke(payload))

|는 마법이 아니라 RunnableSequence를 만드는 연산자입니다. 왼쪽 부품의 출력을 오른쪽 부품에 넘기는 객체를 하나 만들 뿐입니다. chain 자체도 Runnable이므로 다른 체인 안에 부품으로 다시 들어갈 수 있습니다. 이 점이 뒤 편들에서 계속 쓰입니다.

체인 전체가 무엇을 받는지도 물어볼 수 있습니다.

  
print(chain.get_input_jsonschema()["properties"].keys())
# dict_keys(['context', 'question'])

프롬프트 템플릿의 자리 이름이 곧 체인의 입력 스키마가 됩니다. 자리를 하나 늘리면 체인의 입력도 따라 바뀝니다.

규격을 맞추면 딸려오는 것

chain은 Runnable이므로 Runnable의 메서드를 전부 가집니다. 따로 구현한 것이 없는데도 배치와 스트리밍이 됩니다.

  
# 배치. max_concurrency로 동시 요청 수를 제한한다
payloads = [
    {"context": "기본 배송비는 3,000원입니다.", "question": "배송비 얼마예요?"},
    {"context": "평일 오후 2시 이전 결제분은 당일 출고됩니다.", "question": "언제 출고되나요?"},
]
for out in chain.batch(payloads, config={"max_concurrency": 5}):
    print(out)

# 스트리밍
for piece in chain.stream(payload):
    print(piece, end="", flush=True)

스트리밍에서 눈여겨볼 것은 조각이 파서를 통과해서도 계속 흐른다는 점입니다. StrOutputParser는 들어오는 조각을 모아 두지 않고 그때그때 문자열로 바꿔 내보내도록 만들어져 있습니다. 반대로 전체 출력이 모여야 결과를 낼 수 있는 부품이 체인 중간에 있으면 그 지점에서 스트리밍이 끊기고, 앞에서 아무리 조각을 흘려보내도 사용자는 한 번에 받게 됩니다. 어떤 부품이 그런지는 3편에서 구조화 출력을 다루며 봅니다.

async도 마찬가지로 그냥 있습니다.

  
import asyncio

async def main():
    print(await chain.ainvoke(payload))

asyncio.run(main())

값을 그대로 흘려보내기

여기까지는 입력이 dict 하나로 딱 맞아떨어졌습니다. 실제로는 앞 단계의 출력을 뒷단계의 여러 자리에 나눠 넣어야 하는 경우가 많습니다.

체인에 질문 문자열 하나만 넘기고 싶다고 해봅니다. 프롬프트는 context와 question 두 자리를 요구하므로, 들어온 문자열을 두 갈래로 흘려보내야 합니다.

  
from langchain_core.runnables import RunnablePassthrough

def lookup(question: str) -> str:
    """일단은 고정 문자열을 돌려준다. 4편에서 리트리버로 바꾼다."""
    return "골드 등급부터는 주문 금액과 관계없이 배송비가 무료입니다."

chain2 = (
    {"context": lookup, "question": RunnablePassthrough()}
    | prompt
    | model
    | parser
)

print(chain2.invoke("골드 등급이면 배송비가 무료인가요?"))

두 가지가 조용히 일어났습니다.

dict가 부품이 되었습니다. 체인 자리에 놓인 dict는 RunnableParallel로 바뀝니다. 들어온 값을 각 키의 부품에 똑같이 한 번씩 넘기고, 결과를 같은 키의 dict로 모읍니다. 키가 여럿이면 동시에 실행됩니다.

함수가 부품이 되었습니다. 체인 자리에 놓인 평범한 함수는 RunnableLambda로 감싸집니다. lookup은 Runnable을 상속하지 않았지만 파이프 안에서는 부품처럼 동작합니다.

RunnablePassthrough는 받은 값을 그대로 내보내는 부품입니다. 위에서는 질문 문자열을 question 자리에 그대로 꽂는 역할을 합니다.

명시적으로 쓰면 이렇게 됩니다. 동작은 같습니다.

  
from langchain_core.runnables import RunnableParallel, RunnableLambda

chain2 = (
    RunnableParallel(context=RunnableLambda(lookup), question=RunnablePassthrough())
    | prompt | model | parser
)

평소에는 dict 축약형을 쓰고, 갈래가 복잡해져 읽기 어려울 때 명시형으로 풀어 쓰면 됩니다.

중간에 값을 덧붙이기

들어온 dict를 유지한 채 키를 하나 더하고 싶을 때는 assign을 씁니다. 앞 단계 결과를 버리지 않아야 하는 경우에 자주 쓰입니다.

  
enriched = RunnablePassthrough.assign(
    context=lambda x: lookup(x["question"])
)

print(enriched.invoke({"question": "골드 등급이면 배송비가 무료인가요?"}))
# {'question': '골드 등급이면 배송비가 무료인가요?',
#  'context': '골드 등급부터는 주문 금액과 관계없이 배송비가 무료입니다.'}

{"context": ..., "question": ...}가 새 dict를 만드는 것이라면, assign은 원래 dict에 키를 얹는 것입니다. 5편에서 답과 함께 출처를 반환할 때 이 차이가 결정적입니다.

함정

dict 안의 값이 부품이 아니면 상수가 됩니다. {"context": lookup(question)}처럼 함수를 호출해서 넣으면 체인을 만들 때 한 번 실행된 결과가 박제됩니다. 호출하지 않고 함수 자체를 넘겨야 매 입력마다 실행됩니다. 괄호 하나 차이라 발견하기 어렵습니다.

RunnableLambda로 감싸지는 함수는 인자를 하나만 받습니다. 인자가 둘 이상인 함수를 그대로 꽂으면 입력 dict 전체가 첫 인자로 들어갑니다. lambda x: f(x["a"], x["b"])로 풀거나 functools.partial로 고정합니다.

체인 정의는 실행이 아닙니다. prompt | model | parser를 만드는 시점에는 API가 호출되지 않습니다. 이 줄이 오래 걸린다면 원인은 다른 데 있습니다.

체인 그림이 보고 싶으면 별도 패키지가 필요합니다. chain.get_graph().print_ascii()는 grandalf가 설치되어 있어야 동작합니다. 없으면 위에서 쓴 chain.steps로 충분합니다.

정리

Runnable은 invoke, batch, stream과 그 async 짝을 갖춘 규격이고, 프롬프트와 모델과 파서가 전부 이 규격을 따른다
|는 RunnableSequence를 만드는 연산자이고, 만들어진 체인 자체도 Runnable이라 다시 부품이 된다
규격을 맞췄기 때문에 배치와 스트리밍과 async를 따로 구현하지 않아도 된다
체인 자리의 dict는 RunnableParallel, 평범한 함수는 RunnableLambda로 자동 변환된다
새 dict를 만들 때는 dict 축약형, 기존 dict에 키를 더할 때는 RunnablePassthrough.assign

다음 편에서 체인의 양끝을 정리합니다. 프롬프트 템플릿에 대화 이력을 끼우는 방법과 예시를 넣는 방법, 그리고 모델의 답을 문자열이 아니라 검증된 객체로 받는 방법을 봅니다.

다음 글: LangChain 기초 (3) - Prompts and Output Parsers: 템플릿과 구조화 출력

AI, LangChain