my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Multi-Agent Personal Learning System: Mem0 + LlamaIndex AgentWorkflow Example
#3
#4	INSTALLATIONS:
#5	!pip install llama-index-core llama-index-memory-mem0 openai
#6
#7	You need MEM0_API_KEY and OPENAI_API_KEY to run the example.
#8	"""
#9
#10	import asyncio
#11	import logging
#12	from datetime import datetime
#13
#14	from dotenv import load_dotenv
#15
#16	# LlamaIndex imports
#17	from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent
#18	from llama_index.core.tools import FunctionTool
#19	from llama_index.llms.openai import OpenAI
#20
#21	# Memory integration
#22	from llama_index.memory.mem0 import Mem0Memory
#23
#24	load_dotenv()
#25
#26	# Configure logging
#27	logging.basicConfig(
#28	level=logging.INFO,
#29	format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
#30	handlers=[logging.StreamHandler(), logging.FileHandler("learning_system.log")],
#31	)
#32	logger = logging.getLogger(__name__)
#33
#34
#35	class MultiAgentLearningSystem:
#36	"""
#37	Multi-Agent Architecture:
#38	- TutorAgent: Main teaching and explanations
#39	- PracticeAgent: Exercises and skill reinforcement
#40	- Shared Memory: Both agents learn from student interactions
#41	"""
#42
#43	def __init__(self, student_id: str):
#44	self.student_id = student_id
#45	self.llm = OpenAI(model="gpt-4.1-nano-2025-04-14", temperature=0.2)
#46
#47	# Memory context for this student
#48	self.memory_context = {"user_id": student_id, "app": "learning_assistant"}
#49	self.memory = Mem0Memory.from_client(context=self.memory_context)
#50
#51	self._setup_agents()
#52
#53	def _setup_agents(self):
#54	"""Setup two agents that work together and share memory"""
#55
#56	# TOOLS
#57	async def assess_understanding(topic: str, student_response: str) -> str:
#58	"""Assess student's understanding of a topic and save insights"""
#59	# Simulate assessment logic
#60	if "confused" in student_response.lower() or "don't understand" in student_response.lower():
#61	assessment = f"STRUGGLING with {topic}: {student_response}"
#62	insight = f"Student needs more help with {topic}. Prefers step-by-step explanations."
#63	elif "makes sense" in student_response.lower() or "got it" in student_response.lower():
#64	assessment = f"UNDERSTANDS {topic}: {student_response}"
#65	insight = f"Student grasped {topic} quickly. Can move to advanced concepts."
#66	else:
#67	assessment = f"PARTIAL understanding of {topic}: {student_response}"
#68	insight = f"Student has basic understanding of {topic}. Needs reinforcement."
#69
#70	return f"Assessment: {assessment}\nInsight saved: {insight}"
#71
#72	async def track_progress(topic: str, success_rate: str) -> str:
#73	"""Track learning progress and identify patterns"""
#74	progress_note = f"Progress on {topic}: {success_rate} - {datetime.now().strftime('%Y-%m-%d')}"
#75	return f"Progress tracked: {progress_note}"
#76
#77	# Convert to FunctionTools
#78	tools = [
#79	FunctionTool.from_defaults(async_fn=assess_understanding),
#80	FunctionTool.from_defaults(async_fn=track_progress),
#81	]
#82
#83	# === AGENTS ===
#84	# Tutor Agent - Main teaching and explanation
#85	self.tutor_agent = FunctionAgent(
#86	name="TutorAgent",
#87	description="Primary instructor that explains concepts and adapts to student needs",
#88	system_prompt="""
#89	You are a patient, adaptive programming tutor. Your key strength is REMEMBERING and BUILDING on previous interactions.
#90
#91	Key Behaviors:
#92	1. Always check what the student has learned before (use memory context)
#93	2. Adapt explanations based on their preferred learning style
#94	3. Reference previous struggles or successes
#95	4. Build progressively on past lessons
#96	5. Use assess_understanding to evaluate responses and save insights
#97
#98	MEMORY-DRIVEN TEACHING:
#99	- "Last time you struggled with X, so let's approach Y differently..."
#100	- "Since you prefer visual examples, here's a diagram..."
#101	- "Building on the functions we covered yesterday..."
#102
#103	When student shows understanding, hand off to PracticeAgent for exercises.
#104	""",
#105	tools=tools,
#106	llm=self.llm,
#107	can_handoff_to=["PracticeAgent"],
#108	)
#109
#110	# Practice Agent - Exercises and reinforcement
#111	self.practice_agent = FunctionAgent(
#112	name="PracticeAgent",
#113	description="Creates practice exercises and tracks progress based on student's learning history",
#114	system_prompt="""
#115	You create personalized practice exercises based on the student's learning history and current level.
#116
#117	Key Behaviors:
#118	1. Generate problems that match their skill level (from memory)
#119	2. Focus on areas they've struggled with previously
#120	3. Gradually increase difficulty based on their progress
#121	4. Use track_progress to record their performance
#122	5. Provide encouraging feedback that references their growth
#123
#124	MEMORY-DRIVEN PRACTICE:
#125	- "Let's practice loops again since you wanted more examples..."
#126	- "Here's a harder version of the problem you solved yesterday..."
#127	- "You've improved a lot in functions, ready for the next level?"
#128
#129	After practice, can hand back to TutorAgent for concept review if needed.
#130	""",
#131	tools=tools,
#132	llm=self.llm,
#133	can_handoff_to=["TutorAgent"],
#134	)
#135
#136	# Create the multi-agent workflow
#137	self.workflow = AgentWorkflow(
#138	agents=[self.tutor_agent, self.practice_agent],
#139	root_agent=self.tutor_agent.name,
#140	initial_state={
#141	"current_topic": "",
#142	"student_level": "beginner",
#143	"learning_style": "unknown",
#144	"session_goals": [],
#145	},
#146	)
#147
#148	async def start_learning_session(self, topic: str, student_message: str = "") -> str:
#149	"""
#150	Start a learning session with multi-agent memory-aware teaching
#151	"""
#152
#153	if student_message:
#154	request = f"I want to learn about {topic}. {student_message}"
#155	else:
#156	request = f"I want to learn about {topic}."
#157
#158	# The magic happens here - multi-agent memory is automatically shared!
#159	response = await self.workflow.run(user_msg=request, memory=self.memory)
#160
#161	return str(response)
#162
#163	async def get_learning_history(self) -> str:
#164	"""Show what the system remembers about this student"""
#165	try:
#166	# Search memory for learning patterns
#167	memories = self.memory.search(user_id=self.student_id, query="learning machine learning")
#168
#169	if memories and len(memories):
#170	history = "\n".join(f"- {m['memory']}" for m in memories)
#171	return history
#172	else:
#173	return "No learning history found yet. Let's start building your profile!"
#174
#175	except Exception as e:
#176	return f"Memory retrieval error: {str(e)}"
#177
#178
#179	async def run_learning_agent():
#180	learning_system = MultiAgentLearningSystem(student_id="Alexander")
#181
#182	# First session
#183	logger.info("Session 1:")
#184	response = await learning_system.start_learning_session(
#185	"Vision Language Models",
#186	"I'm new to machine learning but I have good hold on Python and have 4 years of work experience.",
#187	)
#188	logger.info(response)
#189
#190	# Second session - multi-agent memory will remember the first
#191	logger.info("\nSession 2:")
#192	response2 = await learning_system.start_learning_session("Machine Learning", "what all did I cover so far?")
#193	logger.info(response2)
#194
#195	# Show what the multi-agent system remembers
#196	logger.info("\nLearning History:")
#197	history = await learning_system.get_learning_history()
#198	logger.info(history)
#199
#200
#201	if __name__ == "__main__":
#202	"""Run the example"""
#203	logger.info("Multi-agent Learning System powered by LlamaIndex and Mem0")
#204
#205	async def main():
#206	await run_learning_agent()
#207
#208	asyncio.run(main())
#209

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public