LLM을 활용한 챗봇 체인구성 분석하기

2star_·2024년 11월 29일

Shelter_PJ

목록 보기

8/12

비상 대처 매뉴얼 및 대피소 안내 챗봇의 코드를 짜면서, 가장 이해가 안가는 부분은 체인을 구성할때와 이들의 호출구조였다. 하나씩 살펴보며 다시 이해하는 시간을 갖자.

코드 리뷰

체인 구조

chat_chain = (
    ModelInvocation()
    | ResponseProcessor()
)

ModelInvocation 클래스

class ModelInvocation(Runnable):
    def invoke(self, input, config=None):
        # 1단계: 프롬프트 템플릿을 사용하여 사용자 입력을 메시지로 포맷합니다.
        user_message = chat_template.format_messages(user_input=input)
        
        # 2단계: 메모리에서 대화 기록을 로드합니다.
        messages = memory.load_memory_variables({})["history"] + user_message
        
        # 3단계: 결합된 메시지로 모델을 호출합니다.
        response = model.invoke(messages)
        
        # 4단계: 현재 대화 컨텍스트를 메모리에 저장합니다.
        memory.save_context({"input": input}, {"output": response.content})
        
        # 5단계: 모델의 응답을 반환합니다.
        return response

체인 구조에서 제일 먼저 활용(호출)되는 ModelInvocation class를 살펴보자.

먼저 config=None 이건 기본값이다. config는 데이터 처리에 필요한 추가적인 설정 값들을 전달하는 역할을 한다. ex) 우선 처리순위, 모델 세부 설정, 디버그 정보 등등

순서대로 주석을 보면서 따라가다 보면, user_message는 어떤 형식으로 구성되어 있는지 궁금하다.

모델 호출 후의 response 객체 구조
타입: AIMessage
속성:
content: 모델이 생성한 텍스트 응답.
additional_kwargs: 모델이 함수를 호출하려는 경우 function_call 키를 포함하는 딕셔너리.

예시

response는 아래와 같은 구조를 할 수 있다.


AIMessage(
    content="",
    additional_kwargs={
        "function_call": {
            "name": "find_nearest_shelters",
            "arguments": '{"location": "사용자의 현재 위치"}'
        }
    }
)

ResponseProcessor 클래스

class ResponseProcessor(Runnable):
    def invoke(self, input, config=None):
        # 모델의 응답을 처리합니다.
        process_response(input)
        return input

process_response 함수

def process_response(response):
    if response.additional_kwargs.get("function_call"):
        # 함수 호출이 있으면 처리합니다.
        handle_function_call(response)
    else:
        # 그렇지 않으면 응답 내용을 출력합니다.
        print(f"챗봇: {response.content}")

모델의 functions?

코드

model = ChatOpenAI(
    model="gpt-4o",
    temperature=0.4,
    streaming=True,
    model_kwargs={"functions": functions}
)

functions를 설정한 경우:
입력:
사용자가 "내 주변 대피소 알려줘."라고 요청.

response = {
    "content": "대피소 정보를 검색 중입니다.",
    "additional_kwargs": {
        "function_call": {
            "name": "find_nearest_shelters",
            "arguments": '{"address": "서울특별시 중구"}'
        }
    }
}

functions를 설정하지 않은 경우:
입력:
동일한 요청: "내 주변 대피소 알려줘."

response = {
    "content": "죄송합니다. 주변 대피소 정보를 제공할 수 없습니다.",
    "additional_kwargs": {}
}

이렇게 function calling을 위해 함수를 지정하고 모델 정의부분에서 호출하면, 응답 구조에 additional_kwargs가 모델의 판단하에 추가될 수 있다.

함수 호출 요청이 필요한 상황을 어떻게 판단을 해?

모델이 함수 호출 요청이 필요한 상황을 판단하는 과정은 **프롬프트(지침)**와 모델의 논리적 추론 능력에 따라 결정됩니다.

함수 호출 처리

그럼 위와같이 함수가 호출되어야 한다고 모델이 판단하면 어떤 로직으로 흘러가는지 알아보자

def handle_function_call(response):
    function_call = response.additional_kwargs.get("function_call")
    if function_call:
        # 함수 이름과 인수를 추출합니다.
        function_name = function_call.get("name")
        function_args = function_call.get("arguments")
        
        # 함수 인수를 파싱합니다.
        if isinstance(function_args, str):
            function_args = json.loads(function_args)
        
        # 해당 함수를 호출합니다.
        if function_name == "find_nearest_shelters":
            function_result = find_nearest_shelters(**function_args)
            function_message = FunctionMessage(
                name=function_name,
                content=function_result
            )
            
            # 함수 호출과 결과를 메모리에 저장합니다.
            memory.save_context(
                {"input": function_message.content},
                {"output": function_result}
            )
            
            # 함수 결과를 사용하여 최종 응답을 생성합니다.
            final_response = model.invoke(memory.load_memory_variables({})["history"])
            print(f"챗봇: {final_response.content}")
            
            # 최종 응답을 메모리에 저장합니다.
            memory.save_context(
                {"input": function_result},
                {"output": final_response.content}
            )
        else:
            print(f"알 수 없는 함수 호출: {function_name}")
    else:
        print(f"챗봇: {response.content}")

response 객체의 변환

함수 호출 전:
response는 additional_kwargs에 function_call을 포함하고 있습니다.
함수 호출 후:
함수의 출력을 나타내는 새로운 FunctionMessage가 생성됩니다.
대화 기록이 함수 결과로 업데이트됩니다.
함수의 출력을 기반으로 모델을 다시 호출하여 최종 응답을 생성합니다.
최종 response는 함수 결과를 고려한 AI의 응답을 포함합니다.
예시
앞서의 예시를 계속해서, 함수 호출을 처리한 후:

함수 결과: find_nearest_shelters 함수가 "가장 가까운 대피소는 서울역 대피소입니다."를 반환했다고 합시다.
최종 응답: 모델은 "가장 가까운 대피소는 서울역 대피소입니다. 안전하게 대피하시길 바랍니다."와 같은 응답을 생성합니다.

main 함수

def main():
    # 1단계: 초기 AI 메시지를 출력합니다.
    initial_messages = chat_template.format_messages(user_input="")
    response = model.invoke(initial_messages)
    process_response(response)
    
    while True:
        # 2단계: 사용자 입력을 받습니다.
        user_input = get_user_input()
        
        # 종료 조건.
        if user_input.lower() in ["종료", "exit", "quit"]:
            print("대화를 종료합니다.")
            break
        
        # 3단계: 사용자 입력으로 채팅 체인을 호출합니다.
        response = chat_chain.invoke(user_input)