记忆也分“快慢”短期会话 vs 长期档案在动手写代码前咱们得先把概念掰扯清楚。Agent的记忆跟咱们人类的记忆其实挺像大致分两种短期记忆就是当前的对话上下文。Agent得知道你刚才说了啥才好接着你的话茬往下聊。这玩意儿要求的就是个“快”字像闪电一样。长期记忆好比你的用户档案。你叫啥你喜欢用React还是Vue你对猫毛过敏……这些信息得持久化地存下来下次你登录哪怕隔了一个月Agent也能像老朋友一样记得你。 第一步给每场对话上个“会话锁”要实现记忆最关键的第一块积木就是会话隔离。不能让张三聊的内容跑到李四的对话里去那不乱套了嘛。我的做法很简单就是生成一个唯一的 session_id。在FastAPI里这活儿可以交给一个轻量级的依赖项来做干净又利落。你可能会问“用Cookie或Header带过来不就行了”对逻辑是这样而且千万别偷懒用那种自增的简单ID太容易被遍历攻击了。务必用uuid4这种几乎不可能碰撞的字符串。from fastapi import Header, HTTPException from uuid import UUID, uuid4 async def get_session_id(x_session_id: str Header(None, aliasX-Session-ID)): 从 Header 里提取并校验 session_id logger.debug(f客户端 session_id: {x_session_id}) # 如果请求里带了有效的session_id直接返回 if x_session_id: try: UUID(x_session_id) return x_session_id, False # False 表示不是新会话 except ValueError: raise HTTPException(400, 会话ID格式不太对哦得是标准的UUID。) # 如果没带生成一个新的返回 return str(uuid4()), True # True 表示是新会话⚡️ 第二步选好“记事本”快慢分离好会话锁有了接着选“记事本”。工具的选择好比选螺丝刀不是最贵的就好而是最顺手的。 短期记忆Redis 异步存储对于秒级读写、需要频繁更新的对话历史我首选Redis。咱们用异步Redis客户端跟FastAPI的异步特性简直是天作之合。每次对话来就把新的消息追加到历史列表里并设置一个合理的过期时间比如30分钟不用操心内存溢出。️ 长期记忆PostgreSQL 持久存档当一轮对话结束或者中途提取到了关键的用户信息比如“对了我对猫毛过敏”就需要异步地写到 PostgreSQL 里存起来。这是咱们的用户档案库讲究的是可靠和结构清晰。根据以往的经验这里有个容易翻车的点别在请求的主流程里直接干这活儿会让用户觉得你的Agent反应迟钝。推荐用 BackgroundTasks 或一个外部队列来异步处理悄无声息地完成存档。 第三步FastAPI 依赖注入全程优雅“记忆”接下来重点来了怎么让我们的服务端处理每一步逻辑时都能轻松拿到属于“这个用户、这个会话”的记忆呢用FastAPI的依赖项注入。我们可以设计一个函数它依赖刚才的 session_id 和咱们初始化的 Redis 连接自动读取或新生成一个会话对象。async def get_session_context( session_data: tuple Depends(get_session_id), redis: Redis Depends(get_redis), ): 核心依赖给你一个会“记事儿”的上下文对象 # 拼 Redis 键每个会话一个列表 session_id, is_new session_data key fchat:{session_id} if is_new: history [] else: raw_messages await redis.lrange(key, 0, -1) # 反序列化历史消息 history [json.loads(msg) for msg in raw_messages] # # 没捞到去数据库里翻翻旧账伪代码展示核心思路 # if not history: # user_profile await load_from_pg(session_id) # return {history: [], profile: user_profile} # 把需要用到的东东全部塞进一个字典返回 return { is_new_session: is_new, # 把是否新消息会话标记传出去用于响应头 session_id: session_id, history: history, redis: redis, } 实战秀跨轮次对话记忆不丢失咱们来模拟一下完整流程你就知道这事儿有多酷了。1️⃣第一轮对话用户带着 session_id 来了说“嘿我想吃意大利菜。”Agent通过依赖注入拿到空白会话把这句话存入Redis历史然后推荐了一家意面馆。同时后台任务默默记下“该用户偏好西餐”。2️⃣第二轮对话跨轮次过了一会同一个 session_id 又来了问“刚才那家店在哪儿”Agent再次通过依赖项从Redis秒读出历史“你想吃意大利菜”于是准确地给出地址。3️⃣新的会话长期记忆验证几天后用户换了个设备带着新的 session_id 登录。问“随便推荐点吃的。”Agent一看Redis里历史是空的但依赖项从PostgreSQL里加载了用户画像贴心地问“之前看你喜欢西餐要不要试试新开的牛排馆”原理理解了真正写接口时才是一波三折呀这里给出完整代码防止一个解析不对直接给个500错误很多踩坑点直接在注释里面说明了router.post(/chat) async def chat( req: ChatRequest, ctx: dict Depends(get_session_context), # 自动注入当前会话上下文 config: Settings Depends(get_settings) ): 核心接口聊天自动读写记忆 is_new ctx[is_new_session] session_id ctx[session_id] redis: Redis ctx[redis] history: list ctx[history] # 这个消息列表就是短期记忆 logger.debug(fRedis 缓存历史消息\n{history}) # 1) 把用户刚说的话塞进记忆里 user_msg {role: user, content: req.message} history.append(user_msg) # 2) 调用大模型这里直接使用上一篇里定义的模型调用函数带多轮历史会话消息功能 assistant_reply await call_llm( messageshistory, system_prompt你是一个能根据多轮对话综合思考后给出贴心建议的小助手, configconfig ) logger.debug(f模型回复信息\n{assistant_reply}) history.append(assistant_reply) # 这里一定要注意拼接时的返回格式如果有误下轮会话可能失败 # 3) 把新产生的两条消息异步写入 Redis并刷新过期时间 key fchat:{session_id} # rpush 直接追加到列表尾部因为 history 是从 Redis 里读出来的再追加不会重复 await redis.rpush(key, json.dumps(user_msg, ensure_asciiFalse)) await redis.rpush(key, json.dumps(assistant_reply, ensure_asciiFalse)) await redis.expire(key, 1800) # 30分钟保鲜期自动清理 # 4) 可在这通过 BackgroundTasks 把用户偏好异步写进 PostgreSQL略 # 构建响应 content { reply: assistant_reply, session_id: session_id, memory_rounds: len(history), } if is_new: # 第一次会话告诉客户端以后带这个ID来找我 response JSONResponse(contentcontent) response.headers[X-Session-ID] session_id return response else: return content️ 最后啰嗦几句都是血泪史 Redis连接池耗尽一定、一定、一定要用单例模式管理连接池。我早期图省事在函数里每次读写都新建连接流量一大端口直接耗光。 记忆蒸馏别傻乎乎地把几十轮对话都塞给LLMToken算力都是要钱的呀只取最近N轮或做个摘要这个优化能省下不少预算。用户画像的长时存储也是同样的道理总不能把所有的会话历史都存储吧简单点可以让LLM帮我们把多轮对话的内容提取用户特征或偏好async def extract_key_info(messages: list[dict]) - list[str]: # 取最后几轮对话拼接成 prompt prompt 从以下对话中提炼用户特征或偏好只输出JSON列表不要其他文字... # 调用 LLM response await call_llm(prompt) # 解析返回的列表如 [喜欢 Python, 对猫毛过敏, 想去意大利旅游] return json.loads(response[message][content]) 数据清理