Agent上下文管理实战指南：从基础原理到生产环境避坑

1次阅读

没有评论

共计 2158 个字符，预计需要花费 6 分钟才能阅读完成。

在分布式系统和 AI agent 开发中，上下文管理是一个核心挑战。特别是在长对话场景中，经常会遇到以下几个典型问题：

长对话丢失：由于内存限制或服务重启，导致对话历史被清空
多会话交叉污染：不同用户的对话上下文被错误地混合在一起
版本回溯困难：无法快速回滚到对话的某个特定状态
并发冲突：多个请求同时修改同一上下文导致数据不一致

这些问题如果不解决好，会直接影响用户体验和系统可靠性。

我们对比了三种常见的存储方案：

内存存储
优点：访问速度快(微秒级)，实现简单
缺点：易丢失，不适合生产环境
测试数据：平均读取 0.2ms，写入 0.3ms
Redis
优点：性能好(毫秒级)，支持持久化
缺点：内存成本高，集群配置复杂
测试数据：平均读取 1.5ms，写入 2ms
关系型数据库(如 PostgreSQL)
优点：数据可靠，支持复杂查询
缺点：性能较差(十毫秒级)
测试数据：平均读取 15ms，写入 20ms

对于生产环境，推荐 Redis+ 数据库的混合方案，热数据放 Redis，冷数据持久化到数据库。

from collections import OrderedDict
import time

class ContextManager:
    def __init__(self, max_size=1000):
        self.cache = OrderedDict()
        self.max_size = max_size

    def get(self, session_id):
        if session_id not in self.cache:
            return None

        # 更新访问时间
        context = self.cache.pop(session_id)
        self.cache[session_id] = context
        return context

    def set(self, session_id, context):
        if len(self.cache) >= self.max_size:
            # LRU 淘汰
            self.cache.popitem(last=False)
        self.cache[session_id] = context

时间复杂度分析：
– get 操作：O(1)
– set 操作：O(1)

class VersionedContext:
    def __init__(self):
        self.versions = []  # 版本历史
        self.current = {}   # 当前上下文

    def commit(self):
        # 深拷贝当前状态
        import copy
        snapshot = copy.deepcopy(self.current)
        self.versions.append(snapshot)
        return len(self.versions) - 1  # 返回版本号

    def rollback(self, version):
        if 0 <= version < len(self.versions):
            self.current = copy.deepcopy(self.versions[version])
            return True
        return False

import redis

r = redis.Redis()

def update_context(session_id, new_data):
    while True:
        # 获取当前版本
        version = r.get(f'{session_id}:version') or 0

        # 开启事务
        pipe = r.pipeline()
        pipe.watch(f'{session_id}:version')

        # 检查版本是否变化
        current_version = r.get(f'{session_id}:version')
        if current_version != version:
            pipe.unwatch()
            continue  # 重试

        # 更新数据
        pipe.multi()
        pipe.hmset(f'{session_id}:data', new_data)
        pipe.incr(f'{session_id}:version')
        try:
            pipe.execute()
            break
        except redis.WatchError:
            continue

推荐使用 tracemalloc 进行内存监控：

import tracemalloc

tracemalloc.start()

# ... 运行一段时间后...
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

for stat in top_stats[:10]:
    print(stat)

对比 JSON 和 MessagePack：