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:
"""清理内存实现所使用的所有资源。"""
...