reading LangChain docs (2)

 

 

 (1)편에서 랭체인의 기본 예시를 실행시켜 봤습니다. 그리고 LCEL의 개념을 살펴보고 체인 형태로 파이프라인을 만드는 것을 배웠습니다. 오늘은 이어서 LCEL관련해서 문서를 계속 볼 계획인데 Interface와 How to 까지 살펴보려 합니다.

 

그전에 기존에 우리가 openai에서 제공하는 코드로 api를 호출할 때와 LangChain을 사용했을 때를 비교하는 페이지를 봤습니다.

아래와 같은 코드로 기본적인 질의응답 프로그램을 활용합니다.

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


prompt = ChatPromptTemplate.from_template("Tell me a short joke about {topic}")
model = ChatOpenAI(model="gpt-3.5-turbo")
output_parser = StrOutputParser()

chain = prompt | model | output_parser

 

 

---

Why user LCEL

invoke: 체인을 호출해서 실행시키는 부분이 chain을 통해서 간결하게 표현되는 것을 확인할 수 있습니다.

 

stream: 스트리밍 및 응답을 추출할 때 명시적으로 처리해야 하지만 LCEL은 훨씬 추상화되었습니다. 

 

 batch: 배치단위로 실행할 때도 훨씬 간결하게 작업을 실행시킬 수 있습니다.

 

 

 

---

Interface

 "Runnable" 이라는 프로토콜을 구현해 사용자 정의 체인을 구현하기 쉽게 도와줍니다. Runnable은 표준 인터페이스로써 체인을 정의하고 호출하기 쉽게 해줍니다. 표준 인터페이스는 아래와 같습니다.

 

  - stream: 응답 청크를 stream으로 돌려줍니다.

  - invoke: 체인을 호출합니다.

  - batch: 입력 리스트에 대한 체인을 호출합니다.

   비동기적인 방법들로는 astream,  ainvoke, abatch, astream_log 가 있습니다. 비동기 처리를 한다는 것은 프로그램의 작업을 동시에 여러 개를 처리하도록 하는 것입니다. 예를 들어 파이썬 함수 정의 앞에 async를 붙이면 되죠.

 

 아래의 표는 질의응답시 입력과 출력의 형식을 보여줍니다. Prompt의 경우 dictionary형태로 받아서 PromptValue형태로 출력해주고, LLM의 경우 string 하나를 받거나 message 또는 PromptValue를 받아서 string형태로 출력합니다.

 

PromptTemplate + ChatModel 로 체인을 만드는 코드는 아래와 같습니다.

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

model = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
chain = prompt | model

 

 

Input Schema

input Schema는 프로토콜 Runnable에 의해 들어온 입력데이터의 표현입니다. .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'}}}

 

 모델 역시 input_schema.schema()를 통해 확인이 가능하다. 이하 생략.

model.input_schema.schema()

 

 

Output Schema

출력에서도 ouput_schema.schema()를 사용해 스키마를 확인할 수 있다. 
사용자가 질의하는 부분인 HumanMessage, 응답메세지인 AIMessage, 입력메세지의 첫번째로 들어가는 SystemMessage, function을 불러올 때 사용하는 FunctionMessage, chatbot이나 multi-user 시스템에서 주로 사용되는 ChatMessage 의 스키마를 확인 가능합니다.

 

Stream

  챗봇 모델에 입력을 스트리밍하고 실시간으로 응답을 스트리밍합니다.

for s in chain.stream({"topic": "bears"}):
    print(s.content, end="", flush=True)
    
'''Why don't bears wear shoes?

Because they already have bear feet!'''

 

Invoke

  질의에 대해 chain을 호출해 응답을 받고 결과값을 출력합니다.

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

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

 

Batch

 한번에 여러개의 입력을 배치로 실행시켜 결과값을 출력합니다.

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

'''
[AIMessage(content="Why don't bears wear shoes?\n\nBecause they have bear feet!"),
 AIMessage(content="Sure, here's a cat joke for you:\n\nWhy don't cats play poker in the wild?\n\nToo many cheetahs!")]
'''

 

이어서 async처리를 하는 것과 parallel 처리를 하는 것에 대한 설명은 너무 길어져서 생략합니다.

 

 

---

How to

RunnableParallel: Manipulating data

 

  - 입출력 데이터 조작하기

 

 아래의 코드는 context와 question을 입력으로 하는 방식입니다. context의 경우 벡터스토어로 FAISS를 활용하는데 context를 임베딩하여 FAISS에 저장하고 question에 대해 유사도를 계산해서 유사한 context를 검색해 찾고 이를 활용해서 question에 대한 대답을 출력합니다.  

 

from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

vectorstore = FAISS.from_texts(
    ["harrison worked at kensho"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
template = """Answer the question based only on the following context:
{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()

retrieval_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

retrieval_chain.invoke("where did harrison work?")

# 'Harrison worked at Kensho.'

 

 

  - Using itemgetter as shorthand

  

  아래의 코드는 itemgetter를 활용해서 RunnalbleMap에서 데이터를 추출할 수 있음을 보여줍니다.

from operator import itemgetter

from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

vectorstore = FAISS.from_texts(
    ["harrison worked at kensho"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()

template = """Answer the question based only on the following context:
{context}

Question: {question}

Answer in the following language: {language}
"""
prompt = ChatPromptTemplate.from_template(template)

chain = (
    {
        "context": itemgetter("question") | retriever,
        "question": itemgetter("question"),
        "language": itemgetter("language"),
    }
    | prompt
    | model
    | StrOutputParser()
)

chain.invoke({"question": "where did harrison work", "language": "italian"})

#'Harrison ha lavorato a Kensho.'

 

 

 

  - Parallelize steps. 병렬처리하기.

 

  아래의 코드는 RunnableParallel 을 사용해서 병렬로 Runnable을 조작합니다. joke_chain과 poem_chain이 눈에 띕니다. ChatPromptTemplate.from_template를 사용하는 것은 익숙했지만 여기서는 각 체인을 문자열을 사용해서 프롬프트를 만듭니다. 그리고 | model 을 통해서 생성된 프롬프트를 model과 파이프라인 연결을 짓습니다.

  쉽게 말하면 지금 단어 곰을 가지고 유머를 만들라는 chain이 하나 뿐이었는데 짧은 시도 하나 만들어달라는 chain이 하나 더 추가된 겁니다. 그리고 단어 곰을 한번만 invoke해도 두 개의 chain에 병렬로 실행되어 결과를 한번에 받아볼 수 있게 됩니다. 

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 forest's embrace, bear roams with might,\nNature's guardian, a majestic sight.")}
'''

 

  - joke_chain, poem_chain 각각 실행했을 때와 마지막에 둘을 동시에 실행했을 때 걸리는 속도를 비교해 봅니다. 한번에 두개의 chain의 작업을 수행하는 것을 확인했습니다.

 

 

 

---

RunnalblePassthrough: Passing data through

 

  - 입력을 변경하지 않거나 추가 키를 더해서 전달할 수 있습니다. 일반적으로 RunnableParallel과 함께 map의 새 key에 데이터를 할당합니다. RunablePassthrough()가 자체적으로 호출되면 입력을 받아 전달하기만 하면 됩니다. assign과 함께 RunablePassthrough called(RunablePasthrough.assign(...)) 입력을 받고 assign 함수에 전달된 추가 인수를 추가합니다. 

 

 아래의 코드는 RunnableParallel을 상요해 여러 하위작업을 병렬로 실행하고 하위 작업들의 결과를 반환하는 코드입니다. RunnableParallel에 포함된 하위 작업은 passed, extra, modified 입니다. 

  passed는 RunnablePassthrough()는 입력값을 그대로 전달만 하는 역할입니다.

  extra는 RunnablePassthough.assign()을 사용해 mult라는 함수를 추가로 할당합니다.

  modified는 입력받은 num의 값을 1증가시키는 함수로 바꿉니다.

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}

 

 

 

 Retrievale Example

 

  RunnableMap에서 RunnablePassingthough를 사용하는 예시 코드입니다.

  기본적인 content, question을 입력해 검색을 활용한 질의응답 코드인데, question에 인자를 보낼 때 RunnablePassthough()로 값그대로 전달합니다.

 

from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

vectorstore = FAISS.from_texts(
    ["harrison worked at kensho"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
template = """Answer the question based only on the following context:
{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()

retrieval_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

retrieval_chain.invoke("where did harrison work?")

# 'Harrison worked at Kensho.'

 

 

---

 

RunnableLambda: Run CustomFunctions

  파이프라인에서 특수한 함수를 사용할 수 있습니다. 많이 사용될 것으로 보이므로 여기까지는 꼭 보고 넘어가야 할 듯 합니다. 안보면 (보더라도) 나중에 다시 해당 페이지에 돌아올 것만 같은 느낌이 듭니다.

  다수의 인자를 받는 함수가 있다면 wrapper를 써야합니다. wrapper는 하나의 입력을 받고, 다수의 인자로 포장을 푸는 역할을 합니다.

  아래의 코드를 살펴봅시다. RunnableLambda를 import해온 것을 확인했습니다. 그리고 함수 3개를 선언했는데요. 전부 하나의 출력값을 뱉고 있구요. 입력은 하나의 인자를 받거나, 두개의 인자를 받거나, 두개의 값이 포함된 하나의 인자를 받는 경우입니다. 간단한 함수.

  

  ChatPromptTemplate을 사용하고 있고요. 템플릿 내용은 "what is {a} + {b}" 가 됩니다.

 

  chain1: 첫번째 체인입니다. a, b 변수를 받고 모델에 전달하게 됩니다.

  chain: 두번째 체인입니다. prompt에 들어가기 전에 더 복잡한 작업을 수행합니다. 템플릿의 a는 itemgetter("foo")를 사용해서 invoke할 때 들어올 딕셔너리의 foo라는 필드에서 값('bar')을 추출한 다음 RunnableLambda(length_function)을 수행합니다. 이어서 템플릿 "b"의 경우 invoke에서 마찬가지로 "foo", "bar" 필드에 대한 값을 받아온 다음에 각각 2,3 번째 함수에 적용한 결과를 각각 "a", "b"로 인식해서 prompt에 있는 {a}, {b}에 입력합니다. 그리고 prompt를 모델에 넘긴 뒤 모델은 이를 실행합니다.

  

  요약하면 prompt에 매개변수값을 invoke할 때 바로 지정해서 넣어줄 수도 있지만, 좀 변형해서 넣어주고 싶거나 할 때 함수를 한번 태워서 나온 반환값을 prompt 매개변수에 넣어주는 거라고 볼 수 있습니다.

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
)

 

 

  itemgetter가 받는 필드에 대한 값을 딕셔너리로 넣어줍니다.

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

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

 

  chain1은 프롬프트를 바로 모델로 넘기는데 한번 해볼까요? prompt의 a, b 매개변수를 invoke할 때 직접 지정해주었습니다. 말이 되게끔 bar + tender 가 뭔지 물어보는 프롬프트를 날렸고 아래와 같은 결과를 얻었네요.

chain1.invoke({"a":"bar", "b":"tender"})

'''
AIMessage(content='A bartender is a person who prepares and serves alcoholic beverages in a bar, pub, or restaurant. They are responsible for taking orders, mixing and serving drinks, and interacting with customers. Bartenders often have knowledge of different types of alcohol, cocktail recipes, and customer service skills.')
'''

 

 

 

Accepting a Runnable Config

  Runnable lambdas는 RunnableConfig를 옵션으로 사용할 수 있습니다. callback이나 tag 그리고 기타 설정 정보를 중첩된 실행에 전달 가능합니다.

  아래의 코드는 위에서 chain을 만들고 chain.invoke()하는 방식과는 다릅니다. 더 복잡한 구조로 되어있죠. 보통 체인만들고 체인.invoke() 하는데 여긴 좀 다릅니다.

  크게 parse_or_fix 함수와 RunnableLambda(parse_or_fix).invoke()로 나누어 지는데요. 먼저 parse_or_fix는 JSON 형식의 문자열을 파싱하거나 수정하는 역할을 합니다. 인자로는 text와 config를 받고 있네요. text는 JSON 문자열로 파싱 또는 수정하려는 텍스트이고, config는 RunnableConfig 객체로 실행 구성 정보를 포함할 수 있습니다.

  fixing_chain은 JSON 파싱에 실패한 경우 텍스트를 수정하기 위한 작업 체인을 정의합니다. 오류가 있는 텍스트와 오류 메세지를 입력으로 받습니다.

  for 루프에서 JSON 파싱을 시도합니다. 성공하면 JSON객체를 반환하고, 실패하면 fixing_chain을 이용하여 텍스트를 수정하고 3번까지 시도합니다.

 

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"

 

 

  with get_openai_callback() 은 openAI의 콜백을 가져옵니다. callback은 api 가격과 토큰이 얼만지 자세히 알려주니 아주 죻죠. with 문으로 묶어주고 객체 cb를 print하기만 하면 끝입니다.

  RunnableLambda를 사용해서 parse_or_fix 함수를 실행합니다. 첫번째 인자로는 실행할 함수를 전달하고 두번째 인자로는 실행 구성 정보를 포함하는 딕셔너리를 전달합니다. invoke 메서드로 함수를 실행하고 output변수에 저장합니다. 

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)

 

 

  비영어는 토큰수가 더 많이 소요되고 한글의 경우 한글자에서도 토큰이 쪼개질 수 있어 토큰이 많이 듭니다. 영어로 테스트를 하는게 절약하는데 도움이 될 것 같습니다.

 

---

이밖에 다른 Runnable 기능들)

  

  RunnableBranch: Dynamically route logic based on input

  -> 입력값을 기반으로 로직을 동적으로 라우팅합니다. 이를 통해 입력값에 따라 다른 Runnable 또는 작업을 실행할 수 있습니다.

 

  Bind runtime args

  -> 런타임 시에 인수를 바인딩합니다. 동적으로 인수를 설정하고 작업을 실행할 수 있습니다.

 

  Configure chain internals at runtime

  -> 런타임에서 작업 체인 내부를 구성합니다. 작업 체인의 동작을 변경하거나 조절하는데 사용합니다.

 

  Create a runnable with the `@chain` decorator

  -> 데코레이터를 사용하여 Runnable을 생성하고 커스터마이징합니다.

 

  Add fall backs

  -> 예외 처리나 에러 핸들링을 위한 Fallback 작업을 추가하는데 사용합니다.

 

  Stream custom generator functions

  -> 커스텀 생성기 함수를 스트림 처리에 사용합니다.

 

  Inspect your runnables

  -> Runnable 상태와 동작을 검사합니다.

 

  Add message history (memory)

  -> 이전 메세지 기록을 유지하고 검색할 수 있도록 메모리 기능을 추가합니다.