LLM 애플리케이션 개발 훑어보기 - LangChain #2 LangChain Expression Language (LCEL)

LangChain Experssion Language(LCEL)

LangChain Expression Language(LCEL)은 체인을 쉽게 구성할 수 있는 선언적 방식입니다. LCEL은 가장 간단한 "프롬프트 + LLM" 체인부터 가장 복잡한 체인까지 코드 변경 없이 프로토타입을 프로덕션에 적용하는 것을 지원하도록 설계되었습니다. 

LangChain 공식 문서에서는 LCEL을 사용해야 하는 이유를 다음과 같이 소개하고 있습니다.

• Streaming support: LCEL로 체인을 구축하면 첫 번째 토큰에 도달하는 시간(첫 번째 출력 청크가 나올 때까지 경과한 시간)을 최대한 단축할 수 있습니다.
• Async support: LCEL로 구축된 모든 체인은 동기식 API(예: 프로토타이핑 중 Jupyter 노트북에서)와 비동기식 API(예: LangServe 서버에서)로 모두 호출할 수 있습니다. 이를 통해 프로토타입과 프로덕션에서 동일한 코드를 사용할 수 있으며, 뛰어난 성능과 함께 동일한 서버에서 많은 동시 요청을 처리할 수 있습니다.
• Optimized parallel execution: LCEL 체인에 병렬로 실행할 수 있는 단계가 있을 때마다(예: 여러 검색기에서 문서를 가져오는 경우) 동기화 및 비동기 인터페이스 모두에서 자동으로 실행하여 지연 시간을 최소화합니다.
• Retries and fallbacks: LCEL 체인의 모든 부분에 대해 재시도 및 폴백을 구성할 수 있습니다. 이는 체인을 대규모로 더욱 안정적으로 만들 수 있는 좋은 방법입니다.
• Access intermediate results: 더 복잡한 체인의 경우, 최종 결과물이 생성되기 전에도 중간 단계의 결과에 액세스하는 것이 매우 유용할 때가 많습니다. 이는 최종 사용자에게 어떤 일이 일어나고 있음을 알리거나 체인을 디버깅하는 데 사용할 수 있습니다.
• Input and output schemas: 입력 및 출력 스키마는 모든 LCEL 체인에 체인의 구조로부터 추론된 Pydantic 및 JSONSchema 스키마를 제공합니다. 이는 입력 및 출력의 유효성 검사에 사용할 수 있습니다.

 

---

 

LCEL Interface

LCEL에는 custom chain을 최대한 쉽게 만들 수 있도록 Runnable 프로토콜이 구현되어 있습니다. Runnable 프로토콜은 대부분의 컴포넌트에 구현되어 있으며, 이는 표준 인터페이스로 custom chain을 쉽게 정의하고 표준화된 방식으로 호출할 수 있게 해줍니다. 표준 인터페이스에는 다음이 포함됩니다:

• stream: 응답 결과의 청크를 스트리밍합니다.
• invoke: 단일 입력 값에 대한 chain을 호출합니다.
• batch: 여러 입력 값에 대한 chain을 호출합니다.

또한 위의 표준 인터페이스에 대한 비동기 메서드도 제공합니다:

• astream: 응답 결과의 청크를 비동기 스트리밍합니다.
• ainvoke: 단일 입력 값에 대한 chain을 비동기로 호출합니다.
• abatch: 여러 입력 값에 대한 chain을 비동기로 호출합니다.
• astream_log: 최종 응답 외에도 중간 단계가 발생하면 스트리밍합니다.

 

표준 인터페이스의 입력 유형과 출력 유형은 컴포넌트에 따라 다릅니다:

Component	Input Type	Output Type
Prompt	Dictionary	PromptValue
ChatModel	Single string, list of chat messages or a PromptValue	ChatMessage
LLM	Single string, list of chat messages or a PromptValue	String
OutputParser	The output of an LLM or ChatModel	Depends on the parser
Retrieval	Single string	List of Documents
Tool	Single string or dictionary, depending on the tool	Depends on the tool

 

Input Schema

Runnable이 받아들이는 입력에 대한 설명입니다. 이것은 모든 Runnable의 구조에서 동적으로 생성되는 Pydantic Model입니다. .schema()를 호출하여 JSONSchema 표현을 얻을 수 있습니다.

# The input schema of the chain is the input schema of its first part, the prompt.
chain.input_schema.schema()


# {'title': 'PromptInput',
#  'type': 'object',
#  'properties': {'topic': {'title': 'Topic', 'type': 'string'}}}

prompt.input_schema.schema()

# {'title': 'PromptInput',
#  'type': 'object',
#  'properties': {'topic': {'title': 'Topic', 'type': 'string'}}}

 

Output Schema

Runnable이 생성하는 출력에 대한 설명입니다. 이것은 모든 Runnable의 구조에서 동적으로 생성되는 Pydantic Model입니다. .schema()를 호출하여 JSONSchema 표현을 얻을 수 있습니다.

 

Stream

응답 결과의 청크를 스트리밍합니다.

for s in chain.stream({"topic": "bears"}):
    print(s.content, end="", flush=True)
    
# output
# Why don't bears wear shoes?
# 
# Because they already have bear feet!

Stream를 사용하면 기존 코드와 비교해서 다음과 같이 간단하게 구현할 수 있습니다.

 

Invoke

단일 입력값에 대해 chain을 호출합니다.

chain.invoke({"topic": "bears"})

# AIMessage(content="Why don't bears wear shoes?\n\nBecause they already have bear feet!")

Invoke를 사용하면 기존 코드와 비교해서 다음과 같이 간단하게 구현할 수 있습니다.

 

Batch

여러 입력값에 대해 chain을 호출합니다.

chain.batch([{"topic": "bears"}, {"topic": "cats"}])

# [AIMessage(content="Why don't bears wear shoes?\n\nBecause they have bear feet!"),
#  AIMessage(content="Why don't cats play poker in the wild?\n\nToo many cheetahs!")]

max_concurrency 파라미터를 사용해서 동시 요청 수를 설정할 수 있습니다.

chain.batch([{"topic": "bears"}, {"topic": "cats"}], config={"max_concurrency": 5})

Batch를 사용하면 기존 코드와 비교해서 다음과 같이 간단하게 구현할 수 있습니다.

 

Async Stream

응답 결과의 청크를 비동기 스트리밍합니다.

async for s in chain.astream({"topic": "bears"}):
    print(s.content, end="", flush=True)

 

Async Invoke

단일 입력 값에 대한 chain을 비동기로 호출합니다.

await chain.ainvoke({"topic": "bears"})

Ainvoke를 사용하면 기존 코드와 비교해서 다음과 같이 간단하게 구현할 수 있습니다.

 

Async Batch

여러 입력 값에 대한 chain을 비동기로 호출합니다.

await chain.abatch([{"topic": "bears"}])

 

---

 

RunnableParallel

RunnableParallel은 여러 스텝으로 이루어진 Runnable Map을 병렬로 수행합니다. 생성자에서는 다양한 입력 유형에 대응하여 steps에 Runnable Mapping을 초기화합니다.

class RunnableParallel(RunnableSerializable[Input, Dict[str, Any]]):
    """
    A runnable that runs a mapping of runnables in parallel,
    and returns a mapping of their outputs.
    """

    steps: Mapping[str, Runnable[Input, Any]]

    def __init__(
        self,
        __steps: Optional[
            Mapping[
                str,
                Union[
                    Runnable[Input, Any],
                    Callable[[Input], Any],
                    Mapping[str, Union[Runnable[Input, Any], Callable[[Input], Any]]],
                ],
            ]
        ] = None,
        **kwargs: Union[
            Runnable[Input, Any],
            Callable[[Input], Any],
            Mapping[str, Union[Runnable[Input, Any], Callable[[Input], Any]]],
        ],
    ) -> None:
        merged = {**__steps} if __steps is not None else {}
        merged.update(kwargs)
        super().__init__(
            steps={key: coerce_to_runnable(r) for key, r in merged.items()}
        )

 

RunnableParallel의 수행은 ContextThreadPool 하위에서 병렬적으로 수행됩니다.

    def invoke(
        self, input: Input, config: Optional[RunnableConfig] = None
    ) -> Dict[str, Any]:
        ...

        # gather results from all steps
        try:
            # copy to avoid issues from the caller mutating the steps during invoke()
            steps = dict(self.steps)
            with get_executor_for_config(config) as executor:
                futures = [
                    executor.submit(
                        step.invoke,
                        input,
                        # mark each step as a child run
                        patch_config(
                            config,
                            callbacks=run_manager.get_child(f"map:key:{key}"),
                        ),
                    )
                    for key, step in steps.items()
                ]
                output = {key: future.result() for key, future in zip(steps, futures)}
        ...

 

이러한 RunnableParallel은 시퀀스에서 다음 Runnable의 입력 형식과 일치하도록 하나의 Runnable의 출력을 조작하는 데 유용할 수 있습니다. chain을 구성함에 있어서 아래 세 가지 표현은 같은 의미를 내포합니다:

{"context": retriever, "question": RunnablePassthrough()}

RunnableParallel({"context": retriever, "question": RunnablePassthrough()})

RunnableParallel(context=retriever, question=RunnablePassthrough())

 

위에서 본 input, output에 대한 manipulating 외에도 여러 Runnable을 병렬로 쉽게 실행하고 이러한 Runnable의 출력을 맵으로 반환할 수 있습니다.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI

model = ChatOpenAI()
joke_chain = ChatPromptTemplate.from_template("tell me a joke about {topic}") | model
poem_chain = (
    ChatPromptTemplate.from_template("write a 2-line poem about {topic}") | model
)

map_chain = RunnableParallel(joke=joke_chain, poem=poem_chain)

map_chain.invoke({"topic": "bear"})

{'joke': AIMessage(content="Why don't bears wear shoes?\n\nBecause they have bear feet!"),
 'poem': AIMessage(content="In the wild's embrace, bear roams free,\nStrength and grace, a majestic decree.")}

 

---

 

RunnablePassthrough

RunnablePassthrough를 사용하면 변경 없이 또는 추가 키를 추가하여 입력을 전달할 수 있습니다. 이는 일반적으로 RunnableParallel과 함께 사용되어 맵의 새 키에 데이터를 할당합니다. RunnablePassthrough()는 자체적으로 호출되며 단순히 입력을 받아 통과시킵니다.

assign 함수를 활용하면, RunnablePassthrough 객체는 입력을 받아 assign 함수에 추가 인수를 전달합니다.

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

runnable = RunnableParallel(
    passed=RunnablePassthrough(),
    extra=RunnablePassthrough.assign(mult=lambda x: x["num"] * 3),
    modified=lambda x: x["num"] + 1,
)

runnable.invoke({"num": 1})

{'passed': {'num': 1}, 'extra': {'num': 1, 'mult': 3}, 'modified': 2}

 

---

 

RunnableLambda

RunnableLambda는 파이썬 Callable을 Runnable로 변환합니다. Callable을 RunnableLambda로 wrapping하면 동기 또는 비동기 컨텍스트 내에서 Callable을 사용할 수 있습니다. RunnableLambda는 다른 Runnable과 마찬가지로 구성할 수 있습니다.

from operator import itemgetter

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


def length_function(text):
    return len(text)


def _multiple_length_function(text1, text2):
    return len(text1) * len(text2)


def multiple_length_function(_dict):
    return _multiple_length_function(_dict["text1"], _dict["text2"])


prompt = ChatPromptTemplate.from_template("what is {a} + {b}")
model = ChatOpenAI()

chain1 = prompt | model

chain = (
    {
        "a": itemgetter("foo") | RunnableLambda(length_function),
        "b": {"text1": itemgetter("foo"), "text2": itemgetter("bar")}
        | RunnableLambda(multiple_length_function),
    }
    | prompt
    | model
)

chain.invoke({"foo": "bar", "bar": "gah"})

AIMessage(content='3 + 9 equals 12.')

 

RunnableLambda는 선택적으로 콜백, 태그 및 기타 구성 정보를 중첩된 실행에 전달하는 데 사용할 수 있는 RunnableConfig를 받을 수 있습니다.

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableConfig
import json


def parse_or_fix(text: str, config: RunnableConfig):
    fixing_chain = (
        ChatPromptTemplate.from_template(
            "Fix the following text:\n\n```text\n{input}\n```\nError: {error}"
            " Don't narrate, just respond with the fixed data."
        )
        | ChatOpenAI()
        | StrOutputParser()
    )
    for _ in range(3):
        try:
            return json.loads(text)
        except Exception as e:
            text = fixing_chain.invoke({"input": text, "error": e}, config)
    return "Failed to parse"

from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    output = RunnableLambda(parse_or_fix).invoke(
        "{foo: bar}", {"tags": ["my-tag"], "callbacks": [cb]}
    )
    print(output)
    print(cb)

 

---

 

RunnableBranch

RunnableBranch는 조건에 따라 실행할 브랜치를 선택하는 Runnable입니다. RunnableBranch은 (condition, Runnable) 쌍의 리스트와 default branch로 초기화됩니다. 입력에 대해 연산할 때 True로 평가되는 첫 번째 조건이 선택되고 해당 Runnable이 입력에 대해 실행됩니다. True로 평가되는 조건이 없으면 default branch가 입력에 대해 실행됩니다.

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: isinstance(x, str), lambda x: x.upper()),
    (lambda x: isinstance(x, int), lambda x: x + 1),
    (lambda x: isinstance(x, float), lambda x: x * 2),
    lambda x: "goodbye",
)

branch.invoke("hello") # "HELLO"
branch.invoke(None) # "goodbye"

 

RunnableBranch를 사용하면 라우팅을 쉽게 구현할 수 있습니다. 라우팅을 사용하면 이전 단계의 출력이 다음 단계를 정의하는 비결정적 체인을 생성할 수 있습니다. 라우팅은 LLM과의 상호 작용에 대한 구조와 일관성을 제공하는 데 도움이 됩니다.

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "anthropic" in x["topic"].lower(), anthropic_chain),
    (lambda x: "langchain" in x["topic"].lower(), langchain_chain),
    general_chain,
)

 

---

 

RunnableSequence

Runnable의 시퀀스로, 각각의 출력이 다음의 입력이 되는 시퀀스입니다. RunnableSequence는 거의 모든 체인에서 사용되기 때문에 LangChain에서 가장 중요한 구성 연산자입니다. RunnableSequence는 직접 인스턴스화할 수도 있고, 일반적으로 `|` 연산자를 사용하여 왼쪽 또는 오른쪽 피연산자 중 하나(또는 둘 다)가 Runnable이어야 하는 경우 인스턴스화할 수도 있습니다.

{"equation_statement": RunnablePassthrough()} | prompt | model | StrOutputParser()

 

---

 

Runnable.bind()

때때로 시퀀스 내 이전 Runnable의 출력에 포함되지 않고 사용자 입력에 포함되지 않는 상수 인수를 사용하여 RunnableSequence 내에서 Runnable을 호출하고 싶을 때가 있습니다. 이러한 인수를 쉽게 전달하기 위해 Runnable.bind()를 사용할 수 있습니다.

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "Write out the following equation using algebraic symbols then solve it. Use the format\n\nEQUATION:...\nSOLUTION:...\n\n",
        ),
        ("human", "{equation_statement}"),
    ]
)
model = ChatOpenAI(temperature=0)
runnable = (
    {"equation_statement": RunnablePassthrough()} | prompt | model | StrOutputParser()
)

print(runnable.invoke("x raised to the third plus seven equals 12"))

"""
EQUATION: x^3 + 7 = 12

SOLUTION:
Subtracting 7 from both sides of the equation, we get:
x^3 = 12 - 7
x^3 = 5

Taking the cube root of both sides, we get:
x = ∛5

Therefore, the solution to the equation x^3 + 7 = 12 is x = ∛5.
"""

runnable = (
    {"equation_statement": RunnablePassthrough()}
    | prompt
    | model.bind(stop="SOLUTION")
    | StrOutputParser()
)
print(runnable.invoke("x raised to the third plus seven equals 12"))

"""
EQUATION: x^3 + 7 = 12

"""

 

---

 

Runnable.configurable_alternatives

종종 작업을 수행하는 여러 가지 다른 방법을 실험하거나 최종 사용자에게 노출시키고 싶을 수도 있습니다. 이 경험을 가능한 한 쉽게 만들기 위해 두 가지 방법을 정의했습니다.

첫째, configurable_fields 메서드를 통해 Runnable의 특정 필드를 구성할 수 있습니다.
둘째, configurable_alternatives 메서드를 사용하면 런타임 중에 특정 Runnable에 대한 alternative를 설정할 수 있습니다.

from langchain.prompts import PromptTemplate
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatOpenAI(temperature=0).configurable_fields(
    temperature=ConfigurableField(
        id="llm_temperature",
        name="LLM Temperature",
        description="The temperature of the LLM",
    )
)

model.invoke("pick a random number")

"""
AIMessage(content='7')
"""

model.with_config(configurable={"llm_temperature": 0.9}).invoke("pick a random number")

"""
AIMessage(content='34')
"""

 

이는 prompt에도 적용할 수 있습니다.

llm = ChatAnthropic(temperature=0)
prompt = PromptTemplate.from_template(
    "Tell me a joke about {topic}"
).configurable_alternatives(
    # This gives this field an id
    # When configuring the end runnable, we can then use this id to configure this field
    ConfigurableField(id="prompt"),
    # This sets a default_key.
    # If we specify this key, the default LLM (ChatAnthropic initialized above) will be used
    default_key="joke",
    # This adds a new option, with name `poem`
    poem=PromptTemplate.from_template("Write a short poem about {topic}"),
    # You can add more configuration options here
)
chain = prompt | llm

chain.invoke({"topic": "bears"})

"""
AIMessage(content=" Here's a silly joke about bears:\n\nWhat do you call a bear with no teeth?\nA gummy bear!")
"""

chain.with_config(configurable={"prompt": "poem"}).invoke({"topic": "bears"})

"""
AIMessage(content=' Here is a short poem about bears:\n\nThe bears awaken from their sleep\nAnd lumber out into the deep\nForests filled with trees so tall\nForaging for food before nightfall \nTheir furry coats and claws so sharp\nSniffing for berries and fish to nab\nLumbering about without a care\nThe mighty grizzly and black bear\nProud creatures, wild and free\nRuling their domain majestically\nWandering the woods they call their own\nBefore returning to their dens alone')
"""