你是不是也遇到过这种让人抓狂的场景？🎯 你正跟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斗智斗勇的战友，咱们下期见！