히스토리 관리

---

🏁 학습할 내용

• 히스토리 관리가 필요한 이유
  
  • 히스토리 관리 시작하기
• 최적화를 위한 삭제
  
  • Remove 담당 Node
• Human-in-the-loop

---

📖 히스토리

 

1️⃣ 히스토리 관리가 필요한 이유

• 중간에 장애가 발생해도 복구할 수 있음 
• 대화 / 작업 흐름을 유지할 수 있음
• Human-in-the-loop: 개발자가 개입하여, 검토 및 수정을 할 수 있음
• 디버깅 및 추적이 간편함

 

2️⃣ 히스토리 관리 시작하기

checkpoints를 활용, Snapshot을 만듬, 우리는 간단한 MemorySaver checkpointer를 이용함

checkpointer는 매 실행마다, 상태 스냅샷(checkpoint)를 thread에 누적하게 됨

 

💡 Thread란?

checkpointer가 저장하는 checkpoint들을 묶는 고유 ID
그래프를 실행할 때, configuable.thread_id를 반드시 지정해야함

 

# 그래프 구성

from langgraph.graph import START, END


graph_builder.add_edge(START, 'agent')
graph_builder.add_conditional_edges(
    'agent',
    should_continue,
    ['tools', 'summarize_messages']
)
graph_builder.add_edge('tools', 'agent')
graph_builder.add_edge('summarize_messages', 'delete_messages')
graph_builder.add_edge('delete_messages', END)


# 체크 포인터 지정, compile함수의 매개변수로 넘겨줌

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

graph= graph_builder.compile(checkpointer=checkpointer)


# config주기 


from langchain_core.messages import HumanMessage

config = {
    'configurable': {
        'thread_id': 'summarize_paper' # thread_id
    }
}

query = 'jasonkang14@gmail.com으로 Attention Is All You Need 논문을 요약해서 이메일 초안을 작성해주세요'
for chunk in graph.stream({'messages': [HumanMessage(query)], 'summary': ''}, config=config, stream_mode='values'):
    chunk['messages'][-1].pretty_print()

---

🗑️ 최적화를 위한  메시지 삭제

 

한번 다음과 같은 모습을 상상해보자.

 

현재 메시지 리스트가 다음과 같이 있을 때, llm에게 저 내용 전체를 넘겨주는게 좋을까, 아니면 불필요할까?

 

보통은 그렇게 좋은 편은아니다.

 

물론 llm은 메시지 내용 전체를 알고 있으면, 조금더 답변의 정확도나 내용이 좋아지는 건 맞지만

우리는 한정된 리소스안에서 문제를 최대한 해결해야한다.

 

그렇기 떄문에 여기서 우리는, 메시지 전부를 넘겨줄 필요는 없다.

이 때, 삭제하기전에 앞의 내용을 요약한 후, 마지막 메시지와 함께 넘겨주게되면

 

두마리 토끼를 잡을 수 있다.

 

한번 살펴보자.

 

1️⃣ RemoveMessage 이용하기

요약하기

def summarize_messages(state: AgentState) -> AgentState:
    """
    주어진 state의 메시지를 요약합니다.

    Args:
        state (AgentState): 메시지와 요약을 포함하는 state.

    Returns:
        AgentState: 요약된 메시지를 포함하는 딕셔너리.
    """
    # state에서 메시지와 요약을 가져옵니다.
    messages = state['messages']
    summary = state['summary']
    
    # 요약 프롬프트를 생성합니다.
    summary_prompt = f'summarize this chat history below: \n\nchat_history:{messages}'
    
    # 기존 요약이 있으면, 요약을 포함한 프롬프트를 생성합니다.
    if summary != '':
        summary_prompt = f'''summarize this chat history below while looking at the summary of earlier conversations
chat_history:{messages}
summary:{summary}'''
    
    # LLM을 사용하여 요약을 생성합니다.
    summary = small_llm.invoke(summary_prompt)
    
    # 요약된 메시지를 반환합니다.
    return {'summary': summary.content}

 

삭제하기

rom langchain_core.messages import RemoveMessage

def delete_messages(state: AgentState) -> AgentState:
    """
    주어진 state에서 오래된 메시지를 삭제합니다.

    Args:
        state (AgentState): 메시지를 포함하는 state.

    Returns:
        AgentState: 삭제된 메시지를 포함하는 새로운 state.
    """
    # state에서 메시지를 가져옵니다.
    messages = state['messages']
    # 마지막 세 개의 메시지를 제외한 나머지 메시지를 삭제합니다.
    delete_messages = [RemoveMessage(id=message.id) for message in messages[:-3]]
    # 삭제된 메시지를 포함하는 새로운 state를 반환합니다.
    return {'messages': delete_messages}

 

그래프

from typing import Literal

def should_continue(state: AgentState) -> Literal['tools', 'summarize_messages']:
    """
    주어진 state에 따라 다음 단계로 진행할지를 결정합니다.

    Args:
        state (AgentState): 메시지와 도구 호출 정보를 포함하는 state.

    Returns:
        Literal['tools', 'summarize_messages']: 다음 단계로 'tools' 또는 'summarize_messages'를 반환합니다.
    """
    # state에서 메시지를 가져옵니다.
    messages = state['messages']
    # 마지막 AI 메시지를 확인합니다.
    last_ai_message = messages[-1]
    
    # 마지막 AI 메시지가 도구 호출을 포함하고 있는지 확인합니다.
    if last_ai_message.tool_calls:
        # 도구 호출이 있으면 'tools'를 반환합니다.
        return 'tools'
    
    # 도구 호출이 없으면 'summarize_messages'를 반환합니다.
    return 'summarize_messages'

graph_builder.add_node('agent', agent)
graph_builder.add_node('tools', tool_node)
graph_builder.add_node(delete_messages)
graph_builder.add_node(summarize_messages)


from langgraph.graph import START, END


graph_builder.add_edge(START, 'agent')
graph_builder.add_conditional_edges(
    'agent',
    should_continue,
    ['tools', 'summarize_messages']
)
graph_builder.add_edge('tools', 'agent')
graph_builder.add_edge('summarize_messages', 'delete_messages')
graph_builder.add_edge('delete_messages', END)

 

• agent를 tools를 사용하여 llm 답변을 생성하고
• 이후 요약

---

🥷Human-in-the-loop

 

⭐ 정의

에이전트가 “도구(tool)”를 실행하려고 할 때,
사람이 중간에 끼어들어 승인/수정/거절을 할 수 있게 하는 미들웨어

🧩 역할

• aprrove: 도구 호출을 그대로 실행
• edit: 도구의 args등을 수정한 후, 실행
• rejcet: 실행하지 않고 거절 사유/피드백을 대화에 추가

 

 🚨HITL(Human-in-the-loop)를 할 떄 체크포인터가 없을 경우, 에러가 발생함

 

 

🧱 구성 요소

• interrupt(input_payload)
  • 중단
  • input_payload: 유저에게 전달될 내용 (딕셔너리 형태)
  • 유저는 이걸보고, 다음 스탭을 어떻게할 지 정함
• Command(resume=human_input)
  • human_input: interrupt 함수의 리턴값 (딕셔너리 형태)
  • a = interrupt() , a에 human_input 값이 들아감
• Command(goto='다음 목적지 노드이름', update={업데이트 내용})
  • goto: 다음 목적지 이름
  • update: 업데이트될 내용 (딕셔너리 형태)

 

 

다음 3가지 경우를 한번 살펴보자.

• LLM이 재대로 된 답변을 했을 경우 -> 그대로 진행 
• 도구를 잘 선택했지만, args를 수정할 경우 -> args만 수정 후 실행
• 도구를 잘못 선택했을 경우 -> 도구를 수정해줌

from typing import Literal
from langgraph.types import interrupt, Command

def human_review(state: AgentState) -> Command[Literal['tools', 'agent']]:
    """
    human_review node는 LLM의 도구 호출에 대해 사람의 검토를 요청합니다.

    Args:
        state (AgentState): 메시지 기록을 포함하는 state.

    Returns:
        Command: 다음 node로 이동하기 위한 Command를 반환합니다.
    """
    messages = state['messages']
    last_message = messages[-1]
    tool_call = last_message.tool_calls[-1]
    human_review = interrupt({
        'question': '이렇게 진행하면 될까요?',
        'tool_call': tool_call
    })
    review_action = human_review['action']
    review_data = human_review.get('data', None)
    
    if review_action == 'continue':
        # 에이전트의 판단이 맞다면, 도구를 사용하기 위해 아무것도 수정하지 않고 `tools` 노드로 이동합니다
        return Command(goto='tools')
    
    if review_action == 'update_args':
        # 도구를 더 효율적으로 사용하기 위해 AIMessage의 `tool_calls` 필드를 업데이트합니다
        updated_ai_message = {
            'id': last_message.id,
            'role': 'ai',
            'content': last_message.content,
            'tool_calls': [{
                'id': tool_call['id'],
                'name': tool_call['name'],
                'args': review_data
            }],
        }
        return Command(goto='tools', update={'messages': [updated_ai_message]})
    
    if review_action == 'update_tool':
        # 다른 도구를 사용하기 위해 `ToolMessage`를 업데이트합니다 
        updated_tool_message = {
            'tool_call_id': tool_call['id'],
            'name': tool_call['name'],
            'role': 'tool',
            'content': review_data
        }
        return Command(goto='agent', update={'messages': [updated_tool_message]})

 

이후 커맨드,

for chunk in graph.stream(
    Command(resume={"action": "continue"}),
    config,
    stream_mode="updates", # 값을 업데이트 할 때, 여기가 꼭 updates로
):
    print(f'chunk == {chunk}')

---

출처

https://wikidocs.net/261582

1-4. 메모리 (Memory)

https://docs.langchain.com/oss/python/langgraph/persistence

Persistence - Docs by LangChain

https://github.com/jasonkang14/inflearn-langgraph-agent/blob/main/3.5%20Agent%EC%9D%98%20%ED%9E%88%EC%8A%A4%ED%86%A0%EB%A6%AC%EB%A5%BC%20%EA%B4%80%EB%A6%AC%ED%95%98%EB%8A%94%20%EB%B0%A9%EB%B2%95.ipynb

inflearn-langgraph-agent/3.5 Agent의 히스토리를 관리하는 방법.ipynb at main · jasonkang14/inflearn-langgraph-agent

https://github.com/jasonkang14/inflearn-langgraph-agent/blob/main/3.6%20human-in-the-loop%20%EC%82%AC%EB%9E%8C%EC%9D%B4%20Agent%EC%99%80%20%EC%86%8C%ED%86%B5%ED%95%98%EB%8A%94%20%EB%B0%A9%EB%B2%95.ipynb

inflearn-langgraph-agent/3.6 human-in-the-loop 사람이 Agent와 소통하는 방법.ipynb at main · jasonkang14/inflearn-lang