LangGraph 공식문서를 번역한 내용입니다. 필요한 경우 부연 설명을 추가하였고 이해하기 쉽게 예제를 일부 변경하였습니다. 문제가 되면 삭제하겠습니다.
https://langchain-ai.github.io/langgraph/how-tos/memory/manage-conversation-history/
영속성의 가장 일반적인 사용 사례 중 하나는 대화 기록을 추적하는 것이다. 즉, 대화를 쉽게 이어갈 수 있게 한다. 그러나 대화가 길어지면 대화 기록이 쌓여서 컨텍스트 공간을 점점 더 많이 차지하게 된다. 간혹 의도하지 않게 동작하며, LLM에 대한 호출이 더 비싸지고 길어지며, 오류가 발생할 수도 있다. 이런 일이 발생하지 않도록 하려면 대화 기록을 적절하게 관리해야 한다.
참고: 이 가이드는 LangGraph에서 이를 어떻게 처리할 수 있는지에 대해 설명한다. LangGraph에서는 완전히 커스트마이징할 수 있다. 만약 즉시 사용할 수 있는 방법을 원한다면, LangChain에서 제공하는 기능을 살펴볼 수 있다.
준비
필요한 패키지를 설치하자.
pip install langgraph langchain_openaiagent 생성
간단한 ReAct 스타일 에이전트를 만들자.
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.prebuilt import ToolNode
load_dotenv()
memory = MemorySaver()
@tool
def search(query: str):
"""Call to surf the web."""
return "It's sunny in San Francisco, but you better look out if you're a Gemini 😈."
tools = [search]
tool_node = ToolNode(tools)
model = ChatOpenAI(model_name="gpt-4o-mini")
bound_model = model.bind_tools(tools)
def should_continue(state: MessagesState):
"""Return the next node to execute."""
last_message = state["messages"][-1]
# 만일 함수 호출이 없다면, 끝낸다.
if not last_message.tool_calls:
return END
# 만일 함수 호출이 있다면, 계속한다.
return "action"
# 모델을 호출하는 함수를 정의한다.
def call_model(state: MessagesState):
response = bound_model.invoke(state["messages"])
# 리스트를 반환한다. 기존 리스트에 추가될 것이기 때문이다.
return {"messages": response}
workflow = StateGraph(MessagesState)
# 두 개의 노드가 서로를 순환하도록 정의한다.
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
# 진입점을 'agent'로 설정한다.
# 이것은 이 노드가 처음으로 호출되는 것을 의미한다.
workflow.add_edge(START, "agent")
# 조건부 엣지를 추가한다.
workflow.add_conditional_edges(
# 우선, 시작 노드를 정의한다. 'agent'를 사용한다.
# 이것은 'agent' 노드가 호출된 후에 호출되는 엣지를 의미한다.
"agent",
# 다음으로, 다음에 호출될 노드를 결정할 함수를 전달한다.
should_continue,
# 다음으로, 경로 맵을 전달한다. 이 엣지가 갈 수 있는 모든 노드들이다.
["action", END],
)
# tools에서 agent로의 일반 엣지를 추가한다.
# 이것은 tools가 호출된 후에 agent 노드가 호출된다는 것을 의미한다.
workflow.add_edge("action", "agent")
# 마지막으로, 컴파일한다!
# 이것은 LangChain Runnable로 컴파일된다.
# 이것은 다른 Runnable처럼 사용할 수 있다는 것을 의미한다.
app = workflow.compile(checkpointer=memory)from langchain_core.messages import HumanMessage
config = {"configurable": {"thread_id": "2"}}
input_message = HumanMessage(content="안녕! 내 이름은 홍길동이야")
for event in app.stream({"messages": [input_message]}, config, stream_mode="values"):
event["messages"][-1].pretty_print()
input_message = HumanMessage(content="내 이름이 뭐야?")
for event in app.stream({"messages": [input_message]}, config, stream_mode="values"):
event["messages"][-1].pretty_print()================================ Human Message =================================
안녕! 내 이름은 홍길동이야
================================== Ai Message ==================================
안녕하세요, 홍길동님! 어떻게 도와드릴까요?
================================ Human Message =================================
내 이름이 뭐야?
================================== Ai Message ==================================
당신의 이름은 홍길동입니다!메시지 필터링
대화 기록이 지나치게 커지는 것을 방지하기 위한 간단한 방법은 LLM에 전달되기 전에 메시지 목록을 필터링하는 것이다. 이는 두 가지 부분으로 나눌 수 있다. 메시지를 필터링하는 함수 정의와 이를 그래프에 추가하는 것이다. 아래 예시에서는 매우 간단한 filter_messages 함수를 정의하고 이를 사용하는 방법을 보여준다.
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.prebuilt import ToolNode
load_dotenv()
memory = MemorySaver()
@tool
def search(query: str):
"""Call to surf the web."""
return "It's sunny in San Francisco, but you better look out if you're a Gemini 😈."
tools = [search]
tool_node = ToolNode(tools)
model = ChatOpenAI(model_name="gpt-4o-mini")
bound_model = model.bind_tools(tools)
def should_continue(state: MessagesState):
"""Return the next node to execute."""
last_message = state["messages"][-1]
# 만일 함수 호출이 없다면, 끝낸다.
if not last_message.tool_calls:
return END
# 만일 함수 호출이 있다면, 계속한다.
return "action"
def filter_messages(messages: list):
# 마지막 메시지만 사용하는 매우 간단한 도우미 함수이다.
return messages[-1:]
# 모델을 호출하는 함수를 정의한다.
def call_model(state: MessagesState):
messages = filter_messages(state["messages"])
response = bound_model.invoke(messages)
# 리스트를 반환한다. 기존 리스트에 추가될 것이기 때문이다.
return {"messages": response}
workflow = StateGraph(MessagesState)
# 두 개의 노드가 서로를 순환하도록 정의한다.
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
# 진입점을 'agent'로 설정한다.
# 이것은 이 노드가 처음으로 호출되는 것을 의미한다.
workflow.add_edge(START, "agent")
# 조건부 엣지를 추가한다.
workflow.add_conditional_edges(
# 우선, 시작 노드를 정의한다. 'agent'를 사용한다.
# 이것은 'agent' 노드가 호출된 후에 호출되는 엣지를 의미한다.
"agent",
# 다음으로, 다음에 호출될 노드를 결정할 함수를 전달한다.
should_continue,
# 다음으로, 경로 맵을 전달한다. 이 엣지가 갈 수 있는 모든 노드들이다.
["action", END],
)
# tools에서 agent로의 일반 엣지를 추가한다.
# 이것은 tools가 호출된 후에 agent 노드가 호출된다는 것을 의미한다.
workflow.add_edge("action", "agent")
# 마지막으로, 컴파일한다!
# 이것은 LangChain Runnable로 컴파일된다.
# 이것은 다른 Runnable처럼 사용할 수 있다는 것을 의미한다.
app = workflow.compile(checkpointer=memory)from langchain_core.messages import HumanMessage
config = {"configurable": {"thread_id": "2"}}
input_message = HumanMessage(content="안녕! 내 이름은 홍길동이야")
for event in app.stream({"messages": [input_message]}, config, stream_mode="values"):
event["messages"][-1].pretty_print()
input_message = HumanMessage(content="내 이름이 뭐야?")
for event in app.stream({"messages": [input_message]}, config, stream_mode="values"):
event["messages"][-1].pretty_print()================================ Human Message =================================
안녕! 내 이름은 홍길동이야
================================== Ai Message ==================================
안녕하세요, 홍길동님! 어떻게 도와드릴까요?
================================ Human Message =================================
내 이름이 뭐야?
================================== Ai Message ==================================
죄송하지만, 당신의 이름을 알 수 없습니다. 어떻게 도와드릴까요?위 예시에서는 filter_messages 함수를 직접 정의했다. 또한 LangChain에서는 메시지를 자르고 필터링하는 즉시 사용할 수 있는 방법을 제공한다.
반응형
