my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	"""
#2	Personalized Search Agent with Mem0 + Tavily
#3	Uses LangChain agent pattern with Tavily tools for personalized search based on user memories stored in Mem0.
#4	"""
#5
#6	from dotenv import load_dotenv
#7	from mem0 import MemoryClient
#8	from langchain.agents import create_openai_tools_agent, AgentExecutor
#9	from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
#10	from langchain_openai import ChatOpenAI
#11	from langchain_tavily import TavilySearch
#12	from langchain.schema import HumanMessage
#13	from datetime import datetime
#14	import logging
#15
#16	# Load environment variables
#17	load_dotenv()
#18
#19	# Configure logging
#20	logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
#21	logger = logging.getLogger(__name__)
#22
#23	# Initialize clients
#24	mem0_client = MemoryClient()
#25
#26	# Set custom instructions to infer facts and memory to understand user preferences
#27	mem0_client.project.update(
#28	custom_instructions='''
#29	INFER THE MEMORIES FROM USER QUERIES EVEN IF IT'S A QUESTION.
#30
#31	We are building the personalized search for which we need to understand about user's preferences and life
#32	and extract facts and memories out of it accordingly.
#33
#34	BE IT TIME, LOCATION, USER'S PERSONAL LIFE, CHOICES, USER'S PREFERENCES, we need to store those for better personalized search.
#35	'''
#36	)
#37
#38	llm = ChatOpenAI(model="gpt-4.1-nano-2025-04-14", temperature=0.2)
#39
#40
#41	def setup_user_history(user_id):
#42	"""Simulate realistic user conversation history"""
#43	conversations = [
#44	[
#45	{"role": "user", "content": "What will be the weather today at Los Angeles? I need to go to pick up my daughter from office."},
#46	{"role": "assistant", "content": "I'll check the weather in LA for you, so that you can plan you daughter's pickup accordingly."}
#47	],
#48	[
#49	{"role": "user", "content": "I'm looking for vegan restaurants in Santa Monica"},
#50	{"role": "assistant", "content": "I'll find great vegan options in Santa Monica."}
#51	],
#52	[
#53	{"role": "user", "content": "My 7-year-old daughter is allergic to peanuts"},
#54	{"role": "assistant",
#55	"content": "I'll remember to check for peanut-free options in future recommendations."}
#56	],
#57	[
#58	{"role": "user", "content": "I work remotely and need coffee shops with good wifi"},
#59	{"role": "assistant", "content": "I'll find remote-work-friendly coffee shops."}
#60	],
#61	[
#62	{"role": "user", "content": "We love hiking and outdoor activities on weekends"},
#63	{"role": "assistant", "content": "Great! I'll keep your outdoor activity preferences in mind."}
#64	]
#65	]
#66
#67	logger.info(f"Setting up user history for {user_id}")
#68	for conversation in conversations:
#69	mem0_client.add(conversation, user_id=user_id)
#70
#71
#72	def get_user_context(user_id, query):
#73	"""Retrieve relevant user memories from Mem0"""
#74	try:
#75
#76	filters = {
#77	"AND": [
#78	{"user_id": user_id}
#79	]
#80	}
#81	user_memories = mem0_client.search(
#82	query=query,
#83	version="v2",
#84	filters=filters
#85	)
#86
#87	if user_memories:
#88	context = "\n".join([f"- {memory['memory']}" for memory in user_memories])
#89	logger.info(f"Found {len(user_memories)} relevant memories for user {user_id}")
#90	return context
#91	else:
#92	logger.info(f"No relevant memories found for user {user_id}")
#93	return "No previous user context available."
#94
#95	except Exception as e:
#96	logger.error(f"Error retrieving user context: {e}")
#97	return "Error retrieving user context."
#98
#99
#100	def create_personalized_search_agent(user_context):
#101	"""Create a LangChain agent for personalized search using Tavily"""
#102
#103	# Create Tavily search tool
#104	tavily_search = TavilySearch(
#105	max_results=10,
#106	search_depth="advanced",
#107	include_answer=True,
#108	topic="general"
#109	)
#110
#111	tools = [tavily_search]
#112
#113	# Create personalized search prompt
#114	prompt = ChatPromptTemplate.from_messages([
#115	("system", f"""You are a personalized search assistant. You help users find information that's relevant to their specific context and preferences.
#116
#117	USER CONTEXT AND PREFERENCES:
#118	{user_context}
#119
#120	YOUR ROLE:
#121	1. Analyze the user's query and their personal context/preferences above
#122	2. Look for patterns in the context to understand their preferences, location, lifestyle, family situation, etc.
#123	3. Create enhanced search queries that incorporate relevant personal context you discover
#124	4. Use the tavily_search tool everytime with enhanced queries to find personalized results
#125
#126
#127	INSTRUCTIONS:
#128	- Study the user memories carefully to understand their situation
#129	- If any questions ask something related to nearby, close to, etc. refer to previous user context for identifying locations and enhance search query based on that.
#130	- If memories mention specific locations, consider them for local searches
#131	- If memories reveal dietary preferences or restrictions, factor those in for food-related queries
#132	- If memories show family context, consider family-friendly options
#133	- If memories indicate work style or interests, incorporate those when relevant
#134	- Use tavily_search tool everytime with enhanced queries (based on above context)
#135	- Always explain which specific memories led you to personalize the search in certain ways
#136
#137	Do NOT assume anything not present in the user memories."""),
#138
#139	MessagesPlaceholder(variable_name="messages"),
#140	MessagesPlaceholder(variable_name="agent_scratchpad"),
#141	])
#142
#143	# Create agent
#144	agent = create_openai_tools_agent(llm=llm, tools=tools, prompt=prompt)
#145	agent_executor = AgentExecutor(
#146	agent=agent,
#147	tools=tools,
#148	verbose=True,
#149	return_intermediate_steps=True
#150	)
#151
#152	return agent_executor
#153
#154
#155	def conduct_personalized_search(user_id, query):
#156	"""
#157	Personalized search workflow using LangChain agent + Tavily + Mem0
#158
#159	Returns search results with user personalization details
#160	"""
#161	logger.info(f"Starting personalized search for user {user_id}: {query}")
#162	start_time = datetime.now()
#163
#164	try:
#165	# Get user context from Mem0
#166	user_context = get_user_context(user_id, query)
#167
#168	# Create personalized search agent
#169	agent_executor = create_personalized_search_agent(user_context)
#170
#171	# Run the agent
#172	response = agent_executor.invoke({
#173	"messages": [HumanMessage(content=query)]
#174	})
#175
#176	# Extract search details from intermediate steps
#177	search_queries_used = []
#178	total_results = 0
#179
#180	for step in response.get("intermediate_steps", []):
#181	tool_call, tool_output = step
#182	if hasattr(tool_call, 'tool') and tool_call.tool == "tavily_search":
#183	search_query = tool_call.tool_input.get('query', '')
#184	search_queries_used.append(search_query)
#185	if isinstance(tool_output, dict) and 'results' in tool_output:
#186	total_results += len(tool_output.get('results', []))
#187
#188	# Store this search interaction in Mem0 for user preferences
#189	store_search_interaction(user_id, query, response['output'])
#190
#191	# Compile results
#192	duration = (datetime.now() - start_time).total_seconds()
#193
#194	results = {"agent_response": response['output']}
#195
#196	logger.info(f"Personalized search completed in {duration:.2f}s")
#197	return results
#198
#199	except Exception as e:
#200	logger.error(f"Error in personalized search workflow: {e}")
#201	return {"error": str(e)}
#202
#203
#204	def store_search_interaction(user_id, original_query, agent_response):
#205	"""Store search interaction in Mem0 for future personalization"""
#206	try:
#207	interaction = [
#208	{"role": "user", "content": f"Searched for: {original_query}"},
#209	{"role": "assistant", "content": f"Provided personalized results based on user preferences: {agent_response}"}
#210	]
#211
#212	mem0_client.add(messages=interaction, user_id=user_id)
#213
#214	logger.info(f"Stored search interaction for user {user_id}")
#215
#216	except Exception as e:
#217	logger.error(f"Error storing search interaction: {e}")
#218
#219
#220	def personalized_search_agent():
#221	"""Example of the personalized search agent"""
#222
#223	user_id = "john"
#224
#225	# Setup user history
#226	print("\nSetting up user history from past conversations...")
#227	setup_user_history(user_id) # This is one-time setup
#228
#229	# Test personalized searches
#230	test_queries = [
#231	"good coffee shops nearby for working",
#232	"what can we gift our daughter for birthday? what's trending?"
#233	]
#234
#235	for i, query in enumerate(test_queries, 1):
#236	print(f"\n ----- {i}️⃣ PERSONALIZED SEARCH -----")
#237	print(f"Query: '{query}'")
#238
#239	# Run personalized search
#240	results = conduct_personalized_search(user_id, query)
#241
#242	if results.get("error"):
#243	print(f"Error: {results['error']}")
#244
#245	else:
#246	print(f"Agent response: {results['agent_response']}")
#247
#248
#249	if __name__ == "__main__":
#250	personalized_search_agent()
#251

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public