EmailAgentwithMemory / app /tools /context_agent_tools.py
Gaykar's picture
chnages
873e08b
Raw
History Blame
2.25 kB
from typing import Any
from langmem import create_search_memory_tool
from langchain.tools import tool
from typing import Dict, Any, Optional,Annotated
from langchain.tools import tool
from langgraph.prebuilt import InjectedState
from langgraph.store.base import BaseStore
@tool
def search_sender_memory_tool(
query: str,
limit: int = 3,
# 1. Inject the entire Graph state at runtime (Corrected ✅)
state: Annotated[Dict[str, Any], InjectedState] = InjectedState,
# 2. Inject the compiled graph storage layer (Fixed line below 👇)
store: Annotated[BaseStore, InjectedState("store")] = InjectedState("store")
) -> str:
"""Search long-term memory for specific historical email contexts.
This tool automatically scopes the search to the active sender interaction.
"""
active_user = state.get("user_id")
# Using fallback to match alternative naming conventions in your state
sender_email = state.get("sender_email_id") or state.get("senders_email")
if not sender_email:
return "Error: Cannot isolate history. Active sender_email_id is missing from state context."
metadata_filter = {
"receiver_email_id": sender_email
}
results = store.search(
namespace=("email", active_user, "collection"),
query=query,
filter=metadata_filter,
limit=limit
)
if not results:
return f"No prior email context found specifically for sender: {sender_email}."
formatted_memories = []
for item in results:
val = item.value
formatted_memories.append(
f"--- Past Interaction Summary ---\n"
f"Sender: {val.get('user_email_id')}\n"
f"Receiver: {val.get('receiver_email_id')}\n"
f"Context Summary: {val.get('summary')}\n"
)
return "\n".join(formatted_memories)
@tool
def give_previous_context(memory_summary: str) -> str:
"""
Args:
memory_summary: Structured summary containing sender identity,
past context, new facts stored, and suggested tone.
"""
return memory_summary
context_agent_tools=[search_sender_memory_tool,give_previous_context]