autogen_core.memory._base_memory 源代码

from abc import ABC, abstractmethod
from enum import Enum
from typing import Any, Dict, List, Union

from pydantic import BaseModel, ConfigDict, field_serializer

from .._cancellation_token import CancellationToken
from .._component_config import ComponentBase
from .._image import Image
from ..model_context import ChatCompletionContext


[文档] class MemoryMimeType(Enum): """支持的内存内容 MIME 类型。""" TEXT = "text/plain" JSON = "application/json" MARKDOWN = "text/markdown" IMAGE = "image/*" BINARY = "application/octet-stream"
ContentType = Union[str, bytes, Dict[str, Any], Image]
[文档] class MemoryContent(BaseModel): """一个内存内容项。""" content: ContentType """内存项的内容。可以是字符串、字节、字典或 :class:`~autogen_core.Image`。""" mime_type: MemoryMimeType | str """内存内容的 MIME 类型。""" metadata: Dict[str, Any] | None = None """与内存项关联的元数据。""" model_config = ConfigDict(arbitrary_types_allowed=True)
[文档] @field_serializer("mime_type") def serialize_mime_type(self, mime_type: MemoryMimeType | str) -> str: """将 MIME 类型序列化为字符串。""" if isinstance(mime_type, MemoryMimeType): return mime_type.value return mime_type
[文档] class MemoryQueryResult(BaseModel): """内存 :meth:`~autogen_core.memory.Memory.query` 操作的返回结果。""" results: List[MemoryContent]
[文档] class UpdateContextResult(BaseModel): """内存 :meth:`~autogen_core.memory.Memory.update_context` 操作的结果。""" memories: MemoryQueryResult
[文档] class Memory(ABC, ComponentBase[BaseModel]): """定义内存实现接口的协议。 内存是用于存储数据的容器,可用于丰富或修改模型上下文。 内存实现可以使用任何存储机制,例如列表、数据库或文件系统。 也可以使用任何检索机制,例如向量搜索或文本搜索。 具体如何存储和检索数据由实现决定。 内存实现还需负责根据当前模型上下文查询内存存储, 并用相关内存内容更新模型上下文。 示例实现可参考 :class:`~autogen_core.memory.ListMemory`。 """ component_type = "memory"
[文档] @abstractmethod async def update_context( self, model_context: ChatCompletionContext, ) -> UpdateContextResult: """ 使用相关记忆内容更新提供的模型上下文。 Args: model_context: 待更新的上下文 Returns: 包含相关记忆的UpdateContextResult """ ...
[文档] @abstractmethod async def query( self, query: str | MemoryContent, cancellation_token: CancellationToken | None = None, **kwargs: Any, ) -> MemoryQueryResult: """ 查询记忆存储并返回相关条目。 Args: query: 查询内容项 cancellation_token: 可选的取消操作令牌 **kwargs: 特定实现的额外参数 Returns: 包含带有相关性分数的记忆条目的MemoryQueryResult """ ...
[文档] @abstractmethod async def add(self, content: MemoryContent, cancellation_token: CancellationToken | None = None) -> None: """ 向记忆中添加新内容。 Args: content: 要添加的记忆内容 cancellation_token: 可选的取消操作令牌 """ ...
[文档] @abstractmethod async def clear(self) -> None: """清除内存中的所有条目。""" ...
[文档] @abstractmethod async def close(self) -> None: """清理内存实现所使用的所有资源。""" ...