使用Autogen和Zep构建具有长期记忆的代理

import os
import uuid
from typing import Dict, Union

from dotenv import load_dotenv

from autogen import Agent, ConversableAgent

load_dotenv()

config_list = [
    {
        "model": "gpt-4o-mini",
        "api_key": os.environ.get("OPENAI_API_KEY"),
        "max_tokens": 1024,
    }
]

from zep_cloud import FactRatingExamples, FactRatingInstruction, Message
from zep_cloud.client import AsyncZep

MIN_FACT_RATING = 0.3

# Configure Zep
zep = AsyncZep(api_key=os.environ.get("ZEP_API_KEY"))

def convert_to_zep_messages(chat_history: list[dict[str, str | None]]) -> list[Message]:
    """
    Convert chat history to Zep messages.

    Args:
    chat_history (list): List of dictionaries containing chat messages.

    Returns:
    list: List of Zep Message objects.
    """
    return [
        Message(
            role_type=msg["role"],
            role=msg.get("name", None),
            content=msg["content"],
        )
        for msg in chat_history
    ]

class ZepConversableAgent(ConversableAgent):
    """
    A custom ConversableAgent that integrates with Zep for long-term memory.
    """

    def __init__(
        self,
        name: str,
        system_message: str,
        llm_config: dict,
        function_map: dict,
        human_input_mode: str,
        zep_session_id: str,
    ):
        super().__init__(
            name=name,
            system_message=system_message,
            llm_config=llm_config,
            function_map=function_map,
            human_input_mode=human_input_mode,
        )
        self.zep_session_id = zep_session_id
        # store the original system message as we will update it with relevant facts from Zep
        self.original_system_message = system_message
        self.register_hook("a_process_last_received_message", self.persist_user_messages)
        self.register_hook("a_process_message_before_send", self.persist_assistant_messages)

    async def persist_assistant_messages(
        self, sender: Agent, message: Union[Dict, str], recipient: Agent, silent: bool
    ):
        """Agent sends a message to the user. Add the message to Zep."""

        # Assume message is a string
        zep_messages = convert_to_zep_messages([{"role": "assistant", "name": self.name, "content": message}])
        await zep.memory.add(session_id=self.zep_session_id, messages=zep_messages)

        return message

    async def persist_user_messages(self, messages: list[dict[str, str]] | str):
        """
        User sends a message to the agent. Add the message to Zep and
        update the system message with relevant facts from Zep.
        """
        # Assume messages is a string
        zep_messages = convert_to_zep_messages([{"role": "user", "content": messages}])
        await zep.memory.add(session_id=self.zep_session_id, messages=zep_messages)

        memory = await zep.memory.get(self.zep_session_id, min_rating=MIN_FACT_RATING)

        # Update the system message with the relevant facts retrieved from Zep
        self.update_system_message(
            self.original_system_message
            + f"\n\nRelevant facts about the user and their prior conversation:\n{memory.relevant_facts}"
        )

        return messages

bot_name = "CareBot"
user_name = "Cathy"

user_id = user_name + str(uuid.uuid4())[:4]
session_id = str(uuid.uuid4())

await zep.user.add(user_id=user_id)

fact_rating_instruction = """Rate the facts by poignancy. Highly poignant
 facts have a significant emotional impact or relevance to the user.
 Low poignant facts are minimally relevant or of little emotional significance.
"""

fact_rating_examples = FactRatingExamples(
    high="The user received news of a family member's serious illness.",
    medium="The user completed a challenging marathon.",
    low="The user bought a new brand of toothpaste.",
)

await zep.memory.add_session(
    user_id=user_id,
    session_id=session_id,
    fact_rating_instruction=FactRatingInstruction(
        instruction=fact_rating_instruction,
        examples=fact_rating_examples,
    ),
)

chat_history = [
    {
        "role": "assistant",
        "name": "carebot",
        "content": "Hi Cathy, how are you doing today?",
    },
    {
        "role": "user",
        "name": "Cathy",
        "content": "To be honest, I've been feeling a bit down and demotivated lately. It's been tough.",
    },
    {
        "role": "assistant",
        "name": "CareBot",
        "content": "I'm sorry to hear that you're feeling down and demotivated, Cathy. It's understandable given the challenges you're facing. Can you tell me more about what's been going on?",
    },
    {
        "role": "user",
        "name": "Cathy",
        "content": "Well, I'm really struggling to process the passing of my mother.",
    },
    {
        "role": "assistant",
        "name": "CareBot",
        "content": "I'm deeply sorry for your loss, Cathy. Losing a parent is incredibly difficult. It's normal to struggle with grief, and there's no 'right' way to process it. Would you like to talk about your mother or how you're coping?",
    },
    {
        "role": "user",
        "name": "Cathy",
        "content": "Yes, I'd like to talk about my mother. She was a kind and loving person.",
    },
]

# Convert chat history to Zep messages
zep_messages = convert_to_zep_messages(chat_history)

await zep.memory.add(session_id=session_id, messages=zep_messages)

response = await zep.memory.get_session_facts(session_id=session_id, min_rating=MIN_FACT_RATING)

for r in response.facts:
    print(r)

carebot_system_message = """
You are a compassionate mental health bot and caregiver. Review information about the user and their prior conversation below and respond accordingly.
Keep responses empathetic and supportive. And remember, always prioritize the user's well-being and mental health. Keep your responses very concise and to the point.
"""

agent = ZepConversableAgent(
    bot_name,
    system_message=carebot_system_message,
    llm_config={"config_list": config_list},
    function_map=None,  # No registered functions, by default it is None.
    human_input_mode="NEVER",  # Never ask for human input.
    zep_session_id=session_id,
)

cathy = ConversableAgent(
    user_name,
    system_message="You are returning to your conversation with CareBot, a mental health bot. Ask the bot about your previous session.",
    llm_config={"config_list": config_list},
    human_input_mode="NEVER",  # Never ask for human input.
)

result = await agent.a_initiate_chat(
    cathy,
    message="Hi Cathy, nice to see you again. How are you doing today?",
    max_turns=3,
)

response = await zep.memory.get_session_facts(session_id, min_rating=MIN_FACT_RATING)

for r in response.facts:
    print(r)

response = await zep.memory.search_sessions(
    text="What do you know about Cathy's family?",
    user_id=user_id,
    search_scope="facts",
    min_fact_rating=MIN_FACT_RATING,
)

for r in response.results:
    print(r.fact)