qwen_agent/docs/agent_memory_cache.md

Agent 内存缓存系统

本文档描述了基于 cachetools 实现的 Agent 内存缓存系统。

概述

Agent 内存缓存系统使用 cachetools 库在内存中缓存 Agent 实例，提供以下功能：

• ✅ 高性能访问：基于内存存储，访问速度极快（纳秒级）
• ✅ 自动过期：支持设置 TTL（Time To Live），自动清理过期项
• ✅ LRU 淘汰策略：当缓存满时，自动淘汰最久未使用的项
• ✅ 访问时自动续期：每次访问时自动延长过期时间
• ✅ 线程安全：使用 RLock 确保多线程并发安全
• ✅ 统计信息：提供详细的缓存命中率等统计信息

配置

环境变量

可以通过以下环境变量配置缓存行为：

• TOOL_CACHE_MAX_SIZE：最大缓存项数（默认：1000）
• TOOL_CACHE_TTL：默认过期时间，秒（默认：180 = 3分钟）
• TOOL_CACHE_AUTO_RENEW：是否自动续期（默认：true）

示例：

export TOOL_CACHE_MAX_SIZE="500"
export TOOL_CACHE_TTL="300"  # 5分钟
export TOOL_CACHE_AUTO_RENEW="false"

代码配置

from agent.agent_memory_cache import AgentMemoryCacheManager

# 创建缓存管理器
cache = AgentMemoryCacheManager(
    max_size=1000,      # 最多缓存 1000 个 Agent
    default_ttl=180,    # 3分钟过期
    auto_renew=True     # 访问时自动续期
)

使用方法

1. 自动缓存（推荐）

缓存系统已集成到 init_agent 函数中，会自动根据配置生成缓存键并缓存 Agent。

from agent.deep_assistant import init_agent
from agent.agent_config import AgentConfig

# 创建配置
config = AgentConfig(
    bot_id="your-bot-id",
    session_id="user-session-123",
    # ... 其他配置
)

# 初始化 Agent（会自动使用内存缓存）
agent = await init_agent(config)

2. 手动使用缓存

from agent.agent_memory_cache import get_memory_cache_manager

# 获取全局缓存管理器
cache = get_memory_cache_manager()

# 设置缓存
cache.set("my-key", {"data": "value"}, ttl=300)  # 5分钟过期

# 获取缓存
value = cache.get("my-key")
if value is not None:
    print("缓存命中！")
    print(f"数据: {value}")
else:
    print("缓存未命中")

# 删除缓存
cache.delete("my-key")

# 获取统计信息
stats = cache.get_stats()
print(f"命中率: {stats['hit_rate_percent']}%")
print(f"内存使用: {stats['memory_usage_mb']} MB")
print(f"缓存项数: {stats['total_items']}")

缓存特性详解

1. TTL（Time To Live）过期

每个缓存项都有生存时间，超过时间后自动失效：

# 设置 10 秒过期的缓存
cache.set("temp_data", "value", ttl=10)

# 10 秒后访问会返回 None

2. 自动续期

当 auto_renew=True 时，每次访问缓存会自动延长其过期时间：

cache.set("data", {"key": "value"}, ttl=60)  # 1分钟过期

# 50 秒后访问
value = cache.get("data")  # 成功获取，且过期时间重置为 60 秒

3. LRU 淘汰策略

当缓存达到最大容量时，会自动淘汰最久未使用的项：

cache = AgentMemoryCacheManager(max_size=2)

cache.set("key1", "value1")  # 缓存: [key1]
cache.set("key2", "value2")  # 缓存: [key1, key2]
cache.set("key3", "value3")  # 淘汰 key1，缓存: [key2, key3]

4. 线程安全

缓存管理器使用 RLock 确保线程安全，可以在多线程环境中安全使用：

import threading

def worker(cache, thread_id):
    for i in range(100):
        cache.set(f"key_{thread_id}_{i}", f"value_{i}")
        value = cache.get(f"key_{thread_id}_{i}")

threads = []
for i in range(10):
    t = threading.Thread(target=worker, args=(cache, i))
    threads.append(t)
    t.start()

for t in threads:
    t.join()  # 安全等待所有线程完成

缓存管理工具

项目提供了缓存管理脚本：

# 查看缓存统计
poetry run python scripts/cache_manager.py stats

# 清空所有缓存
poetry run python scripts/cache_manager.py clear

# 列出缓存键
poetry run python scripts/cache_manager.py list

# 删除特定缓存键
poetry run python scripts/cache_manager.py delete my-cache-key

性能考虑

内存使用估算

内存缓存会占用系统内存，每个 Agent 对象的大小取决于其内部状态：

# 获取内存使用估算
stats = cache.get_stats()
print(f"内存使用: {stats['memory_usage_mb']} MB")

性能优化建议

1. 合理的缓存大小：

   • 根据可用内存设置 max_size
   • 建议不超过可用内存的 20-30%

2. 适当的 TTL：

   • 频繁访问的 Agent 可以设置较长的 TTL
   • 不常用的 Agent 使用较短的 TTL

3. 监控命中率：

   stats = cache.get_stats()
   if stats['hit_rate_percent'] < 70:
       logger.warning("缓存命中率较低，考虑调整 TTL 或缓存大小")
   

4. 定期清理：

   # 清理超过 1 小时的缓存
   cache.cleanup_old_entries(max_age_seconds=3600)
   

与磁盘缓存的比较

特性	内存缓存	磁盘缓存
访问速度	极快（纳秒级）	较慢（毫秒级）
持久化	进程重启后丢失	持久化保存
内存占用	较高	较低
并发性能	优秀	受 I/O 限制
使用场景	高频访问、临时缓存	长期存储、低频访问

注意事项

1. 内存限制：内存缓存受限于系统可用内存，注意监控内存使用
2. 进程隔离：每个进程有独立的缓存，多进程环境下不共享
3. 重启丢失：服务重启后缓存会丢失，首次访问需要重建
4. 合理配置：根据实际需求调整缓存大小和 TTL

故障排查

缓存命中率低

1. 检查 TTL 是否过短
2. 检查缓存大小是否过小
3. 查看日志中的缓存键是否一致

内存使用过高

1. 减少 max_size 配置
2. 缩短 TTL 时间
3. 定期调用 cleanup_old_entries

性能问题

1. 检查是否有大量并发访问
2. 考虑使用连接池或分布式缓存方案
3. 监控 GC（垃圾回收）压力

示例输出

INFO: AgentMemoryCacheManager initialized with max_size: 1000, default_ttl: 180s, auto_renew: True
INFO: Using cached agent for session: user-123, cache_key: agent_cache_a1b2c3d4
INFO: Cached agent for key: agent_cache_e5f6g7h8, ttl: 180s
INFO: Cache stats - Total items: 15, Hit rate: 85.5%, Memory: 12.34 MB