AI原生应用领域短期记忆:提升应用性能的秘诀

关键词:AI原生应用;短期记忆;上下文缓存;中间结果存储;性能优化;缓存策略;大语言模型
摘要:在AI原生应用(如聊天机器人、代码助手、多模态生成)中,“短期记忆"是提升性能的关键秘密武器。它就像厨师炒菜时的"备菜盘”——把常用的食材(上下文、中间结果)放在手边,不用每次都跑回冰箱(调用大模型或数据库),从而大幅缩短响应时间、减少重复计算。本文将用"厨房炒菜"的类比,一步步拆解AI短期记忆的核心概念、实现原理和实战技巧,帮你搞懂如何用短期记忆让AI应用"跑得更快、更聪明"。

背景介绍

目的和范围

AI原生应用(如ChatGPT、GitHub Copilot、DALL·E 3)的核心特点是实时交互(比如多轮对话)、复杂任务处理(比如生成代码/图片)和上下文依赖(比如记得用户之前说的话)。这些应用的性能瓶颈往往不是"模型不够强",而是"重复劳动太多"——比如每次对话都要重新输入所有历史记录,每次生成图片都要重新处理用户的描述。

本文的目的是:用通俗易懂的语言解释"AI短期记忆"的本质,教你如何通过缓存上下文、存储中间结果、管理状态等方式,让AI应用"少做无用功",提升响应速度和用户体验。

范围覆盖:短期记忆的核心概念、常见类型、实现策略(代码示例)、实际应用场景,以及未来发展趋势。

预期读者

  • AI应用开发者(想提升产品性能的工程师);
  • 产品经理(想理解AI应用底层逻辑的从业者);
  • 对AI技术感兴趣的爱好者(想搞懂"为什么ChatGPT能记住我的对话")。

文档结构概述

本文按照"问题引入→概念拆解→原理讲解→实战演示→应用场景"的逻辑展开,具体结构如下:

  1. 用"厨师炒菜"的故事引出短期记忆的重要性;
  2. 拆解AI短期记忆的3个核心类型(上下文缓存、中间结果存储、状态管理);
  3. 讲解短期记忆的实现原理(缓存策略、代码示例);
  4. 用"AI代码助手"的实战项目演示如何落地;
  5. 分析短期记忆在不同应用场景中的作用;
  6. 展望未来发展趋势。

术语表

核心术语定义
  • AI短期记忆:AI应用在处理任务时,临时存储的上下文信息、中间计算结果或状态数据(比如用户的对话历史、代码片段、生成中的图片草稿)。它的特点是"快速访问、有限容量、时效性强"(类似电脑的"内存")。
  • 上下文窗口:大语言模型(如GPT-4)能处理的最大输入长度(比如8k、16k tokens)。短期记忆的作用是"压缩"上下文,让模型不用处理全部历史记录。
  • 缓存命中率:从短期记忆中找到所需数据的请求比例(比如100次请求中有80次命中缓存,命中率就是80%)。命中率越高,性能提升越明显。
相关概念解释
  • 长期记忆:AI应用中持久存储的历史数据(比如用户的所有对话记录、训练好的模型参数),类似电脑的"硬盘"。短期记忆是长期记忆的"子集",用于处理即时需求。
  • 中间结果:AI处理任务时的"半成品"(比如生成代码时的语法检查结果、生成图片时的草稿)。存储中间结果可以避免重复计算。
缩略词列表
  • LLM:大语言模型(Large Language Model);
  • Redis:一种高性能键值对缓存数据库(常用作短期记忆存储);
  • API:应用程序编程接口(Application Programming Interface,用于调用LLM)。

核心概念与联系:AI的"备菜盘"到底是什么?

故事引入:为什么厨师需要"备菜盘"?

假设你是一位厨师,要做一道"番茄炒蛋"。如果没有备菜盘,你得:

  1. 从冰箱里拿出番茄(跑一趟);
  2. 洗番茄、切番茄(耗时);
  3. 从冰箱里拿出鸡蛋(再跑一趟);
  4. 打鸡蛋、搅拌(再耗时);
  5. 炒菜(终于开始)。

但如果有了备菜盘,你可以提前把切好的番茄、打好的鸡蛋放在手边,炒菜时直接拿过来用——节省了来回跑冰箱的时间,也避免了重复切菜/打鸡蛋。

AI原生应用的"短期记忆",本质就是这个备菜盘:它把AI处理任务时需要的"常用食材"(上下文、中间结果)提前存起来,不用每次都"跑回冰箱"(调用LLM或数据库),从而提升效率。

比如,ChatGPT能记住你之前的对话,就是因为它把对话历史存在了"短期记忆"里;GitHub Copilot能推荐符合你代码风格的片段,就是因为它把你之前写的代码存在了"短期记忆"里。

核心概念解释:AI短期记忆的3种"食材"

AI短期记忆的"备菜盘"里,主要装着3种"食材",每种都有不同的作用:

1. 上下文缓存:记住"刚才说的话"(类比:备菜盘里的"切好的番茄")

定义:存储用户与AI之间的近期对话历史(比如最近5条消息),让AI能理解"上下文"。
生活例子:你和朋友聊天时,朋友说"我昨天去了公园",你接着说"好玩吗?“——你不用重复问"你昨天去了哪里?”,因为你"记住了上下文"。
AI中的作用:避免每次对话都把全部历史记录传给LLM(比如ChatGPT的上下文窗口是8k tokens,如果每次都传100条历史,会超出限制,而且慢)。

比如,用户和AI的对话:

  • 用户:“我想买一台笔记本电脑,预算5000元。”
  • AI:“推荐联想小新Pro 14,性价比高。”
  • 用户:“它的续航怎么样?”

如果没有上下文缓存,AI会把"它的续航怎么样?“单独传给LLM,LLM不知道"它"指的是"联想小新Pro 14”,可能会回答"请问你说的是哪款产品?“。
如果有了上下文缓存,AI会把之前的对话(“我想买一台笔记本电脑…推荐联想小新Pro 14”)和当前问题一起传给LLM,LLM就能正确回答"联想小新Pro 14的续航是12小时左右”。

2. 中间结果存储:保存"半成品"(类比:备菜盘里的"打好的鸡蛋")

定义:存储AI处理任务时的中间计算结果(比如生成代码时的语法检查结果、生成图片时的草稿),避免重复计算。
生活例子:你做数学题时,把"12×3=36"写在草稿纸上,后面计算"36+5=41"时不用再算一遍"12×3"——草稿纸就是你的"中间结果存储"。
AI中的作用:减少对LLM的调用次数(因为LLM调用很贵、很慢)。

比如,用户让AI"生成一个Python的冒泡排序代码",AI生成后,用户又说"把它改成快速排序"。如果没有中间结果存储,AI需要重新调用LLM生成快速排序代码;如果有了中间结果存储,AI可以把之前生成的冒泡排序代码存在缓存里,然后让LLM基于"冒泡排序"修改成"快速排序"——这样LLM的工作量更小,响应更快。

3. 状态管理:跟踪"当前在做什么"(类比:备菜盘里的"炒菜步骤")

定义:存储AI应用的当前状态(比如用户的会话阶段、任务进度),让AI能"延续之前的工作"。
生活例子:你玩游戏时,保存了"第3关"的进度,下次打开游戏不用从头开始——游戏的"进度保存"就是状态管理。
AI中的作用:处理复杂的多步骤任务(比如"帮我订机票→选座位→确认订单")。

比如,智能客服机器人处理用户的"订机票"请求:

  • 状态1:询问用户"出发地和目的地";
  • 状态2:询问用户"出发日期";
  • 状态3:推荐航班;
  • 状态4:确认订单。

如果没有状态管理,用户每发一条消息,机器人都要重新问"你要订机票吗?“;如果有了状态管理,机器人能记住"当前在状态2”,直接问"请问出发日期是哪天?"。

核心概念之间的关系:"备菜盘"的协作逻辑

上下文缓存、中间结果存储、状态管理这三个"食材",不是孤立的,而是互相配合的,就像厨师炒菜时需要"切好的番茄"+“打好的鸡蛋”+"炒菜步骤"一起用。

1. 上下文缓存是"基础":没有它,中间结果和状态都没用

比如,状态管理需要知道"当前在状态2",但如果没有上下文缓存,机器人不知道"状态2"对应的是"订机票"的请求,可能会把"状态2"用到"查天气"的请求上——这就乱了。

2. 中间结果存储是"进阶":让上下文缓存更高效

比如,上下文缓存里存了"用户想要Python的冒泡排序代码",中间结果存储里存了"生成的冒泡排序代码"。当用户要求"改成快速排序"时,AI可以把"上下文缓存"(用户的需求)和"中间结果"(冒泡排序代码)一起传给LLM,让LLM更快生成结果。

3. 状态管理是"高层":让短期记忆更"智能"

比如,状态管理知道"用户正在订机票的状态3"(推荐航班),这时上下文缓存里存了"用户的出发地、目的地、日期",中间结果存储里存了"推荐的航班列表"。AI可以直接用这些数据,不用再问用户——这就是"智能"的体现。

核心概念原理和架构的文本示意图

AI原生应用的短期记忆架构,就像一个"三层备菜盘":

  • 第一层(最底层):上下文缓存:存储近期对话历史(比如用户ID→最近5条消息);
  • 第二层(中间层):中间结果存储:存储中间计算结果(比如问题哈希→生成的代码/图片草稿);
  • 第三层(最上层):状态管理:存储应用状态(比如用户ID→当前会话阶段)。

当用户发送请求时,AI应用的处理流程是:

  1. 状态管理中获取用户当前的状态(比如"订机票的状态3");
  2. 上下文缓存中获取近期对话历史(比如"出发地:北京,目的地:上海,日期:2024-05-01");
  3. 中间结果存储中获取相关的中间结果(比如"推荐的航班列表:CA123、MU456");
  4. 把这些数据拼接成LLM的输入(比如"用户正在订机票,需要确认航班,之前推荐了CA123和MU456");
  5. 调用LLM生成回答(比如"请问你想选CA123还是MU456?");
  6. 更新上下文缓存(添加新的对话)、中间结果存储(如果有新的中间结果)、状态管理(如果状态变化,比如从"状态3"到"状态4")。

Mermaid 流程图:AI应用处理请求的流程

graph TD
    A[用户发送请求] --> B[获取用户ID]
    B --> C[从状态管理中获取当前状态]
    C --> D[从上下文缓存中获取近期对话]
    D --> E[从中间结果存储中获取相关中间结果]
    E --> F[拼接LLM输入(状态+上下文+中间结果)]
    F --> G{是否需要调用LLM?}
    G -->|是| H[调用LLM生成回答]
    G -->|否| I[直接用缓存的结果生成回答]
    H --> J[更新中间结果存储(如果有新中间结果)]
    I --> J
    J --> K[更新上下文缓存(添加新对话)]
    K --> L[更新状态管理(如果状态变化)]
    L --> M[返回回答给用户]

核心算法原理 & 具体操作步骤:如何搭建AI的"备菜盘"?

1. 上下文缓存:用Redis实现"对话历史存储"

算法原理:上下文缓存的核心是"保留近期的对话",常用的策略是固定窗口大小(比如保留最近5条消息)或时间窗口(比如保留最近10分钟的消息)。
工具选择:Redis(分布式、支持哈希表,适合存储用户对话历史)。
具体步骤

  • 步骤1:安装Redis(sudo apt install redis-server);
  • 步骤2:用Python的redis库连接Redis;
  • 步骤3:定义"上下文缓存"的键(比如user:context:{user_id});
  • 步骤4:当用户发送消息时,从Redis中获取该用户的对话历史,添加新消息,然后保留最近5条;
  • 步骤5:把更新后的对话历史存回Redis。

代码示例(Python)

import redis
from typing import List

# 连接Redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def update_context(user_id: str, message: str, max_length: int = 5) -> List[str]:
    """更新用户的上下文缓存(保留最近max_length条消息)"""
    # 定义Redis键
    key = f"user:context:{user_id}"
    # 获取当前对话历史(列表)
    context = redis_client.lrange(key, 0, -1)  # 从Redis中取所有元素
    context = [msg.decode('utf-8') for msg in context]  # 转成字符串
    # 添加新消息
    context.append(message)
    # 保留最近max_length条
    if len(context) > max_length:
        context = context[-max_length:]
    # 存回Redis(先删除旧的,再存新的)
    redis_client.delete(key)
    for msg in context:
        redis_client.rpush(key, msg)  # 从右往左存,保持顺序
    return context

# 测试:用户发送消息
user_id = "user_123"
message1 = "我想买一台笔记本电脑,预算5000元。"
message2 = "推荐联想小新Pro 14,性价比高。"
message3 = "它的续航怎么样?"

# 更新上下文
update_context(user_id, message1)
update_context(user_id, message2)
update_context(user_id, message3)

# 获取上下文(应该是最近3条)
context = redis_client.lrange(f"user:context:{user_id}", 0, -1)
context = [msg.decode('utf-8') for msg in context]
print(context)
# 输出:['我想买一台笔记本电脑,预算5000元。', '推荐联想小新Pro 14,性价比高。', '它的续航怎么样?']

2. 中间结果存储:用LRU策略优化缓存

算法原理:中间结果存储的核心是"保留常用的结果",常用的策略是LRU(最近最少使用)——当缓存满时,删除最久没用到的结果。
工具选择:Python的functools.lru_cache(简单场景)或Redis的expire命令(设置过期时间)。
具体步骤

  • 步骤1:定义中间结果的键(比如result:hash:{question_hash});
  • 步骤2:当处理用户问题时,先计算问题的哈希值(比如MD5);
  • 步骤3:检查Redis中是否有该哈希值对应的结果;
  • 步骤4:如果有,直接返回;如果没有,调用LLM生成结果,并存到Redis(设置过期时间,比如1小时)。

代码示例(Python)

import hashlib
import redis
from typing import Optional

# 连接Redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def get_middle_result(question: str, expire_time: int = 3600) -> Optional[str]:
    """获取中间结果(用LRU策略,过期时间1小时)"""
    # 计算问题的哈希值(MD5)
    question_hash = hashlib.md5(question.encode('utf-8')).hexdigest()
    # 定义Redis键
    key = f"result:hash:{question_hash}"
    # 检查缓存是否存在
    result = redis_client.get(key)
    if result:
        return result.decode('utf-8')
    else:
        # 调用LLM生成结果(这里用假数据代替)
        llm_result = "联想小新Pro 14的续航是12小时左右。"
        # 存到Redis,设置过期时间
        redis_client.set(key, llm_result, ex=expire_time)
        return llm_result

# 测试:用户问"联想小新Pro 14的续航怎么样?"
question = "联想小新Pro 14的续航怎么样?"
result1 = get_middle_result(question)
print(f"第一次调用:{result1}")  # 输出:联想小新Pro 14的续航是12小时左右。(调用LLM)
result2 = get_middle_result(question)
print(f"第二次调用:{result2}")  # 输出:联想小新Pro 14的续航是12小时左右。(从缓存取)

3. 状态管理:用枚举类跟踪会话状态

算法原理:状态管理的核心是"定义清晰的状态转移规则",比如"从状态1(询问出发地)到状态2(询问目的地)“的条件是"用户提供了出发地”。
工具选择:Python的enum类(定义状态)+ Redis(存储用户当前状态)。
具体步骤

  • 步骤1:用enum定义会话状态(比如BookingState);
  • 步骤2:当用户发送消息时,从Redis中获取该用户的当前状态;
  • 步骤3:根据当前状态和用户消息,判断下一个状态;
  • 步骤4:更新Redis中的用户状态。

代码示例(Python)

import enum
import redis
from typing import Optional

# 定义会话状态(订机票的流程)
class BookingState(enum.Enum):
    INIT = "init"  # 初始状态(未开始订机票)
    ASK_ORIGIN = "ask_origin"  # 询问出发地
    ASK_DESTINATION = "ask_destination"  # 询问目的地
    ASK_DATE = "ask_date"  # 询问出发日期
    RECOMMEND_FLIGHT = "recommend_flight"  # 推荐航班
    CONFIRM = "confirm"  # 确认订单

# 连接Redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def update_state(user_id: str, new_state: BookingState) -> None:
    """更新用户的会话状态"""
    key = f"user:state:{user_id}"
    redis_client.set(key, new_state.value)

def get_state(user_id: str) -> BookingState:
    """获取用户的当前会话状态(默认初始状态)"""
    key = f"user:state:{user_id}"
    state = redis_client.get(key)
    if state:
        return BookingState(state.decode('utf-8'))
    else:
        return BookingState.INIT

# 测试:用户开始订机票
user_id = "user_123"
# 初始状态是INIT
current_state = get_state(user_id)
print(f"初始状态:{current_state}")  # 输出:BookingState.INIT

# 用户发送"我想订机票",转移到ASK_ORIGIN状态
update_state(user_id, BookingState.ASK_ORIGIN)
current_state = get_state(user_id)
print(f"更新后状态:{current_state}")  # 输出:BookingState.ASK_ORIGIN

# 用户发送"出发地是北京",转移到ASK_DESTINATION状态
update_state(user_id, BookingState.ASK_DESTINATION)
current_state = get_state(user_id)
print(f"更新后状态:{current_state}")  # 输出:BookingState.ASK_DESTINATION

数学模型和公式:如何衡量短期记忆的效果?

短期记忆的效果主要用缓存命中率性能提升比来衡量。

1. 缓存命中率(Hit Ratio)

定义:从短期记忆中找到所需数据的请求比例。
公式
命中率=命中次数总请求次数×100% \text{命中率} = \frac{\text{命中次数}}{\text{总请求次数}} \times 100\% 命中率=总请求次数命中次数×100%
例子:假设100次请求中有80次命中缓存,命中率就是80%。

2. 性能提升比(Performance Improvement Ratio)

定义:使用短期记忆后的性能(比如响应时间)与不使用短期记忆的性能之比。
公式
性能提升比=不使用缓存的平均响应时间使用缓存的平均响应时间 \text{性能提升比} = \frac{\text{不使用缓存的平均响应时间}}{\text{使用缓存的平均响应时间}} 性能提升比=使用缓存的平均响应时间不使用缓存的平均响应时间
例子:不使用缓存时,平均响应时间是1500ms(调用LLM);使用缓存时,平均响应时间是200ms(从缓存取)。性能提升比是1500/200=7.5倍。

3. 缓存的收益分析

假设:

  • 未命中缓存的处理时间(T_miss):1500ms(调用LLM);
  • 命中缓存的处理时间(T_hit):200ms(从缓存取);
  • 命中率(h):80%。

平均响应时间T_avg)的计算公式:
Tavg=h×Thit+(1−h)×Tmiss T_{\text{avg}} = h \times T_{\text{hit}} + (1-h) \times T_{\text{miss}} Tavg=h×Thit+(1h)×Tmiss

代入数值:
Tavg=0.8×200+0.2×1500=160+300=460ms T_{\text{avg}} = 0.8 \times 200 + 0.2 \times 1500 = 160 + 300 = 460 \text{ms} Tavg=0.8×200+0.2×1500=160+300=460ms

结论:使用缓存后,平均响应时间从1500ms降到460ms,提升了3.26倍(1500/460≈3.26)。

项目实战:用短期记忆打造高性能AI代码助手

项目目标

打造一个AI代码助手,能记住用户之前的代码片段和问题,提升后续回答的准确性和响应速度。

开发环境搭建

  • 后端框架:FastAPI(轻量级、高性能);
  • 缓存工具:Redis(存储上下文、中间结果、状态);
  • LLM:调用OpenAI的GPT-3.5-turbo API(生成代码);
  • 开发语言:Python 3.10+。

源代码详细实现和代码解读

1. 项目结构
ai-code-assistant/
├── main.py  # 后端入口(FastAPI)
├── cache.py  # 缓存操作(上下文、中间结果、状态)
├── llm.py    # LLM调用(OpenAI API)
└── requirements.txt  # 依赖库(fastapi、uvicorn、redis、openai)
2. 依赖安装
pip install fastapi uvicorn redis openai python-dotenv
3. 配置文件(.env)
OPENAI_API_KEY=your-openai-api-key  # 替换成你的OpenAI API密钥
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
4. 缓存操作(cache.py)
import redis
from enum import Enum
from typing import List, Optional
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()

# 连接Redis
redis_client = redis.Redis(
    host=os.getenv("REDIS_HOST"),
    port=int(os.getenv("REDIS_PORT")),
    db=int(os.getenv("REDIS_DB"))
)

# 定义会话状态(代码助手的流程)
class CodeAssistantState(Enum):
    INIT = "init"  # 初始状态(未开始提问)
    ASK_PROBLEM = "ask_problem"  # 询问问题(比如"我想生成冒泡排序代码")
    ASK_DETAILS = "ask_details"  # 询问细节(比如"需要Python还是Java?")
    GENERATE_CODE = "generate_code"  # 生成代码
    MODIFY_CODE = "modify_code"  # 修改代码

def update_context(user_id: str, message: str, max_length: int = 5) -> List[str]:
    """更新用户的上下文缓存(保留最近max_length条消息)"""
    key = f"user:context:{user_id}"
    context = redis_client.lrange(key, 0, -1)
    context = [msg.decode('utf-8') for msg in context]
    context.append(message)
    if len(context) > max_length:
        context = context[-max_length:]
    redis_client.delete(key)
    for msg in context:
        redis_client.rpush(key, msg)
    return context

def get_context(user_id: str) -> List[str]:
    """获取用户的上下文缓存"""
    key = f"user:context:{user_id}"
    context = redis_client.lrange(key, 0, -1)
    return [msg.decode('utf-8') for msg in context]

def set_middle_result(question_hash: str, result: str, expire_time: int = 3600) -> None:
    """存储中间结果(设置过期时间)"""
    key = f"result:hash:{question_hash}"
    redis_client.set(key, result, ex=expire_time)

def get_middle_result(question_hash: str) -> Optional[str]:
    """获取中间结果"""
    key = f"result:hash:{question_hash}"
    result = redis_client.get(key)
    return result.decode('utf-8') if result else None

def update_state(user_id: str, new_state: CodeAssistantState) -> None:
    """更新用户的会话状态"""
    key = f"user:state:{user_id}"
    redis_client.set(key, new_state.value)

def get_state(user_id: str) -> CodeAssistantState:
    """获取用户的当前会话状态"""
    key = f"user:state:{user_id}"
    state = redis_client.get(key)
    return CodeAssistantState(state.decode('utf-8')) if state else CodeAssistantState.INIT
5. LLM调用(llm.py)
from openai import OpenAI
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()

# 初始化OpenAI客户端
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def generate_code(prompt: str) -> str:
    """调用GPT-3.5-turbo生成代码"""
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
        max_tokens=500
    )
    return response.choices[0].message.content.strip()
6. 后端入口(main.py)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from cache import (
    update_context, get_context, set_middle_result, get_middle_result,
    update_state, get_state, CodeAssistantState
)
from llm import generate_code
import hashlib

app = FastAPI(title="AI Code Assistant", version="1.0")

# 请求模型(用户发送的消息)
class UserMessage(BaseModel):
    user_id: str
    message: str

# 响应模型(AI的回答)
class AIResponse(BaseModel):
    answer: str
    state: str

@app.post("/chat", response_model=AIResponse)
async def chat(user_message: UserMessage):
    """处理用户的聊天请求"""
    user_id = user_message.user_id
    message = user_message.message

    # 1. 获取当前状态
    current_state = get_state(user_id)
    print(f"用户{user_id}的当前状态:{current_state}")

    # 2. 更新上下文缓存(添加用户的新消息)
    update_context(user_id, f"用户:{message}")
    context = get_context(user_id)
    print(f"用户{user_id}的上下文:{context}")

    # 3. 根据状态处理请求
    if current_state == CodeAssistantState.INIT:
        # 初始状态:用户第一次提问,转移到ASK_PROBLEM状态
        answer = "你想生成什么代码?比如'生成Python的冒泡排序代码'。"
        new_state = CodeAssistantState.ASK_PROBLEM
    elif current_state == CodeAssistantState.ASK_PROBLEM:
        # 询问问题状态:用户提供了问题,转移到ASK_DETAILS状态
        answer = "需要用什么语言?比如Python、Java。"
        new_state = CodeAssistantState.ASK_DETAILS
        # 存储用户的问题(中间结果)
        question_hash = hashlib.md5(message.encode('utf-8')).hexdigest()
        set_middle_result(question_hash, message)
    elif current_state == CodeAssistantState.ASK_DETAILS:
        # 询问细节状态:用户提供了语言,生成代码,转移到GENERATE_CODE状态
        # 获取之前的问题(中间结果)
        question_hash = hashlib.md5(context[-2].split(":")[-1].encode('utf-8')).hexdigest()  # 取倒数第二条消息(用户的问题)
        problem = get_middle_result(question_hash)
        if not problem:
            raise HTTPException(status_code=500, detail="未找到之前的问题")
        # 生成代码的提示词
        prompt = f"生成{message}语言的{problem}代码,并解释每一步的作用。"
        # 调用LLM生成代码
        code = generate_code(prompt)
        answer = f"以下是{message}语言的{problem}代码:\n```\n{code}\n```\n需要修改吗?"
        new_state = CodeAssistantState.GENERATE_CODE
        # 存储生成的代码(中间结果)
        code_hash = hashlib.md5(f"{problem}_{message}".encode('utf-8')).hexdigest()
        set_middle_result(code_hash, code)
    elif current_state == CodeAssistantState.GENERATE_CODE:
        # 生成代码状态:用户要求修改代码,转移到MODIFY_CODE状态
        if "修改" in message or "调整" in message:
            # 获取之前生成的代码(中间结果)
            problem = context[-3].split(":")[-1]  # 取倒数第三条消息(用户的问题)
            language = context[-2].split(":")[-1]  # 取倒数第二条消息(用户的语言)
            code_hash = hashlib.md5(f"{problem}_{language}".encode('utf-8')).hexdigest()
            code = get_middle_result(code_hash)
            if not code:
                raise HTTPException(status_code=500, detail="未找到之前的代码")
            # 生成修改代码的提示词
            prompt = f"修改以下{language}代码,要求:{message}\n原代码:\n```\n{code}\n```"
            # 调用LLM修改代码
            modified_code = generate_code(prompt)
            answer = f"修改后的代码:\n```\n{modified_code}\n```\n需要进一步修改吗?"
            new_state = CodeAssistantState.MODIFY_CODE
        else:
            # 用户没有要求修改,回到INIT状态
            answer = "好的,如果你有其他问题,请随时问我。"
            new_state = CodeAssistantState.INIT
    elif current_state == CodeAssistantState.MODIFY_CODE:
        # 修改代码状态:用户继续要求修改,或者结束
        if "修改" in message or "调整" in message:
            # 获取之前修改的代码(中间结果)
            problem = context[-4].split(":")[-1]  # 取倒数第四条消息(用户的问题)
            language = context[-3].split(":")[-1]  # 取倒数第三条消息(用户的语言)
            code_hash = hashlib.md5(f"{problem}_{language}_modified".encode('utf-8')).hexdigest()
            code = get_middle_result(code_hash)
            if not code:
                raise HTTPException(status_code=500, detail="未找到之前的修改代码")
            # 生成修改代码的提示词
            prompt = f"继续修改以下{language}代码,要求:{message}\n原代码:\n```\n{code}\n```"
            # 调用LLM修改代码
            modified_code = generate_code(prompt)
            answer = f"再次修改后的代码:\n```\n{modified_code}\n```\n需要进一步修改吗?"
            new_state = CodeAssistantState.MODIFY_CODE
            # 存储再次修改的代码(中间结果)
            set_middle_result(code_hash, modified_code)
        else:
            # 用户结束修改,回到INIT状态
            answer = "好的,如果你有其他问题,请随时问我。"
            new_state = CodeAssistantState.INIT
    else:
        # 未知状态,回到INIT状态
        answer = "抱歉,我有点混乱,请重新开始。"
        new_state = CodeAssistantState.INIT

    # 4. 更新状态
    update_state(user_id, new_state)
    # 5. 更新上下文缓存(添加AI的回答)
    update_context(user_id, f"AI:{answer}")

    # 返回响应
    return AIResponse(answer=answer, state=new_state.value)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

代码解读与分析

  • 缓存操作cache.py中的函数负责管理上下文、中间结果和状态,使用Redis作为存储介质,保证了快速访问和分布式支持。
  • LLM调用llm.py中的generate_code函数调用OpenAI的GPT-3.5-turbo API,生成代码或修改代码。
  • 后端逻辑main.py中的/chat接口处理用户的聊天请求,根据当前状态(CodeAssistantState)决定下一步操作:
    • 初始状态(INIT):引导用户提问;
    • 询问问题状态(ASK_PROBLEM):获取用户的问题;
    • 询问细节状态(ASK_DETAILS):获取用户需要的语言;
    • 生成代码状态(GENERATE_CODE):调用LLM生成代码;
    • 修改代码状态(MODIFY_CODE):调用LLM修改代码。

测试效果

  • 当用户第一次发送"我想生成冒泡排序代码",AI会问"需要用什么语言?";
  • 用户发送"Python",AI会生成Python的冒泡排序代码,并问"需要修改吗?";
  • 用户发送"把它改成快速排序",AI会基于之前生成的冒泡排序代码,修改成快速排序代码——这一步没有重新调用LLM生成全部代码,而是基于中间结果(冒泡排序代码)进行修改,提升了响应速度。

实际应用场景:短期记忆在哪里发挥作用?

短期记忆在AI原生应用中的应用非常广泛,以下是几个典型场景:

1. 聊天机器人(多轮对话)

场景:用户和ChatGPT聊"旅游计划",从"推荐目的地"到"订酒店"再到"安排行程",ChatGPT能记住之前的对话,不用重复问"你想去哪里?"。
短期记忆的作用:上下文缓存存储对话历史,状态管理跟踪"旅游计划"的进度,中间结果存储推荐的目的地、酒店列表。

2. 代码助手(代码生成/修改)

场景:GitHub Copilot帮用户写"用户登录接口",用户先写了"获取用户名和密码"的代码,Copilot能记住这段代码,推荐"验证用户名和密码"的代码。
短期记忆的作用:上下文缓存存储用户写的代码片段,中间结果存储生成的代码,状态管理跟踪"登录接口"的开发进度。

3. 多模态生成(图片/视频生成)

场景:DALL·E 3帮用户生成"一只在海边的猫",用户说"把猫的颜色改成橙色",DALL·E 3能记住之前的图片描述,直接修改颜色,不用重新生成全部图片。
短期记忆的作用:上下文缓存存储用户的图片描述,中间结果存储生成的图片草稿,状态管理跟踪"图片生成"的进度。

4. 智能客服(问题处理)

场景:智能客服帮用户"查询订单状态",用户先提供了"订单号",客服能记住订单号,直接查询状态,不用重复问"你的订单号是多少?"。
短期记忆的作用:上下文缓存存储用户的订单号,中间结果存储查询的订单状态,状态管理跟踪"订单查询"的进度。

工具和资源推荐

1. 缓存工具

  • Redis:分布式键值对缓存数据库,支持多种数据结构(列表、哈希表、字符串),适合存储上下文、中间结果和状态。
  • Memcached:简单高效的键值对缓存数据库,适合小容量、高并发的场景。
  • LangChain Memory:LangChain框架中的记忆模块,提供多种记忆类型(比如ConversationBufferMemoryConversationSummaryMemory),方便快速集成到LLM应用中。

2. LLM框架

  • LangChain:用于构建LLM应用的框架,提供记忆、工具调用、链等组件,简化短期记忆的实现。
  • LlamaIndex:用于构建上下文增强的LLM应用的框架,提供ContextStore组件,管理上下文信息。

3. 监控工具

  • Prometheus:开源的监控系统,用于收集缓存的metrics(比如命中率、响应时间)。
  • Grafana:开源的可视化工具,用于展示Prometheus收集的metrics,帮助分析缓存性能。

4. 学习资源

  • 《LangChain实战:构建大语言模型应用》:讲解如何用LangChain实现短期记忆;
  • 《Redis实战》:深入讲解Redis的使用场景和优化技巧;
  • OpenAI官方文档:了解LLM的上下文窗口和调用方式。

未来发展趋势与挑战

1. 未来发展趋势

  • 智能缓存策略:用机器学习预测用户的下一个请求,提前缓存所需数据(比如用户刚问了"北京的天气",预测他接下来会问"上海的天气",提前缓存上海的天气数据)。
  • 短期与长期记忆融合:用向量数据库(比如Pinecone、Weaviate)保存长期记忆(比如用户的所有对话记录),短期记忆保存最近的上下文,结合两者提升性能(比如用户问"我去年去的公园在哪里?",短期记忆找不到,就从长期记忆中取)。
  • 边缘设备优化:在手机、平板等边缘设备上,用本地缓存(比如SQLite)存储短期记忆,减少网络请求(比如离线使用的AI语音助手,用本地缓存保存用户的对话历史)。
  • 动态容量调整:根据用户请求量动态调整短期记忆的容量(比如高峰时段增加缓存容量,低谷时段减少缓存容量),避免资源浪费。

2. 挑战

  • 缓存一致性:短期记忆中的数据可能会过时(比如用户的订单状态从"未发货"变成"已发货"),需要及时更新缓存(比如用"过期时间"或"事件驱动"的方式)。
  • 容量限制:短期记忆的容量有限(比如Redis的内存限制),需要平衡缓存命中率和内存占用(比如用LRU策略删除不常用的缓存项)。
  • 隐私问题:短期记忆中可能存储用户的敏感信息(比如订单号、身份证号),需要加密存储(比如用Redis的加密模块),避免数据泄露。

总结:学到了什么?

核心概念回顾

  • AI短期记忆:AI应用的"备菜盘",存储上下文、中间结果和状态,提升性能;
  • 上下文缓存:记住"刚才说的话",避免重复输入历史记录;
  • 中间结果存储:保存"半成品",避免重复计算;
  • 状态管理:跟踪"当前在做什么",处理复杂多步骤任务。

概念关系回顾

  • 上下文缓存是基础,中间结果存储是进阶,状态管理是高层;
  • 三者互相配合,让AI应用"跑得更快、更聪明"。

关键结论

  • 短期记忆是AI原生应用提升性能的关键,比"升级模型"更有效(因为模型升级需要更多资源,而短期记忆能减少重复劳动);
  • 选择合适的缓存策略(比如LRU)和工具(比如Redis),能大幅提升缓存命中率;
  • 结合应用场景设计短期记忆结构(比如聊天机器人需要上下文缓存,代码助手需要中间结果存储),才能发挥最大效果。

思考题:动动小脑筋

  1. 如果你开发一个实时语音助手,需要处理用户的连续语音命令(比如"打开空调→调到26度→关闭卧室灯"),如何设计短期记忆来提升响应速度?
  2. 如果你的AI应用的短期记忆容量有限(比如Redis的内存只有1GB),如何平衡缓存的命中率和内存占用?
  3. 短期记忆中的数据可能会过时(比如用户的订单状态变化),如何处理这种情况?(比如用"过期时间"或"事件驱动"的方式)
  4. 如果你开发一个多模态AI应用(比如同时处理文本、图片、语音),如何设计短期记忆来存储不同类型的中间结果?(比如文本用字符串,图片用二进制数据)

附录:常见问题与解答

Q1:短期记忆和长期记忆有什么区别?

A1:短期记忆是临时的、快速访问的(类似电脑的内存),用于处理即时需求;长期记忆是持久的、容量大的(类似电脑的硬盘),用于存储历史数据。比如,ChatGPT的短期记忆存储最近的10条对话,长期记忆存储用户的所有对话记录。

Q2:如何选择缓存策略?

A2:根据应用场景选择:

  • 如果你需要保留近期的对话,用固定窗口大小策略(比如保留最近5条消息);
  • 如果你需要保留常用的结果,用LRU策略(删除最久没用到的结果);
  • 如果你需要保留时效性强的数据,用时间窗口策略(比如保留最近10分钟的消息)。

Q3:如何监控短期记忆的性能?

A3:用监控工具(比如Prometheus+Grafana)收集以下metrics:

  • 缓存命中率(Hit Ratio):衡量缓存的效果;
  • 响应时间(Response Time):衡量使用缓存后的性能提升;
  • 缓存容量(
Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐