LangChain을 이용한 RAG - (5) 생성 편

2024.07.15 - [데이터 분석/LLM] - LangChain을 이용한 RAG - (4) 검색 편

Augment & Generate

Retriever를 통해 찾은 문서를 프롬프트에 증강(Augment)하고 LLM을 통해 답변을 생성(Generate)하는 과정이다. 이로써 RAG의 R, A, G가 모두 제 역할을 마치게 된다.

 

 

Chain을 이용한 생성

LangChain에 왜 Chain이 붙었는지에 대한 이유를 여기서 알 수 있게 된다.

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough
from langchain import hub

prompt = hub.pull("rlm/rag-prompt")

# Model
llm = ChatOpenAI(
    model='gpt-3.5-turbo-0125',
    temperature=0,
    max_tokens=500,
)

def format_docs(docs):
    return '\n\n'.join([d.page_content for d in docs])

# Chain
chain = (
    {'context': ensemble_retriever | format_docs, 'question': RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

query = "리셀은 암표상과 어떤 점이 유사한가요?"

# Run
response = chain.invoke(query)
response
"""
리셀은 암표상과 유사한 효과를 초래할 수 있으며, 창작자의 의도를 무시하고 상업적 이익만을 추구하는 측면이 있습니다. 그러나, 소비자의 선택권을 확대하고 시장 메커니즘을 통해 가격 안정화를 이끌어낼 수 있는 긍정적인 기능도 가지고 있습니다.
"""

LangChain hub로부터 프롬프트를 가져오고 (line 6)

사용할 LLM 선택 후 (line 9-13)

체인을 선언한 뒤 (line 19-24)

답변을 만들어낸다. (line 29)

 

 

체인 작동 방식

1. `query`는 `chain`의 `ensemble_retriever`로 전달되어 유사 문서를 검색
2. `ensemble_retriever`의 결과인 유사 문서들은 `format_docs` 함수를 거쳐 처리됨
3. 1, 2와 동시에 `query`는 `RunnablePassthrough()`를 통해 `question`으로 그대로 전달
4. `format_docs`의 결과와 `question`은 딕셔너리 형태로 `prompt`에 전달
5. 이 딕셔너리는 `prompt`를 거쳐 처리되고, 그 결과는 `llm`에 전달

6. `llm`을 통해 나온 결과는 `StrOutputParser()`를 거쳐 string 형태로 나오게 됨

 

여기서 이해가 잘 되지 않는 부분은 `|` 연산자를 통해 값들이 다음으로 계속 전달이 된다는 것이다. pandas에서는 or 연산자로 사용되고, 보통은 bitwise or로 사용되기 때문에 LangChain을 처음 봤을 때 많이 헷갈렸다. 공부해보니 `LangChain` 클래스 성질 때문에 가능하다는 것을 알게 되었다.

 

`|` 연산자

위와 같은 연산이 가능한 이유는 클래스 내에서 `__or__`과 `__ror__` 메서드가 정의되었기 때문이다. 예시를 통해 이해를 해보자.

def square_numbers(numbers):
    return [n ** 2 for n in numbers]


def sum_numbers(numbers):
    return sum(numbers)


numbers = [1, 2, 3, 4, 5]

리스트의 값들을 제곱하는 함수와 리스트의 값들을 합하는 함수가 있다. 두 함수와 숫자 리스트를 엮으려면 일반적으로 아래와 같이 코드를 작성해야 한다.

result = sum_numbers(square_numbers(numbers))

 

위 `langchain`처럼 `|` 연산자로 묶으면 아래와 같은 오류가 발생한다.

result = numbers | comp1 | comp2

`TypeError: unsupported operand type(s) for |: 'list' and 'function'`

 

`langchain`과 같이 작성하기 위해서는 함수를 특별한 클래스로 감싸주어야 한다.

class PipelineComponent:
    def __init__(self, func):
        self.func = func

    def __or__(self, other):
        def combined_func(input):
            return other.func(self.func(input))
        return PipelineComponent(combined_func)

    def __call__(self, input):
        return self.func(input)

    def __ror__(self, other):
        return self.func(other)

위와 같이 `__or__`과 `__ror__` 메서드를 정의한 후

 

comp1 = PipelineComponent(square_numbers)
comp2 = PipelineComponent(sum_numbers)

함수를 위 클래스로 감싸주면

 

result = numbers | comp1 | comp2

위 코드가 정상적으로 작동하게 되며 작동 방식은 다음과 같다.

1. `numbers | comp1`이 가장 먼저 작동하고 `__ror__`에 의해 `comp1.func(numbers)`가 작동한다. 즉 `square_numbers(numbers)` 결과가 나온다.

2. `square_numbers(numbers) | comp2`가 수행되며 `sum_numbers(square_numbers(numbers))`가 수행된다.

 

 

한편 아래와 같은 코드로도 가능하다.

pipe = comp1 | comp2
result = pipe(numbers)

 

1. `comp1`의 `__or__` 메서드가 호출되어 `PipelineComponent(combined_func)` 값이 `pipe`가 된다.

2. 여기서 `combined_func`은 `other.func(self.func(input))`으로, `other`은 `comp2`이므로 결국 다음과 같다. `sum_numbers(square_numbers(input))`

3. 따라서 `pipe(numbers)`는 `sum_numbers(square_numbers(numbers))`와 같다.

 

 

물론 `langchain`에서는 더 복잡하게 메서드를 정의했을 것이다. 이렇게 `|` 연산자로 묶는 이유는 가독성이 좋기 때문일 것이다. 함수 안에 함수를 넣고 하는 것과 비교하여 입력되는 값이 앞부터 순서대로 있다보니 어떤 순서로 흘러가는지 파악하기에는 확실히 보기 좋다.

 

 

stream 모드

토큰 단위로 답변이 생성되는 걸 바로바로 확인하고 싶다면 아래와 같이 코드 몇 줄만 수정/추가하면 된다.

llm = ChatOpenAI(
    model='gpt-3.5-turbo-0125',
    temperature=0,
    max_tokens=500,
    streaming=True,
)

# chain 구성 생략

for chunk in chain.stream(query):
    print(chunk, end='')

`llm`에 `streaming=True` 파라미터를 추가해주고, `chain`은 `invoke`가 아닌 `stream` 메서드를 호출하면 된다.

 

프롬프트 엔지니어링

앞서 허브에서 프롬프트를 가져왔다고 했는데, 프롬프트가 어떻게 생겼는지 확인해보자.

print(prompt.messages[0].prompt.template)

You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Use three sentences maximum and keep the answer concise.
Question: {question}
Context: {context}
Answer:

 

LangChain에서는 저렇게 프롬프트에 `{question}`, `{context}`를 해두고 딕셔너리를 통해 `question`과 `context` 값을 `prompt`에 넘겨주면 그 안에 값을 넣어준다.

 

그러면 내가 직접 프롬프트를 짜고 싶을 땐 어떻게 할까? `ChatPromptTemplate`를 이용하여 프롬프트를 구성할 수 있다.

from langchain.prompts import ChatPromptTemplate

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

Question: {question}
'''

prompt = ChatPromptTemplate.from_template(template)

 

 

LLM 선택

LangChain에서 지원하는 LLM은 공식 문서에서 확인할 수 있다. 각 LLM마다 문서가 있으니 해당 문서를 보고 `llm`을 갈아끼우기만 하면 된다.

 

문제는 AWS의 Bedrock의 경우 AWS 내에서 사용하는 것을 상정하고 작성되어 외부에서는 어떻게 사용할 수 있는지 써있지 않았다.

from langchain_aws import ChatBedrock

llm = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    model_kwargs=dict(temperature=0),
    # other params...
)

위와 같이 선언하면 된다는데, AWS 외부에서 Bedrock을 포함한 서비스를 사용하려면 aws access key가 필요하다. 하지만 LangChain 공식 문서에는 키를 어디에 어떻게 입력해야 하는지를 가르쳐 주지 않고 있으며, 실제로 키를 입력하지 않으면 `Error raised by bedrock service: Unable to locate credentials` 에러를 뱉는다.

 

구글 검색을 통해 알아본 결과, 아래처럼 `boto3`를 통해 bedrock-runtime client를 선언하고 llm에 넣어주면 된다. (bedrock client가 아님에 주의하자.)

from langchain_aws import ChatBedrock

client = boto3.client('bedrock-runtime', 
    region_name="us-east-1",
    aws_access_key_id='your_access_key',
    aws_secret_access_key='your_secret_key'
)
    
llm = ChatBedrock(
    model_id="anthropic.claude-3-5-sonnet-20240620-v1:0",
    model_kwargs={"temperature": 0.1},
    streaming=True,
    client=client,
)

 

 

참고자료

https://teddylee777.github.io/langchain/rag-tutorial/