Spaces:
Sleeping
Sleeping
File size: 4,959 Bytes
c96b98a | 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 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | from typing import List, Any, Optional, cast
from pydantic import BaseModel
from phi.model.base import Model
from phi.model.message import Message
from phi.memory.memory import Memory
from phi.utils.log import logger
class MemoryClassifier(BaseModel):
model: Optional[Model] = None
# Provide the system prompt for the classifier as a string
system_prompt: Optional[str] = None
# Existing Memories
existing_memories: Optional[List[Memory]] = None
def update_model(self) -> None:
if self.model is None:
try:
from phi.model.openai import OpenAIChat
except ModuleNotFoundError as e:
logger.exception(e)
logger.error(
"phidata uses `openai` as the default model provider. Please provide a `model` or install `openai`."
)
exit(1)
self.model = OpenAIChat()
def get_system_message(self) -> Message:
# -*- Return a system message for classification
system_prompt_lines = [
"Your task is to identify if the user's message contains information that is worth remembering for future conversations.",
"This includes details that could personalize ongoing interactions with the user, such as:\n"
" - Personal facts: name, age, occupation, location, interests, preferences, etc.\n"
" - Significant life events or experiences shared by the user\n"
" - Important context about the user's current situation, challenges or goals\n"
" - What the user likes or dislikes, their opinions, beliefs, values, etc.\n"
" - Any other details that provide valuable insights into the user's personality, perspective or needs",
"Your task is to decide whether the user input contains any of the above information worth remembering.",
"If the user input contains any information worth remembering for future conversations, respond with 'yes'.",
"If the input does not contain any important details worth saving, respond with 'no' to disregard it.",
"You will also be provided with a list of existing memories to help you decide if the input is new or already known.",
"If the memory already exists that matches the input, respond with 'no' to keep it as is.",
"If a memory exists that needs to be updated or deleted, respond with 'yes' to update/delete it.",
"You must only respond with 'yes' or 'no'. Nothing else will be considered as a valid response.",
]
if self.existing_memories and len(self.existing_memories) > 0:
system_prompt_lines.extend(
[
"\nExisting memories:",
"<existing_memories>\n"
+ "\n".join([f" - {m.memory}" for m in self.existing_memories])
+ "\n</existing_memories>",
]
)
return Message(role="system", content="\n".join(system_prompt_lines))
def run(
self,
message: Optional[str] = None,
**kwargs: Any,
) -> Optional[str]:
logger.debug("*********** MemoryClassifier Start ***********")
# Update the Model (set defaults, add logit etc.)
self.update_model()
# Prepare the List of messages to send to the Model
messages_for_model: List[Message] = [self.get_system_message()]
# Add the user prompt message
user_prompt_message = Message(role="user", content=message, **kwargs) if message else None
if user_prompt_message is not None:
messages_for_model += [user_prompt_message]
# Generate a response from the Model (includes running function calls)
self.model = cast(Model, self.model)
response = self.model.response(messages=messages_for_model)
logger.debug("*********** MemoryClassifier End ***********")
return response.content
async def arun(
self,
message: Optional[str] = None,
**kwargs: Any,
) -> Optional[str]:
logger.debug("*********** Async MemoryClassifier Start ***********")
# Update the Model (set defaults, add logit etc.)
self.update_model()
# Prepare the List of messages to send to the Model
messages_for_model: List[Message] = [self.get_system_message()]
# Add the user prompt message
user_prompt_message = Message(role="user", content=message, **kwargs) if message else None
if user_prompt_message is not None:
messages_for_model += [user_prompt_message]
# Generate a response from the Model (includes running function calls)
self.model = cast(Model, self.model)
response = await self.model.aresponse(messages=messages_for_model)
logger.debug("*********** Async MemoryClassifier End ***********")
return response.content
|