Spaces:
Runtime error
Runtime error
File size: 2,248 Bytes
015bec7 c6421b9 ec9cdad 015bec7 c6421b9 015bec7 a590be9 015bec7 873e08b ec9cdad 873e08b 015bec7 873e08b 015bec7 c6421b9 015bec7 c6421b9 873e08b c6421b9 a590be9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | 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] |