메모리를 가진 에이전트
우리는 라우터를 사용하여 이메일을 분류하고, 이메일을 에이전트에 전달하여 응답을 생성하는 이메일 어시스턴트를 만들었습니다. 또한 평가를 수행하고, 특정 도구 호출을 검토하기 위해 휴먼-인-더-루프(HITL)를 추가했습니다. 이제 메모리를 추가하여 어시스턴트가 HITL 피드백을 기억할 수 있는 능력을 부여합니다!
환경 변수 로드
from dotenv import load_dotenv
load_dotenv("../../.env", override=True)False
LangGraph의 메모리
스레드 범위 메모리와 스레드 간 메모리
먼저, LangGraph에서 메모리가 작동하는 방식을 설명할 필요가 있습니다. LangGraph는 상호 보완적인 목적을 제공하는 두 가지 별개의 메모리 타입을 제공합니다:
스레드 범위 메모리(단기 메모리) 는 단일 대화 스레드의 경계 내에서 작동합니다. 그래프의 상태의 일부로 자동으로 관리되며 스레드 범위 체크포인트를 통해 유지됩니다. 이 메모리 타입은 대화 히스토리, 업로드된 파일, 검색된 문서, 그리고 상호작용 중에 생성된 기타 아티팩트를 보관합니다. 이를 특정 대화 내에서 컨텍스트를 유지하는 작업 메모리로 생각하면 되며, 에이전트가 매번 처음부터 시작하지 않고도 이전 메시지나 작업을 참조할 수 있게 합니다.
스레드 간 메모리(장기 메모리) 는 개별 대화를 넘어 확장되어 여러 세션에 걸쳐 지속되는 지식 베이스를 만듭니다. 이 메모리는 메모리 스토어에 JSON 문서로 저장되며, 네임스페이스(폴더처럼)와 고유한 키(파일명처럼)로 구성됩니다. 스레드 범위 메모리와 달리, 이 정보는 대화가 끝난 후에도 지속되어 시스템이 사용자 선호도, 과거 결정, 누적된 지식을 기억할 수 있게 합니다. 이것이 에이전트가 각 상호작용을 독립적으로 처리하는 대신 시간이 지남에 따라 진정으로 학습하고 적응할 수 있게 하는 기능입니다.
Store는 이 아키텍처의 기초로, 메모리를 구성하고, 검색하고, 업데이트할 수 있는 유연한 데이터베이스를 제공합니다. 이 접근 방식이 강력한 이유는 어떤 메모리 타입을 사용하든 동일한 Store 인터페이스가 일관된 접근 패턴을 제공하기 때문입니다. 이를 통해 개발 중에는 간단한 인메모리 구현을 사용하든, 배포 시에는 프로덕션급 데이터베이스를 사용하든, 에이전트 코드는 변경되지 않고 동일하게 유지될 수 있습니다.
LangGraph Store
LangGraph는 배포 방식에 따라 다양한 Store 구현을 제공합니다:
-
인메모리 (예: 노트북):
from langgraph.store.memory import InMemoryStore사용- 순수하게 메모리 내 Python 딕셔너리로 영속성 없음
- 프로세스가 종료되면 데이터 손실
- 빠른 실험과 테스트에 유용
- 시맨틱 검색은 여기에 표시된 대로 구성 가능
-
langgraph dev를 사용한 로컬 개발:- InMemoryStore와 유사하지만 유사 영속성 제공
- 재시작 간에 데이터가 로컬 파일시스템에 피클로 저장됨
- 가볍고 빠르며, 외부 데이터베이스 필요 없음
- 시맨틱 검색은 여기에 표시된 대로 구성 가능
- 개발에 적합하지만 프로덕션 용도로 설계되지 않음
-
LangGraph Platform 또는 프로덕션 배포:
- 프로덕션급 영속성을 위해 pgvector가 있는 PostgreSQL 사용
- 신뢰할 수 있는 백업이 있는 완전 영속적 데이터 스토리지
- 더 큰 데이터셋에 대해 확장 가능
- 시맨틱 검색은 여기에 표시된 대로 구성 가능
- 기본 거리 메트릭은 코사인 유사도 (사용자 정의 가능)
노트북에서는 InMemoryStore를 사용하겠습니다!
from langgraph.store.memory import InMemoryStore
in_memory_store = InMemoryStore()메모리는 튜플로 네임스페이스를 지정하며, 이 특정 예제에서는 (<user_id>, “memories”)가 됩니다. 네임스페이스는 어떤 길이든 가능하고 무엇이든 나타낼 수 있으며, 반드시 사용자 특정일 필요는 없습니다.
user_id = "1"
namespace_for_memory = (user_id, "memories")store.put 메서드를 사용하여 스토어의 네임스페이스에 메모리를 저장합니다. 이렇게 할 때, 위에서 정의한 대로 네임스페이스와 메모리의 키-값 쌍을 지정합니다: 키는 단순히 메모리의 고유 식별자(memory_id)이고 값(딕셔너리)은 메모리 자체입니다.
import uuid
memory_id = str(uuid.uuid4())
memory = {"food_preference": "I like pizza"}
in_memory_store.put(namespace_for_memory, memory_id, memory)store.search 메서드를 사용하여 네임스페이스의 메모리를 읽을 수 있으며, 이는 주어진 사용자에 대한 모든 메모리를 리스트로 반환합니다. 가장 최근 메모리는 리스트의 마지막에 있습니다. 각 메모리 타입은 특정 속성을 가진 Python 클래스(Item)입니다. .dict를 통해 변환하여 딕셔너리로 접근할 수 있습니다. 아래와 같은 속성들이 있지만, 가장 중요한 것은 일반적으로 value입니다.
memories = in_memory_store.search(namespace_for_memory)
memories[-1].dict(){'namespace': ['1', 'memories'],
'key': 'afeb46b1-9a2c-4b10-b52d-e53f93f7fb7a',
'value': {'food_preference': 'I like pizza'},
'created_at': '2025-06-05T18:21:36.440950+00:00',
'updated_at': '2025-06-05T18:21:36.440955+00:00',
'score': None}
그래프에서 이것을 사용하려면, 스토어와 함께 그래프를 컴파일하기만 하면 됩니다:
# We need this because we want to enable threads (conversations)
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()# We need this because we want to enable across-thread memory
from langgraph.store.memory import InMemoryStore
in_memory_store = InMemoryStore()
# Compile the graph with the checkpointer and store
# graph = graph.compile(checkpointer=checkpointer, store=in_memory_store)그러면 스토어는 아래에서 볼 수 있듯이 그래프의 모든 노드에서 접근할 수 있습니다!
어시스턴트에 메모리 추가하기
HITL이 있는 그래프를 가져와서 메모리를 추가해봅시다. 이것은 이전에 가지고 있던 것과 매우 유사할 것입니다. 우리는 사용자로부터 피드백을 받을 때 스토어의 메모리를 업데이트하기만 하면 됩니다.
%load_ext autoreload
%autoreload 2
from datetime import datetime
from typing import Literal
from email_assistant.prompts import (
MEMORY_UPDATE_INSTRUCTIONS,
MEMORY_UPDATE_INSTRUCTIONS_REINFORCEMENT,
agent_system_prompt_hitl_memory,
default_background,
default_cal_preferences,
default_response_preferences,
default_triage_instructions,
triage_system_prompt,
triage_user_prompt,
)
from email_assistant.schemas import RouterSchema, State, StateInput
from email_assistant.tools.default.prompt_templates import HITL_MEMORY_TOOLS_PROMPT
from email_assistant.utils import format_email_markdown, format_for_display, parse_email
from langchain_core.tools import tool
from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model
from langgraph.graph import END, START, StateGraph
from langgraph.store.base import BaseStore
from langgraph.types import Command, interrupt
# Agent tools
@tool
def write_email(to: str, subject: str, content: str) -> str:
"""Write and send an email."""
# Placeholder response - in real app would send email
return f"Email sent to {to} with subject '{subject}' and content: {content}"
@tool
def schedule_meeting(
attendees: list[str],
subject: str,
duration_minutes: int,
preferred_day: datetime,
start_time: int,
) -> str:
"""Schedule a calendar meeting."""
# Placeholder response - in real app would check calendar and schedule
date_str = preferred_day.strftime("%A, %B %d, %Y")
return f"Meeting '{subject}' scheduled on {date_str} at {start_time} for {duration_minutes} minutes with {len(attendees)} attendees"
@tool
def check_calendar_availability(day: str) -> str:
"""Check calendar availability for a given day."""
# Placeholder response - in real app would check actual calendar
return f"Available times on {day}: 9:00 AM, 2:00 PM, 4:00 PM"
@tool
class Question(BaseModel):
"""Question to ask user."""
content: str
@tool
class Done(BaseModel):
"""E-mail has been sent."""
done: bool
# All tools available to the agent
tools = [write_email, schedule_meeting, check_calendar_availability, Question, Done]
tools_by_name = {tool.name: tool for tool in tools}
# Initialize the LLM for use with router / structured output
llm = init_chat_model("openai:gpt-4.1", temperature=0.0)
llm_router = llm.with_structured_output(RouterSchema)
# Initialize the LLM, enforcing tool use (of any available tools) for agent
llm = init_chat_model("openai:gpt-4.1", temperature=0.0)
llm_with_tools = llm.bind_tools(tools, tool_choice="required")The autoreload extension is already loaded. To reload it, use:
%reload_ext autoreload
이제 중요한 부분입니다! 현재 그래프에서는 사용자로부터의 피드백을 캡처하지 않습니다.
메모리 관리
우리가 하고 싶은 것은 매우 간단합니다: 피드백을 메모리 Store에 추가하고 싶습니다. 그래프를 Store와 함께 컴파일하면, 어떤 노드에서든 접근할 수 있습니다. 그래서 문제가 되지 않습니다!
하지만 두 가지 질문에 답해야 합니다:
- 메모리를 어떻게 구조화하고 싶은가?
- 메모리를 어떻게 업데이트하고 싶은가?
1)의 경우 간단하게 하기 위해 메모리를 문자열로 저장하겠습니다. 아래 함수에서, 스토어에서 메모리를 문자열로 가져오고 존재하지 않으면 기본값으로 초기화합니다.
from rich.markdown import Markdown
Markdown(default_triage_instructions)"\nEmails that are not worth responding to:\n- Marketing newsletters and promotional emails\n- Spam or suspicious emails\n- CC'd on FYI threads with no direct questions\n\nThere are also other things that should be known about, but don't require an email response. For these, you should notify (using the `notify` response). Examples of this include:\n- Team member out sick or on vacation\n- Build system notifications or deployments\n- Project status updates without action items\n- Important company announcements\n- FYI emails that contain relevant information for current projects\n- HR Department deadline reminders\n- Subscription status / renewal reminders\n- GitHub notifications\n\nEmails that are worth responding to:\n- Direct questions from team members requiring expertise\n- Meeting requests requiring confirmation\n- Critical bug reports related to team's projects\n- Requests from management requiring acknowledgment\n- Client inquiries about project status or features\n- Technical questions about documentation, code, or APIs (especially questions about missing endpoints or features)\n- Personal reminders related to family (wife / daughter)\n- Personal reminder related to self-care (doctor appointments, etc)\n"
Markdown(default_cal_preferences)'\n30 minute meetings are preferred, but 15 minute meetings are also acceptable.\n'
Markdown(default_response_preferences)"\nUse professional and concise language. If the e-mail mentions a deadline, make sure to explicitly acknowledge and reference the deadline in your response.\n\nWhen responding to technical questions that require investigation:\n- Clearly state whether you will investigate or who you will ask\n- Provide an estimated timeline for when you'll have more information or complete the task\n\nWhen responding to event or conference invitations:\n- Always acknowledge any mentioned deadlines (particularly registration deadlines)\n- If workshops or specific topics are mentioned, ask for more specific details about them\n- If discounts (group or early bird) are mentioned, explicitly request information about them\n- Don't commit \n\nWhen responding to collaboration or project-related requests:\n- Acknowledge any existing work or materials mentioned (drafts, slides, documents, etc.)\n- Explicitly mention reviewing these materials before or during the meeting\n- When scheduling meetings, clearly state the specific day, date, and time proposed\n\nWhen responding to meeting scheduling requests:\n- If times are proposed, verify calendar availability for all time slots mentioned in the original email and then commit to one of the proposed times based on your availability by scheduling the meeting. Or, say you can't make it at the time proposed.\n- If no times are proposed, then check your calendar for availability and propose multiple time options when available instead of selecting just one.\n- Mention the meeting duration in your response to confirm you've noted it correctly.\n- Reference the meeting's purpose in your response.\n"
def get_memory(store, namespace, default_content=None):
"""Get memory from the store or initialize with default if it doesn't exist.
Args:
store: LangGraph BaseStore instance to search for existing memory
namespace: Tuple defining the memory namespace, e.g. ("email_assistant", "triage_preferences")
default_content: Default content to use if memory doesn't exist
Returns:
str: The content of the memory profile, either from existing memory or the default
"""
# Search for existing memory with namespace and key
user_preferences = store.get(namespace, "user_preferences")
# If memory exists, return its content (the value)
if user_preferences:
return user_preferences.value
# If memory doesn't exist, add it to the store and return the default content
else:
# Namespace, key, value
store.put(namespace, "user_preferences", default_content)
user_preferences = default_content
# Return the default content
return user_preferences- 메모리 업데이트의 경우, GPT-4.1 프롬프팅 가이드의 몇 가지 트릭을 사용하여 메모리를 업데이트할 수 있습니다:
- 최적의 성능을 위해 프롬프트의 시작과 끝에 핵심 지침을 반복
- 명확하고 명시적인 지침 작성
- 구조를 위한 XML 구분자 사용
- 예제 제공
Markdown(MEMORY_UPDATE_INSTRUCTIONS)┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ ┃ Role and Objective ┃ ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ You are a memory profile manager for an email assistant agent that selectively updates user preferences based on feedback messages from human-in-the-loop interactions with the email assistant. ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ ┃ Instructions ┃ ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ • NEVER overwrite the entire memory profile • ONLY make targeted additions of new information • ONLY update specific facts that are directly contradicted by feedback messages • PRESERVE all other existing information in the profile • Format the profile consistently with the original style • Generate the profile as a string ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ ┃ Reasoning Steps ┃ ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ 1 Analyze the current memory profile structure and content 2 Review feedback messages from human-in-the-loop interactions 3 Extract relevant user preferences from these feedback messages (such as edits to emails/calendar invites, explicit feedback on assistant performance, user decisions to ignore certain emails) 4 Compare new information against existing profile 5 Identify only specific facts to add or update 6 Preserve all other existing information 7 Output the complete updated profile ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ ┃ Example ┃ ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ <memory_profile> RESPOND: • wife • specific questions • system admin notifications NOTIFY: • meeting invites IGNORE: • marketing emails • company-wide announcements • messages meant for other teams </memory_profile> <user_messages> "The assistant shouldn't have responded to that system admin notification." </user_messages> <updated_profile> RESPOND: • wife • specific questions NOTIFY: • meeting invites • system admin notifications IGNORE: • marketing emails • company-wide announcements • messages meant for other teams </updated_profile> ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ ┃ Process current profile for {namespace} ┃ ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ <memory_profile> {current_profile} </memory_profile> Think step by step about what specific feedback is being provided and what specific information should be added or updated in the profile while preserving everything else. Think carefully and update the memory profile based upon these user messages:
Markdown(MEMORY_UPDATE_INSTRUCTIONS_REINFORCEMENT)Remember: • NEVER overwrite the entire memory profile • ONLY make targeted additions of new information • ONLY update specific facts that are directly contradicted by feedback messages • PRESERVE all other existing information in the profile • Format the profile consistently with the original style • Generate the profile as a string
class UserPreferences(BaseModel):
"""Updated user preferences based on user's feedback."""
chain_of_thought: str = Field(
description="Reasoning about which user preferences need to add / update if required"
)
user_preferences: str = Field(description="Updated user preferences")
def update_memory(store, namespace, messages):
"""Update memory profile in the store.
Args:
store: LangGraph BaseStore instance to update memory
namespace: Tuple defining the memory namespace, e.g. ("email_assistant", "triage_preferences")
messages: List of messages to update the memory with
"""
# Get the existing memory
user_preferences = store.get(namespace, "user_preferences")
# Update the memory
llm = init_chat_model("openai:gpt-4.1", temperature=0.0).with_structured_output(
UserPreferences
)
result = llm.invoke(
[
{
"role": "system",
"content": MEMORY_UPDATE_INSTRUCTIONS.format(
current_profile=user_preferences.value, namespace=namespace
),
},
]
+ messages
)
# Save the updated memory to the store
store.put(namespace, "user_preferences", result.user_preferences)이전과 같이 트리아지 라우터를 설정하되, 작은 변경사항이 하나 있습니다
def triage_router(
state: State, store: BaseStore
) -> Command[Literal["triage_interrupt_handler", "response_agent", "__end__"]]:
"""Analyze email content to decide if we should respond, notify, or ignore.
The triage step prevents the assistant from wasting time on:
- Marketing emails and spam
- Company-wide announcements
- Messages meant for other teams
"""
# Parse the email input
author, to, subject, email_thread = parse_email(state["email_input"])
user_prompt = triage_user_prompt.format(
author=author, to=to, subject=subject, email_thread=email_thread
)
# Create email markdown for Agent Inbox in case of notification
email_markdown = format_email_markdown(subject, author, to, email_thread)
# Search for existing triage_preferences memory
triage_instructions = get_memory(
store, ("email_assistant", "triage_preferences"), default_triage_instructions
)
# Format system prompt with background and triage instructions
system_prompt = triage_system_prompt.format(
background=default_background,
triage_instructions=triage_instructions,
)
# Run the router LLM
result = llm_router.invoke(
[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
]
)
# Decision
classification = result.classification
# Process the classification decision
if classification == "respond":
print("📧 Classification: RESPOND - This email requires a response")
# Next node
goto = "response_agent"
# Update the state
update = {
"classification_decision": result.classification,
"messages": [
{"role": "user", "content": f"Respond to the email: {email_markdown}"}
],
}
elif classification == "ignore":
print("🚫 Classification: IGNORE - This email can be safely ignored")
# Next node
goto = END
# Update the state
update = {
"classification_decision": classification,
}
elif classification == "notify":
print("🔔 Classification: NOTIFY - This email contains important information")
# Next node
goto = "triage_interrupt_handler"
# Update the state
update = {
"classification_decision": classification,
}
else:
raise ValueError(f"Invalid classification: {classification}")
return Command(goto=goto, update=update)
사용자가 피드백을 제공할 때 메모리를 업데이트하도록 인터럽트 핸들러에 작은 변경만 하면 됩니다.
def triage_interrupt_handler(
state: State, store: BaseStore
) -> Command[Literal["response_agent", "__end__"]]:
"""Handles interrupts from the triage step"""
# Parse the email input
author, to, subject, email_thread = parse_email(state["email_input"])
# Create email markdown for Agent Inbox in case of notification
email_markdown = format_email_markdown(subject, author, to, email_thread)
# Create messages
messages = [
{"role": "user", "content": f"Email to notify user about: {email_markdown}"}
]
# Create interrupt for Agent Inbox
request = {
"action_request": {
"action": f"Email Assistant: {state['classification_decision']}",
"args": {},
},
"config": {
"allow_ignore": True,
"allow_respond": True,
"allow_edit": False,
"allow_accept": False,
},
# Email to show in Agent Inbox
"description": email_markdown,
}
# Send to Agent Inbox and wait for response
response = interrupt([request])[0]
# If user provides feedback, go to response agent and use feedback to respond to email
if response["type"] == "response":
# Add feedback to messages
user_input = response["args"]
messages.append(
{
"role": "user",
"content": f"User wants to reply to the email. Use this feedback to respond: {user_input}",
}
)
# This is new: update triage_preferences with feedback
update_memory(
store,
("email_assistant", "triage_preferences"),
[
{
"role": "user",
"content": f"The user decided to respond to the email, so update the triage preferences to capture this.",
}
]
+ messages,
)
goto = "response_agent"
# If user ignores email, go to END
elif response["type"] == "ignore":
# Make note of the user's decision to ignore the email
messages.append(
{
"role": "user",
"content": f"The user decided to ignore the email even though it was classified as notify. Update triage preferences to capture this.",
}
)
# This is new: triage_preferences with feedback
update_memory(store, ("email_assistant", "triage_preferences"), messages)
goto = END
# Catch all other responses
else:
raise ValueError(f"Invalid response: {response}")
# Update the state
update = {
"messages": messages,
}
return Command(goto=goto, update=update)LLM 응답에 메모리 통합하기
이제 메모리 관리자를 설정했으므로, 응답을 생성할 때 저장된 선호도를 사용할 수 있습니다
def llm_call(state: State, store: BaseStore):
"""LLM decides whether to call a tool or not"""
# Search for existing cal_preferences memory
cal_preferences = get_memory(
store, ("email_assistant", "cal_preferences"), default_cal_preferences
)
# Search for existing response_preferences memory
response_preferences = get_memory(
store, ("email_assistant", "response_preferences"), default_response_preferences
)
return {
"messages": [
llm_with_tools.invoke(
[
{
"role": "system",
"content": agent_system_prompt_hitl_memory.format(
tools_prompt=HITL_MEMORY_TOOLS_PROMPT,
background=default_background,
response_preferences=response_preferences,
cal_preferences=cal_preferences,
),
}
]
+ state["messages"]
)
]
}인터럽트 핸들러에 메모리 통합하기
마찬가지로, 인터럽트 핸들러에도 메모리를 추가하겠습니다!
def interrupt_handler(
state: State, store: BaseStore
) -> Command[Literal["llm_call", "__end__"]]:
"""Creates an interrupt for human review of tool calls"""
# Store messages
result = []
# Go to the LLM call node next
goto = "llm_call"
# Iterate over the tool calls in the last message
for tool_call in state["messages"][-1].tool_calls:
# Allowed tools for HITL
hitl_tools = ["write_email", "schedule_meeting", "Question"]
# If tool is not in our HITL list, execute it directly without interruption
if tool_call["name"] not in hitl_tools:
# Execute tool without interruption
tool = tools_by_name[tool_call["name"]]
observation = tool.invoke(tool_call["args"])
result.append(
{
"role": "tool",
"content": observation,
"tool_call_id": tool_call["id"],
}
)
continue
# Get original email from email_input in state
email_input = state["email_input"]
author, to, subject, email_thread = parse_email(email_input)
original_email_markdown = format_email_markdown(
subject, author, to, email_thread
)
# Format tool call for display and prepend the original email
tool_display = format_for_display(tool_call)
description = original_email_markdown + tool_display
# Configure what actions are allowed in Agent Inbox
if tool_call["name"] == "write_email":
config = {
"allow_ignore": True,
"allow_respond": True,
"allow_edit": True,
"allow_accept": True,
}
elif tool_call["name"] == "schedule_meeting":
config = {
"allow_ignore": True,
"allow_respond": True,
"allow_edit": True,
"allow_accept": True,
}
elif tool_call["name"] == "Question":
config = {
"allow_ignore": True,
"allow_respond": True,
"allow_edit": False,
"allow_accept": False,
}
else:
raise ValueError(f"Invalid tool call: {tool_call['name']}")
# Create the interrupt request
request = {
"action_request": {"action": tool_call["name"], "args": tool_call["args"]},
"config": config,
"description": description,
}
# Send to Agent Inbox and wait for response
response = interrupt([request])[0]
# Handle the responses
if response["type"] == "accept":
# Execute the tool with original args
tool = tools_by_name[tool_call["name"]]
observation = tool.invoke(tool_call["args"])
result.append(
{
"role": "tool",
"content": observation,
"tool_call_id": tool_call["id"],
}
)
elif response["type"] == "edit":
# Tool selection
tool = tools_by_name[tool_call["name"]]
initial_tool_call = tool_call["args"]
# Get edited args from Agent Inbox
edited_args = response["args"]["args"]
# Update the AI message's tool call with edited content (reference to the message in the state)
ai_message = state["messages"][
-1
] # Get the most recent message from the state
current_id = tool_call["id"] # Store the ID of the tool call being edited
# Create a new list of tool calls by filtering out the one being edited and adding the updated version
# This avoids modifying the original list directly (immutable approach)
updated_tool_calls = [
tc for tc in ai_message.tool_calls if tc["id"] != current_id
] + [
{
"type": "tool_call",
"name": tool_call["name"],
"args": edited_args,
"id": current_id,
}
]
# Create a new copy of the message with updated tool calls rather than modifying the original
# This ensures state immutability and prevents side effects in other parts of the code
# When we update the messages state key ("messages": result), the add_messages reducer will
# overwrite existing messages by id and we take advantage of this here to update the tool calls.
result.append(
ai_message.model_copy(update={"tool_calls": updated_tool_calls})
)
# Save feedback in memory and update the write_email tool call with the edited content from Agent Inbox
if tool_call["name"] == "write_email":
# Execute the tool with edited args
observation = tool.invoke(edited_args)
# Add only the tool response message
result.append(
{"role": "tool", "content": observation, "tool_call_id": current_id}
)
# This is new: update the memory
update_memory(
store,
("email_assistant", "response_preferences"),
[
{
"role": "user",
"content": f"User edited the email response. Here is the initial email generated by the assistant: {initial_tool_call}. Here is the edited email: {edited_args}. Follow all instructions above, and remember: {MEMORY_UPDATE_INSTRUCTIONS_REINFORCEMENT}.",
}
],
)
# Save feedback in memory and update the schedule_meeting tool call with the edited content from Agent Inbox
elif tool_call["name"] == "schedule_meeting":
# Execute the tool with edited args
observation = tool.invoke(edited_args)
# Add only the tool response message
result.append(
{"role": "tool", "content": observation, "tool_call_id": current_id}
)
# This is new: update the memory
update_memory(
store,
("email_assistant", "cal_preferences"),
[
{
"role": "user",
"content": f"User edited the calendar invitation. Here is the initial calendar invitation generated by the assistant: {initial_tool_call}. Here is the edited calendar invitation: {edited_args}. Follow all instructions above, and remember: {MEMORY_UPDATE_INSTRUCTIONS_REINFORCEMENT}.",
}
],
)
# Catch all other tool calls
else:
raise ValueError(f"Invalid tool call: {tool_call['name']}")
elif response["type"] == "ignore":
if tool_call["name"] == "write_email":
# Don't execute the tool, and tell the agent how to proceed
result.append(
{
"role": "tool",
"content": "User ignored this email draft. Ignore this email and end the workflow.",
"tool_call_id": tool_call["id"],
}
)
# Go to END
goto = END
# This is new: update the memory
update_memory(
store,
("email_assistant", "triage_preferences"),
state["messages"]
+ result
+ [
{
"role": "user",
"content": f"The user ignored the email draft. That means they did not want to respond to the email. Update the triage preferences to ensure emails of this type are not classified as respond. Follow all instructions above, and remember: {MEMORY_UPDATE_INSTRUCTIONS_REINFORCEMENT}.",
}
],
)
elif tool_call["name"] == "schedule_meeting":
# Don't execute the tool, and tell the agent how to proceed
result.append(
{
"role": "tool",
"content": "User ignored this calendar meeting draft. Ignore this email and end the workflow.",
"tool_call_id": tool_call["id"],
}
)
# Go to END
goto = END
# This is new: update the memory
update_memory(
store,
("email_assistant", "triage_preferences"),
state["messages"]
+ result
+ [
{
"role": "user",
"content": f"The user ignored the calendar meeting draft. That means they did not want to schedule a meeting for this email. Update the triage preferences to ensure emails of this type are not classified as respond. Follow all instructions above, and remember: {MEMORY_UPDATE_INSTRUCTIONS_REINFORCEMENT}.",
}
],
)
elif tool_call["name"] == "Question":
# Don't execute the tool, and tell the agent how to proceed
result.append(
{
"role": "tool",
"content": "User ignored this question. Ignore this email and end the workflow.",
"tool_call_id": tool_call["id"],
}
)
# Go to END
goto = END
# This is new: update the memory
update_memory(
store,
("email_assistant", "triage_preferences"),
state["messages"]
+ result
+ [
{
"role": "user",
"content": f"The user ignored the Question. That means they did not want to answer the question or deal with this email. Update the triage preferences to ensure emails of this type are not classified as respond. Follow all instructions above, and remember: {MEMORY_UPDATE_INSTRUCTIONS_REINFORCEMENT}.",
}
],
)
else:
raise ValueError(f"Invalid tool call: {tool_call['name']}")
elif response["type"] == "response":
# User provided feedback
user_feedback = response["args"]
if tool_call["name"] == "write_email":
# Don't execute the tool, and add a message with the user feedback to incorporate into the email
result.append(
{
"role": "tool",
"content": f"User gave feedback, which can we incorporate into the email. Feedback: {user_feedback}",
"tool_call_id": tool_call["id"],
}
)
# This is new: update the memory
update_memory(
store,
("email_assistant", "response_preferences"),
state["messages"]
+ result
+ [
{
"role": "user",
"content": f"User gave feedback, which we can use to update the response preferences. Follow all instructions above, and remember: {MEMORY_UPDATE_INSTRUCTIONS_REINFORCEMENT}.",
}
],
)
elif tool_call["name"] == "schedule_meeting":
# Don't execute the tool, and add a message with the user feedback to incorporate into the email
result.append(
{
"role": "tool",
"content": f"User gave feedback, which can we incorporate into the meeting request. Feedback: {user_feedback}",
"tool_call_id": tool_call["id"],
}
)
# This is new: update the memory
update_memory(
store,
("email_assistant", "cal_preferences"),
state["messages"]
+ result
+ [
{
"role": "user",
"content": f"User gave feedback, which we can use to update the calendar preferences. Follow all instructions above, and remember: {MEMORY_UPDATE_INSTRUCTIONS_REINFORCEMENT}.",
}
],
)
elif tool_call["name"] == "Question":
# Don't execute the tool, and add a message with the user feedback to incorporate into the email
result.append(
{
"role": "tool",
"content": f"User answered the question, which can we can use for any follow up actions. Feedback: {user_feedback}",
"tool_call_id": tool_call["id"],
}
)
else:
raise ValueError(f"Invalid tool call: {tool_call['name']}")
# Update the state
update = {
"messages": result,
}
return Command(goto=goto, update=update)나머지는 이전과 동일합니다!
from email_assistant.utils import show_graph
# Conditional edge function
def should_continue(
state: State, store: BaseStore
) -> Literal["interrupt_handler", END]:
"""Route to tool handler, or end if Done tool called"""
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
for tool_call in last_message.tool_calls:
if tool_call["name"] == "Done":
return END
else:
return "interrupt_handler"
# Build workflow
agent_builder = StateGraph(State)
# Add nodes - with store parameter
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("interrupt_handler", interrupt_handler)
# Add edges
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
"llm_call",
should_continue,
{
"interrupt_handler": "interrupt_handler",
END: END,
},
)
# Compile the agent
response_agent = agent_builder.compile()
# Build overall workflow with store and checkpointer
overall_workflow = (
StateGraph(State, input=StateInput)
.add_node(triage_router)
.add_node(triage_interrupt_handler)
.add_node("response_agent", response_agent)
.add_edge(START, "triage_router")
)
email_assistant = overall_workflow.compile()
show_graph(email_assistant)메모리가 있는 에이전트 테스트하기
이제 이메일 어시스턴트에 메모리를 구현했으므로, 시스템이 사용자 피드백으로부터 어떻게 학습하고 시간이 지남에 따라 적응하는지 테스트해봅시다. 이 테스트 섹션은 다양한 유형의 사용자 상호작용이 어떻게 어시스턴트의 향후 성능을 향상시키는 별개의 메모리 업데이트를 생성하는지 탐구합니다.
이 테스트를 통해 답하는 핵심 질문들:
- 시스템이 사용자 선호도를 어떻게 캡처하고 저장하는가?
- 이러한 저장된 선호도가 향후 결정에 어떻게 영향을 미치는가?
- 어떤 상호작용 패턴이 어떤 유형의 메모리 업데이트로 이어지는가?
먼저, 테스트 전반에 걸쳐 메모리가 어떻게 진화하는지 추적할 수 있도록 메모리 콘텐츠를 표시하는 헬퍼 함수를 만들어봅시다:
import uuid
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.types import Command
# Helper function to display memory content
def display_memory_content(store, namespace=None):
# Display current memory content for all namespaces
print("\n======= CURRENT MEMORY CONTENT =======")
if namespace:
memory = store.get(namespace, "user_preferences")
if memory:
print(f"\n--- {namespace[1]} ---")
print(memory.value)
else:
print(f"\n--- {namespace[1]} ---")
print("No memory found")
else:
for namespace in [
("email_assistant", "triage_preferences"),
("email_assistant", "response_preferences"),
("email_assistant", "cal_preferences"),
("email_assistant", "background"),
]:
memory = store.get(namespace, "user_preferences")
if memory:
print(f"\n--- {namespace[1]} ---")
print(memory.value)
else:
print(f"\n--- {namespace[1]} ---")
print("No memory found")
print("=======================================\n")write_email과 schedule_meeting 승인하기
첫 번째 테스트는 사용자가 수정 없이 에이전트의 작업을 승인할 때 어떤 일이 발생하는지 검사합니다. 이 기준 케이스는 피드백이 제공되지 않을 때 시스템이 어떻게 동작하는지 이해하는 데 도움이 됩니다:
- 이전 테스트에서 사용한 것과 동일한 세금 계획 이메일을 사용합니다
- 시스템은 이것을 “RESPOND”로 분류하고 회의 일정 잡기를 제안합니다
- 변경 없이 회의 일정을 승인합니다
- 에이전트는 회의를 확인하는 이메일을 생성합니다
- 변경 없이 이메일을 승인합니다
이 테스트는 메모리가 활성화된 시스템의 기본 동작을 보여줍니다. 사용자가 단순히 제안된 작업을 승인할 때, 학습할 명시적인 피드백이 없으므로 최소한의 메모리 업데이트 또는 업데이트가 없을 것으로 예상됩니다. 그러나 시스템은 응답을 생성할 때 기존 메모리(있는 경우)를 여전히 활용합니다.
# Respond - Meeting Request Email
email_input_respond = {
"to": "Lance Martin <[email protected]>",
"author": "Project Manager <[email protected]>",
"subject": "Tax season let's schedule call",
"email_thread": "Lance,\n\nIt's tax season again, and I wanted to schedule a call to discuss your tax planning strategies for this year. I have some suggestions that could potentially save you money.\n\nAre you available sometime next week? Tuesday or Thursday afternoon would work best for me, for about 45 minutes.\n\nRegards,\nProject Manager",
}
# Compile the graph
checkpointer = MemorySaver()
store = InMemoryStore()
graph = overall_workflow.compile(checkpointer=checkpointer, store=store)
thread_id_1 = uuid.uuid4()
thread_config_1 = {"configurable": {"thread_id": thread_id_1}}
# Run the graph until the first interrupt
# Email will be classified as "respond"
# Agent will create a schedule_meeting and write_email tool call
print("Running the graph until the first interrupt...")
for chunk in graph.stream({"email_input": email_input_respond}, config=thread_config_1):
# Inspect interrupt object if present
if "__interrupt__" in chunk:
Interrupt_Object = chunk["__interrupt__"][0]
print("\nINTERRUPT OBJECT:")
print(f"Action Request: {Interrupt_Object.value[0]['action_request']}")
# Check memory after first interrupt
display_memory_content(store)schedule_meeting 도구 호출 승인하기
초기 schedule_meeting 제안을 검토할 때, 시스템이 기존 메모리를 사용하여 결정을 알리는 방법에 주목하세요:
- 기본 캘린더 선호도는 30분 회의를 선호하지만, 이메일은 45분을 요청합니다
- 에이전트는 여전히 발신자의 특정 요청을 존중하여 45분 회의를 제안합니다
- 수정 없이 이 제안을 승인하여 단순 승인이 메모리 업데이트를 트리거하는지 확인합니다
이 단계를 실행한 후, 메모리 내용을 확인하여 승인만으로 메모리 업데이트가 발생하는지 확인합니다. 단순 승인은 기준 사용자 경험을 나타냅니다 - 시스템은 조정이 필요 없이 의도한 대로 작동합니다.
print(
f"\nSimulating user accepting the {Interrupt_Object.value[0]['action_request']['action']} tool call..."
)
for chunk in graph.stream(Command(resume=[{"type": "accept"}]), config=thread_config_1):
# Inspect interrupt object if present
if "__interrupt__" in chunk:
Interrupt_Object = chunk["__interrupt__"][0]
print("\nINTERRUPT OBJECT:")
print(f"Action Request: {Interrupt_Object.value[0]['action_request']}")write_email 도구 호출 승인하기
이제 회의 일정을 확인하는 이메일 초안을 승인합니다:
- 이메일 초안은 캘린더 선호도를 알고 있는 상태로 생성됩니다
- 회의 시간, 기간, 목적에 대한 세부사항을 포함합니다
- 변경 없이 승인하여 기준 테스트 케이스를 완료합니다
승인 후, 모든 메모리 스토어를 확인하여 업데이트가 발생했는지 확인합니다. 예상대로, 에이전트의 제안을 단순히 승인하는 것은 강력한 학습 신호를 제공하지 않습니다 - 사용자가 에이전트의 접근 방식에 대해 좋아하는 것이나 싫어하는 것에 대한 명확한 피드백이 없습니다.
트레이스 링크는 전체 워크플로우 실행을 보여주며, 응답 생성을 위한 LLM 호출에서 메모리가 사용되는 것을 볼 수 있지만, 단순 승인에 대한 예상 동작인 메모리 업데이트는 발생하지 않습니다.
print(
f"\nSimulating user accepting the {Interrupt_Object.value[0]['action_request']['action']} tool call..."
)
for chunk in graph.stream(Command(resume=[{"type": "accept"}]), config=thread_config_1):
# Inspect response_agent most recent message
if "response_agent" in chunk:
chunk["response_agent"]["messages"][-1].pretty_print()
# Inspect interrupt object if present
if "__interrupt__" in chunk:
Interrupt_Object = chunk["__interrupt__"][0]
print("\nINTERRUPT OBJECT:")
print(f"Action Request: {Interrupt_Object.value[0]['action_request']}")
# Check memory after accepting the write_email tool call
display_memory_content(store)전체 메시지와 트레이스를 볼 수 있습니다:
https://smith.langchain.com/public/86ff6474-29fe-452e-8829-b05a91b458eb/r
응답을 위한 LLM 호출에서 메모리가 사용되는 것을 확인할 수 있습니다.
그러나 메모리 스토어는 업데이트되지 않습니다, 왜냐하면 HITL을 통해 피드백을 추가하지 않았기 때문입니다.
state = graph.get_state(thread_config_1)
for m in state.values["messages"]:
m.pretty_print()write_email과 schedule_meeting 수정하기
이 테스트는 시스템이 제안된 작업에 대한 직접 수정으로부터 어떻게 학습하는지 탐구합니다. 사용자가 에이전트의 제안을 수정하면, 그들의 선호도에 대한 명확하고 구체적인 학습 신호가 생성됩니다:
- 이전과 동일한 세금 계획 이메일을 사용합니다
- 에이전트가 45분 회의를 제안하면, 다음과 같이 수정합니다:
- 기간을 30분으로 변경 (저장된 선호도와 일치)
- 제목을 더 간결하게 만들기
- 에이전트가 이메일 초안을 작성하면, 다음과 같이 수정합니다:
- 더 짧고 덜 공식적으로
- 구조를 다르게
수정은 사용자 선호도에 대한 가장 명시적인 피드백을 제공하여, 시스템이 원하는 변경사항을 정확히 학습할 수 있게 합니다. 이러한 수정을 반영하는 메모리 스토어에 특정하고 목표가 명확한 업데이트가 표시될 것으로 예상됩니다.
# Same email as before
email_input_respond = {
"to": "Lance Martin <[email protected]>",
"author": "Project Manager <[email protected]>",
"subject": "Tax season let's schedule call",
"email_thread": "Lance,\n\nIt's tax season again, and I wanted to schedule a call to discuss your tax planning strategies for this year. I have some suggestions that could potentially save you money.\n\nAre you available sometime next week? Tuesday or Thursday afternoon would work best for me, for about 45 minutes.\n\nRegards,\nProject Manager",
}
# Compile the graph with new thread
checkpointer = MemorySaver()
store = InMemoryStore()
graph = overall_workflow.compile(checkpointer=checkpointer, store=store)
thread_id_2 = uuid.uuid4()
thread_config_2 = {"configurable": {"thread_id": thread_id_2}}
# Run the graph until the first interrupt - will be classified as "respond" and the agent will create a write_email tool call
print("Running the graph until the first interrupt...")
for chunk in graph.stream({"email_input": email_input_respond}, config=thread_config_2):
# Inspect interrupt object if present
if "__interrupt__" in chunk:
Interrupt_Object = chunk["__interrupt__"][0]
print("\nINTERRUPT OBJECT:")
print(f"Action Request: {Interrupt_Object.value[0]['action_request']}")
# Check memory after first interrupt
display_memory_content(store, ("email_assistant", "cal_preferences"))schedule_meeting 도구 호출 수정하기
회의 제안을 수정하면, 우리의 선호도에 대한 직접적이고 명시적인 피드백을 제공합니다. 이것은 시스템에 중요한 학습 기회를 만듭니다:
- 에이전트는 처음에 45분 회의를 제안합니다 (이메일에서 요청한 기간)
- 우리는 이것을 30분으로 수정하고 제목을 “Tax Planning Strategies Discussion”에서 “Tax Planning Discussion”으로 단순화합니다
- 이것은 우리의 시간 선호도와 명명 규칙에 대한 명확하고 구체적인 피드백을 만듭니다
수정 후, 캘린더 선호도 메모리 스토어를 확인하여 어떻게 업데이트되는지 확인합니다. 메모리 업데이트는 다음을 캡처해야 합니다:
- 더 짧은 30분 회의에 대한 우리의 선호도
- 더 간결한 회의 제목에 대한 우리의 선호도
트레이스는 정확한 메모리 업데이트 로직을 보여주며, 시스템이 우리의 제안과 수정 사이의 차이를 분석하여 의미 있는 패턴과 선호도를 추출하는 방법을 보여줍니다. 각 메모리 업데이트에 대한 자세한 정당화를 볼 수 있어 학습 프로세스의 투명성을 보장합니다.
# Now simulate user editing the schedule_meeting tool call
print("\nSimulating user editing the schedule_meeting tool call...")
edited_schedule_args = {
"attendees": ["[email protected]", "[email protected]"],
"subject": "Tax Planning Discussion",
"duration_minutes": 30, # Changed from 45 to 30
"preferred_day": "2025-04-22",
"start_time": 14,
}
for chunk in graph.stream(
Command(resume=[{"type": "edit", "args": {"args": edited_schedule_args}}]),
config=thread_config_2,
):
# Inspect response_agent most recent message
if "response_agent" in chunk:
chunk["response_agent"]["messages"][-1].pretty_print()
# Inspect interrupt object if present
if "__interrupt__" in chunk:
Interrupt_Object = chunk["__interrupt__"][0]
print("\nINTERRUPT OBJECT:")
print(f"Action Request: {Interrupt_Object.value[0]['action_request']}")
# Check memory after editing schedule_meeting
print("\nChecking memory after editing schedule_meeting:")
display_memory_content(store, ("email_assistant", "cal_preferences")){'preferences': '\\n30 minute meetings are preferred, but 15 minute meetings are also acceptable.\\n'}
{'preferences': "30 minute meetings are preferred, but 15 minute meetings are also acceptable.\\n\\nUser prefers 30 minute meetings over longer durations such as 45 minutes. When scheduling, default to 30 minutes unless otherwise specified. Subject lines should be concise (e.g., 'Tax Planning Discussion' instead of 'Tax Planning Strategies Discussion')."}
캘린더 초대를 수정한 후 메모리를 보면 업데이트된 것을 확인할 수 있습니다:
- 시스템은 우리가 더 긴 기간보다 30분 회의를 선호한다는 것을 식별했습니다
- 또한 간결한 회의 제목에 대한 우리의 선호도도 캡처했습니다
이 메모리 업데이트에서 특히 인상적인 점은:
- 단순히 특정 수정을 기록하는 것이 아니라 더 광범위한 선호도 패턴으로 일반화합니다
- 새 정보를 추가하면서 모든 기존 메모리 콘텐츠를 보존합니다
- 단일 수정 상호작용에서 여러 선호도 신호를 추출합니다
이제, 이메일 초안을 수정하여 시스템이 다양한 유형의 커뮤니케이션 선호도를 어떻게 캡처하는지 확인해봅시다:
display_memory_content(store, ("email_assistant", "response_preferences"))
# Now simulate user editing the write_email tool call
print("\nSimulating user editing the write_email tool call...")
edited_email_args = {
"to": "[email protected]",
"subject": "Re: Tax season let's schedule call",
"content": "Thanks! I scheduled a 30-minute call next Thursday at 3:00 PM. Would that work for you?\n\nBest regards,\nLance Martin",
}
for chunk in graph.stream(
Command(resume=[{"type": "edit", "args": {"args": edited_email_args}}]),
config=thread_config_2,
):
# Inspect response_agent most recent message
if "response_agent" in chunk:
chunk["response_agent"]["messages"][-1].pretty_print()
# Inspect interrupt object if present
if "__interrupt__" in chunk:
Interrupt_Object = chunk["__interrupt__"][0]
print("\nINTERRUPT OBJECT:")
print(f"Action Request: {Interrupt_Object.value[0]['action_request']}")
# Check memory after editing write_email
print("\nChecking memory after editing write_email:")
display_memory_content(store, ("email_assistant", "response_preferences"))이메일 수정은 훨씬 더 정교한 학습 능력을 드러냅니다:
- 이메일 내용을 극적으로 단축하고 단순화했습니다
- 톤을 더 캐주얼하게 변경했습니다
- 시간이 맞는다고 가정하는 대신 확인을 요청하는 질문을 추가했습니다
- 회의 세부사항(요일과 시간)을 약간 변경했습니다
업데이트된 메모리를 보면, 시스템이 우리의 커뮤니케이션 스타일에 대한 핵심 통찰력을 추출한 것을 볼 수 있습니다:
회의를 일정 잡을 때, 이미 회의가 일정 잡혔다고 가정하고 명시하는 대신, 제안된 시간이 괜찮은지 수신자에게 확인을 요청하세요.
이것은 시스템의 다음 능력을 보여줍니다:
- 표면적인 수준이 아니라 의도를 이해하기 위해 수정을 분석
- 특정 예제에서 일반화 가능한 원칙 추출
- 새로운 통찰력을 추가하면서 모든 기존 지침 보존
- 메모리의 구성과 구조 유지
이러한 목표가 명확하고 고품질의 메모리 업데이트는 반복적인 수정 없이 향후 모든 상호작용을 개선할 것입니다.
state = graph.get_state(thread_config_2)
for m in state.values["messages"]:
m.pretty_print()write_email과 schedule_meeting에 피드백 제공하기
마지막 테스트 세트는 “response” 피드백 패턴을 탐구합니다 - 직접 수정하거나 승인하지 않고 가이던스를 제공합니다. 이 대화형 피드백 메커니즘은 승인과 수정 사이의 중간 지점을 제공합니다:
- 먼저, 다음을 요청하여 회의 일정에 대한 피드백을 테스트합니다:
- 더 짧은 기간 (45분 대신 30분)
- 오후 회의 시간 (오후 2시 이후)
- 다음으로, 다음을 요청하여 이메일 초안 작성에 대한 피드백을 테스트합니다:
- 더 짧고 덜 공식적인 언어
- 회의를 기대한다는 특정 마무리 문구
- 마지막으로, 다음을 제공하여 질문에 대한 피드백을 테스트합니다:
- 추가 컨텍스트가 있는 직접 답변
- 특정 선호도 (브런치 장소, 시간)
이 자연어 피드백 접근 방식은 사용자가 직접 작업을 하지 않고도 어시스턴트를 가이드할 수 있게 합니다. 우리의 특정 피드백에서 일반 원칙을 추출하는 자세한 메모리 업데이트가 표시될 것으로 예상됩니다.
# Respond - Meeting Request Email
email_input_respond = {
"to": "Lance Martin <[email protected]>",
"author": "Project Manager <[email protected]>",
"subject": "Tax season let's schedule call",
"email_thread": "Lance,\n\nIt's tax season again, and I wanted to schedule a call to discuss your tax planning strategies for this year. I have some suggestions that could potentially save you money.\n\nAre you available sometime next week? Tuesday or Thursday afternoon would work best for me, for about 45 minutes.\n\nRegards,\nProject Manager",
}
# Compile the graph
checkpointer = MemorySaver()
store = InMemoryStore()
graph = overall_workflow.compile(checkpointer=checkpointer, store=store)
thread_id_5 = uuid.uuid4()
thread_config_5 = {"configurable": {"thread_id": thread_id_5}}
# Run the graph until the first interrupt
# Email will be classified as "respond"
# Agent will create a schedule_meeting and write_email tool call
print("Running the graph until the first interrupt...")
for chunk in graph.stream({"email_input": email_input_respond}, config=thread_config_5):
# Inspect interrupt object if present
if "__interrupt__" in chunk:
Interrupt_Object = chunk["__interrupt__"][0]
print("\nINTERRUPT OBJECT:")
print(f"Action Request: {Interrupt_Object.value[0]['action_request']}")
# Check memory after first interrupt
display_memory_content(store, ("email_assistant", "cal_preferences"))schedule_meeting 도구 호출에 피드백 제공하기
회의 제안을 직접 수정하거나 단순히 승인하는 대신, 자연어 피드백을 제공합니다:
- 45분 대신 30분 회의를 요청합니다
- 오후 2시 이후의 오후 회의에 대한 선호도를 표현합니다
- 시스템은 이 피드백을 해석하고 새로운 제안을 생성해야 합니다
이 대화형 접근 방식은 종종 직접 수정보다 더 자연스럽고 효율적이며, 특히 모바일 사용자나 자세한 수정보다 높은 수준의 방향을 제시하는 것을 선호하는 사용자에게 적합합니다.
피드백을 제공한 후, 캘린더 선호도 메모리를 검사하여 이 자연어 가이던스가 어떻게 캡처되는지 확인합니다. 시스템이 회의 기간과 시간대 선호도를 모두 일반 원칙으로 추출할 것으로 예상됩니다.
print(
f"\nSimulating user providing feedback for the {Interrupt_Object.value[0]['action_request']['action']} tool call..."
)
for chunk in graph.stream(
Command(
resume=[
{
"type": "response",
"args": "Please schedule this for 30 minutes instead of 45 minutes, and I prefer afternoon meetings after 2pm.",
}
]
),
config=thread_config_5,
):
# Inspect interrupt object if present
if "__interrupt__" in chunk:
Interrupt_Object = chunk["__interrupt__"][0]
print("\nINTERRUPT OBJECT:")
print(f"Action Request: {Interrupt_Object.value[0]['action_request']}")
# Check memory after providing feedback for schedule_meeting
print("\nChecking memory after providing feedback for schedule_meeting:")
display_memory_content(store, ("email_assistant", "cal_preferences"))피드백을 제공한 후 메모리 확인은 우아하게 단순한 캘린더 선호도 업데이트를 보여줍니다:
30 minute meetings are preferred, but 15 minute meetings are also acceptable.
Afternoon meetings after 2pm are preferred.
시스템은:
- 피드백의 두 측면(기간과 시간대)을 모두 캡처했습니다
- 15분 회의에 대한 기존 선호도를 보존했습니다
- 오후 2시 이후 오후 회의에 대한 선호도를 새 줄로 추가했습니다
- 형식을 깔끔하고 읽기 쉽게 유지했습니다
이 자연어 피드백 메커니즘은 직접 수정과 동일한 품질의 메모리 업데이트를 생성하지만 사용자의 노력은 덜 필요합니다. 시스템은 비구조화된 피드백에서 구조화된 선호도를 추출할 수 있어, 대화형 상호작용으로부터 학습할 수 있는 능력을 보여줍니다.
이 수정된 회의 제안을 승인하고 이메일 초안으로 이동합시다:
print(
f"\nSimulating user accepting the {Interrupt_Object.value[0]['action_request']['action']} tool call..."
)
for chunk in graph.stream(Command(resume=[{"type": "accept"}]), config=thread_config_5):
# Inspect interrupt object if present
if "__interrupt__" in chunk:
Interrupt_Object = chunk["__interrupt__"][0]
print("\nINTERRUPT OBJECT:")
print(f"Action Request: {Interrupt_Object.value[0]['action_request']}")
# Check memory after accepting schedule_meeting after feedback
print("\nChecking memory after accepting schedule_meeting after feedback:")
display_memory_content(store, ("email_assistant", "response_preferences"))write_email 도구 호출에 피드백 제공하기
회의 피드백과 유사하게, 이제 이메일 초안에 대한 자연어 가이던스를 제공합니다:
- “더 짧고 덜 공식적인” 언어를 요청합니다 - 스타일 선호도
- 회의를 기대한다는 특정 마무리 문구를 요청합니다
- 시스템은 이 가이던스를 해석하고 이메일을 그에 따라 다시 작성해야 합니다
이 피드백을 제공한 후, 응답 선호도 메모리를 확인하여 이러한 스타일과 구조 선호도가 어떻게 캡처되는지 확인합니다. 이메일 간결성, 공식성, 마무리 문구에 대한 일반화 가능한 가이드라인이 선호도 프로필에 추가될 것으로 예상됩니다.
print(
f"\nSimulating user providing feedback for the {Interrupt_Object.value[0]['action_request']['action']} tool call..."
)
for chunk in graph.stream(
Command(
resume=[
{
"type": "response",
"args": "Shorter and less formal. Include a closing statement about looking forward to the meeting!",
}
]
),
config=thread_config_5,
):
# Inspect response_agent most recent message
if "response_agent" in chunk:
chunk["response_agent"]["messages"][-1].pretty_print()
# Inspect interrupt object if present
if "__interrupt__" in chunk:
Interrupt_Object = chunk["__interrupt__"][0]
print("\nINTERRUPT OBJECT:")
print(f"Action Request: {Interrupt_Object.value[0]['action_request']}")
# Check memory after providing feedback for write_email
print("\nChecking memory after providing feedback for write_email:")
display_memory_content(store, ("email_assistant", "response_preferences"))이메일 피드백 후 메모리 업데이트는 회의 일정과 이메일 작성 선호도 모두에 대한 매우 정교한 학습을 보여줍니다:
-
시스템은 “이메일 응답 작성 시”라는 제목의 완전히 새로운 섹션을 응답 선호도에 추가했으며 두 가지 핵심 선호도가 있습니다:
- “컨텍스트가 공식성을 요구하지 않는 한, 가능한 경우 더 짧고 덜 공식적인 언어를 선호”
- “약속을 확인할 때 회의나 대화를 기대한다는 마무리 문구를 포함”
-
또한 “회의 일정 요청에 응답할 때” 섹션에 새로운 글머리 기호를 추가했습니다:
- “회의를 일정 잡을 때, 가능한 경우 오후 2시 이후의 오후 시간을 선호하고, 달리 지정되지 않는 한 30분 기간을 기본으로 사용”
이것은 시스템의 다음 능력을 보여줍니다:
- 학습된 선호도를 적절한 카테고리로 구성
- 단일 피드백 인스턴스에서 여러 통찰력 추출
- 회의 선호도를 캘린더와 이메일 컨텍스트 모두에 적용
- 적절한 한정자로 뉘앙스 캡처 (“가능한 경우”, “달리 지정되지 않는 한”)
- 메모리의 계층 구조 유지
결과 이메일은 이러한 모든 선호도가 적용된 것을 보여줍니다: 더 짧고, 덜 공식적이며, 채팅을 기대한다는 마무리 문구를 포함하고, 30분 회의 시간을 올바르게 참조합니다.
print(
f"\nSimulating user accepting the {Interrupt_Object.value[0]['action_request']['action']} tool call..."
)
for chunk in graph.stream(Command(resume=[{"type": "accept"}]), config=thread_config_5):
# Inspect interrupt object if present
if "__interrupt__" in chunk:
Interrupt_Object = chunk["__interrupt__"][0]
print("\nINTERRUPT OBJECT:")
print(f"Action Request: {Interrupt_Object.value[0]['action_request']}")
# Check memory after accepting write_email after feedback
print("\nChecking memory after accepting write_email after feedback:")
display_memory_content(store, ("email_assistant", "response_preferences"))전체 메시지 히스토리를 확인합니다.
state = graph.get_state(thread_config_5)
for m in state.values["messages"]:
m.pretty_print()로컬 배포
src/email_assistant 디렉토리에서 메모리 통합이 있는 이 그래프를 찾을 수 있습니다:
src/email_assistant/email_assistant_hitl_memory.py
테스트할 이메일:
{
"author": "Alice Smith <[email protected]>",
"to": "John Doe <[email protected]>",
"subject": "Quick question about API documentation",
"email_thread": "Hi John,\nI was reviewing the API documentation for the new authentication service and noticed a few endpoints seem to be missing from the specs. Could you help clarify if this was intentional or if we should update the docs?\nSpecifically, I'm looking at:\n- /auth/refresh\n- /auth/validate\nThanks!\nAlice"
}
이전과 같이, dev.agentinbox.ai로 이동하면 그래프에 쉽게 연결할 수 있습니다:
- Graph name:
langgraph.json파일의 이름 (email_assistant_hitl_memory) - Graph URL:
http://127.0.0.1:2024/
LangGraph Studio의 Memory 탭은 각 상호작용마다 선호도가 어떻게 캡처되고 업데이트되는지 실시간으로 볼 수 있게 합니다:
계속 사용하면서 시스템은 점점 더 개인화됩니다:
- 응답하고 싶은, 알림 받고 싶은, 또는 무시하고 싶은 이메일을 학습합니다
- 커뮤니케이션 스타일 선호도에 적응합니다
- 일정 선호도를 기억합니다
- 각 상호작용마다 이해를 개선합니다
이러한 HITL과 메모리의 조합은 자동화와 제어의 균형을 맞추는 시스템을 만듭니다 - 일상적인 작업을 자동으로 처리하면서 피드백으로부터 학습하여 시간이 지남에 따라 선호도에 더 부합하게 됩니다.
Gmail 도구와 함께 호스팅 배포
실제로 자신의 이메일에서 이것을 실행하고 싶다면, Gmail 도구와 함께 그래프를 배포할 수 있습니다.
여기를 따라 Gmail 자격 증명을 설정하세요.
Gmail 도구와 함께 설정된 그래프가 있습니다:
python src/email_assistant/email_assistant_hitl_memory_gmail.py배포 옵션 중 하나는 hosted입니다, 로컬 배포에서 수행한 것처럼 배포된 그래프 URL을 Agent Inbox에 연결하기만 하면 됩니다.
메모리 개선하기
현재 메모리 스키마와 업데이트는 매우 간단합니다:
- 스키마는 문자열입니다
- 항상 기존 메모리를 새 문자열로 덮어씁니다
스토어는 메모리 컬렉션에 대한 시맨틱 검색으로 쉽게 구성할 수 있습니다.
또한 더 고급 메모리 관리를 위해 LangMem 사용을 고려하세요.