[AI 서비스 개발] Langchain 대화 내용 기록 RunnableWithMessageHistory (2) Runnable, invoke

- Example where the wrapped Runnable takes a dictionary input:

https://python.langchain.com/v0.2/api_reference/core/runnables/langchain_core.runnables.history.RunnableWithMessageHistory.html

 

from typing import Optional

from langchain_community.chat_models import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory


prompt = ChatPromptTemplate.from_messages([ #1
    ("system", "You're an assistant who's good at {ability}"),
    MessagesPlaceholder(variable_name="history"), #2
    ("human", "{question}"),
])

chain = prompt | ChatAnthropic(model="claude-2") #3

chain_with_history = RunnableWithMessageHistory(
    chain,
    # Uses the get_by_session_id function defined in the example
    # above.
    get_by_session_id,
    input_messages_key="question",
    history_messages_key="history",
)

print(chain_with_history.invoke(  # noqa: T201
    {"ability": "math", "question": "What does cosine mean?"},
    config={"configurable": {"session_id": "foo"}}
))

# Uses the store defined in the example above.
print(store)  # noqa: T201

print(chain_with_history.invoke(  # noqa: T201
    {"ability": "math", "question": "What's its inverse"},
    config={"configurable": {"session_id": "foo"}}
))

print(store)  # noqa: T201

 

1. ChatPromptTemplate.from_messages

- 여러 메세지를 입력받아 ChatPromptTemplate객체를 반환한다.

- 메시지 구조: 역할(role)과 콘텐츠(content)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You're an assistant who's good at {ability}"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}"),
])

예)
inputs = {"ability": "swimming", "question": "접영하는 법 알려줘"}

# prompt.format(**inputs)
# **inputs는 ability="swimming", question="접영하는 법 알려줘"로 변환되어 함수에 전달된다.
# format(ability="swimming", question="접영하는 법 알려줘") 이렇게 언팩됨

 

2.  MessagesPlaceholder(variable_name="history")

- 대화의 맥락을 유지하기 위해 이전 대화 메시지들을 저장하고, 필요한 위치에 삽입하는데 사용

- variable_name: 자리 표시자로 사용할 변수의 이름을 정의

- history에 이전의 메시지 흐름이 포함된다

- 예시)

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory

# 메모리 생성 (대화 기록 저장소)
memory = ConversationBufferMemory(return_messages=True)

# ChatPromptTemplate 생성
chat_prompt = ChatPromptTemplate.from_messages([
    MessagesPlaceholder(variable_name="history"),  # 이전 대화 메시지 기록 삽입
    {"role": "user", "content": "What is the weather in {location}?"}
])

# 메모리에 저장된 대화 기록을 활용
inputs = {"history": memory.chat_memory.messages, "location": "New York"}

# 프롬프트 생성
formatted_prompt = chat_prompt.format(**inputs)
print(formatted_prompt)

 

3. chain = prompt | ChatAnthropic(model="claude-2")

- prompt의 결과물(템플릿이 아닌, input값을 넣어 생성된 최종 프롬프트)가 chatAnthropic모델로 전잘된다.

- | 연산자: 앞 단계의 출력값을 다음 단계의 입력값으로 전달 python 3.10이상에서 지원되는 기능

- chatAnthropic: ChatOpenAI 같은 언어모델

- ChatAnthropic( prompt )로 쓰지 않는 이유: 명시적으로 각 단계를 시작적으로 "연결"하기 위해서

- 프롬프트 생성 → 모델호출 흐름을 명확히 표현하기 위함

- Runnable 인터페이스를 구현

 

1). Runnable이란?

- 함수나 메서드를 실행할 수 있는 클래스나 함수 객체

- 입력 데이터를 받아 특정 동작을 수행하고 결과를 반환하는 구조

- chatAnthropic은 Runnable 인터페이스를 구현하여 invoke메서드를 통해 입력을 받아 처리하고 결과를 반환

- invoke: 입력 데이터를 받아 처리하고 결과를 반환하는 메서드

- Langchain에서 Runnable은 작업을 처리하는 표준 인터페이스로 이를 구현하거나 확장하는 객체들은 모두 invoke메서드를 사용할 수 있다.

from langchain.chat_models import ChatOpenAI
from langchain.llms import HuggingFaceHub
from langchain.schema.runnable import InputTransform # 입력 데이터를 전처리하거나 변환
from langchain.schema.runnable import OutputParser # JSON 응답을 python 객체로 변환
from langchain.schema.runnable import RunnableMap # 모델 여러개 동시에 처리
from langchain.schema.runnable import RunnableSequence # 여러 runnable 순차적 실행
from langchain.schema.runnable import RunnableLambda # 간단한 람다함수 실행
from langchain.memory import ConversationBufferMemory # 대화 기록 관리, 이전 대화 맥락 저장

* json과 python 

- json은 데이터 교환 형식(텍스트 기반)이고, python 딕셔너리는 python 언어의 데이터 타입(내부 자료 구조)

 

2) wrapper란?

- 다른 객체나 함수를 감싸 추가적인 기능을 제공하는 구조

- 원래의 함수나 객체가 하는 일을 변경하지 않으면서, 부가적인 동작을 추가하거나 제어할 수 있는 구조

- 원래 객체의 기능을 확장하거나 수정하지 않고, 기존 동작을 감싸서 확장, 외부에서 새로운 기능을 덧붙이는 방식

* 원래의 기능을 수정하거나 완전히 대체: 오버라이딩

 

4. RunnableWithMessageHistory

chain_with_history = RunnableWithMessageHistory(
    chain,
    # Uses the get_by_session_id function defined in the example
    # above.
    get_by_session_id,
    input_messages_key="question",
    history_messages_key="history",
)

1) RunnableWithMessageHistory

- 대화 히트로리를 관리하고 사용할 수 있는 runnable 클래스

- chain: 대화의 입력 데이터를 처리하고 결과를 반환

- get_by_session_id: 세션ID를 기반으로 대화 히스토리를 가져오는 함수

* RunnableWithMessageHistory은 invoke 호출 시 session_id를 전달 해줘야 함

- input_messages_key: 사용자 입력 데이터를 가리키는 키 이름

- history_messages_key: 대화 히스토리 데이터를 가리키는 키 이름, 이전 대화 기록이 체인에 전달됨

- 이전 대화의 문맥을 기반으로 새 입력 데이터를 처리