大多数人做 AI 助手做到第三步就卡住了换一个新会话AI 什么都不记得。这不是模型不够聪明而是你没有给 AI 设计记忆系统。这次我们直接动手从零实现一个跨会话记忆系统采用三层记忆架构让 AI 在多个会话之间仍然能记住用户的名字、偏好和历史关键信息。这套方案重点解决三个问题第一新会话不再“失忆”第二记忆按重要程度分层不是把所有聊天记录都塞进上下文第三代码结构能直接接到你自己的 FastAPI 服务、Web 应用或本地模型上。文末会给出完整的功能测试流程、接口调用示例和常见问题排查清单。1. 三层记忆系统核心能力速览能力项说明项目类型AI 助手跨会话记忆系统从零手写实现记忆架构工作记忆短期 情景记忆长期 核心记忆用户画像技术栈Python、FastAPI、OpenAI 兼容接口、SentenceTransformer、NumPy大模型接入支持本地模型Ollama / vLLM / llama.cpp和云端 OpenAI 兼容 API启动方式命令行启动 FastAPI 服务是否支持 API支持提供/chat对话接口、/memory/{user_id}记忆查询接口是否支持批量任务接口天然支持批量调用注意并发下的记忆写入排锁记忆检索向量相似度检索默认使用BAAI/bge-small-zh-v1.5中文向量模型隐私安全支持按用户维度隔离记忆生产环境需增加鉴权和删除接口适合场景个人助手、客服机器人、知识库对话、角色扮演、工作流 Agent这套设计不依赖特定大模型LLM 部分只是走标准 OpenAI 兼容协议。你可以用云端模型调试逻辑也可以切换到本地模型做隐私部署。2. 为什么 AI 助手会“失忆”三层记忆设计与选型思路普通对话接口是“无状态”的。每一次请求模型只看到当前传入的消息列表请求结束之后上下文就丢掉了。你新建一个会话之前的对话内容就完全消失。要解决这个问题核心思路是把有价值的对话信息沉淀到外部存储中在后续请求时重新注入。近期一些开源记忆框架例如 RippleMem 这类项目也在强调同一个方向记忆系统不是把更多历史消息检索出来堆给模型而是让 agent 学会“回忆”。所谓回忆就是按重要程度提取、存储、检索和遗忘。基于这个思路我把记忆拆成三层。2.1 第一层工作记忆短期会话上下文工作记忆就是当前会话内的消息列表。它负责维持对话的连贯性直接拼到 prompt 里交给大模型。这层的特点是时效性强、信息密度低。如果全部会话历史都塞进去很快会超出模型上下文窗口。所以必须设置上限比如最近 20 条消息超过之后就丢弃最旧的消息或者用 LLM 对旧消息做摘要压缩。2.2 第二层情景记忆跨会话事件检索情景记忆负责记录有价值的历史信息例如用户在上一段对话里提到过“最近在准备考研”“喜欢刘慈欣的小说”“项目下周上线”。这些信息不会直接全部塞进 prompt而是先做向量化存储。用户发起新问题时把当前问题转成向量从已有记忆里检索出语义最相关的 top_k 条再拼入系统提示词。这层解决的是“跨会话回忆”问题用户在新会话中问起旧话题AI 能通过向量检索找到相关历史而不是靠模型自己去猜。2.3 第三层核心记忆用户画像与稳定事实核心记忆保存的是用户的基本事实和稳定偏好例如姓名、职业、所在地、兴趣爱好、说话风格偏好。这些信息每次对话都应该被加载作为系统提示词的一部分。和情景记忆不同核心记忆不依赖语义检索而是按用户维度直接读取。它的特点是稳定、长期、结构化适合用 JSON 文件或数据库表维护。场景上可以理解为“AI 对用户的长期认知”。三层记忆的配合关系如下核心记忆保证 AI 始终知道“用户是谁”情景记忆保证 AI 能回忆起“过去聊过什么”工作记忆保证当前这轮对话上下文完整。三层各司其职不会互相干扰。3. 环境准备与前置条件3.1 基础依赖本次演示使用 Python 3.10 以上版本。项目核心依赖如下fastapi uvicorn openai numpy sentence-transformers创建项目目录并安装依赖mkdir ai-memory-system cd ai-memory-system pip install fastapi uvicorn openai numpy sentence-transformers如果你本机没有可用的 GPU也没问题。SentenceTransformer 的向量化可以用 CPU 跑bge-small-zh-v1.5是轻量级模型CPU 推理速度可以接受。真正的大模型推理走的是外部接口不占用本地推理资源。3.2 LLM 接口说明本项目的LLMClient直接使用 OpenAI 兼容接口格式。我建议先准备一个可用的模型服务例如本地 Ollamaollama run qwen2.5:7bOllama 启动后默认暴露的接口是http://127.0.0.1:11434/v1可以通过 OpenAI SDK 直接调用。如果你使用的是其他模型服务只要它兼容/v1/chat/completions接口格式都可以通过修改base_url接入。3.3 向量检索方案为了便于理解原理这版实现没有引入独立的向量数据库而是用 SentenceTransformer 生成向量再用 NumPy 计算余弦相似度最后把向量和原始文本一起持久化到 JSON 文件。这种实现方式的好处是零额外依赖适合学习原理和做功能验证。如果未来记忆量达到几万条以上建议替换为 FAISS 或 ChromaDB检索性能会更好。这个点会在后续章节详细说明。4. 代码实现从零搭建三层记忆系统先给出完整项目结构后面按文件逐个实现ai-memory-system/ ├── main.py ├── requirements.txt ├── memory/ │ ├── __init__.py │ ├── agent.py │ ├── llm_client.py │ ├── working_memory.py │ ├── episodic_memory.py │ └── semantic_memory.py__init__.py保持空文件即可。下面直接写核心代码。4.1 LLM 客户端封装封装一个兼容 OpenAI 协议的大模型调用客户端。这样后续替换模型时只需要改配置参数不用改业务代码。# memory/llm_client.py from openai import OpenAI class LLMClient: def __init__( self, base_url: str http://127.0.0.1:11434/v1, api_key: str ollama, model: str qwen2.5:7b, ): self.model model self.client OpenAI(base_urlbase_url, api_keyapi_key) def chat(self, messages: list[dict], temperature: float 0.7) - str: resp self.client.chat.completions.create( modelself.model, messagesmessages, temperaturetemperature, ) return resp.choices[0].message.content如果你的模型服务不需要 api_key随便填一个占位字符串即可。OpenAI SDK 的base_url参数就是服务地址本地和云端用法一致。4.2 第一层实现工作记忆模块工作记忆负责保存当前会话的消息列表。这里用 JSON 文件持久化文件名为{session_id}.json。超过最大消息数时丢弃最旧的消息。# memory/working_memory.py import json import time from pathlib import Path class WorkingMemory: def __init__(self, storage_dir: str data/sessions, max_messages: int 20): self.storage_dir Path(storage_dir) self.storage_dir.mkdir(parentsTrue, exist_okTrue) self.max_messages max_messages def _file_path(self, session_id: str) - Path: return self.storage_dir / f{session_id}.json def get_messages(self, session_id: str) - list[dict]: file_path self._file_path(session_id) if not file_path.exists(): return [] data json.loads(file_path.read_text(encodingutf-8)) return data.get(messages, []) def add_message(self, session_id: str, role: str, content: str, user_id: str ): file_path self._file_path(session_id) if file_path.exists(): data json.loads(file_path.read_text(encodingutf-8)) else: data {session_id: session_id, user_id: user_id, messages: []} data[messages].append({ role: role, content: content, timestamp: time.time(), }) if len(data[messages]) self.max_messages: data[messages] data[messages][-self.max_messages:] file_path.write_text( json.dumps(data, ensure_asciiFalse, indent2), encodingutf-8, )生产环境建议把工作记忆放在 Redis 这类带过期策略的存储里会话关闭后自动释放。演示项目为了部署简单直接用文件存储。4.3 第二层实现情景记忆与向量检索情景记忆是跨会话记忆的核心模块。它负责保存“值得长期记住”的信息并提供向量检索能力。# memory/episodic_memory.py import json import time import uuid from pathlib import Path import numpy as np from sentence_transformers import SentenceTransformer class EpisodicMemory: def __init__( self, storage_dir: str data/episodic, model_name: str BAAI/bge-small-zh-v1.5, ): self.storage_dir Path(storage_dir) self.storage_dir.mkdir(parentsTrue, exist_okTrue) self.index_file self.storage_dir / index.json self.embedder SentenceTransformer(model_name) self.items self._load() def _load(self) - list[dict]: if self.index_file.exists(): return json.loads(self.index_file.read_text(encodingutf-8)) return [] def _save(self): self.index_file.write_text( json.dumps(self.items, ensure_asciiFalse, indent2), encodingutf-8, ) def add_memory( self, user_id: str, content: str, memory_type: str event, ): vector self.embedder.encode(content).tolist() item { id: str(uuid.uuid4()), user_id: user_id, content: content, memory_type: memory_type, vector: vector, created_at: time.time(), } self.items.append(item) self._save() def recall( self, query: str, user_id: str | None None, top_k: int 5, ) - list[dict]: query_vec self.embedder.encode(query) scored [] for item in self.items: if user_id and item[user_id] ! user_id: continue v np.array(item[vector]) cosine float( np.dot(query_vec, v) / (np.linalg.norm(query_vec) * np.linalg.norm(v) 1e-8) ) scored.append((cosine, item)) scored.sort(keylambda x: x[0], reverseTrue) return [item for _, item in scored[:top_k]] def remove_by_user(self, user_id: str): self.items [item for item in self.items if item[user_id] ! user_id] self._save()向量化模型首次运行时会自动从 Hugging Face 下载权重。你需要确保本机网络能正常访问模型仓库或者预先配置好镜像源。4.4 第三层实现核心画像记忆核心记忆按用户 ID 保存稳定的画像信息。数据结构分为facts和preferences两部分分别存储事实性信息和偏好性信息。# memory/semantic_memory.py import json from pathlib import Path class SemanticMemory: def __init__(self, storage_dir: str data/semantic): self.storage_dir Path(storage_dir) self.storage_dir.mkdir(parentsTrue, exist_okTrue) def _file_path(self, user_id: str) - Path: return self.storage_dir / f{user_id}.json def get_profile(self, user_id: str) - dict: file_path self._file_path(user_id) if not file_path.exists(): return {user_id: user_id, preferences: {}, facts: []} return json.loads(file_path.read_text(encodingutf-8)) def update_profile( self, user_id: str, new_facts: list[str] | None None, preferences: dict | None None, ): profile self.get_profile(user_id) if preferences: profile[preferences].update(preferences) if new_facts: existing set(profile[facts]) existing.update(new_facts) profile[facts] list(existing) file_path self._file_path(user_id) file_path.write_text( json.dumps(profile, ensure_asciiFalse, indent2), encodingutf-8, ) def remove_user(self, user_id: str): file_path self._file_path(user_id) if file_path.exists(): file_path.unlink()核心记忆是“每次对话都注入”的所以它的数据质量直接影响 AI 的用户认知。后文会讲如何用大模型自动抽取并更新画像。4.5 记忆抽取与更新策略跨会话记忆不是把对话原封不动存起来而是让 agent 学会提炼。每一轮对话结束后我会调用一次大模型让它判断这段对话中有没有值得长期记忆的信息然后输出结构化 JSON。# memory/agent.py 中的记忆抽取方法 import json import logging import re logger logging.getLogger(__name__) def extract_memories(self, user_id: str, user_message: str, assistant_reply: str): prompt f你是记忆提取器。请从用户消息中提取需要长期记忆的信息。 输出格式必须是 JSON不要输出其他内容 {{ preferences: [用户偏好1, 用户偏好2], facts: [用户事实1, 用户事实2] }} 如果没有值得记忆的信息输出{{preferences: [], facts: []}} 用户消息{user_message} 助手回复{assistant_reply} try: resp self.llm_client.chat( [ {role: system, content: 你是一个严格的记忆提取器只输出 JSON。}, {role: user, content: prompt}, ], temperature0, ) json_str resp.strip() if json_str.startswith(): json_str re.sub(r^.*$, , json_str, flagsre.MULTILINE).strip( \n ) data json.loads(json_str) preferences data.get(preferences, []) facts data.get(facts, []) if preferences: self.semantic_memory.update_profile( user_id, preferences{兴趣偏好: preferences}, ) if facts: self.semantic_memory.update_profile(user_id, new_factsfacts) for item in preferences facts: self.episodic_memory.add_memory( user_id, item, memory_typeextracted, ) except Exception as e: logger.warning(记忆抽取失败: %s, e)这段代码的核心是让大模型自己判断“什么值得记”。比如用户说“我叫小明”大模型会抽出事实用户姓名是小明更新到核心画像中。用户说“我喜欢看科幻小说”会抽出偏好并同时写入情景记忆方便后续相关性检索。这里是演示版生产环境可以每隔几轮做一次批量抽取减少大模型调用次数。4.6 记忆编排把三层串成统一 Agent现在把三层记忆组合起来实现完整的对话编排流程。流程顺序是读取核心画像检索情景记忆获取工作记忆历史组装系统提示词调用大模型最后保存消息并抽取新记忆。# memory/agent.py from .episodic_memory import EpisodicMemory from .llm_client import LLMClient from .semantic_memory import SemanticMemory from .working_memory import WorkingMemory class MemoryAgent: def __init__( self, llm_client: LLMClient, working_memory: WorkingMemory, episodic_memory: EpisodicMemory, semantic_memory: SemanticMemory, ): self.llm_client llm_client self.working_memory working_memory self.episodic_memory episodic_memory self.semantic_memory semantic_memory def _build_system_prompt( self, profile: dict, related_memories: list[dict], ) - str: prompt_parts [你是一个有长期记忆的 AI 助手请用自然、简洁的方式回答用户。] if profile.get(facts): prompt_parts.append(关于用户已经确认的事实) for fact in profile[facts]: prompt_parts.append(f- {fact}) if profile.get(preferences): prompt_parts.append(用户偏好) for key, values in profile[preferences].items(): if isinstance(values, list): prompt_parts.append(f- {key}: {, .join(values)}) else: prompt_parts.append(f- {key}: {values}) if related_memories: prompt_parts.append(以下是与你当前问题相关的历史记忆) for mem in related_memories: prompt_parts.append(f- [{mem[memory_type]}] {mem[content]}) prompt_parts.append(回答时不要说明你使用了记忆系统。) return \n.join(prompt_parts) def chat(self, user_id: str, session_id: str, user_message: str) - dict: profile self.semantic_memory.get_profile(user_id) related self.episodic_memory.recall( user_message, user_iduser_id, top_k5, ) history self.working_memory.get_messages(session_id) system_prompt self._build_system_prompt(profile, related) messages [ {role: system, content: system_prompt}, *history, {role: user, content: user_message}, ] reply self.llm_client.chat(messages) self.working_memory.add_message(session_id, user, user_message, user_id) self.working_memory.add_message(session_id, assistant, reply, user_id) self.extract_memories(user_id, user_message, reply) return { reply: reply, memories_used: [mem[content] for mem in related], }注意这里有一个细节history是当前会话历史消息user_message是当前这条新消息。组装 messages 时先放系统提示词再放历史消息最后放当前用户消息顺序不能颠倒。4.7 FastAPI 接口服务最后用 FastAPI 把整个 Agent 暴露成 HTTP 接口。# main.py from fastapi import FastAPI from pydantic import BaseModel from memory.agent import MemoryAgent from memory.episodic_memory import EpisodicMemory from memory.llm_client import LLMClient from memory.semantic_memory import SemanticMemory from memory.working_memory import WorkingMemory app FastAPI(titleAI Memory System) class ChatRequest(BaseModel): user_id: str session_id: str message: str llm_client LLMClient( base_urlhttp://127.0.0.1:11434/v1, api_keyollama, modelqwen2.5:7b, ) working_memory WorkingMemory(storage_dirdata/sessions, max_messages20) episodic_memory EpisodicMemory(storage_dirdata/episodic) semantic_memory SemanticMemory(storage_dirdata/semantic) agent MemoryAgent( llm_clientllm_client, working_memoryworking_memory, episodic_memoryepisodic_memory, semantic_memorysemantic_memory, ) app.post(/chat) def chat(req: ChatRequest): result agent.chat(req.user_id, req.session_id, req.message) return { session_id: req.session_id, reply: result[reply], memories_used: result[memories_used], profile: semantic_memory.get_profile(req.user_id), } app.get(/memory/{user_id}) def get_memory(user_id: str): profile semantic_memory.get_profile(user_id) memories [ item for item in episodic_memory.items if item[user_id] user_id ] safe_memories [ { content: item[content], memory_type: item[memory_type], created_at: item[created_at], } for item in memories ] return {profile: profile, memories: safe_memories}启动服务uvicorn main:app --host 127.0.0.1 --port 8000启动后控制台会输出 FastAPI 访问地址。第一次运行EpisodicMemory会加载向量模型如果模型还没下载到本地会先显示下载进度耐心等待即可。5. 跨会话记忆功能测试服务启动后用 Python 脚本模拟多个会话验证跨会话记忆是否真正生效。下面给出完整的测试流程。5.1 测试场景一跨会话记住用户基本信息先新建一个会话告诉 AI 自己的名字然后新建另一个会话询问 AI 是否记得自己。import requests API http://127.0.0.1:8000 # 第一个会话告诉 AI 基本信息 r1 requests.post( f{API}/chat, json{ user_id: u_001, session_id: session_1, message: 我叫小明是一名后端工程师最近在准备考研。, }, ) print(会话1, r1.json()[reply]) # 第二个会话新 session跨会话提问 r2 requests.post( f{API}/chat, json{ user_id: u_001, session_id: session_2, message: 你还记得我是谁吗, }, ) print(会话2, r2.json()[reply])预期现象第二次会话中模型能说出“你是小明一名后端工程师”。判断标准是模型回复中出现之前提到的姓名和职业信息。如果模型没有记住先调用/memory/u_001接口查看核心画像数据是否已经写入。5.2 测试场景二跨会话更新用户偏好继续模拟用户偏好变化。用户先表示喜欢科幻小说之后表示开始看推理小说最后请求推荐。# 会话 A表达兴趣 requests.post( f{API}/chat, json{ user_id: u_001, session_id: session_3, message: 我最近特别喜欢看科幻小说尤其是刘慈欣的作品。, }, ) # 会话 B表达新兴趣 requests.post( f{API}/chat, json{ user_id: u_001, session_id: session_4, message: 不过我最近也开始看推理小说了东野圭吾的看得比较多。, }, ) # 会话 C跨会话推荐 r3 requests.post( f{API}/chat, json{ user_id: u_001, session_id: session_5, message: 结合我的兴趣推荐几本我可能会喜欢的书。, }, ) print(会话5, r3.json()[reply]) print(使用到的记忆, r3.json()[memories_used])这里要注意推荐结果好不好除了看记忆系统还看底层大模型的理解能力。如果模型推荐结果里同时涵盖科幻和推理题材说明记忆抽取、存储、检索链条是通的。5.3 测试场景三长期记忆检索相关性直接调用情景记忆的recall方法验证向量检索是否返回正确内容。top_memories episodic_memory.recall( 用户喜欢什么类型的小说, user_idu_001, top_k3, ) for mem in top_memories: print(f[{mem[memory_type]}] {mem[content]})预期输出应包含“科幻小说”“刘慈欣”“推理小说”“东野圭吾”等相关记忆。如果检索结果不相关可以调整top_k或更换向量模型。5.4 测试场景四长对话上下文压缩模拟连续多轮对话验证只保留最近消息时对话不会中断。for i in range(30): response requests.post( f{API}/chat, json{ user_id: u_001, session_id: long_session, message: f这是第 {i 1} 轮测试消息, }, ) print(f第 {i 1} 轮完成)如果max_messages设置为 20那么存储文件里最多只有最近的 20 条消息。这样能保证请求体不会无限膨胀。这个测试主要验证系统在高频对话下是否稳定以及工作记忆淘汰机制是否生效。6. 接口 API 与扩展接入当前服务提供两个核心接口POST /chat和GET /memory/{user_id}。POST /chat请求体{ user_id: u_001, session_id: session_1, message: 你好 }响应体{ session_id: session_1, reply: 你好有什么可以帮你的, memories_used: [], profile: { user_id: u_001, preferences: {}, facts: [] } }GET /memory/{user_id}用于调试返回该用户当前的核心画像和情景记忆列表。开发阶段非常有用。curl 调用示例curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d { user_id: u_001, session_id: curl_test, message: 我叫小明 }批量任务场景下你可以把一批消息按user_id和session_id分好循环调用/chat接口。注意目前的文件存储没有加锁高并发写入同一个用户时可能出现数据竞争生产环境需要把工作记忆和核心画像迁移到 Redis 或 PostgreSQL。7. 资源占用与性能观察性能观察主要看三个部分向量化耗时、大模型推理耗时、记忆文件读写耗时。SentenceTransformer默认在 CPU 上运行bge-small属于轻量级中文向量模型单条短文本向量化一般在几十毫秒到几百毫秒之间。如果你的服务器有多核 CPU可以设置torch.set_num_threads来优化。实际耗时以本机测试为准我这边不给出固定数值。大模型推理耗时取决于你使用的模型服务。本地 7B 量化模型和云端大模型差距明显但记忆系统本身引入的额外 token 并不多主要就是系统提示词里的画像和检索结果。控制检索条数top_k是控制 token 消耗最直接的方式。显存和内存占用方面本文的项目本体几乎没有显存需求向量模型内存占用很小。但如果你在本地跑 7B 或更大参数的模型显存占用会很高具体数值取决于量化等级、上下文长度和批处理大小。建议先用小模型验证记忆逻辑确认没有问题后再换大模型。日志规范建议每个/chat请求都打印请求耗时、memories_used数量和profile长度。这样能快速判断是记忆检索拖慢请求还是大模型推理拖慢请求。8. 常见问题与排查清单问题现象可能原因排查方式解决方案新会话仍然记不住用户核心画像未写入或未注入调用/memory/{user_id}查看画像检查记忆抽取流程是否报错看系统提示词是否包含画像检索到的记忆与问题无关embedding 模型语义能力不够top_k过大打印memories_used列表降低top_k或更换更强的向量模型请求超时大模型推理慢或向量化耗时过长查看接口日志中的耗时分布切换更快的模型服务减少历史消息数工作记忆被截断max_messages设置过小检查 session 文件中的消息数量适当调大max_messages或开启摘要压缩记忆抽取结果为空大模型未能按 JSON 格式输出查看日志中的抽取报错优化 extraction prompt增加 few-shot 示例并发写入文件出错文件存储没有加锁观察日志是否有写入冲突迁移到 Redis或为文件写入加线程锁向量模型下载失败网络无法访问模型仓库检查模型下载日志手动下载模型权重到本地缓存目录用户删除记忆后仍恢复旧记忆工作记忆、情景记忆、核心记忆删除不彻底检查三个 data 目录删除用户时清理三层存储并在生产环境增加 DELETE 接口排查这类记忆系统问题核心原则是先确认数据到底存没存进去再确认存进去的数据有没有被注入到 prompt最后才能怀疑模型能力。9. 最佳实践与合规使用建议第一每次对话都调用记忆抽取会增加大量大模型请求生产环境建议做一个节流策略例如每 5 轮做一次抽取或者在用户主动表达关键信息时才触发。第二记忆数据的去重和合并很重要。用户在不同时间说出相似偏好如果不做去重画像会越来越冗余。可以定期让大模型对画像做一次合并压缩。第三向量检索在数据量小时用 NumPy 扫描没问题但记忆量超过上万条后应切换到 FAISS 或向量数据库。隐私和数据合规方面必须重点强调记忆系统存储的是用户对话内容和个人信息上线前必须获得用户的明示授权并且提供用户自助查看和删除记忆的入口。涉及姓名、联系方式、身份证号、人脸、声音等敏感信息时建议在记忆抽取阶段就直接过滤或脱敏不要进入长期存储。测试阶段应使用虚构数据不要拿真实用户信息做实验。如果你的助手涉及换脸、声音克隆、数字人等能力还需要额外确认肖像权和声音授权避免侵权风险。接口服务默认绑定在127.0.0.1生产环境如果要暴露到局域网或公网必须加鉴权、限流和 HTTPS否则用户记忆数据会直接泄露。10. 结束后可以继续做什么这套三层记忆系统虽然实现简单但架构上是完整的。你可以继续扩展几个方向一是在系统提示词中加入记忆写入和更新的过程反思让模型自己判断哪些记忆已经失效需要替换二是增加时间衰减因子让太久远且不再被检索的记忆自动弱化三是做一个记忆可视化页面让用户直接看到 AI 记住了自己哪些信息。如果你打算接入微信、飞书或 Web 小程序只需要在回调逻辑里调用这个/chat接口把用户的 open_id 映射为user_id把单聊会话关联到固定的session_id跨会话记忆就能直接复用。建议先收藏本文的实现框架在本地跑通一次完整测试后再接入具体业务场景。
