my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	# Copyright (c) 2023 - 2024, Owners of https://github.com/autogen-ai
#2	#
#3	# SPDX-License-Identifier: Apache-2.0
#4	#
#5	# Portions derived from https://github.com/microsoft/autogen are under the MIT License.
#6	# SPDX-License-Identifier: MIT
#7	# forked from autogen.agentchat.contrib.capabilities.teachability.Teachability
#8
#9	from typing import Dict, Optional, Union
#10
#11	from autogen.agentchat.assistant_agent import ConversableAgent
#12	from autogen.agentchat.contrib.capabilities.agent_capability import AgentCapability
#13	from autogen.agentchat.contrib.text_analyzer_agent import TextAnalyzerAgent
#14	from termcolor import colored
#15
#16	from mem0 import Memory
#17
#18
#19	class Mem0Teachability(AgentCapability):
#20	def __init__(
#21	self,
#22	verbosity: Optional[int] = 0,
#23	reset_db: Optional[bool] = False,
#24	recall_threshold: Optional[float] = 1.5,
#25	max_num_retrievals: Optional[int] = 10,
#26	llm_config: Optional[Union[Dict, bool]] = None,
#27	agent_id: Optional[str] = None,
#28	memory_client: Optional[Memory] = None,
#29	):
#30	self.verbosity = verbosity
#31	self.recall_threshold = recall_threshold
#32	self.max_num_retrievals = max_num_retrievals
#33	self.llm_config = llm_config
#34	self.analyzer = None
#35	self.teachable_agent = None
#36	self.agent_id = agent_id
#37	self.memory = memory_client if memory_client else Memory()
#38
#39	if reset_db:
#40	self.memory.reset()
#41
#42	def add_to_agent(self, agent: ConversableAgent):
#43	self.teachable_agent = agent
#44	agent.register_hook(hookable_method="process_last_received_message", hook=self.process_last_received_message)
#45
#46	if self.llm_config is None:
#47	self.llm_config = agent.llm_config
#48	assert self.llm_config, "Teachability requires a valid llm_config."
#49
#50	self.analyzer = TextAnalyzerAgent(llm_config=self.llm_config)
#51
#52	agent.update_system_message(
#53	agent.system_message
#54	+ "\nYou've been given the special ability to remember user teachings from prior conversations."
#55	)
#56
#57	def process_last_received_message(self, text: Union[Dict, str]):
#58	expanded_text = text
#59	if self.memory.get_all(agent_id=self.agent_id):
#60	expanded_text = self._consider_memo_retrieval(text)
#61	self._consider_memo_storage(text)
#62	return expanded_text
#63
#64	def _consider_memo_storage(self, comment: Union[Dict, str]):
#65	response = self._analyze(
#66	comment,
#67	"Does any part of the TEXT ask the agent to perform a task or solve a problem? Answer with just one word, yes or no.",
#68	)
#69
#70	if "yes" in response.lower():
#71	advice = self._analyze(
#72	comment,
#73	"Briefly copy any advice from the TEXT that may be useful for a similar but different task in the future. But if no advice is present, just respond with 'none'.",
#74	)
#75
#76	if "none" not in advice.lower():
#77	task = self._analyze(
#78	comment,
#79	"Briefly copy just the task from the TEXT, then stop. Don't solve it, and don't include any advice.",
#80	)
#81
#82	general_task = self._analyze(
#83	task,
#84	"Summarize very briefly, in general terms, the type of task described in the TEXT. Leave out details that might not appear in a similar problem.",
#85	)
#86
#87	if self.verbosity >= 1:
#88	print(colored("\nREMEMBER THIS TASK-ADVICE PAIR", "light_yellow"))
#89	self.memory.add(
#90	[{"role": "user", "content": f"Task: {general_task}\nAdvice: {advice}"}], agent_id=self.agent_id
#91	)
#92
#93	response = self._analyze(
#94	comment,
#95	"Does the TEXT contain information that could be committed to memory? Answer with just one word, yes or no.",
#96	)
#97
#98	if "yes" in response.lower():
#99	question = self._analyze(
#100	comment,
#101	"Imagine that the user forgot this information in the TEXT. How would they ask you for this information? Include no other text in your response.",
#102	)
#103
#104	answer = self._analyze(
#105	comment, "Copy the information from the TEXT that should be committed to memory. Add no explanation."
#106	)
#107
#108	if self.verbosity >= 1:
#109	print(colored("\nREMEMBER THIS QUESTION-ANSWER PAIR", "light_yellow"))
#110	self.memory.add(
#111	[{"role": "user", "content": f"Question: {question}\nAnswer: {answer}"}], agent_id=self.agent_id
#112	)
#113
#114	def _consider_memo_retrieval(self, comment: Union[Dict, str]):
#115	if self.verbosity >= 1:
#116	print(colored("\nLOOK FOR RELEVANT MEMOS, AS QUESTION-ANSWER PAIRS", "light_yellow"))
#117	memo_list = self._retrieve_relevant_memos(comment)
#118
#119	response = self._analyze(
#120	comment,
#121	"Does any part of the TEXT ask the agent to perform a task or solve a problem? Answer with just one word, yes or no.",
#122	)
#123
#124	if "yes" in response.lower():
#125	if self.verbosity >= 1:
#126	print(colored("\nLOOK FOR RELEVANT MEMOS, AS TASK-ADVICE PAIRS", "light_yellow"))
#127	task = self._analyze(
#128	comment, "Copy just the task from the TEXT, then stop. Don't solve it, and don't include any advice."
#129	)
#130
#131	general_task = self._analyze(
#132	task,
#133	"Summarize very briefly, in general terms, the type of task described in the TEXT. Leave out details that might not appear in a similar problem.",
#134	)
#135
#136	memo_list.extend(self._retrieve_relevant_memos(general_task))
#137
#138	memo_list = list(set(memo_list))
#139	return comment + self._concatenate_memo_texts(memo_list)
#140
#141	def _retrieve_relevant_memos(self, input_text: str) -> list:
#142	search_results = self.memory.search(input_text, agent_id=self.agent_id, limit=self.max_num_retrievals)
#143	memo_list = [result["memory"] for result in search_results if result["score"] <= self.recall_threshold]
#144
#145	if self.verbosity >= 1 and not memo_list:
#146	print(colored("\nTHE CLOSEST MEMO IS BEYOND THE THRESHOLD:", "light_yellow"))
#147	if search_results["results"]:
#148	print(search_results["results"][0])
#149	print()
#150
#151	return memo_list
#152
#153	def _concatenate_memo_texts(self, memo_list: list) -> str:
#154	memo_texts = ""
#155	if memo_list:
#156	info = "\n# Memories that might help\n"
#157	for memo in memo_list:
#158	info += f"- {memo}\n"
#159	if self.verbosity >= 1:
#160	print(colored(f"\nMEMOS APPENDED TO LAST MESSAGE...\n{info}\n", "light_yellow"))
#161	memo_texts += "\n" + info
#162	return memo_texts
#163
#164	def _analyze(self, text_to_analyze: Union[Dict, str], analysis_instructions: Union[Dict, str]):
#165	self.analyzer.reset()
#166	self.teachable_agent.send(
#167	recipient=self.analyzer, message=text_to_analyze, request_reply=False, silent=(self.verbosity < 2)
#168	)
#169	self.teachable_agent.send(
#170	recipient=self.analyzer, message=analysis_instructions, request_reply=True, silent=(self.verbosity < 2)
#171	)
#172	return self.teachable_agent.last_message(self.analyzer)["content"]
#173

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public