当前位置：首页 > ds >正文

Langflow Memory 技术深度分析

ds 2025/9/1 9:31:45

Langflow Memory 技术深度分析

1. Memory 技术概述和设计理念

1.1 技术概述

Langflow 的 Memory 系统是一个多层次的记忆管理框架，专门设计用于处理对话历史、上下文状态和会话数据的存储与检索。该系统采用了分层架构设计，支持多种记忆类型和存储后端，为 AI 应用提供了强大的记忆能力。

1.2 设计理念

分层抽象：通过基类和接口定义统一的记忆操作规范
多后端支持：支持数据库存储、外部记忆服务和第三方记忆系统
异步优先：全面采用异步编程模型，提升并发性能
类型安全：使用 Pydantic 模型确保数据类型安全和验证
会话隔离：基于 session_id 实现多会话数据隔离
灵活扩展：支持自定义记忆组件和存储策略

2. 核心架构和记忆存储模型

2.1 架构层次图

┌─────────────────────────────────────────────────────────────┐
│                    应用层 (Components)                        │
├─────────────────────────────────────────────────────────────┤
│  MemoryComponent  │  Mem0MemoryComponent  │  MessageStore   │
├─────────────────────────────────────────────────────────────┤
│                    抽象层 (Base Classes)                      │
├─────────────────────────────────────────────────────────────┤
│  BaseMemoryComponent  │  LCChatMemoryComponent              │
├─────────────────────────────────────────────────────────────┤
│                    核心层 (Core Memory)                       │
├─────────────────────────────────────────────────────────────┤
│  Memory Functions  │  LCBuiltinChatMemory                   │
├─────────────────────────────────────────────────────────────┤
│                    数据层 (Data Models)                       │
├─────────────────────────────────────────────────────────────┤
│  Message  │  MessageTable  │  MessageRead  │  MessageCreate │
├─────────────────────────────────────────────────────────────┤
│                    存储层 (Storage)                           │
└─────────────────────────────────────────────────────────────┘
│  Database  │  External Memory  │  Third-party Services      │
└─────────────────────────────────────────────────────────────┘

2.2 存储模型设计

消息基础模型

class MessageBase(SQLModel):timestamp: Annotated[datetime, str_to_timestamp_validator] = Field(default_factory=lambda: datetime.now(timezone.utc))sender: str                    # 发送者类型 (Machine/User)sender_name: str              # 发送者名称session_id: str               # 会话标识符text: str = Field(sa_column=Column(Text))  # 消息内容files: list[str] = Field(default_factory=list)  # 附件文件error: bool = Field(default=False)        # 错误标记edit: bool = Field(default=False)         # 编辑标记properties: Properties = Field(default_factory=Properties)  # 扩展属性category: str = Field(default="message")  # 消息分类content_blocks: list[ContentBlock] = Field(default_factory=list)  # 内容块

数据库存储模型

class MessageTable(MessageBase, table=True):id: UUID = Field(default_factory=uuid4, primary_key=True)job_id: UUID | None = Field(default=None)tenant_id: UUID | None = Field(default=None)organization_id: UUID | None = Field(default=None)business_domain_id: UUID | None = Field(default=None)flow_id: UUID | None = Field(default=None)# JSON 字段存储复杂数据files: list[str] = Field(sa_column=Column(JSON))properties: dict | Properties = Field(default_factory=lambda: Properties().model_dump(), sa_column=Column(JSON))content_blocks: list[dict | ContentBlock] = Field(default_factory=list, sa_column=Column(JSON))

3. 记忆基类和抽象接口分析

3.1 BaseMemoryComponent 基类

class BaseMemoryComponent(CustomComponent):display_name = "Chat Memory"description = "Retrieves stored chat messages given a specific Session ID."beta: bool = Trueicon = "history"def build_config(self):return {"sender": {"options": [MESSAGE_SENDER_AI, MESSAGE_SENDER_USER, "Machine and User"],"display_name": "Sender Type",},"sender_name": {"display_name": "Sender Name", "advanced": True},"n_messages": {"display_name": "Number of Messages","info": "Number of messages to retrieve.",},"session_id": {"display_name": "Session ID","info": "Session ID of the chat history.","input_types": ["Message"],},"order": {"options": ["Ascending", "Descending"],"display_name": "Order","info": "Order of the messages.","advanced": True,},"data_template": {"display_name": "Data Template","multiline": True,"info": "Template to convert Data to Text.","advanced": True,},}def get_messages(self, **kwargs) -> list[Data]:raise NotImplementedErrordef add_message(self, sender: str, sender_name: str, text: str, session_id: str, metadata: dict | None = None, **kwargs) -> None:raise NotImplementedError

3.2 LangChain 集成抽象类

class LCChatMemoryComponent(Component):trace_type = "chat_memory"outputs = [Output(display_name="Memory",name="memory",method="build_message_history",)]def build_base_memory(self) -> BaseChatMemory:"""构建基础记忆对象"""return ConversationBufferMemory(chat_memory=self.build_message_history())@abstractmethoddef build_message_history(self) -> Memory:"""构建聊天消息历史记忆"""pass

4. 记忆存储和持久化机制

4.1 异步存储函数

async def aadd_messages(messages: Message | list[Message], flow_id: str | UUID | None = None):"""异步添加消息到记忆系统"""if not isinstance(messages, list):messages = [messages]if not all(isinstance(message, Message) for message in messages):types = ", ".join([str(type(message)) for message in messages])msg = f"The messages must be instances of Message. Found: {types}"raise ValueError(msg)try:messages_models = [MessageTable.from_message(msg, flow_id=flow_id) for msg in messages]async with session_scope() as session:messages_models = await aadd_messagetables(messages_models, session)return [await Message.create(**message.model_dump()) for message in messages_models]except Exception as e:logger.exception(e)raise

4.2 消息存储核心逻辑

async def aadd_messagetables(messages: list[MessageTable], session: AsyncSession):try:try:for message in messages:session.add(message)await session.commit()except asyncio.CancelledError:await session.rollback()return await aadd_messagetables(messages, session)for message in messages:await session.refresh(message)except asyncio.CancelledError as e:logger.exception(e)error_msg = "Operation cancelled"raise ValueError(error_msg) from eexcept Exception as e:logger.exception(e)raise# 处理 JSON 字段反序列化new_messages = []for msg in messages:msg.properties = json.loads(msg.properties) if isinstance(msg.properties, str) else msg.propertiesmsg.content_blocks = [json.loads(j) if isinstance(j, str) else j for j in msg.content_blocks]msg.category = msg.category or ""new_messages.append(msg)return [MessageRead.model_validate(message, from_attributes=True) for message in new_messages]

4.3 数据一致性保障

事务管理：使用数据库事务确保操作原子性
异常处理：完善的异常捕获和回滚机制
并发控制：通过 asyncio.CancelledError 处理并发取消
数据验证：Pydantic 模型确保数据类型安全

5. 记忆检索和查询系统

5.1 查询构建器

def _get_variable_query(sender: str | None = None,sender_name: str | None = None,session_id: str | UUID | None = None,order_by: str | None = "timestamp",order: str | None = "DESC",flow_id: UUID | None = None,limit: int | None = None,
):stmt = select(MessageTable).where(MessageTable.error == False)if sender:stmt = stmt.where(MessageTable.sender == sender)if sender_name:stmt = stmt.where(MessageTable.sender_name == sender_name)if session_id:stmt = stmt.where(MessageTable.session_id == session_id)if flow_id:stmt = stmt.where(MessageTable.flow_id == flow_id)if order_by:col = getattr(MessageTable, order_by).desc() if order == "DESC" else getattr(MessageTable, order_by).asc()stmt = stmt.order_by(col)if limit:stmt = stmt.limit(limit)return stmt

5.2 异步检索实现

async def aget_messages(sender: str | None = None,sender_name: str | None = None,session_id: str | UUID | None = None,order_by: str | None = "timestamp",order: str | None = "DESC",flow_id: UUID | None = None,limit: int | None = None,
) -> list[Message]:"""异步检索消息"""async with session_scope() as session:stmt = _get_variable_query(sender, sender_name, session_id, order_by, order, flow_id, limit)messages = await session.exec(stmt)return [await Message.create(**d.model_dump()) for d in messages]

5.3 查询优化策略

索引优化：在 session_id、timestamp、sender 等字段建立索引
分页查询：通过 limit 参数控制查询结果数量
条件过滤：支持多维度条件组合查询
排序控制：支持升序和降序排列

6. 上下文管理和会话状态

6.1 会话隔离机制

class Message(Data):session_id: str | UUID | None = Field(default="")sender: str | None = Nonesender_name: str | None = Nonetimestamp: Annotated[str, timestamp_to_str_validator] = Field(default_factory=lambda: datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S %Z"))flow_id: str | UUID | None = None

6.2 上下文状态管理

会话标识：通过 session_id 实现会话级别的数据隔离
时间戳管理：自动记录消息创建和更新时间
发送者追踪：区分 AI 和用户消息来源
流程关联：通过 flow_id 关联特定的工作流

6.3 状态持久化

async def astore_message(message: Message,flow_id: str | UUID | None = None,
) -> list[Message]:"""存储消息到记忆系统"""if not message:logger.warning("No message provided.")return []if not message.session_id or not message.sender or not message.sender_name:msg = (f"All of session_id, sender, and sender_name must be provided. "f"Session ID: {message.session_id}, Sender: {message.sender}, "f"Sender Name: {message.sender_name}")raise ValueError(msg)if hasattr(message, "id") and message.id:try:return await aupdate_messages([message])except ValueError as e:logger.error(e)if flow_id and not isinstance(flow_id, UUID):flow_id = UUID(flow_id)return await aadd_messages([message], flow_id=flow_id)

7. 记忆压缩和优化策略

7.1 消息更新机制

async def aupdate_messages(messages: Message | list[Message]) -> list[Message]:if not isinstance(messages, list):messages = [messages]async with session_scope() as session:updated_messages: list[MessageTable] = []for message in messages:msg = await session.get(MessageTable, message.id)if msg:msg = msg.sqlmodel_update(message.model_dump(exclude_unset=True, exclude_none=True))# 转换 flow_id 为 UUID 防止保存错误if msg.flow_id and isinstance(msg.flow_id, str):msg.flow_id = UUID(msg.flow_id)session.add(msg)await session.commit()await session.refresh(msg)updated_messages.append(msg)else:error_message = f"Message with id {message.id} not found"logger.warning(error_message)raise ValueError(error_message)return [MessageRead.model_validate(message, from_attributes=True) for message in updated_messages]

7.2 数据清理策略

async def adelete_messages(session_id: str) -> None:"""删除指定会话的所有消息"""async with session_scope() as session:stmt = (delete(MessageTable).where(col(MessageTable.session_id) == session_id).execution_options(synchronize_session="fetch"))await session.exec(stmt)async def delete_message(id_: str) -> None:"""删除单个消息"""async with session_scope() as session:message = await session.get(MessageTable, id_)if message:await session.delete(message)await session.commit()

7.3 优化策略

批量操作：支持批量添加、更新和删除消息
增量更新：只更新变更的字段，减少数据传输
定期清理：提供会话级别的数据清理功能
索引优化：在关键字段建立数据库索引

8. 多类型记忆支持

8.1 内置记忆类型

聊天记忆组件

class MemoryComponent(Component):display_name = "聊天历史"description = "获取聊天历史。"mode_config = {"Store": ["message", "memory", "sender", "sender_name", "session_id"],"Retrieve": ["n_messages", "order", "template", "memory"],}async def store_message(self) -> Message:"""存储消息模式"""message = Message(text=self.message) if isinstance(self.message, str) else self.messagemessage.session_id = self.session_id or message.session_idmessage.sender = self.sender or message.sender or MESSAGE_SENDER_AImessage.sender_name = self.sender_name or message.sender_name or MESSAGE_SENDER_NAME_AIif self.memory:self.memory.session_id = message.session_idlc_message = message.to_lc_message()await self.memory.aadd_messages([lc_message])stored_messages = await self.memory.aget_messages() or []stored_messages = [Message.from_lc_message(m) for m in stored_messages] if stored_messages else []else:await astore_message(message, flow_id=self.graph.flow_id)stored_messages = await aget_messages(session_id=message.session_id, sender_name=message.sender_name, sender=message.sender) or []return stored_messages[0] if stored_messages else Noneasync def retrieve_messages(self) -> Data:"""检索消息模式"""sender_type = self.sender_typeif sender_type == "Machine and User":sender_type = Noneif self.memory:self.memory.session_id = self.session_idstored = await self.memory.aget_messages()if self.order == "Descending":stored = stored[::-1]if self.n_messages:stored = stored[-self.n_messages:] if self.order == "ASC" else stored[:self.n_messages]stored = [Message.from_lc_message(m) for m in stored]else:stored = await aget_messages(sender=sender_type,sender_name=self.sender_name,session_id=self.session_id,limit=self.n_messages,order="DESC" if self.order == "Descending" else "ASC",)return cast(Data, stored)

Mem0 外部记忆集成

class Mem0MemoryComponent(LCChatMemoryComponent):display_name = "Mem0 Chat Memory"description = "Retrieves and stores chat messages using Mem0 memory storage."def build_mem0(self) -> Memory:"""初始化 Mem0 记忆实例"""if self.openai_api_key:os.environ["OPENAI_API_KEY"] = self.openai_api_keytry:if not self.mem0_api_key:return Memory.from_config(config_dict=dict(self.mem0_config)) if self.mem0_config else Memory()if self.mem0_config:return MemoryClient.from_config(api_key=self.mem0_api_key, config_dict=dict(self.mem0_config))return MemoryClient(api_key=self.mem0_api_key)except ImportError as e:msg = "Mem0 is not properly installed. Please install it with 'pip install -U mem0ai'."raise ImportError(msg) from edef ingest_data(self) -> Memory:"""摄取数据到 Mem0 记忆"""mem0_memory = self.existing_memory or self.build_mem0()if not self.ingest_message or not self.user_id:logger.warning("Missing 'ingest_message' or 'user_id'; cannot ingest data.")return mem0_memorymetadata = self.metadata or {}logger.info("Ingesting message for user_id: %s", self.user_id)try:mem0_memory.add(self.ingest_message, user_id=self.user_id, metadata=metadata)except Exception:logger.exception("Failed to add message to Mem0 memory.")raisereturn mem0_memorydef build_search_results(self) -> Data:"""搜索相关记忆"""mem0_memory = self.ingest_data()search_query = self.search_queryuser_id = self.user_idtry:if search_query:related_memories = mem0_memory.search(query=search_query, user_id=user_id)else:related_memories = mem0_memory.get_all(user_id=user_id)except Exception:logger.exception("Failed to retrieve related memories from Mem0.")raisereturn related_memories

8.2 记忆类型特点

短期记忆：基于会话的临时存储，会话结束后可清理
长期记忆：持久化存储，跨会话保持数据
工作记忆：当前对话上下文的活跃记忆
语义记忆：通过 Mem0 等服务实现的语义搜索记忆

9. 错误处理和数据一致性

9.1 异常处理机制

class ErrorMessage(Message):"""专门用于错误消息的消息类"""def __init__(self,exception: BaseException,session_id: str | None = None,source: Source | None = None,trace_name: str | None = None,flow_id: UUID | str | None = None,) -> None:if exception.__class__.__name__ == "ExceptionWithMessageError" and exception.__cause__ is not None:exception = exception.__cause__plain_reason = self._format_plain_reason(exception)markdown_reason = self._format_markdown_reason(exception)super().__init__(session_id=session_id,sender=source.display_name if source else None,sender_name=source.display_name if source else None,text=plain_reason,properties=Properties(text_color="red",background_color="red",edited=False,source=source,icon="error",allow_markdown=False,targets=[],),category="error",error=True,content_blocks=[ContentBlock(title="Error",contents=[ErrorContent(type="error",component=source.display_name if source else None,field=str(exception.field) if hasattr(exception, "field") else None,reason=markdown_reason,solution=str(exception.solution) if hasattr(exception, "solution") else None,traceback=traceback.format_exc(),)],)],flow_id=flow_id,)

9.2 数据验证机制

@field_validator("flow_id", mode="before")
@classmethod
def validate_flow_id(cls, value):if value is None:return valueif isinstance(value, str):value = UUID(value)return value@field_validator("properties", "content_blocks", mode="before")
@classmethod
def validate_properties_or_content_blocks(cls, value):if isinstance(value, list):return [cls.validate_properties_or_content_blocks(item) for item in value]if hasattr(value, "model_dump"):return value.model_dump()if isinstance(value, str):return json.loads(value)return value

9.3 一致性保障策略

事务完整性：使用数据库事务确保操作原子性
数据验证：Pydantic 模型提供强类型验证
异常恢复：完善的异常捕获和恢复机制
日志记录：详细的操作日志便于问题追踪

10. 与 LangChain Memory 的集成

10.1 LangChain 兼容性

class LCBuiltinChatMemory(BaseChatMessageHistory):"""与 LangChain 兼容的内置聊天记忆"""def __init__(self, flow_id: str, session_id: str) -> None:self.flow_id = flow_idself.session_id = session_id@propertydef messages(self) -> list[BaseMessage]:messages = get_messages(session_id=self.session_id)return [m.to_lc_message() for m in messages if not m.error]async def aget_messages(self) -> list[BaseMessage]:messages = await aget_messages(session_id=self.session_id)return [m.to_lc_message() for m in messages if not m.error]def add_messages(self, messages: Sequence[BaseMessage]) -> None:for lc_message in messages:message = Message.from_lc_message(lc_message)message.session_id = self.session_idstore_message(message, flow_id=self.flow_id)async def aadd_messages(self, messages: Sequence[BaseMessage]) -> None:for lc_message in messages:message = Message.from_lc_message(lc_message)message.session_id = self.session_idawait astore_message(message, flow_id=self.flow_id)def clear(self) -> None:delete_messages(self.session_id)async def aclear(self) -> None:await adelete_messages(self.session_id)

10.2 消息格式转换

def to_lc_message(self) -> BaseMessage:"""转换为 LangChain BaseMessage"""if self.text is None or not self.sender:logger.warning("Missing required keys ('text', 'sender') in Message, defaulting to HumanMessage.")text = "" if not isinstance(self.text, str) else self.textif self.sender == MESSAGE_SENDER_USER or not self.sender:if self.files:contents = [{"type": "text", "text": text}]contents.extend(self.get_file_content_dicts())human_message = HumanMessage(content=contents)else:human_message = HumanMessage(content=text)return human_messagereturn AIMessage(content=text)@classmethod
def from_lc_message(cls, lc_message: BaseMessage) -> Message:"""从 LangChain BaseMessage 创建 Message"""if lc_message.type == "human":sender = MESSAGE_SENDER_USERsender_name = MESSAGE_SENDER_NAME_USERelif lc_message.type == "ai":sender = MESSAGE_SENDER_AIsender_name = MESSAGE_SENDER_NAME_AIelif lc_message.type == "system":sender = "System"sender_name = "System"else:sender = lc_message.typesender_name = lc_message.typereturn cls(text=lc_message.content, sender=sender, sender_name=sender_name)

11. 内置记忆类型分析

11.1 基础聊天记忆

功能：存储和检索基本的对话历史
特点：支持会话隔离、发送者过滤、时间排序
适用场景：简单的问答对话、客服系统

11.2 Mem0 智能记忆

功能：基于语义的智能记忆存储和检索
特点：支持语义搜索、用户个性化、元数据关联
适用场景：个性化助手、知识管理系统

11.3 消息存储组件

功能：专门用于消息存储的轻量级组件
特点：简化的接口、快速存储、批量操作
适用场景：日志记录、消息归档

12. 自定义记忆组件开发

12.1 开发指南

class CustomMemoryComponent(BaseMemoryComponent):display_name = "Custom Memory"description = "Custom memory implementation"def build_config(self):config = super().build_config()# 添加自定义配置config.update({"custom_param": {"display_name": "Custom Parameter","info": "Custom parameter description",}})return configdef get_messages(self, **kwargs) -> list[Data]:# 实现自定义消息检索逻辑passdef add_message(self, sender: str, sender_name: str, text: str, session_id: str, metadata: dict | None = None, **kwargs) -> None:# 实现自定义消息添加逻辑pass

12.2 扩展接口

存储后端扩展：支持 Redis、MongoDB 等不同存储后端
检索算法扩展：支持向量搜索、全文搜索等检索方式
压缩策略扩展：支持自定义的记忆压缩和清理策略
格式转换扩展：支持不同的消息格式和协议

13. 应用示例

13.1 基础聊天机器人记忆系统

from langflow.components.helpers.memory import MemoryComponent
from langflow.schema.message import Messageclass ChatbotMemoryExample:def __init__(self, session_id: str):self.session_id = session_idself.memory = MemoryComponent()self.memory.session_id = session_idself.memory.mode = "Retrieve"self.memory.n_messages = 10self.memory.order = "Ascending"async def get_conversation_history(self) -> list[Message]:"""获取对话历史"""try:# 检索最近10条消息messages_data = await self.memory.retrieve_messages()return messages_data if isinstance(messages_data, list) else []except Exception as e:print(f"Error retrieving messages: {e}")return []async def add_user_message(self, text: str) -> Message:"""添加用户消息"""message = Message(text=text,sender="User",sender_name="User",session_id=self.session_id)# 切换到存储模式self.memory.mode = "Store"self.memory.message = messagetry:stored_message = await self.memory.store_message()return stored_messageexcept Exception as e:print(f"Error storing user message: {e}")return messageasync def add_ai_response(self, text: str) -> Message:"""添加AI响应"""message = Message(text=text,sender="Machine",sender_name="AI",session_id=self.session_id)self.memory.mode = "Store"self.memory.message = messagetry:stored_message = await self.memory.store_message()return stored_messageexcept Exception as e:print(f"Error storing AI message: {e}")return messageasync def format_conversation_context(self) -> str:"""格式化对话上下文"""messages = await self.get_conversation_history()if not messages:return "No conversation history available."context_lines = []for msg in messages[-5:]:  # 只取最近5条消息if hasattr(msg, 'sender_name') and hasattr(msg, 'text'):context_lines.append(f"{msg.sender_name}: {msg.text}")return "\n".join(context_lines)# 使用示例
async def chatbot_example():# 创建聊天机器人记忆实例chatbot_memory = ChatbotMemoryExample("user_session_123")# 添加用户消息user_msg = await chatbot_memory.add_user_message("Hello, how are you?")print(f"User message stored: {user_msg.text}")# 添加AI响应ai_msg = await chatbot_memory.add_ai_response("Hello! I'm doing well, thank you for asking. How can I help you today?")print(f"AI response stored: {ai_msg.text}")# 获取对话上下文context = await chatbot_memory.format_conversation_context()print(f"Conversation context:\n{context}")# 继续对话await chatbot_memory.add_user_message("Can you help me with Python programming?")await chatbot_memory.add_ai_response("Of course! I'd be happy to help you with Python programming. What specific topic or problem would you like assistance with?")# 获取更新后的对话历史history = await chatbot_memory.get_conversation_history()print(f"Total messages in history: {len(history)}")

13.2 智能个人助手记忆系统

from langflow.components.mem0.mem0_chat_memory import Mem0MemoryComponent
from langflow.schema.message import Messageclass PersonalAssistantMemory:def __init__(self, user_id: str, mem0_api_key: str = None):self.user_id = user_idself.mem0_component = Mem0MemoryComponent()self.mem0_component.user_id = user_idself.mem0_component.mem0_api_key = mem0_api_key# 配置 Mem0 设置self.mem0_component.mem0_config = {"graph_store": {"provider": "neo4j","config": {"url": "neo4j+s://your-neo4j-url","username": "neo4j","password": "your-password"}},"version": "v1.1"}async def learn_user_preference(self, preference_text: str, category: str = "general") -> None:"""学习用户偏好"""self.mem0_component.ingest_message = preference_textself.mem0_component.metadata = {"category": category,"type": "preference","timestamp": datetime.now().isoformat()}try:memory = self.mem0_component.ingest_data()print(f"Learned user preference: {preference_text}")except Exception as e:print(f"Error learning preference: {e}")async def recall_related_memories(self, query: str) -> list:"""回忆相关记忆"""self.mem0_component.search_query = querytry:related_memories = self.mem0_component.build_search_results()return related_memories if isinstance(related_memories, list) else []except Exception as e:print(f"Error recalling memories: {e}")return []async def get_user_context(self, current_topic: str) -> dict:"""获取用户上下文信息"""# 搜索相关记忆related_memories = await self.recall_related_memories(current_topic)# 获取用户偏好preferences = await self.recall_related_memories("preference")return {"related_memories": related_memories,"user_preferences": preferences,"user_id": self.user_id,"current_topic": current_topic}async def personalized_response(self, user_query: str) -> str:"""基于记忆生成个性化响应"""# 获取用户上下文context = await self.get_user_context(user_query)# 构建个性化响应（这里简化处理）response_parts = [f"Based on what I know about you, {self.user_id}:"]if context["user_preferences"]:response_parts.append("I remember your preferences include:")for pref in context["user_preferences"][:3]:  # 只显示前3个偏好if hasattr(pref, 'text'):response_parts.append(f"- {pref.text}")if context["related_memories"]:response_parts.append("Related to your query, I recall:")for memory in context["related_memories"][:2]:  # 只显示前2个相关记忆if hasattr(memory, 'text'):response_parts.append(f"- {memory.text}")response_parts.append(f"Now, regarding '{user_query}', let me help you...")return "\n".join(response_parts)# 使用示例
async def personal_assistant_example():# 创建个人助手记忆实例assistant = PersonalAssistantMemory("alice_user_456")# 学习用户偏好await assistant.learn_user_preference("I prefer vegetarian food", "food")await assistant.learn_user_preference("I like morning workouts", "fitness")await assistant.learn_user_preference("I work in software development", "career")await assistant.learn_user_preference("I enjoy reading science fiction books", "hobbies")# 模拟用户查询user_queries = ["Can you recommend a restaurant?","What's a good exercise routine?","I need help with my career development","What book should I read next?"]for query in user_queries:print(f"\nUser Query: {query}")response = await assistant.personalized_response(query)print(f"Assistant Response:\n{response}")print("-" * 50)# 展示记忆检索功能print("\nRecalling memories about 'food':")food_memories = await assistant.recall_related_memories("food")for memory in food_memories:if hasattr(memory, 'text'):print(f"- {memory.text}")# 运行示例
if __name__ == "__main__":import asyncio# 运行聊天机器人示例print("=== Chatbot Memory Example ===")asyncio.run(chatbot_example())print("\n" + "="*60 + "\n")# 运行个人助手示例print("=== Personal Assistant Memory Example ===")asyncio.run(personal_assistant_example())