[langchain] Message 다루기

 

 

0. 개요

메세지는 채팅모델의 입력/출력 입니다.

메세지는 내용(content) 와 역할(role)로 구성되어 있습니다.

 

 

메세지를 다루는 세가지 방법

1. Trim

2. Filter

3. 동일 유형의 메세지 병합

 

 

메세지는 모델과의 대화를 저장하고 추적하는데 주로 사용

 

---

1. Trim 기법

1.1. trim_messages 란

이전 대화 내용을 적절히 잘라서 LLM이 처리할 수 있도록 조정하는 역할

llm은 메세지의 크기가 제한되어 있기 때문에, 토큰의 수를 다듬어야 한다.

 

trim 기법은 크게 두가지가 있다.

• 토큰수 기반 Trimming (Trimming based on token count)
• 메시지 수를 기반 Trimming (Trimming based on message count)

 

---

1.2. Trimming 예시 코드

각 모델마다 메시지에서 사용하는 토큰 수가 다를 수 있기 때문에, 모델에 맞는 토큰 계산기(Token Counter)를 사용하는 것이 중요

trim_messages 유틸리티를 사용

 

파라미터 설명

• token_counter=ChatOpenAI(model="gpt-4o") : 모델의 토큰의 숫자를 기반으로 메세지의 길이를 제한합니다.
• token_counter=len : 메시지의 개수를 기준
• max_tokens: 최대 토근 개수
• allow_partial=True : 메시지 내용이 길어지면, 메시지 일부를 자를 수 있습니다.
• include_system=True : 시스템 메세지를 False로 하면 제거할 수 있습니다.
• strategy="first" : 가장 처음부터 주어진 토큰 수만큼 메시지를 유지, "last"로 설정할 수도 있음

 

from langchain_core.message import trim_messages 를 사용

from langchain_core.messages import (
    AIMessage,  # AI의 응답 메시지
    HumanMessage,  # 사람이 보낸 메시지
    SystemMessage,  # 시스템의 지시 메시지
    ToolMessage,  # 도구와 관련된 메시지
    trim_messages,  # 메시지 다듬기 함수
)
from langchain_openai import ChatOpenAI  # OpenAI 채팅 모델

# 주어진 메시지들
messages = [
    SystemMessage("당신은 훌륭한 보조자입니다. 항상 농담으로 응답합니다."),
    HumanMessage("왜 이것을 LangChain이라고 부를까요?"),
    AIMessage("음... 'WordRope'나 'SentenceString'이 별로 어울리지 않아서 그런 거 아닐까요!"),
    HumanMessage("해리슨은 누구를 쫓고 있나요?"),
    AIMessage("글쎄요, 아마 사무실의 마지막 커피를 쫓고 있는 걸까요!"),
    HumanMessage("말 못하는 앵무새를 뭐라고 부르죠?"),
]

# 메시지 다듬기
trim_messages(
    messages,
    strategy="last",  # 최근 메시지 유지
    token_counter=ChatOpenAI(model="gpt-4o"),  # 토큰 계산을 위한 모델
    max_tokens=45,  # 최대 토큰 수 45개
    start_on="human",  # 사람이 보낸 메시지로 시작하도록 설정
    end_on=("human", "tool"),  # 사람이 보낸 메시지 또는 도구 메시지로 끝나도록 설정
    include_system=True,  # 시스템 메시지를 유지
    allow_partial=False,  # 부분적으로 자르지 않음
)

 

---

1.3. 사용자 정의 token_counter

생략: 상세 설명 링크

How to trim messages | 🦜️🔗 LangChain

 

---

1.4. 체인(Chain)에 반영

from langchain_core.message import trim_messages

llm = ChatOpenAI(model="gpt-4o")

trimmer = trim_messages(
    token_counter=llm,
    strategy="last",
    max_tokens=45,
    start_on="human",
    end_on=("human", "tool"),
    include_system=True,
)

chain = trimmer | llm
chain.invoke(messages)

 

 

---

2. Filter

2.1. Filter 개념

이 글에서는 메시지를 필터링하는 방법

메시지를 추적할 때, 특정 메시지들만 선별

filter_messages 유틸리티를 사용

 

---

2.2. Filter 예시 코드

from langchain_core.message import filter_messages를 사용

 

from langchain_core.messages import (
    AIMessage,  # AI의 응답 메시지
    HumanMessage,  # 사람이 보낸 메시지
    SystemMessage,  # 시스템의 지시 메시지
    filter_messages,  # 메시지 필터링 함수
)

# 예시 메시지 목록
messages = [
    SystemMessage("당신은 훌륭한 보조자입니다.", id="1"),  # 시스템 메시지
    HumanMessage("예시 입력", id="2", name="example_user"),  # 사람이 보낸 예시 메시지
    AIMessage("예시 출력", id="3", name="example_assistant"),  # AI의 예시 응답
    HumanMessage("실제 입력", id="4", name="bob"),  # 사람이 보낸 실제 메시지
    AIMessage("실제 출력", id="5", name="alice"),  # AI의 실제 응답
]

# HumanMessage만 포함하여 필터링
filter_messages(messages, include_types="human")

# 특정 이름을 가진 메시지를 제외하여 필터링
filter_messages(messages, exclude_names=["example_user", "example_assistant"])

# HumanMessage와 AIMessage만 포함하고, id가 "3"인 메시지를 제외하여 필터링
filter_messages(messages, include_types=[HumanMessage, AIMessage], exclude_ids=["3"])

 

---

2.4. 체인(Chain)에 반영

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=0)

filter_ = filter_messages(exclude_names=["example_user", "example_assistant"])
chain = filter_ | llm
chain.invoke(messages)

 

---

3. 동일 유형의 메세지 병합

3.1. 예시 코드

from langchain_core.message import merge_message_runs 사용하면 동일한 유형의 연속된 메시지를 쉽게 병합할 수 있습니다.

from langchain_core.messages import (
    AIMessage,
    HumanMessage,
    SystemMessage,
    merge_message_runs,
)

messages = [
    SystemMessage("you're a good assistant."),
    SystemMessage("you always respond with a joke."),
    HumanMessage([{"type": "text", "text": "i wonder why it's called langchain"}]),
    HumanMessage("and who is harrison chasing anyways"),
    AIMessage(
        'Well, I guess they thought "WordRope" and "SentenceString" just didn\'t have the same ring to it!'
    ),
    AIMessage("Why, he's probably chasing after the last cup of coffee in the office!"),
]

merged = merge_message_runs(messages)
print("\n\n".join([repr(x) for x in merged]))

 

---

3.2. 체인(Chain)에 반영

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=0)
merger = merge_message_runs()

chain = merger | llm
chain.invoke(messages)

 

---

4. ChatMessageHistory과 함께 사용

그동안의 메세지 히스토리와 함께 사용할 수 있다.

• RunnableWithMessageHistory: 대화 기록과 함께 실행할 수 있는 체인을 관리하는 클래스. 대화 히스토리와 함께 모델 호출을 처리할 수 있도록 도와줌.
• InMemoryChatMessageHistory: 메모리에서 대화 기록을 관리하는 클래스. 대화 히스토리를 추적하고, 이를 기반으로 모델에게 전달할 수 있도록 함.

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# 기존 메시지에서 마지막 메시지를 제외한 대화 기록을 메모리에 저장
chat_history = InMemoryChatMessageHistory(messages=messages[:-1])

# 세션 ID에 따라 대화 기록을 반환하는 함수
def dummy_get_session_history(session_id):
    if session_id != "1":
        return InMemoryChatMessageHistory() # 세션 ID가 "1"이 아닌 경우 빈 대화 기록 반환
    else:
    	return chat_history # 세션 ID가 "1"인 경우 기존 대화 기록 반환


llm = ChatOpenAI(model="gpt-4o")

trimmer = trim_messages(
    max_tokens=45,  # 최대 45개의 토큰까지만 메시지를 유지
    strategy="last",  # 마지막 메시지부터 유지하는 전략
    token_counter=llm,  # LLM 모델을 기반으로 토큰 수 계산
    include_system=True,  # 시스템 메시지를 포함하도록 설정
    start_on="human",  # 대화가 HumanMessage로 시작되도록 설정
)

chain = trimmer | llm

# 대화 기록을 포함한 체인 생성, 세션별로 기록 관리
chain_with_history = RunnableWithMessageHistory(chain, dummy_get_session_history)

# 체인을 실행하여 모델에게 응답을 요청
chain_with_history.invoke(
    [HumanMessage("what do you call a speechless parrot")],
    config={"configurable": {"session_id": "1"}},
)

 

 

5. 최종 응용

1. filter_: 먼저 messages에서 특정 사용자의 메시지를 필터링합니다.

2. trimmer: 필터링된 메시지를 받아, 토큰 수가 45개 이하가 되도록 메시지를 다듬습니다.

3. prompt: 다듬어진 메시지에서 질문을 추출해 PromptTemplate을 사용해 최종 프롬프트를 생성

4. Retriever: 추가 정보 검색

4. llm: 언어모델

5. InMemoryChatMessageHistory: 기존 대화 기록을 메모리에 저장

6. RunnableWithMessageHistory: 체인과 대화 기록을 연결하여, 각 세션별 대화 흐름을 관리하고 유지합니다.

from langchain import PromptTemplate
from langchain_core.messages import filter_messages, trim_messages, HumanMessage
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables import Runnable
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_anthropic import ChatAnthropic
from langchain.retrievers import SimpleRetriever

# 유저의 입력 메시지 목록
messages = [
    HumanMessage(content="오늘 날씨가 어때?", name="example_user"),
    HumanMessage(content="내일은 어떤가요?", name="example_user"),
    HumanMessage(content="모레는 비가 오나요?", name="another_user"),
]

# 1. InMemoryChatMessageHistory 생성 (대화 기록을 메모리에 저장)
chat_history = InMemoryChatMessageHistory(messages=messages)

# 2. 필터링 단계 (특정 사용자의 메시지를 제외하는 필터링)
filter_ = filter_messages(exclude_names=["example_user"])

# 3. 메시지 다듬기 단계 (trim 적용)
trimmer = trim_messages(
    max_tokens=50,  # 최대 50개의 토큰만 유지
    strategy="last"  # 최근 메시지부터 유지
)

# 4. PromptTemplate 정의
template = "유저의 질문: {question}. 시스템은 {system_action}을 수행해야 합니다."
prompt_template = PromptTemplate(input_variables=["question", "system_action"], template=template)

# 프롬프트를 생성하는 Runnable 정의
class PromptRunnable(Runnable):
    def invoke(self, trimmed_messages):
        question = trimmed_messages[0].content if len(trimmed_messages) > 0 else "질문이 없습니다."
        return prompt_template.format(question=question, system_action="날씨 정보를 확인")

prompt_runnable = PromptRunnable()

# 5. Retriever 생성 (예시로 SimpleRetriever 사용)
class SimpleRetriever(Runnable):
    def invoke(self, prompt):
        # 예시로 외부 정보를 검색하는 로직 (간단히 프롬프트에 추가 정보 결합)
        additional_info = "현재 날씨 정보: 맑음, 기온: 25도"
        return f"{prompt}\n추가 정보: {additional_info}"

retriever = SimpleRetriever()

# 6. 모델 생성
llm = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=0)

# 7. RunnableWithMessageHistory로 대화 기록 관리
def dummy_get_session_history(session_id):
    # 세션 ID가 "1"인 경우 기존 대화 기록 반환
    if session_id == "1":
        return chat_history
    # 세션 ID가 다르면 빈 대화 기록 반환
    return InMemoryChatMessageHistory()

# 체인 연결: 필터링 -> 다듬기 -> 프롬프트 생성 -> 정보 검색(Retriever) -> LLM 호출
chain = filter_ | trimmer | prompt_runnable | retriever | llm

# RunnableWithMessageHistory로 체인과 대화 기록을 연결
chain_with_history = RunnableWithMessageHistory(chain, dummy_get_session_history)

# 체인을 실행하여 메시지를 처리, 새로운 메시지 전달
response = chain_with_history.invoke(
    [HumanMessage("말 못하는 앵무새를 뭐라고 부르죠?")],
    config={"configurable": {"session_id": "1"}}  # 세션 ID를 통해 대화 기록 관리
)

print(response)