• 让FastAPI Agent真正记住你:聊聊会话记忆与持久化存储的落地实践


    你是不是也遇到过这种让人抓狂的场景?🎯 你正跟AI聊得火热,说了半天自己的喜好、项目的背景,结果你只是去倒杯水的功夫,回来问它一句“刚才咱们聊到哪儿了?”,它给你来一句标准的“您好,我是AI助手,有什么可以帮您?”。

    得,白聊了。那一刻,真的想把键盘吃了的心都有。这不是人工智能,这是“人工智障”。我最开始自己倒腾Agent的时候,也在这个坑里趴了好久。

    今天咱们就直击痛点,聊聊怎么让我们的 FastAPI Agent 真正拥有“记忆”,从一个转身就失忆的“金鱼”,变成一个能陪你深聊的靠谱伙伴。

    🧠 记忆也分“快慢”:短期会话 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, alias="X-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 = f"chat:{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(f"Redis 缓存历史消息:\n{history}")
    
        # 1) 把用户刚说的话塞进记忆里
        user_msg = {"role": "user", "content": req.message}
        history.append(user_msg)
    
        # 2) 调用大模型(这里直接使用上一篇里定义的模型调用函数,带多轮历史会话消息功能)
        assistant_reply = await call_llm(
            messages=history, 
            system_prompt='你是一个能根据多轮对话综合思考后,给出贴心建议的小助手', 
            config=config
        )
        logger.debug(f"模型回复信息:\n{assistant_reply}")
        history.append(assistant_reply) # 这里一定要注意拼接时的返回格式,如果有误,下轮会话可能失败!
    
        # 3) 把新产生的两条消息异步写入 Redis,并刷新过期时间
        key = f"chat:{session_id}"
        # rpush 直接追加到列表尾部,因为 history 是从 Redis 里读出来的,再追加不会重复
        await redis.rpush(key, json.dumps(user_msg, ensure_ascii=False))
        await redis.rpush(key, json.dumps(assistant_reply, ensure_ascii=False))
        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(content=content)
            response.headers["X-Session-ID"] = session_id
            return response
        else:
            return content

    🛡️ 最后啰嗦几句,都是血泪史

    🟠 Redis连接池耗尽:
    一定、一定、一定要用单例模式管理连接池。我早期图省事,在函数里每次读写都新建连接,流量一大,端口直接耗光。

    🟠 记忆蒸馏:
    别傻乎乎地把几十轮对话都塞给LLM,Token算力都是要钱的呀!只取最近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"])

    🟠 数据清理:
    Redis里的历史是临时的,但数据库里的长期档案得有个软删除或遗忘机制,万一用户哪天行使“被遗忘权”呢。


    好啦,今天就先唠到这儿。让Agent拥有记忆,绝对是提升用户体验最实在的一步。从毫无感情的对话工具,到能记住你爱喝咖啡不放糖的伙伴,秘诀就在这几百行代码里。

    这篇文章里藏着不少我运行翻车才换来的经验,别等下次碰到“失忆”的Agent才后悔没细看。赶紧的,收藏起来🌟,或者转给你身边同样在跟Agent斗智斗勇的战友,咱们下期见!

  • 相关阅读:
    TCP的保活机制
    基于Java+SpringBoot+Thymeleaf+Mysql疫情下社区健康评估系统设计与实现
    弹性资源组件elastic-resource设计(一)-架构
    【springboot笔记】程序可用性检测ApplicationAvailability
    Python基础tuple元组定义与函数
    建筑类企业做ISO9001时需要带GB/T50430标准
    我也来一个“羊了个羊”
    vue2给对象新增、删除属性,界面不刷新 。通过数组下标修改数组界面不会刷新
    做知识付费,这十大知识付费平台一定要知道
    设计模式-适配器模式
  • 原文地址:https://www.cnblogs.com/ymtianyu/p/20018960