Runnable 인터페이스

Runnable 인터페이스 (Runnable interface)

LangChain의 컴포넌트들은 각자 역할이 다르지만, 사용하는 방식은 통일되어 있어요. 그 바탕이 되는 게 바로 Runnable 인터페이스예요. 언어 모델, 출력 파서, 리트리버, 컴파일된 LangGraph 그래프 등 수많은 컴포넌트가 이 인터페이스를 구현합니다. 덕분에 서로 다른 컴포넌트를 일관되고 예측 가능한 방식으로 다룰 수 있어요.

출처: 공식문서

Runnable 인터페이스 개요

Runnable 방식은 표준 인터페이스를 정의해서, 러너블 컴포넌트가 다음과 같은 일들을 할 수 있게 해요.

  • Invoke(호출): 단일 입력을 출력으로 변환해요.
  • Batch(배치): 여러 입력을 효율적으로 출력으로 변환해요.
  • Stream(스트리밍): 출력이 만들어지는 대로 스트리밍해요.
  • Inspect(검사): 러너블의 입력·출력·구성(config)에 대한 스키마 정보에 접근할 수 있어요.
  • Compose(조합): LangChain Expression Language(LCEL)로 여러 러너블을 조합해 복잡한 파이프라인을 만들 수 있어요.

Runnable 인터페이스와 LCEL 표현식의 흔한 패턴은 LCEL 치트시트에서 확인할 수 있어요.

최적화된 병렬 실행 (batch)

LangChain 러너블은 여러 입력을 병렬로 처리할 수 있도록 내장 batch(그리고 batch_as_completed) API를 제공해요. 서로 독립적인 여러 입력을 처리해야 할 때 순차 대신 병렬로 처리하면 성능이 크게 좋아져요.

두 가지 배치 옵션이 있어요.

  • batch: 여러 입력을 병렬로 처리하고, 입력 순서와 같은 순서로 결과를 돌려줘요.
  • batch_as_completed: 여러 입력을 병렬로 처리하되, 완료되는 대로 결과를 돌려줘요. 결과가 순서대로 오지 않을 수 있지만, 각 결과에 입력 인덱스가 포함되어 매칭할 수 있어요.

batch와 batch_as_completed의 기본 구현은 thread pool executor로 invoke를 병렬 실행해요. 그래서 사용자가 스레드를 직접 관리하지 않아도 효율적인 병렬 실행이 가능하고, I/O 중심 코드(API 요청, 파일 읽기 등)가 빨라져요. 다만 CPU 중심 작업에서는 Python의 GIL 때문에 진짜 병렬 실행이 안 돼서 효과가 크지 않아요.

일부 러너블은 자신의 사용 사례에 최적화된 batch/batch_as_completed 구현을 직접 제공하기도 해요(예: 모델 프로바이더가 제공하는 batch API를 활용).

참고: abatch, abatch_as_completed 같은 비동기 버전은 asyncio의 gather와 as_completed 함수로 ainvoke를 병렬 실행해요.

팁: batch나 batch_as_completed로 많은 입력을 처리할 때 RunnableConfig dict의 max_concurrency 속성을 설정하면 병렬 호출 수를 제어할 수 있어요. 채팅 모델에는 요청 빈도를 제어하는 내장 rate limiter도 있으니 함께 참고하세요.

비동기 지원 (async)

러너블은 비동기 API를 제공해서 Python에서 await 문법으로 호출할 수 있어요. 비동기 메서드는 "a" 접두사로 구분하면 돼요(예: ainvoke, abatch, astream, abatch_as_completed). LangChain에서의 비동기 프로그래밍에 대한 자세한 내용은 별도 가이드를 참고하세요.

스트리밍 API

스트리밍은 LLM 기반 애플리케이션이 사용자에게 즉각적으로 반응하는 것처럼 느껴지게 하는 데 아주 중요해요. 러너블은 세 가지 스트리밍 API를 제공합니다.

  1. 동기 stream과 비동기 astream: 러너블의 출력이 생성되는 대로 산출(yield)해요.
  2. 비동기 astream_events: 중간 단계와 최종 출력까지 스트리밍하는 더 고급 API예요.
  3. 레거시 비동기 astream_log: 중간 단계와 최종 출력을 스트리밍하는 예전 API예요.

LangChain에서 스트리밍하는 방법에 대한 자세한 내용은 스트리밍 개념 가이드를 참고하세요.

입력과 출력 타입

모든 Runnable은 입력 타입과 출력 타입으로 특징지어져요. 이 타입들은 어떤 Python 객체든 될 수 있고, 러너블 스스로가 정의합니다.

러너블을 실행하는 메서드(invoke, batch, stream, astream_events 등)는 이 입력·출력 타입을 그대로 사용해요.

  • invoke: 입력을 받아 출력을 반환해요.
  • batch: 입력 리스트를 받아 출력 리스트를 반환해요.
  • stream: 입력을 받아 출력을 산출하는 제너레이터를 반환해요.

입력 타입과 출력 타입은 컴포넌트마다 달라요.

컴포넌트 입력 타입 출력 타입
Prompt dict PromptValue
ChatModel 문자열, 채팅 메시지 리스트 또는 PromptValue ChatMessage
LLM 문자열, 채팅 메시지 리스트 또는 PromptValue String
OutputParser LLM 또는 ChatModel의 출력 파서에 따라 다름
Retriever 문자열 Document 리스트
Tool 도구에 따라 문자열 또는 dict 도구에 따라 다름

각 컴포넌트의 입력·출력 타입과 사용법은 해당 컴포넌트 문서를 참고하세요.

스키마 검사하기

참고: 이건 대부분의 사용자에게 불필요한 고급 기능이에요. 러너블의 스키마를 검사해야 할 특별한 이유가 없다면 이 섹션은 건너뛰어도 돼요.

좀 더 고급 사용에서는 러너블이 기대하고 생성하는 입력·출력 타입을 프로그래밍 방식으로 검사(inspect) 하고 싶을 수 있어요. Runnable 인터페이스는 입력·출력 타입의 JSON Schema와 Pydantic 스키마를 얻는 메서드를 제공해요. 이 API는 주로 내부적으로 단위 테스트와 LangServe(입력 검증과 OpenAPI 문서 생성에 사용)에서 쓰여요.

입력·출력 타입 외에도 일부 러너블은 런타임 구성 옵션을 추가로 갖는데, 그 구성 옵션의 Pydantic/JSON 스키마를 얻는 API도 있어요.

메서드 설명
get_input_schema 러너블 입력 스키마의 Pydantic 스키마를 돌려줘요.
get_output_schema 러너블 출력 스키마의 Pydantic 스키마를 돌려줘요.
config_schema 러너블 config 스키마의 Pydantic 스키마를 돌려줘요.
get_input_jsonschema 러너블 입력 스키마의 JSONSchema를 돌려줘요.
get_output_jsonschema 러너블 출력 스키마의 JSONSchema를 돌려줘요.
get_config_jsonschema 러너블 config 스키마의 JSONSchema를 돌려줘요.

with_types

LangChain은 사용 가능한 정보를 바탕으로 러너블의 입력·출력 타입을 자동으로 추론하려 해요. 다만 현재는 LCEL 컴포지션으로 만든 더 복잡한 러너블에서는 추론이 잘 안 돼서 입력·출력 타입이 틀릴 수 있어요. 이럴 때는 with_types 메서드로 추론된 타입을 직접 덮어쓰는 걸 권장해요.

RunnableConfig

러너블을 실행하는 메서드(invoke, batch, stream, astream_events 등)는 RunnableConfig라는 두 번째 인자를 받아요. 이 인자는 러너블 실행 시 런타임에 사용될 구성을 담은 dict예요.

RunnableConfig에는 다음 속성을 정의할 수 있어요.

속성 설명
run_name 해당 러너블에 쓰는 이름 (하위 호출에는 상속되지 않아요).
run_id 이 호출의 고유 식별자. 하위 호출은 자신만의 고유 run id를 가져요.
tags 이 호출과 모든 하위 호출에 붙는 태그.
metadata 이 호출과 모든 하위 호출에 붙는 메타데이터.
callbacks 이 호출과 모든 하위 호출에 대한 콜백.
max_concurrency 만들 병렬 호출의 최대 수 (예: batch에서 사용).
recursion_limit 호출이 재귀할 수 있는 최대 횟수 (예: 러너블을 반환하는 러너블에서 사용).
configurable 러너블의 구성 가능한 속성에 대한 런타임 값.

invoke에 config를 넘기는 방법은 이렇습니다.

some_runnable.invoke(
   some_input,
   config={
      'run_name': 'my_run',
      'tags': ['tag1', 'tag2'],
      'metadata': {'key': 'value'}

   }
)

RunnableConfig 전파

많은 Runnable은 다른 러너블로 구성돼요. 그래서 만든 러너블이 내부적으로 하는 모든 하위 호출에 RunnableConfig가 전파(propagate)되는 게 중요해요. 그래야 부모 러너블에 준 런타임 구성 값이 모든 하위 호출에 상속되거든요. 만약 그렇지 않다면 모든 하위 호출이 상속받기를 기대하는 callbacks나 tags, metadata 같은 구성 값을 설정하고 전파할 수 없게 돼요.

새 Runnable을 만드는 두 가지 주요 패턴이 있어요.

  1. LCEL로 선언적으로 만들기:

    chain = prompt | chat_model | output_parser

  2. 커스텀 러너블(예: RunnableLambda)이나 @tool 데코레이터로 만들기:

    def foo(input):
    # Note that .invoke() is used directly here
    return bar_runnable.invoke(input)
foo_runnable = RunnableLambda(foo)

LangChain은 두 패턴 모두에서 RunnableConfig를 자동으로 전파하려 해요. 두 번째 패턴을 처리할 때는 Python의 contextvars에 의존합니다.

Python 3.11 이상에서는 이게 기본으로 동작해서 RunnableConfig를 하위 호출에 전파하기 위해 특별히 할 일이 없어요. 다만 Python 3.9·3.10에서 비동기 코드를 쓴다면 호출 시 RunnableConfig를 직접 Runnable에 넘겨줘야 해요. 3.9·3.10의 asyncio tasks가 context 인자를 받지 못하는 제약 때문이에요.

async def foo(input, config): # <-- Note the config argument
    return await bar_runnable.ainvoke(input, config=config)

foo_runnable = RunnableLambda(foo)

주의: Python 3.10 이하에서 비동기 코드를 쓴다면 RunnableConfig가 자동으로 전파되지 않아서 직접 해야 해요! 특히 astream_events와 astream_log로 데이터를 스트리밍할 때 흔한 함정이에요. 이 메서드들은 RunnableConfig 안에 정의된 콜백의 정상적인 전파에 의존하기 때문이죠.

커스텀 run 이름·태그·메타데이터 설정

RunnableConfig dict의 run_name, tags, metadata 속성으로 러너블의 run 이름·태그·메타데이터에 커스텀 값을 설정할 수 있어요.

run_name은 run에 붙일 이름을 설정하는 문자열이에요. 로그 등에서 이 run을 식별하는 데 쓰이며, 하위 호출에는 상속되지 않아요.

tags(리스트)와 metadata(dict)는 run에 커스텀 태그와 메타데이터를 붙이는 데 쓰는데, 이 값들은 하위 호출에 상속됩니다. 이 속성들은 실행 추적·디버깅에 유용하고, LangSmith에서 필터링·검색할 수 있는 trace 속성으로 표시돼요. 또한 콜백에도 전파되고, astream_events 같은 스트리밍 API에서 각 이벤트의 일부로 나타나요.

run id 설정

참고: 대부분의 사용자에게 불필요한 고급 기능이에요.

나중에 참조하거나 다른 시스템과 연관 지으려고 특정 run에 커스텀 run_id를 설정해야 할 수 있어요. run_id는 반드시 유효한 UUID 문자열이어야 하고 각 run마다 고유해야 해요. 부모 run을 식별하는 데 쓰이며, 하위 호출은 자동으로 자신만의 고유 run id를 가져요.

import uuid

run_id = uuid.uuid4()

some_runnable.invoke(
   some_input,
   config={
      'run_id': run_id
   }
)

# Do something with the run_id

recursion limit 설정

참고: 대부분의 사용자에게 불필요한 고급 기능이에요.

일부 러너블은 다른 러너블을 반환할 수 있어서 제대로 처리하지 않으면 무한 재귀에 빠질 수 있어요. 이것을 막으려면 RunnableConfig dict에 recursion_limit을 설정하면 돼요. 러너블이 재귀할 수 있는 횟수를 제한해 줍니다.

max concurrency 설정

batch나 batch_as_completed를 쓸 때 RunnableConfig dict의 max_concurrency 속성으로 병렬 호출의 최대 수를 제어할 수 있어요. 서버나 API에 과부하를 주지 않도록 병렬 호출 수를 제한하고 싶을 때 유용해요.

팁: 채팅 모델이 보내는 요청 수를 제한하고 싶다면 max_concurrency 대신 내장 rate limiter를 쓰는 게 더 효과적이에요. 자세한 내용은 rate limit 처리 가이드를 참고하세요.

configurable 설정

configurable 필드는 러너블의 구성 가능한 속성에 런타임 값을 전달할 때 써요. LangGraph의 Persistence·memory 기능과 그리고 RunnableWithMessageHistory에서 대화 기록을 추적하기 위한 session_id/conversation_id를 지정할 때 자주 사용돼요. 또한 직접 만든 Configurable Runnable에 넘길 커스텀 구성 옵션을 지정하는 데도 쓸 수 있어요.

callbacks 설정

이 옵션으로 런타임에 러너블의 콜백을 구성해요. 설정한 콜백은 러너블이 만드는 모든 하위 호출에 전달됩니다.

some_runnable.invoke(
   some_input,
   {
      "callbacks": [
         SomeCallbackHandler(),
         AnotherCallbackHandler(),
      ]
   }
)

중요: Python 3.9·3.10의 비동기 환경이라면 경우에 따라 RunnableConfig를 하위 호출에 직접 전파해야 해요. 자세한 내용은 'RunnableConfig 전파' 섹션을 참고하세요.

함수로 러너블 만들기

임의의 로직을 실행하는 커스텀 Runnable을 만들어야 할 수 있어요. 특히 LCEL로 여러 러너블을 조합할 때 그중 한 단계에 커스텀 처리 로직을 추가하고 싶을 때 유용하죠. 함수로 커스텀 러너블을 만드는 방법은 두 가지예요.

  • RunnableLambda: 스트리밍이 필요 없는 단순 변환에 씁니다.
  • RunnableGenerator: 스트리밍이 필요할 때 더 복잡한 변환에 씁니다.

중요: 새로운 커스텀 Runnable을 만들려고 러너블을 서브클래스(subclass)로 확장하는 것은 피하세요. RunnableLambda나 RunnableGenerator를 쓰는 것보다 훨씬 복잡하고 오류가 나기 쉬워요.

Configurable runnables

참고: 대부분의 사용자에게 불필요한 고급 기능이에요. LCEL로 만든 큰 "체인"의 구성을 도와주고, 배포된 러너블에 대해 LangServe가 활용해요.

때로는 러너블을 다양한 방식으로 동작하도록 실험하거나 최종 사용자에게 여러 옵션을 노출하고 싶을 수 있어요. 예를 들어 채팅 모델의 temperature 같은 파라미터를 조정하거나 아예 다른 채팅 모델로 전환하는 경우죠.

이를 단순화하기 위해 Runnable 인터페이스는 런타임에 구성 가능한 러너블을 만드는 두 메서드를 제공해요.

  • configurable_fields: 러너블의 특정 속성을 구성할 수 있게 해요. 예를 들어 채팅 모델의 temperature 속성 같은 거죠.
  • configurable_alternatives: 런타임에 실행할 대체 러너블을 지정할 수 있게 해요. 예를 들어 사용할 채팅 모델 목록을 지정하는 식이에요.

런타임 체인 내부 구성을 설정하는 방법은 별도 가이드를 참고하세요.