恒美微站 Logo 恒美微站
  • 首页
  • 关于我们
  • 建站服务
  • 主题模板
  • 案例展示
  • 资讯中心
  • 联系我们

企业级Agent记忆系统实战:Langgraph与DeepAgents构建短期与长期记忆

  • 首页
  • 资讯中心
  • /
  • 企业级Agent记忆系统实战:Langgraph与DeepAgents构建短期与长期记忆

相关资讯

STM32实战:基于Y01-3IN1模块的空气质量监测系统设计与实现 2026/9/7 6:19:03
AI生成PPT后处理全攻略:内容审核、版式优化与场景定制 2026/9/7 6:19:03
批处理优化引发的连锁故障:定时任务依赖治理与数据库性能调优指南 2026/9/7 6:19:03

最新资讯

事业单位面试九类题型底层逻辑与答题框架全解析
Uber Maven构建测试实战:从Maven环境搭建到Spark项目CI门禁
AI智能体驱动CATIA V5参数化建模:从自然语言到自动改型
豆包AI漫剧全流程实战:从脚本、分镜到新海诚风格成片
Tinfoil安全enclaves:OpenWhispr如何通过BYOK实现机密云转录
FunASR 语音识别实战:3 步转写会议录音并自动标注说话人

今日推荐

基于YOLOv8和PyQt5的麦穗稻穗检测识别系统设计与实现
UL 1642锂电池安全标准全解析:测试项目、认证流程与避坑指南
BS EN 13814-1-2019游乐设施安全标准:设计与制造核心要点解析

本周热门

超人会飞不算本事:系统稳定依赖清晰规则与边界设计
超人VS蜘蛛侠:拆解超级IP的影响力与传播方法论
基于CNN的调制信号识别:MATLAB实现时频图分类实战

本月精选

自研推理加速器Redwood:两周内实现PyTorch模型高效部署的实战教程
V4L2摄像头采集实战:从camera_client.rar到出图全流程解析
从“谁发明了钢琴键”到知识问答智能体:RAG与记忆工程实践

企业级Agent记忆系统实战:Langgraph与DeepAgents构建短期与长期记忆

发布时间:2026/9/7 6:19:03
企业级Agent记忆系统实战:Langgraph与DeepAgents构建短期与长期记忆 大模型做 Agent聊到最后基本都会卡在同一个问题上记忆。上下文一长就丢人设用户隔几天再来就认不出业务数据没法跨会话复用——这些问题不解决Agent 就只能在 Demo 里待着。这次来看的方案来自码士集团分享的企业级 Agent 记忆系统技术栈是 Langchain Langgraph DeepAgents一套代码同时覆盖短期记忆、长期记忆并且拿电商场景做了完整演示。先给出我对这套方案的整体判断它没有把记忆做成一个孤立的“大模型套娃”而是把记忆拆成两层。短期记忆交给 Langgraph 的线程化状态Checkpointer长期记忆交给“记忆抽取 向量存储 检索注入”业务编排交给 DeepAgents 的 Supervisor 机制。每一层都可以单独替换、单独测试这正是企业级系统最看重的模块化能力。这篇文章会按照“为什么需要记忆 - 短期记忆怎么落 - 长期记忆怎么落 - DeepAgents 怎么编排 - 电商案例怎么跑通”的顺序展开最后补上接口封装、批量任务、资源占用和排错清单。适合正在做智能客服、知识库问答、营销助手或者准备把 Agent 接入生产业务的人可以收藏备用。1. 核心能力速览先把关键信息放在前面。这套 Agent 记忆系统不是单一工具而是一条完整技术链路核心能力如下能力项说明系统类型企业级 Agent 记忆系统包含短期记忆、长期记忆与多 Agent 编排技术栈Langchain模型调用与工具链、Langgraph状态图与短期记忆、DeepAgents多 Agent 编排短期记忆基于 Langgraph Checkpointer按 thread_id 保存会话状态支持用户级会话隔离和恢复长期记忆对对话内容做结构化抽取生成用户画像与偏好记录写入向量库在后续对话中按相关性检索注入业务记忆通过工具方式接入订单、商品、库存等业务数据作为 Agent 的“实时业务上下文”多 Agent 编排基于 DeepAgents 的 AgentSupervisor 模式一个 Supervisor 管理多个 Tool AgentsLLM 接入默认走 LangChain 的 OpenAI 兼容接口也可以替换为本地模型、国产模型、私有化部署模型存储依赖会话记忆用内存 / SQLite Checkpointer长期记忆用向量库Chroma、Redis、Milvus 等均可接口能力可通过 FastAPI 将记忆 Agent 包装成 HTTP 接口支持 curl / Python 调用批量任务长文本记忆补全、历史会话批量抽取、客服会话批量恢复均可做批处理可视化跟踪Langgraph 原生支持状态图可视化方便观察每一步“读到了什么记忆、调了什么工具”部署方式命令启动依赖 Python 环境和模型服务无需独立 Agent 服务器适用场景电商客服、企业知识库、销售助手、营销推荐、用户运营等这里要提醒一点如果你只想要一个“给大模型加系统提示词”的简单方案这套体系对你来说偏重。它解决的是多轮、跨会话、多业务系统联动时的记忆一致性问题价值出现在场景复杂度足够高的时候。2. Agent 记忆体系设计思路2.1 Agent 需要哪几类记忆业界对 Agent 记忆的分类不完全统一但按这套方案可以分成三层短期记忆Working Memory当前会话内的上下文比如用户本轮问了什么、上一轮回答了什么。这是对话连贯性的基础。长期记忆Long-term Memory跨会话的知识沉淀比如用户的偏好、身份、历史需求、关注点。这解决“用户隔一天再来Agent 还记得他”的问题。业务记忆Business Context用户此刻的业务状态比如最新订单、物流节点、优惠券状态、售后进度。这类信息不适合长期固化管理更适合按需从业务系统实时查询然后注入上下文。这套记忆系统的关键设计就是把这三类记忆用不同技术承载短期记忆靠 Langgraph 状态和 Checkpointer长期记忆靠向量库业务记忆靠工具调用。三者不混在一起逻辑清晰排查也容易。2.2 Langgraph 在记忆系统中的定位Langgraph 本质上是一个“可持久化状态”的图执行框架。Agent 的每次运行都被拆成节点节点之间通过共享状态传递信息。在这个方案里节点负责执行具体工作比如“判断用户意图”、“生成回复”、“更新记忆”。状态负责保存对话消息和业务数据。Checkpointer 负责把状态持久化到存储后端而这是短期记忆的技术载体。因为 Langgraph 天然支持按 thread_id 隔离状态所以短期记忆不需要额外写一套 Redis 缓存逻辑。只要在编译图时挂一个 Checkpointer每次调用时带上 thread_idLanggraph 就会自动把会话历史保存下来并在下一次调用时恢复。这是整个记忆系统里最省事、最可靠的一层。2.3 DeepAgents 在记忆系统中的定位DeepAgents 是 LangChain 生态在多 Agent 编排方向的一个重要实现。它没有另起炉灶而是基于 Langgraph 的底层机制扩展出 AgentSupervisor 模式。一个 Supervisor 负责理解用户意图、分配任务、汇聚结果多个 Tool Agents 分别承担不同职责比如客服对话、订单查询、商品推荐、售后处理。在记忆系统中DeepAgents 的价值在于它让“记忆”不再只是跟在主 Agent 后面的一堆上下文而是变成所有子 Agent 都可以读取的共享信息层。Supervisor 在派发任务前先决定“这个问题需要读取哪些记忆”Tool Agents 在执行任务时也可以主动触发记忆更新。记忆不是某个 Agent 独有的而是整个多 Agent 团队共享的基础设施。2.4 记忆系统不是“把更多资料检索出来”现在很多 Agent 的“记忆”其实做成了挂载一堆文档再检索。但真正的长期记忆应该更接近“学会回忆”。这套方案里有几个和单纯 RAG 完全不同的设计点记忆是结构化的不是把历史对话塞进上下文而是抽取出“用户是谁、喜欢什么、最近买了什么、有哪些未解决的问题”以结构化数据存下来。记忆是选择写入的不是每句话都入库而是由抽取模型判断“这段对话里有没有值得长期记住的信息”有才写入。这样不会产生记忆污染。记忆是主动召回的回答之前先判断“当前问题需要哪些历史记忆”再精确检索而不是把积累的几个月的记忆全部堆给大模型。记忆是允许遗忘和合并的后续检索到相似记忆时可以合并冲突记录、删除过期偏好保证用户画像不被旧数据干扰。这个思路和社区最近讨论较多的 RippleMem、Agent 记忆反思机制是同一个方向核心都是让 Agent 具备“回忆能力”而不是“检索能力”。3. 适用场景与使用边界3.1 适合谁用从这套方案的技术选型来看最适合的是以下四类场景电商客服跨会话记住用户的收货偏好、尺码、风格偏好减少用户反复描述需求。企业知识库助手记住员工身份、所属部门、近期项目让同一个知识库对不同人给出差异化回答。销售与营销助手记录客户跟进状态、关注点、历史报价减少销售手工翻阅记录的时间。用户运营系统基于用户历史行为和偏好让推荐和触达内容更个性化。3.2 不适合什么场景单轮问答没有跨会话需求时长期记忆是多余开销。极低成本场景记忆抽取、向量检索、状态持久化都会增加 Token 消耗和存储成本。强实时场景如果要求毫秒级响应记忆读取和业务工具编排的全链路延迟需要专门优化普通方案可能不满足。一次性工具类 Agent比如只做一次代码解释或格式转换的 Agent根本不存在“认识用户”的需求。3.3 合规与安全边界涉及用户记忆就必须把隐私合规放在第一位。下面是这套方案在落地时必须要处理的边界明确告知记录用户偏好和历史行为前应在产品协议或交互流程中告知用户并在合理范围内征得同意。脱敏处理身份证号、手机号、家庭住址、支付信息等敏感字段不建议写入长期记忆必要时先做脱敏或只存业务侧脱敏 ID。数据可删除记忆系统必须提供“删除指定用户记忆”的能力否则用户提出注销请求时无法响应。授权边界涉及人脸、声音、肖像等生物识别或个性化生成内容时必须确认已获得明确授权不能拿未授权素材做画像或跨场景使用。业务数据权限接入订单、优惠券等领域数据时要遵循最小权限原则Agent 只能读取它实际需要的数据。4. 环境准备与前置条件4.1 软件环境这套方案是基于 Python 生态的建议准备好以下环境项目建议配置操作系统Linux / macOS / WindowsWindows 建议使用 WSL2 或 PowerShellPython3.10 或 3.11建议用 conda 或 uv 建独立虚拟环境模型服务OpenAI 兼容 API。可以是 OpenAI 官方服务、国产模型、本地部署的 vLLM / Ollama 服务向量数据库小型场景用 Chroma生产场景用 Redis 或 Milvus可选组件LangSmith可选用于链路跟踪、FastAPI服务化注意LangChain、Langgraph、DeepAgents 的版本迭代比较快安装时建议直接安装最新稳定版。版本差异导致的接口变动以官方文档为准。4.2 创建虚拟环境并安装依赖# 创建虚拟环境 python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate # 安装核心依赖这里给出通用安装命令版本以官方最新为准 pip install langchain pip install langchain-openai pip install langgraph pip install langgraph-checkpoint-sqlite pip install langchain-deepagents pip install chromadb pip install fastapi uvicorn如果你的项目需要数据库持久化和 API 服务可以额外安装pip install sqlalchemy pip install pydantic4.3 模型与存储配置在项目中新建.env文件按实际使用的模型服务填写OPENAI_API_KEYyour_api_key_here OPENAI_BASE_URLhttps://your-model-service.example.com/v1 LLM_MODELgpt-4o-mini # 记忆向量库目录 MEMORY_DB_PATH./memory_db CHECKPOINT_DB_PATH./checkpoints.db这里的OPENAI_BASE_URL可以指向任何 OpenAI 兼容服务。如果你用本地模型可以把地址改成http://127.0.0.1:8000/v1。后面代码统一通过环境变量读取模型配置方便切换厂商。5. 短期记忆落地Langgraph Checkpoint 会话持久化5.1 为什么短期记忆要用 Checkpointer在对话应用中最常见的做法是手动把历史消息拼进 prompt用户一刷新就丢了。Langgraph 的 Checkpointer 解决了两个核心问题自动保存每次图执行结束状态自动写入存储后端。按会话恢复带上 thread_id就能把历史状态完整恢复。这意味着你不需要自己设计“会话历史”的数据库表不需要手动 truncate 历史Langgraph 会按图执行状态来管理。5.2 带 Checkpointer 的最小客服 Agentfrom langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import MessagesState from langchain_openai import ChatOpenAI # 初始化模型参数通过环境变量读取 model ChatOpenAI(modelgpt-4o-mini) # 定义一个最简单的 Agent 节点 def agent_node(state: MessagesState): response model.invoke(state[messages]) return {messages: [response]} # 构建图 graph StateGraph(MessagesState) graph.add_node(agent, agent_node) graph.add_edge(START, agent) graph.add_edge(agent, END) # 注入 Checkpointer这是短期记忆的关键 checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) # 第一轮对话用户自我介绍 config {configurable: {thread_id: user-001}} app.invoke( {messages: [(user, 我叫小李喜欢黑色机械键盘预算一千以内。)]}, config, ) # 第二轮对话Agent 应该还能记得用户身份和偏好 result app.invoke( {messages: [(user, 帮我看一下合适的键盘品牌)]}, config, ) print(result[messages][-1].content)运行逻辑很简单同一个thread_id下的所有调用共享状态Agent 直接利用历史消息生成回答。这是短期记忆最小可用的实现。5.3 从内存 Checkpointer 切换到 SQLite生产环境不能只用MemorySaver服务一重启记忆就没了。切换 SQLite 版本from langgraph.checkpoint.sqlite import SqliteSaver # DB 文件会持久化到磁盘 with SqliteSaver.from_conn_string(./checkpoints.db) as checkpointer: app graph.compile(checkpointercheckpointer) # 后续逻辑与内存版一致切换后即使服务重启只要 thread_id 不变会话状态还能恢复。5.4 短期记忆验证清单同一 thread_id 连续两轮对话Agent 能回答“我是谁”。换一个 thread_idAgent 不再记得上一轮的内容确认会话隔离生效。重启服务后用 SQLite Checkpointer 仍能恢复历史会话。历史消息越长观察是否出现延迟或上下文超限设计截断策略。6. 长期记忆落地用户画像与偏好存储短期记忆解决的是“当前会话连贯”长期记忆解决的是“跨会话认识用户”。长期记忆的核心流程是抽取 - 存储 - 检索 - 注入。6.1 定义长期记忆的数据模型长期记忆不是原样保存对话文本而是保存结构化的用户画像。可以按下面的模式设计{ user_id: user-001, preferences: { keyboard_style: mechanical, color: black, budget_range: 1000 }, facts: [ 用户叫小李, 用户喜欢机械键盘, 用户偏好黑色 ], last_interaction: 2025-02-15 10:30:00 }这里的原则是能结构化的字段尽量结构化不能结构化的信息以短句形式保存为 facts再配合向量检索召回。6.2 从对话中抽取记忆每次对话结束后可以调用一个“记忆抽取模型”判断这段对话里有没有值得长期记住的信息from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field class MemoryItem(BaseModel): memory_type: str Field(description记忆类型如 preference / fact / requirement) content: str Field(description记忆内容) importance: int Field(description重要程度1-10) class MemoryExtractionOutput(BaseModel): memories: list[MemoryItem] extract_prompt 用户与客服发生了以下对话。 请从中抽取值得长期记住的用户偏好或事实。 如果没有值得记住的信息返回空列表。 只输出结构化结果。 对话 {conversation} def extract_memories(user_id: str, conversation: str) - list[MemoryItem]: llm ChatOpenAI(modelgpt-4o-mini) llm_with_structure llm.with_structured_output(MemoryExtractionOutput) memories llm_with_structure.invoke(extract_prompt.format(conversationconversation)) return memories.memories抽取模型需要独立调一次 LLM会产生额外 Token 消耗。优化方向是只在对话长度达到一定阈值、或者用户明确表达了偏好时才触发抽取而不是每一轮都抽。6.3 记忆写入与向量检索抽取出的记忆先做规范化处理再写入向量库import chromadb from langchain_openai import OpenAIEmbeddings # 初始化向量库 client chromadb.PersistentClient(path./memory_db) collection client.get_or_create_collection(user_memories) embeddings OpenAIEmbeddings(modeltext-embedding-3-small) def write_memory(user_id: str, memory: MemoryItem): doc_id f{user_id}-{memory.memory_type}-{len(memory.content)} collection.upsert( ids[doc_id], documents[memory.content], metadatas[ { user_id: user_id, memory_type: memory.memory_type, importance: memory.importance, } ], embeddings[embeddings.embed_query(memory.content)], )查询时先按用户 ID 过滤再按向量相似度召回def recall_memories(user_id: str, query: str, top_k: int 5) - list[str]: results collection.query( query_embeddings[embeddings.embed_query(query)], where{user_id: user_id}, n_resultstop_k, ) return results[documents][0]检索时的query建议使用“当前用户问题 用户 ID”而不是只传原始问题。比如用户说“帮我推荐键盘”原始问题里没有“黑色”“机械”这些关键词但向量检索仍能根据语义把相关偏好召回。这就是长期记忆比关键词过滤更实用的原因。6.4 记忆写入后如何参与生成召回的记忆不是直接塞进系统提示词而是经过拼接加入 Agent 的上下文def build_context_with_memory(user_id: str, query: str) - str: memories recall_memories(user_id, query) if not memories: return return 根据用户的历史记忆\n \n.join(f- {mem} for mem in memories)在实际 Langgraph 中应该把“记忆读取”设计成一个节点放在主对话节点之前执行。它从向量库召回记忆写入当前状态然后再交给 DeepAgents 的 Supervisor 做意图判断。6.5 长期记忆的两个常见问题记忆冲突用户上一周说喜欢黑色这周说想换白色。写入新记忆的同时应该对相似旧记忆做合并或失效。记忆膨胀随着时间推移用户画像会越来越大。建议定期对同一用户的记忆做一次压缩总结只保留最新的偏好和重要事实。7. DeepAgents 多 Agent 编排与记忆共享7.1 DeepAgents 解决什么单 Agent 系统里记忆、工具、模型是绑在一起的。业务复杂后会出现三种问题工具太多导致提示词过长、不同职责混在一个 Agent 里难以维护、记忆被不同类型任务争夺。DeepAgents 的设计模式是拆成多个专用 Tool Agents由一个 Agent Supervisor 统一调度。在这套企业级 Agent 记忆系统中DeepAgents 不是替代 Langgraph而是跑在 Langgraph 之上的一层编排策略。7.2 DeepAgents 的最小接入示例下面的代码是基于 DeepAgents 通用设计模式的示例。不同版本的导入路径和构造函数可能不同请以官方文档为准。from langchain_deepagents import AgentSupervisor, ToolAgent # 1. 定义多个专业 Tool Agent agent_01_tools [ # 商品查询工具、订单查询工具等 ] agent_02_tools [ # 售后工具、物流工具等 ] # 2. 创建 Supervisor管理两个子 Agent supervisor AgentSupervisor( namecustomer_service_supervisor, sub_agents[ ToolAgent(nameproduct_browsing, toolsagent_01_tools), ToolAgent(nameafter_sales_service, toolsagent_02_tools), ], modelmodel, # 这里可以传入对话历史短期记忆会展开成上下文 prompt你是电商客服中心的主管负责判断用户请求应该交给哪个 Agent 处理。, )7.3 记忆在 DeepAgents 中的打通方式要让记忆和 DeepAgents 配合最简单的方式是记忆读取节点先行用户请求 - 记忆读取节点从向量库召回用户偏好 - Supervisor 编排 - 商品浏览 Agent带记忆上下文 - 订单查询 Agent调业务工具 - 结果汇总 - 记忆更新节点抽取新记忆并写入这样安排的好处是记忆上下文对所有子 Agent 可见子 Agent 不需要各自重复查询用户画像同时每个子 Agent 保持职责单一后续加新的业务 Agent 不影响已有链路。这也是这套体系能支撑电商多业务线的原因。8. 电商案例带记忆的智能客服 Agent 实战8.1 案例需求现在把这套记忆系统落地到一个电商客服场景。需求如下用户首次对话时能记住用户姓名、商品偏好、预算。用户再次咨询时不需要重复描述需求Agent 能直接引用历史偏好推荐。能实时查询订单状态、物流信息。涉及敏感操作改收货地址、退款时调用业务工具前置校验权限。8.2 图结构设计用 Langgraph 实现时图节点可以这样设计START - load_memory读取用户长期记忆 - supervisorDeepAgents 判断意图并调用子 Agent - store_memory抽取并写入新记忆 - END8.3 核心代码下面的代码是一个可以运行的参考骨架直接把上面的短期记忆和长期记忆代码串起来import os from langgraph.graph import StateGraph, START, END, MessagesState from langgraph.checkpoint.memory import MemorySaver from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolNode class AgentState(MessagesState): user_id: str memory_context: str # 工具 1查询订单 def query_order(order_id: str) - str: # 生产环境接入 ERP / 订单中心这里用演示数据返回 return f订单 {order_id} 已发货预计 3 天后送达。 # 工具 2推荐商品 def recommend_product(query: str) - str: # 生产环境接入商品中心这里用演示数据返回 return f根据需求{query}推荐 87 键黑色机械键盘价格 899 元。 # 记忆读取节点 def load_memory(state: AgentState) - dict: user_id state.get(user_id, ) raw_query state[messages][-1].content memories recall_memories(user_id, raw_query) memory_context if memories: memory_context 用户历史记忆\n \n.join(f- {m} for m in memories) return {memory_context: memory_context} # 记忆更新节点 def store_memory(state: AgentState) - dict: user_id state.get(user_id, ) conversation \n.join([f{m.type}: {m.content} for m in state[messages]]) # 为了控制成本可以只抽取最近两轮 extracted extract_memories(user_id, conversation[-2000:]) for mem in extracted: if mem.importance 6: # 只存重要记忆 write_memory(user_id, mem) return {} # Agent 节点把记忆和工具一起交给 Supervisor / 模型 def agent_node(state: AgentState): model_with_tools ChatOpenAI(modelgpt-4o-mini).bind_tools([query_order, recommend_product]) prompt f你是电商客服助手。 {state.get(memory_context, )} 请基于用户历史记忆和当前问题使用工具回答。 response model_with_tools.invoke( [{role: system, content: prompt}] state[messages] ) return {messages: [response]} # 工具执行节点 tools ToolNode([query_order, recommend_product]) graph StateGraph(AgentState) graph.add_node(load_memory, load_memory) graph.add_node(agent, agent_node) graph.add_node(tools, tools) graph.add_node(store_memory, store_memory) graph.add_edge(START, load_memory) graph.add_edge(load_memory, agent) graph.add_edge(agent, tools) graph.add_edge(tools, agent) graph.add_edge(agent, store_memory) graph.add_edge(store_memory, END) checkpointer MemorySaver() app graph.compile(checkpointercheckpointer)实际调用config {configurable: {thread_id: user-001}} # 第一轮用户自我介绍并提出偏好 app.invoke( { messages: [(user, 我叫小李想买一款黑色机械键盘预算一千以内。)], user_id: user-001, }, config, ) # 第二轮用户不重复需求直接问推荐 result app.invoke( { messages: [(user, 有什么适合我的键盘推荐吗)], user_id: user-001, }, config, ) print(result[messages][-1].content)这个案例跑通后你能观察到一个关键现象第一轮对话结束后store_memory节点已经抽取并写入了“黑色”“机械键盘”“预算一千以内”等记忆第二轮对话时load_memory节点把记忆读出来加入上下文即使第二轮的原始问题没有包含这些偏好词模型也会基于记忆生成更精准的推荐。8.4 案例验证清单第一轮说“我喜欢黑色机械键盘”第二轮问“推荐键盘”模型能否输出包含黑色、机械特征的答案。修改 thread_id 后再问同样问题模型是否不再认识用户证明记忆隔离有效。查看 SQLite / Chroma 是否写入了新记忆。删除对应用户的向量记录后模型是否回到无记忆状态确认删除逻辑可用。9. 服务化部署FastAPI 接口与批量任务9.1 为什么需要服务化电商客服系统通常不是给终端用户直接跑 Python 脚本而是把 Agent 包装成内部服务供网页、小程序、客服工作台调用。用 FastAPI 包装是最直接的方式。9.2 FastAPI 封装示例from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): user_id: str message: str thread_id: str app.post(/agent/chat) def chat(req: ChatRequest): config {configurable: {thread_id: req.thread_id or req.user_id}} result app_graph.invoke( { messages: [(user, req.message)], user_id: req.user_id, }, config, ) return { reply: result[messages][-1].content, user_id: req.user_id, thread_id: req.thread_id, }启动服务uvicorn main:app --host 127.0.0.1 --port 80009.3 curl 调用示例服务启动后可以用 curl 验证接口是否可用curl -X POST http://127.0.0.1:8000/agent/chat \ -H Content-Type: application/json \ -d {user_id: user-001, message: 帮我推荐一款键盘, thread_id: }接口正常返回后就可以把http://127.0.0.1:8000/agent/chat接入前端、企业微信机器人、微信客服、钉钉机器人等渠道。9.4 批量任务设计记忆系统里有两类典型批量任务历史会话批量记忆抽取每天定时把前一天所有客服会话拉出来逐段抽取记忆并写入向量库。记忆失效与合并定期扫描过期记忆、冲突记忆进行合并或删除。批量任务建议用独立脚本或消息队列触发避免阻塞在线接口# 批量抽取历史会话记忆 python scripts/backfill_memories.py --date 2025-02-14 --batch-size 100脚本里每个批次处理 100 条会话处理完成后写日志、记录游标支持断点续跑。若某批次失败直接把该批次重新入队即可。10. 资源占用与性能观察这套系统的瓶颈通常不在显存而在 LLM 推理延迟、向量库检索延迟和 Token 消耗。10.1 内存与磁盘占用短期记忆使用 SQLite Checkpointer 时磁盘占用取决于会话消息量。生产环境建议定期清理超过 30 天的旧会话或转移到冷存储。长期记忆Chroma 等向量库的磁盘占用取决于记忆条数和 embedding 维度。用户量大的场景建议按用户 ID 分片存储。内存FastAPI 服务和向量库都会占用内存。可以观察进程 RSS如果峰值过高把向量库查询改为异步或加连接池限制。10.2 Token 消耗如何控制记忆系统最大的成本点是自动记忆抽取和长期记忆注入。建议这样控制记忆抽取只在对话包含明显偏好时触发或者用较低的采样频率。检索回传的记忆条数控制在 3 到 5 条不要把所有历史都注入上下文。长对话做截断或摘要避免把整个历史消息都传给模型。对记忆抽取结果做缓存同一用户一分钟内不重复抽取。10.3 延迟观察点调用一次带记忆的 Agent典型链路是FastAPI 收到请求 - 读取用户记忆向量检索 - LLM 思考 - 调用业务工具 - LLM 再次生成 - 返回响应每一步都会产生延迟。建议在关键节点打印耗时日志或者在 LangSmith 中配置链路追踪观察哪一步最慢。工具调用如果超过 3 秒业务方会明显感受到卡顿需要考虑缓存或异步预取。11. 常见问题与排查方法问题现象可能原因排查方式解决方案长期记忆没有生效向量库没有写入数据或检索条件错误检查 Chroma 集合数量确认 user_id 是否一致先调用写入函数验证 collection再检查检索 where 条件同一 thread_id 下历史错乱Checkpointer 存储未切换成持久化版本确认是否用了 MemorySaver 且服务重启切换为 SqliteSaver 或 Postgres CheckpointerAgent 回答时上下文过长历史消息全部注入没有截断查看请求日志中的 token 数量增加历史消息截断或摘要节点记忆抽取不准确抽取提示词不明确或模型能力不足打印抽取模型输出增加示例数据到提示词或换更强模型向量检索召回结果与问题无关embedding 模型不合适或查询关键词过短打印召回结果查询时结合用户记忆上下文换更高质量的 embedding 模型DeepAgents 导入失败包名或版本不匹配查看安装版本与官方文档按官方最新版本调整导入路径接口返回超时LLM 调用慢或工具调用慢分步打印耗时对工具结果缓存LLM 使用流式响应批量任务失败后无法续跑没有记录处理游标查看日志确认失败批次增加游标表和失败重试队列修改模型服务后 Agent 无法调用BASE_URL 或 API Key 配置错误用 curl 直接测试模型服务检查环境变量和模型服务可用性记忆删除后仍能召回向量库未执行删除或索引未重建检查删除接口返回调用 collection.delete 并重新加载索引12. 最佳实践与使用建议12.1 技术侧建议第一次接入先小规模验证只用单个用户的 thread_id 跑通“短期记忆 长期记忆 工具调用”三条链路再扩展到全量用户。保留一套最小可运行配置方便复现问题。把环境变量、向量库目录、Checkpointer 路径整理成示例配置提交到团队知识库。记忆存储分目录管理模型配置、向量库、会话检查点、批处理日志分别放不同目录避免互相污染。批量任务的失败重试机制要提前设计建议使用游标表记录每个用户的处理状态失败批次可重新入队。接口服务不要直接暴露到公网内部调用也要加 Access Token 或 IP 白名单。12.2 业务与合规侧建议用户画像和偏好数据要有明确的保存周期。建议在数据模型中保存created_at和expires_at由定时任务负责清理过期数据。涉及用户隐私的问答Agent 应主动拒绝记录和回答敏感信息。接入电商订单、发票等业务数据时先判断当前用户的会话权限防止越权查询他人订单。如果未来要把记忆数据用于模型微调或数据分析需要单独确认数据使用授权。12.3 维护建议记忆抽取模型随着业务变化会逐渐出现偏差。建议每个月抽一批真实客服会话人工检查记忆抽取准确性及时调整提示词或模型。向量库也是长期运行的服务建议定期做快照备份防止数据损坏后用户画像全部丢失。13. 总结与下一步这套基于 Langchain Langgraph DeepAgents 的企业级 Agent 记忆系统最值得尝试的是它把短期记忆、长期记忆、业务记忆分成了三个独立层次真正做到了“该记住的记住、该查询的查询、该遗忘的遗忘”。相比单纯给模型加长上下文它更接近生产级 Agent 的形态。如果现在准备动手建议最先验证一个最小闭环用同一个 thread_id 跑两轮对话再插入load_memory和store_memory两个节点观察从第一轮抽取偏好到第二轮引用偏好的完整链路。最容易踩的坑是向量库的user_id过滤条件和 Langgraph 的thread_id没有对齐导致记忆读不到或读到别人的记忆。后续可以沿着三个方向扩展一是把 Checkpointer 替换为 Postgres支持多实例部署和会话共享二是把长期记忆的检索升级为混合检索向量 关键词 业务标签三是在 DeepAgents 中增加专门的“记忆维护 Agent”自动完成记忆合并、冲突解决和过期清理。这套体系跑通后再把客服能力扩展到营销、售后、私域运营就只是增加子 Agent 的事情了。建议收藏备用等技术栈更新后回来对照调整。

关于恒美微站

恒美微站专注于为个体商户、工作室提供极简自助建站服务,让每个人都能轻松拥有专业网站。

快速链接

  • 关于我们
  • 建站服务
  • 主题模板
  • 案例展示
  • 资讯中心

服务项目

  • 可视化建站
  • 拖拽编辑
  • 主题定制
  • SEO 优化
  • 网站托管

联系方式

  • 📍 地址:北京市朝阳区建国路 88 号
  • 📞 电话:400-888-8888
  • ✉️ 邮箱:info@hmyw.cn
  • 🕐 时间:周一至周日 9:00-18:00

© 2024 恒美微站 hmyw.cn 版权所有 | 京 ICP 备 12345678 号