短期记忆 - AG-Kit Documentation

AG-Kit 中的短期记忆管理活动会话中的对话历史和上下文。它提供高效的存储和检索最近消息的功能,使 Agent 能够在具备上下文感知的情况下保持连贯的对话。

概述

短期记忆(也称为会话记忆或对话记忆)存储:

最近的对话消息 - 用户和助手的交流
工具调用历史 - 工具调用和结果的记录
会话状态 - 自定义元数据和上下文信息
时间上下文 - 基于时间的消息排序和过滤

记忆实现

AG-Kit 为不同的用例和部署场景提供多种短期记忆实现:

InMemoryMemory

易失性内存存储,适用于开发、测试和单实例部署。 特性:

快速读写操作
基于内容的相似度搜索
令牌感知的消息修剪
多会话支持
零外部依赖

用例:

开发和测试
单服务器应用
临时会话
原型开发

TDAIMemory

通过 TDAI 服务提供生产级可扩展性的云端持久化存储。 特性:

持久化云存储
高级语义搜索
分布式会话管理
可选的本地缓存
生产就绪的可靠性

用例:

生产环境部署
多服务器应用
长期运行的会话
企业应用

TypeORMMemory

支持多种数据库并具有可自定义 schema 的灵活 ORM 存储。 特性:

多数据库支持(MySQL、PostgreSQL、SQLite 等)
自定义实体定义
文档转换系统
分支和摘要架构
TypeScript 类型安全
迁移支持

用例:

自定义数据库 schema
企业数据库集成
复杂数据关系
类型安全开发

MySQLMemory

扩展 TypeORMMemory 并具有 MySQL 特定功能的优化 MySQL 实现。 特性:

MySQL 特定优化
连接池
事务支持
索引优化
性能监控

用例:

基于 MySQL 的应用
高性能需求
企业 MySQL 部署

MongoDBMemory

具有灵活 schema 和水平扩展能力的 NoSQL 文档存储。 特性:

基于文档的存储
灵活的 schema 设计
自动连接设置 - 无需手动创建客户端
连接字符串配置
水平扩展
聚合管道
GridFS 支持大数据
连接池和选项

用例:

NoSQL 应用
灵活的数据结构
水平扩展需求
面向文档的工作流
最小设置的快速原型开发

CloudBaseMemory

腾讯云 CloudBase 集成,用于无服务器云存储。 特性:

无服务器架构
自动扩展
腾讯云集成
实时同步
内置安全性

用例:

腾讯云部署
无服务器应用
中国市场应用
自动扩展需求

快速入门

使用 InMemoryMemory 的基本用法

import { InMemoryMemory } from '@ag-kit/agents';

// 创建记忆实例
const memory = new InMemoryMemory();

// 添加对话事件
await memory.add({
  message: {
    id: 'msg-1',
    role: 'user',
    content: 'What is the weather today?',
    timestamp: new Date()
  },
  state: { userId: 'user-123', location: 'San Francisco' }
});

// 添加助手响应
await memory.add({
  message: {
    id: 'msg-2',
    role: 'assistant',
    content: 'The weather in San Francisco is sunny, 72°F.',
    timestamp: new Date()
  },
  state: { userId: 'user-123' }
});

// 检索对话历史
const events = await memory.list({ limit: 10 });
console.log(events);
// [
//   { message: { id: 'msg-1', role: 'user', content: '...' }, state: {...} },
//   { message: { id: 'msg-2', role: 'assistant', content: '...' }, state: {...} }
// ]

在生产环境使用 TDAIMemory

import { TDAIMemory } from '@ag-kit/agents';

// 创建 TDAI 记忆实例
const memory = new TDAIMemory({
  sessionId: 'user-session-123',
  clientOptions: {
    apiKey: process.env.TDAI_API_KEY!,
    endpoint: 'https://api.tdai.example.com'
  },
  useCache: true // 启用本地缓存以提高性能
});

// 添加对话事件
await memory.add({
  message: {
    id: 'msg-1',
    role: 'user',
    content: 'Tell me about AG-Kit',
    timestamp: new Date()
  },
  state: { source: 'web-chat' }
});

// 列出最近的消息
const recentMessages = await memory.list({
  limit: 20,
  order: 'desc' // 最新的在前
});

使用带有自定义 Schema 的 TypeORMMemory

from agkit.agents import TypeORMMemory
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 创建数据库连接
engine = create_engine('mysql://agkit_user:password@localhost:3306/agkit_memory')
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建 TypeORM 记忆实例
memory = TypeORMMemory(
    engine=engine,
    session_factory=SessionLocal,
    session_id='user-session-123',
    event_table_name='custom_memory_events',
    state_table_name='custom_memory_state',
    summary_table_name='custom_memory_summaries',
    enable_context_management=True
)

# 使用分支和摘要支持
await memory.add({
    'message': {
        'id': 'msg-1',
        'role': 'user',
        'content': 'Hello with custom schema',
        'timestamp': datetime.now()
    },
    'state': {'custom_field': 'value'}
})

使用 MongoDBMemory

from agkit.agents import MongoDBMemory

# 推荐:直接配置(无需手动设置客户端)
memory = MongoDBMemory(
    connection_string='mongodb://localhost:27017',
    database_name='agkit_memory',
    session_id='user-session-123',
    collection_name='memory_events',
    state_collection_name='memory_state',
    summary_collection_name='memory_summaries',
    enable_context_management=True
)

# 高级配置,带连接选项
memory2 = MongoDBMemory(
    connection_string='mongodb://username:password@localhost:27017',
    database_name='agkit_memory',
    session_id='user-session-456',
    client_options={
        'maxPoolSize': 10,
        'serverSelectionTimeoutMS': 5000,
        'socketTimeoutMS': 45000,
    },
    enable_context_management=True
)

# 替代方案:使用现有的 MongoDB 连接(如果需要)
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017')
db = client.agkit_memory

memory3 = MongoDBMemory(
    db=db,
    session_id='user-session-789',
    collection_name='memory_events'
)

# 使用灵活的文档结构添加事件
await memory.add({
    'message': {
        'id': 'msg-1',
        'role': 'user',
        'content': 'MongoDB flexible storage',
        'timestamp': datetime.now(),
        'metadata': {
            'source': 'mobile-app',
            'version': '1.2.0'
        }
    },
    'state': {
        'user_id': 'user-123',
        'preferences': {
            'theme': 'dark',
            'language': 'en'
        }
    }
})

使用 CloudBaseMemory

# CloudBase 集成主要在 TypeScript 中可用

统一架构

分支和摘要系统

现在所有记忆实现都支持统一的分支和摘要架构,能够实现:

对话分支: 创建用于实验的替代对话路径
自动摘要: 压缩长对话同时保留上下文
上下文工程: 智能令牌管理和上下文窗口优化(了解更多)
状态管理: 跨分支和摘要的持久化会话状态

# 从特定事件创建新分支
branch_path = await memory.create_branch('experiment-1', 'msg-10')

# 切换到分支(checkout)
await memory.checkout('experiment-1')

# 向当前分支添加消息
await memory.add({
    'message': {
        'id': 'msg-11-alt',
        'role': 'assistant',
        'content': 'Alternative response for testing',
        'timestamp': datetime.now()
    }
})

# 列出所有分支及元数据
branches = await memory.list_branches()
print(branches)

# 切换回主分支
await memory.checkout('main')

自定义实体和文档转换

基于 TypeORM 的实现支持自定义实体定义和文档转换,以实现灵活的 schema 设计:

from sqlalchemy import Column, String, JSON, Integer, DateTime
from sqlalchemy.ext.declarative import declarative_base
from agkit.agents import TypeORMMemory

Base = declarative_base()

# 定义自定义实体
class CustomMemoryEvent(Base):
    __tablename__ = 'custom_memory_events'
    
    id = Column(String, primary_key=True)
    session_id = Column(String, nullable=False)
    message = Column(JSON, nullable=False)
    state = Column(JSON, nullable=False)
    branch_path = Column(String, nullable=False)
    created_at = Column(DateTime, nullable=False)
    
    # 自定义字段
    priority = Column(Integer, default=0)
    tags = Column(JSON, default=list)
    category = Column(String, default='general')

# 自定义文档转换器
class CustomDocumentConverter:
    def to_document(self, event, session_id, branch_path):
        return {
            'session_id': session_id,
            'message': event['message'],
            'state': event['state'],
            'branch_path': branch_path,
            'created_at': datetime.now(),
            # 自定义字段
            'priority': event['state'].get('priority', 0),
            'tags': event['state'].get('tags', []),
            'category': event['state'].get('category', 'general')
        }
    
    def from_document(self, doc):
        return {
            'message': doc['message'],
            'state': {
                **doc['state'],
                'priority': doc['priority'],
                'tags': doc['tags'],
                'category': doc['category']
            }
        }

# 使用自定义实体和转换器
memory = TypeORMMemory(
    engine=engine,
    session_id='user-123',
    custom_entity=CustomMemoryEvent,
    document_converter=CustomDocumentConverter()
)

核心操作

添加事件

向记忆中添加单个或多个对话事件。

# 添加单个事件
await memory.add({
    'message': {
        'id': 'msg-1',
        'role': 'user',
        'content': 'Hello, how are you?',
        'timestamp': datetime.now()
    },
    'state': {'mood': 'friendly'}
})

# 高效添加多个事件
await memory.add_list([
    {
        'message': {'id': 'msg-1', 'role': 'user', 'content': 'First message'},
        'state': {}
    },
    {
        'message': {'id': 'msg-2', 'role': 'assistant', 'content': 'Response'},
        'state': {}
    }
])

列出事件

使用过滤、分页和令牌限制检索事件。

# 获取所有事件
all_events = await memory.list()

# 获取最近 10 个事件
recent_events = await memory.list(limit=10)

# 使用分页获取事件
page2 = await memory.list(limit=10, offset=10)

# 按降序获取事件(最新的在前)
newest_first = await memory.list(order='desc')

# 按令牌数量限制(对 LLM 上下文窗口有用)
token_limited = await memory.list(max_tokens=2000)
# 返回适合 2000 令牌的事件

搜索事件

基于内容相似度搜索事件。

# 搜索包含特定内容的事件
results = await memory.retrieve('weather forecast')
# 返回按相关性得分排序的事件

# InMemoryMemory 使用基于内容的相似度评分
# TDAIMemory 使用语义搜索功能

删除事件

从记忆中删除特定事件。

# 按消息 ID 删除
await memory.delete('msg-1')

# 按索引删除(仅 InMemoryMemory)
await memory.delete(0)  # 删除第一个事件

# 注意: TDAIMemory 仅支持按 ID 删除

清空记忆

从存储中删除所有事件。

# 清除所有事件
await memory.clear()

# 检查记忆是否为空
is_empty = await memory.is_empty()  # True

# 获取事件数量
count = await memory.get_count()  # 0

多会话支持

两种记忆实现都支持多个隔离的会话。

memory = InMemoryMemory()

# 向不同会话添加事件
await memory.add(
    {'message': {'id': '1', 'role': 'user', 'content': 'Hello'}, 'state': {}},
    session_id='session-A'
)

await memory.add(
    {'message': {'id': '2', 'role': 'user', 'content': 'Hi there'}, 'state': {}},
    session_id='session-B'
)

# 列出特定会话的事件
session_a_events = await memory.list(session_id='session-A')
session_b_events = await memory.list(session_id='session-B')

# 获取所有会话 ID(仅 InMemoryMemory)
session_ids = memory.get_session_ids()  # ['session-A', 'session-B']

# 检查会话是否存在(仅 InMemoryMemory)
has_session = memory.has_session('session-A')  # True

# 清除特定会话
await memory.clear(session_id='session-A')

令牌管理

AG-Kit 提供自动令牌计数和修剪以管理 LLM 上下文窗口。

from agkit.agents import InMemoryMemory, TiktokenTokenizer

# 使用默认分词器(Tiktoken)
memory = InMemoryMemory()

# 或提供自定义分词器
custom_tokenizer = TiktokenTokenizer('gpt-4')
memory_with_custom = InMemoryMemory(custom_tokenizer)

# 在令牌限制内检索事件
events = await memory.list(max_tokens=4000)
# 自动修剪最旧的消息以适应 4000 令牌

与 Agent 集成

短期记忆与 AI Agent 无缝集成,实现自动上下文管理。 了解更多: 完整的 Agent 集成指南

会话分支

会话分支支持对话实验和时间旅行功能。创建实验性对话路径,测试不同响应并回滚到以前的状态。

会话分支目前由 InMemoryMemory 支持。如果调用分支方法,其他实现将抛出错误。

创建分支

创建实验性对话路径而不丢失原始内容:

from agkit.agents import InMemoryMemory

memory = InMemoryMemory()

# 添加对话历史
await memory.add({
    'message': {'id': 'msg-1', 'role': 'user', 'content': 'Help me write an email'},
    'state': {}
})
await memory.add({
    'message': {'id': 'msg-2', 'role': 'assistant', 'content': 'I can help with that!'},
    'state': {}
})

# 创建分支
branch_id = await memory.branch('polite-version')
print(f'Created branch: {branch_id}')

# 切换到分支
await memory.checkout('polite-version')

# 添加实验性内容
await memory.add({
    'message': {
        'id': 'msg-3',
        'role': 'assistant',
        'content': 'I would be delighted to assist you with composing your email!'
    },
    'state': {}
})

# 列出分支
branches = await memory.list_branches()
print(branches)

使用事件检出进行时间旅行

检出到特定事件并删除其后的所有事件:

# 添加多条消息
await memory.add({'message': {'id': 'msg-1', 'role': 'user', 'content': 'Hello'}, 'state': {}})
await memory.add({'message': {'id': 'msg-2', 'role': 'assistant', 'content': 'Hi!'}, 'state': {}})
await memory.add({'message': {'id': 'msg-3', 'role': 'user', 'content': 'How are you?'}, 'state': {}})
await memory.add({'message': {'id': 'msg-4', 'role': 'assistant', 'content': 'Good!'}, 'state': {}})

# 检出到 msg-2(删除 msg-3 和 msg-4)
await memory.checkout('msg-2', type='event')

events = await memory.list()
print(len(events))  # 2(仅保留 msg-1 和 msg-2)

高级分支操作

# A/B 测试不同的响应
await memory.create_branch('friendly-agent', 'msg-5')
await memory.create_branch('professional-agent', 'msg-5')

# 测试友好版本
await memory.checkout('friendly-agent')
await memory.add({
    'message': {'id': 'resp-1', 'role': 'assistant', 'content': 'Hey! Sure thing!'},
    'state': {}
})

# 测试专业版本
await memory.checkout('professional-agent')
await memory.add({
    'message': {'id': 'resp-2', 'role': 'assistant', 'content': 'Certainly, I can assist.'},
    'state': {}
})

# 比较结果并选择更好的分支
friendly_branch = await memory.list_branches()
await memory.checkout('friendly-agent')  # 保留这个

# 撤销/回滚到特定事件
await memory.checkout('msg-3')  # 回滚到消息 ID

# 对话检查点
await memory.add(important_message1)
checkpoint1 = await memory.create_branch('checkpoint-1')

await memory.add(important_message2)
checkpoint2 = await memory.create_branch('checkpoint-2')

# 稍后返回到检查点
await memory.checkout('checkpoint-1')

高级模式

上下文窗口管理

使用智能阈值和令牌限制自动管理 LLM 上下文窗口。有关全面的上下文工程策略,请参阅上下文工程指南。

# 策略 1: 自动基于阈值的管理(即将推出)
# 注意: 高级阈值管理目前仅在 TypeScript 中可用
# Python 实现即将推出

# 策略 2: 固定令牌限制
memory = InMemoryMemory()
events = await memory.list(max_tokens=4000)

# 策略 3: 动态令牌分配
model_max_tokens = 8000
reserved_for_response = 2000
available_for_context = model_max_tokens - reserved_for_response
context_events = await memory.list(max_tokens=available_for_context)

# 策略 4: 带最近消息的滑动窗口
recent_events = await memory.list(
    limit=20,
    order='desc',
    max_tokens=3000
)

上下文管理策略:

自动阈值(推荐) - 具有可配置阈值的智能压缩
固定令牌限制 - 使用 maxTokens 的简单硬限制
动态令牌分配 - 基于模型容量的计算限制
滑动窗口 - 带令牌约束的最近消息

自动阈值的工作原理:

压缩阶段: 在 preRotThreshold 的 80% → 删除冗余/重复内容
摘要阶段: 在 preRotThreshold 的 95% → 将较旧的消息压缩成摘要
保留: 最近的消息(由 recentToKeep 指定)始终完整保留
互补性: 可与 maxTokens 一起使用以提供额外保护

有关详细的上下文工程模式和高级阈值策略,请参阅上下文工程文档。

工具调用历史

存储和检索工具调用历史。

# 添加工具调用事件
await memory.add({
    'message': {
        'id': 'msg-3',
        'role': 'assistant',
        'content': 'Let me check the weather for you.',
        'tool_calls': [{
            'id': 'call-1',
            'type': 'function',
            'function': {
                'name': 'get_weather',
                'arguments': '{"location": "San Francisco"}'
            }
        }]
    },
    'state': {}
})

# 添加工具结果事件
await memory.add({
    'message': {
        'id': 'msg-4',
        'role': 'tool',
        'content': '{"temperature": 72, "condition": "sunny"}',
        'tool_call_id': 'call-1'
    },
    'state': {}
})

自定义状态管理

为每个事件存储自定义元数据。

await memory.add({
    'message': {
        'id': 'msg-1',
        'role': 'user',
        'content': 'Book a flight to Tokyo',
    },
    'state': {
        'user_id': 'user-123',
        'intent': 'flight_booking',
        'entities': {
            'destination': 'Tokyo',
            'travel_class': 'economy'
        },
        'confidence': 0.95,
        'timestamp': datetime.now().isoformat()
    }
})

# 检索并访问自定义状态
events = await memory.list()
print(events[0]['state']['intent'])  # 'flight_booking'
print(events[0]['state']['entities'])  # {'destination': 'Tokyo', ...}

最佳实践

1. 选择正确的实现

使用 InMemoryMemory 用于开发、测试和单实例应用
使用其他实现 用于生产、分布式系统和持久化存储需求

2. 管理令牌限制

始终考虑 LLM 上下文窗口限制:

# 好:限制令牌以适应模型上下文
events = await memory.list(max_tokens=4000)

# 不好:无限制加载所有事件
events = await memory.list()  # 可能超过上下文窗口

3. 一致地使用会话 ID

为多用户应用维护会话隔离:

# 好:使用一致的会话 ID
session_id = f"user-{user_id}-{conversation_id}"
await memory.add(event, AddOptions(session_id=session_id))

# 不好:混合会话
await memory.add(event)  # 使用默认会话

4. 清理旧会话

定期清除不活跃的会话以管理内存:

# 对话结束后清除特定会话
await memory.clear(ClearOptions(session_id='session-123'))

# 或定期清除所有会话
await memory.clear()

5. 优雅地处理错误

try:
    await memory.add(event)
except Exception as error:
    print(f'Failed to add event: {error}')
    # 实现回退或重试逻辑

性能考虑

InMemoryMemory

快速: 所有操作都在内存中
可扩展: 高效处理数千个事件
限制: 进程重启时数据丢失
内存使用: 随事件数量增长

TDAIMemory

持久化: 数据在重启后保留
分布式: 跨多个服务器工作
缓存: 可选的本地缓存以提高性能
网络: 需要对 TDAI 服务进行网络调用

实现对比

特性	InMemory	TDAI	TypeORM	MySQL	MongoDB	CloudBase
存储类型	易失性	云端	数据库	数据库	NoSQL	无服务器
持久化	❌	✅	✅	✅	✅	✅
性能	最快	快速	快速	最快	快速	良好
可扩展性	单一	高	中等	高	很高	自动
自定义 Schema	❌	❌	✅	✅	✅	❌
分支支持	✅	✅	✅	✅	✅	✅
摘要支持	✅	✅	✅	✅	✅	✅
多数据库	❌	❌	✅	❌	❌	❌
设置复杂度	无	API Key	中等	中等	很低	中等
最适合	开发/测试	生产	企业	MySQL 应用	NoSQL 应用	腾讯云

下一步

长期记忆

了解使用向量存储的持久化记忆

会话管理

管理多用户会话和上下文

Agent 集成

记忆-Agent 集成的完整指南

API 参考

详细的 API 文档

快速开始

构建 Agent

部署 Agent

构建 Agent 应用

开放协议

社区

​概述

​记忆实现

​InMemoryMemory

​TDAIMemory

​TypeORMMemory

​MySQLMemory

​MongoDBMemory

​CloudBaseMemory

​快速入门

​使用 InMemoryMemory 的基本用法

​在生产环境使用 TDAIMemory

​使用带有自定义 Schema 的 TypeORMMemory

​使用 MongoDBMemory

​使用 CloudBaseMemory

​统一架构

​分支和摘要系统

​自定义实体和文档转换

​核心操作

​添加事件

​列出事件

​搜索事件

​删除事件

​清空记忆

​多会话支持

​令牌管理

​与 Agent 集成

​会话分支

​创建分支

​使用事件检出进行时间旅行

​高级分支操作

​高级模式

​上下文窗口管理

​工具调用历史

​自定义状态管理

​最佳实践

​1. 选择正确的实现

​2. 管理令牌限制

​3. 一致地使用会话 ID

​4. 清理旧会话

​5. 优雅地处理错误

​性能考虑

​InMemoryMemory

​TDAIMemory

​实现对比

​下一步

长期记忆

会话管理

Agent 集成

API 参考

概述

记忆实现

InMemoryMemory

TDAIMemory

TypeORMMemory

MySQLMemory

MongoDBMemory

CloudBaseMemory

快速入门

使用 InMemoryMemory 的基本用法

在生产环境使用 TDAIMemory

使用带有自定义 Schema 的 TypeORMMemory

使用 MongoDBMemory

使用 CloudBaseMemory

统一架构

分支和摘要系统

自定义实体和文档转换

核心操作

添加事件

列出事件

搜索事件

删除事件

清空记忆

多会话支持

令牌管理

与 Agent 集成

会话分支

创建分支

使用事件检出进行时间旅行

高级分支操作

高级模式

上下文窗口管理

工具调用历史

自定义状态管理

最佳实践

1. 选择正确的实现

2. 管理令牌限制

3. 一致地使用会话 ID

4. 清理旧会话

5. 优雅地处理错误

性能考虑

InMemoryMemory

TDAIMemory

实现对比

下一步