基于Oracle DB与MCP协议构建AI Agent三位一体记忆系统

1. 项目缘起当AI Agent需要记住“一切”最近在折腾一个AI Agent项目遇到了一个非常典型的问题Agent的“记忆力”太差了。这里的记忆力不是指大模型本身的上下文长度而是指Agent在长期运行、与用户多次交互、执行复杂任务过程中对自身状态、历史决策、用户偏好、任务中间结果等信息的持久化存储与精准召回能力。想象一个场景你让Agent帮你分析上个月的销售数据它吭哧吭哧跑了一遍给出了结论。第二天你问它“对比一下这个月和上个月的趋势”如果它完全忘了昨天做了什么、用了哪些数据、得出了什么结论那就只能从头再来。这不仅浪费算力Token更破坏了交互的连贯性和智能体应有的“人格感”。一个健壮的、可投入生产的AI Agent必须拥有一个可靠、高效且结构化的记忆系统。这就是“三位一体记忆体”概念的由来。它不是一个单一的技术点而是一个系统性的工程架构旨在解决Agent记忆的持久化、结构化和高效检索三大核心问题。而我的选择是回归一个久经考验的“老兵”——Oracle Database。为什么是Oracle DB在如今云原生、NoSQL大行其道的背景下这个选择似乎有些“复古”。但经过一番权衡我发现对于企业级AI Agent应用Oracle DB提供了难以替代的优势无与伦比的SQL能力处理复杂关系型记忆、绝佳的稳定性和事务一致性保障记忆不丢失、成熟的生态工具链如SQLcl便于集成与运维。更重要的是我想探索一条将经典企业级数据基础设施与前沿AI Agent架构深度结合的道路。本文将详细拆解我如何利用Oracle DB为AI Agent构建起“工作记忆”、“情景记忆”、“语义记忆”三位一体的记忆体并集成MCPModel Context Protocol协议让Agent的记忆能力变得可插拔、可扩展。这不是一个简单的技术堆砌而是一次从想法到落地工程的完整实践。2. 解构“三位一体”AI Agent需要什么样的记忆在深入技术细节之前我们必须先厘清AI Agent记忆系统的设计目标。直接照搬数据库表结构是行不通的我们需要对记忆进行建模。2.1 记忆的分类与数据模型设计借鉴认知心理学和现有Agent框架如LangChain的AgentExecutor、AutoGen的GroupChat我将Agent记忆分为三个层次对应数据库中的不同实体和关系1. 工作记忆 (Working Memory)是什么Agent单次会话或单个任务执行周期内的临时状态。相当于人类的“短时记忆”。数据特点高频更新、生命周期短、结构相对简单。Oracle DB实现表设计AGENT_SESSION表。核心字段包括SESSION_ID主键、AGENT_ID、USER_ID、CREATED_AT、LAST_ACTIVITY_AT、CONTEXT_HASH当前会话上下文摘要。关联表SESSION_MESSAGES表。记录该会话中的所有消息用户输入、Agent思考、工具调用结果。这是记忆检索的主要素材。字段包括MESSAGE_ID、SESSION_ID、ROLEuser/assistant/tool、CONTENT、TOOL_CALL_ID、TIMESTAMP。这里利用Oracle的CLOB类型存储可能很长的消息内容。策略工作记忆通常与会话绑定。可以设置定时任务清理超过一定时间如30分钟无活动的会话及其消息避免数据无限膨胀。2. 情景记忆 (Episodic Memory)是什么对过去发生的具体事件、任务执行结果的记录。相当于“自传体记忆”回答了“之前发生了什么”的问题。数据特点按事件组织、包含丰富元数据时间、参与者、结果状态、需要支持基于内容的模糊检索。Oracle DB实现表设计AGENT_EPISODE表。核心字段包括EPISODE_ID主键、AGENT_ID、TASK_TYPE如“data_analysis”, “report_generation”、TASK_DESCRIPTION、STATUSsuccess/failed/partial、START_TIME、END_TIME、SUMMARY由LLM生成的本次任务摘要。关键挑战与方案如何从冗长的SESSION_MESSAGES中提炼出可供快速检索的情景这里我引入了向量检索的混合方案。结构化摘要任务结束时触发一个LLM调用将本次任务的关键决策、使用的主要工具、最终结论生成一个文本摘要存入SUMMARY字段。向量化存储使用Oracle的VECTOR数据类型19c以后版本支持或通过PL/SQL调用外部AI服务将SUMMARY和TASK_DESCRIPTION转换为向量存储在EPISODE_VECTOR列中。检索当需要寻找类似历史任务时先将当前问题转换为向量然后利用Oracle的VECTOR_DISTANCE函数进行相似度搜索快速定位相关历史情景。这比单纯的SQLLIKE查询要强大和精准得多。3. 语义记忆 (Semantic Memory)是什么Agent学到的通用知识、事实、用户偏好、领域规则等。相当于“常识”或“长期知识”。数据特点相对稳定、结构化程度高、需要版本管理和有效性验证。Oracle DB实现表设计这是一个更灵活的结构。我创建了AGENT_KNOWLEDGE表包含KNOWLEDGE_ID、AGENT_ID、CATEGORY如“user_preference”, “domain_rule”, “api_schema”、KEY、VALUECLOB、VECTOR可选、EFFECTIVE_FROM、EFFECTIVE_TO。应用例如可以存储“用户A更喜欢图表用柱状图而非饼图”、“某API的响应结构说明”、“项目Y的代码规范”等。通过CATEGORY和KEY可以快速查询。对于非结构化的知识条目同样可以采用向量化存储与检索。版本控制利用EFFECTIVE_FROM和EFFECTIVE_TO实现简单的知识版本管理确保Agent在不同时间点使用的知识是准确的。注意直接存储原始对话消息尤其是包含大量Token的LLM响应会导致数据库迅速膨胀。一个重要的实践是摘要与向量化双轨制。对于需要精确回溯的步骤如工具调用的参数存储原始消息对于需要语义搜索的记忆存储由LLM生成的精炼摘要及其向量。这需要在存储成本和检索精度间取得平衡。2.2 记忆的关联与图谱构建单一的记忆条目价值有限记忆之间的关联更能体现智能。例如一次成功的数据分析任务情景记忆可能用到了某个特定的用户偏好语义记忆并产生了一系列中间步骤工作记忆。我在数据库中通过外键关联和图谱化视图来建立这种连接。SESSION_MESSAGES.SESSION_ID-AGENT_SESSION.SESSION_IDAGENT_EPISODE可以关联到触发它的原始SESSION_ID。在AGENT_KNOWLEDGE中可以通过REF_EPISODE_ID字段关联到产生这条知识的具体任务。更进一步可以创建一个物化视图或应用程序层的缓存构建一个轻量的“记忆图谱”快速回答诸如“用户A在处理销售数据时通常喜欢用什么方法”这类复杂查询。Oracle的高级SQL分析函数如LISTAGG, 递归查询CONNECT BY或RECURSIVE WITH在这里能派上大用场。3. 工程落地Oracle SQLcl与MCP Server的桥梁设计好了数据模型下一步就是让AI Agent能够方便地读写这个记忆库。让Agent直接通过JDBC/ODBC连接Oracle显然不现实也破坏了架构的清晰度。我的方案是构建一个MCPModel Context Protocol Server作为Agent与Oracle记忆体之间的专用桥梁。3.1 为什么选择MCPMCP是一种新兴的协议它允许AI模型或Agent通过标准化的方式发现、调用外部工具和资源。对于记忆体来说MCP化带来了巨大好处解耦记忆体的实现细节Oracle、PostgreSQL、甚至向量数据库对Agent透明。Agent只通过MCP协议定义的标准接口tools与记忆交互。可发现性支持MCP的AI客户端如Claude Code、Cursor可以自动发现记忆服务器提供的工具无需硬编码。标准化提供了统一的错误处理、资源描述和调用方式。3.2 使用Oracle SQLcl构建MCP Server核心Oracle SQLcl是一个基于Java的轻量级命令行工具但它更是一个功能强大的脚本执行环境支持JavaScript、Python等语言。我选择用JavaScript (Nashorn引擎)在SQLcl中快速原型化MCP Server的核心逻辑。步骤一环境准备与基础连接首先确保已安装Oracle SQLcl。然后编写一个基础的JavaScript文件memory_mcp.js建立数据库连接池。SQLcl内置了JDBC连接非常方便。// memory_mcp.js - 核心框架 var DBUtil Java.type(oracle.dbtools.raptor.newscriptrunner.db.DBUtil); var conn DBUtil.getConnection(); // 获取当前SQLcl会话的连接 // 简单的连接池模拟生产环境建议使用更成熟的池化技术 function getConnection() { // 这里简化处理实际应考虑连接的重用和管理 return conn; }步骤二实现MCP工具函数根据之前的设计我们需要实现几个核心的MCP工具Tool。每个工具对应一个JavaScript函数并通过SQLcl的script命令暴露为可调用接口。例如实现一个query_episodic_memory工具// 查询情景记忆 function queryEpisodicMemory(agentId, queryText, limit) { var conn getConnection(); var stmt null; var rs null; try { // 1. 先将查询文本向量化这里需要调用外部向量化服务如OpenAI API或本地模型 // 为简化示例假设有一个函数getVectorEmbedding(text) var queryVector getVectorEmbedding(queryText); // 返回一个数组或序列化字符串 // 2. 使用Oracle VECTOR相似度搜索 (假设向量已存储在EPISODE_VECTOR列) var sql SELECT e.EPISODE_ID, e.TASK_DESCRIPTION, e.SUMMARY, e.STATUS, VECTOR_DISTANCE(e.EPISODE_VECTOR, :1, COSINE) as similarity FROM AGENT_EPISODE e WHERE e.AGENT_ID :2 ORDER BY similarity ASC FETCH FIRST :3 ROWS ONLY ; stmt conn.prepareStatement(sql); // 绑定参数需要根据实际向量存储格式调整 stmt.setObject(1, queryVector); stmt.setString(2, agentId); stmt.setInt(3, limit || 5); rs stmt.executeQuery(); var results []; while (rs.next()) { results.push({ episode_id: rs.getString(EPISODE_ID), description: rs.getString(TASK_DESCRIPTION), summary: rs.getString(SUMMARY), status: rs.getString(STATUS), similarity: rs.getFloat(similarity) }); } return JSON.stringify(results); } catch (e) { return JSON.stringify({error: e.message}); } finally { if (rs) rs.close(); if (stmt) stmt.close(); } } // 将函数注册为SQLcl脚本命令 script.registerFunction(queryEpisodicMemory, queryEpisodicMemory);步骤三封装为MCP Server单纯的SQLcl脚本还不够我们需要一个遵循MCP协议的HTTP/SSE服务器。这里我使用Node.js或Python的FastAPI来构建一个轻量的HTTP服务器这个服务器的后端逻辑通过子进程调用SQLcl并执行上述JavaScript脚本来实现。// Node.js MCP Server 示例 (app.js) const express require(express); const { exec } require(child_process); const app express(); app.use(express.json()); // MCP 标准的 /tools 端点公布可用的工具 app.get(/tools, (req, res) { res.json({ tools: [ { name: query_episodic_memory, description: 根据任务描述语义搜索历史任务情景, inputSchema: { type: object, properties: { agent_id: { type: string }, query: { type: string }, limit: { type: integer, default: 5 } }, required: [agent_id, query] } }, // ... 其他工具定义如 save_working_memory, retrieve_knowledge 等 ] }); }); // MCP 标准的 /tools/call 端点调用具体工具 app.post(/tools/call, (req, res) { const { name, arguments } req.body; if (name query_episodic_memory) { const { agent_id, query, limit } arguments; // 关键调用SQLcl执行JS函数 const sqlclCmd sql -script memory_mcp.js -cmd queryEpisodicMemory(${agent_id}, ${query}, ${limit}); exec(sqlclCmd, (error, stdout, stderr) { if (error) { res.json({ error: { message: stderr } }); return; } // 解析SQLcl的输出通常是JSON字符串 try { const result JSON.parse(stdout.trim()); res.json({ content: [{ type: text, text: JSON.stringify(result, null, 2) }] }); } catch (e) { res.json({ content: [{ type: text, text: stdout }] }); } }); } else { res.status(404).json({ error: { message: Tool ${name} not found } }); } }); app.listen(3000, () console.log(MCP Memory Server running on port 3000));这样一个基于Oracle SQLcl和Node.js的MCP记忆服务器就搭建起来了。AI Agent如通过Codex配置了该MCP Server的Claude Code就可以直接调用query_episodic_memory等工具来访问记忆。3.3 踩坑实录SQLcl集成与性能优化这个过程并非一帆风顺有几个关键的坑需要避开坑1SQLcl进程生命周期与性能每次工具调用都启动一个新的SQLcl进程开销巨大无法满足低延迟要求。解决方案是采用连接池常驻进程模式。修改Node.js服务器使用spawn启动一个常驻的SQLcl子进程并通过标准输入输出stdin/stdout与其进行交互。这需要编写一个更复杂的通信协议例如每行一个JSON命令。在SQLcl的JavaScript脚本中初始化一个真正的数据库连接池例如使用Oracle UCP并在整个进程生命周期内维护它。坑2向量化服务的集成与延迟在SQL层做向量相似度计算虽然方便但向量生成调用OpenAI等Embedding API是网络IO密集型操作放在数据库调用链中会阻塞整个请求。解决方案异步向量化与预处理。当保存新的情景记忆AGENT_EPISODE时不要同步调用Embedding API。将需要向量化的文本如SUMMARY放入一个消息队列如Oracle AQ、RabbitMQ。由一个独立的向量化Worker消费队列生成向量后再异步更新回数据库的EPISODE_VECTOR字段。在向量更新前记忆仍可通过其他字段如时间、任务类型被检索只是缺少了语义搜索能力。这实现了最终一致性。坑3MCP协议细节与错误处理MCP协议要求特定的响应格式。不规范的响应会导致AI客户端无法解析。解决方案严格遵循MCP协议文档定义/tools和/tools/call端点的输入输出格式。错误时返回{ error: { message: “...” } }格式。为每个工具函数实现完善的异常捕获将数据库错误、网络错误等转化为用户友好的消息。4. 在AI Agent框架中集成MCP记忆体记忆服务器Ready了下一步就是让Agent能用上它。我以目前较流行的LangChain框架为例展示集成过程。4.1 将MCP Server封装为LangChain ToolLangChain支持自定义Tool。我们需要创建一个Tool其_run方法内部去调用我们刚构建的MCP Server的HTTP接口。import requests import json from langchain.tools import BaseTool from typing import Type, Optional from pydantic import BaseModel, Field class MCPMemoryQueryInput(BaseModel): 查询情景记忆的输入参数. agent_id: str Field(description智能体ID) query: str Field(description用于搜索记忆的自然语言查询) limit: Optional[int] Field(default5, description返回结果的最大数量) class QueryEpisodicMemoryTool(BaseTool): name query_episodic_memory description 根据自然语言描述语义搜索智能体过去执行过的类似任务情景记忆。 args_schema: Type[BaseModel] MCPMemoryQueryInput mcp_server_url: str http://localhost:3000 def _run(self, agent_id: str, query: str, limit: int 5) - str: 执行工具调用. try: response requests.post( f{self.mcp_server_url}/tools/call, json{ name: self.name, arguments: { agent_id: agent_id, query: query, limit: limit } }, timeout10 ) response.raise_for_status() result response.json() if error in result: return f查询记忆失败{result[error][message]} # MCP返回的内容在content字段中 memory_text result.get(content, [{}])[0].get(text, {}) memories json.loads(memory_text) if isinstance(memories, list): formatted \n.join([f- [{m[status]}] {m[description]} (相似度: {1-m[similarity]:.2f}) for m in memories]) return f找到{len(memories)}条相关历史任务\n{formatted} else: return str(memories) except requests.exceptions.RequestException as e: return f网络请求失败{e} except json.JSONDecodeError as e: return f解析响应失败{e} async def _arun(self, *args, **kwargs): 异步版本可选。 raise NotImplementedError(此工具暂不支持异步调用)4.2 在Agent执行流程中注入记忆查询有了Tool下一步就是在Agent的思考循环中在合适的时机调用它。这通常发生在Agent需要做决策或规划时。from langchain.agents import AgentExecutor, create_react_agent from langchain.prompts import PromptTemplate from langchain_community.llms import OpenAI # 示例可用其他LLM # 1. 初始化LLM llm OpenAI(temperature0, model_namegpt-4) # 2. 准备工具列表包含我们的记忆工具和其他功能工具 tools [QueryEpisodicMemoryTool(), ...你的其他工具...] # 3. 创建自定义Prompt引导Agent在规划时考虑历史记忆 prompt_template 你是一个拥有记忆能力的AI助手。在回答用户问题或执行任务前你可以查询自己的记忆库参考过去的经验。 你有权使用以下工具 {tools} 使用工具时请严格按照以下格式 Thought: 你需要思考现在要做什么 Action: 工具名 Action Input: 工具的输入参数必须是有效的JSON字符串 当你获得工具观察结果后继续 Observation: 工具返回的结果 ...这个Thought/Action/Observation循环可以重复多次 开始如果你认为查询记忆有助于当前任务请首先使用query_episodic_memory工具。 历史对话 {chat_history} 当前用户输入{input} {agent_scratchpad} prompt PromptTemplate.from_template(prompt_template) # 4. 创建Agent并执行 agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 执行一个任务 result agent_executor.invoke({ input: 帮我分析一下本季度和上一季度的销售数据差异并给出可视化建议。, chat_history: , # 这里可以传入持久化的工作记忆 agent_id: sales_analyst_agent_001 })在这个流程中当Agent收到“分析销售数据差异”的任务时Thought过程可能会触发“我应该先看看以前是怎么做销售数据分析的”。于是它会调用query_episodic_memory工具传入agent_id和query:“销售数据分析”。MCP Server会从Oracle DB中返回相似的历史任务。Agent在Observation中看到这些历史记录后就能借鉴过去的成功经验比如用了哪些数据表、生成了哪种图表从而更高效、更一致地完成当前任务。4.3 记忆的写入何时保存保存什么记忆的读取很重要写入同样关键。盲目保存所有对话会塞满数据库。我的策略是事件驱动的记忆写入任务结束时保存情景记忆当Agent完成一个明确的任务如“生成报告”、“分析数据”并成功时触发一个save_episode的MCP工具调用。这个工具会将当前会话的关键消息、最终结论、任务类型等通过SQLcl写入AGENT_EPISODE表并异步触发向量化流程。显式知识沉淀当用户给出明确的指示如“记住我以后都想要PDF格式的报告”Agent应调用save_knowledge工具将这条偏好作为语义记忆存入AGENT_KNOWLEDGE表。工作记忆的自动暂存工作记忆会话消息的保存可以更自动化。可以设置一个后台进程定时或在会话空闲时将活跃会话的消息快照同步到SESSION_MESSAGES表。也可以只在会话中发生重要工具调用或决策点时进行快照。5. 进阶优化与生产级考量将原型推进到生产环境还需要解决更多问题。5.1 记忆的检索优化超越向量搜索单一的向量相似度搜索并非万能。在实践中我采用了**混合检索Hybrid Search**策略关键词过滤先利用Oracle的全文检索CONTEXT索引或普通WHERE子句基于AGENT_ID、TASK_TYPE、时间范围等硬性条件进行第一轮筛选缩小数据集。向量精排对筛选后的结果再进行向量相似度计算和排序。元数据加权在最终排序时综合考虑相似度、任务成功状态STATUSsuccess的优先、任务新鲜度最近发生的任务可能更相关等因素给出一个综合排名。这可以通过在Oracle中编写一个更复杂的PL/SQL函数或存储过程来实现一次性返回混合检索的结果。5.2 记忆的压缩、蒸馏与遗忘机制无限增长的记忆是不可持续的。我们需要“遗忘”机制。工作记忆自动清理基于LAST_ACTIVITY_AT字段定期归档或删除老旧会话。情景记忆摘要化对于非常久远的情景记忆可以只保留SUMMARY摘要和关键元数据删除其关联的详细SESSION_MESSAGES记录释放空间。语义记忆的版本合并对于语义记忆当同一KEY下的知识被多次更新后可以尝试使用LLM对多个版本进行“蒸馏”合并成一个更精炼、更通用的版本删除旧版本。5.3 监控、调试与可观测性一个黑盒的记忆系统是危险的。必须增加可观测性。审计日志在Oracle中创建MEMORY_ACCESS_LOG表记录每一次记忆的读写操作AGENT_ID,OPERATION,TARGET_ID,TIMESTAMP,RESULT_COUNT等。这对于调试Agent的异常行为和优化检索策略至关重要。性能监控监控MCP Server的接口响应时间、SQLcl进程的资源消耗、关键查询语句的执行计划。Oracle的AWR/ASH报告可以帮助定位数据库层面的瓶颈。记忆质量评估定期抽样检查看Agent检索到的记忆是否真的对当前任务有帮助。可以设计一些自动化测试用例或者通过人工评审来评估。构建AI Agent的记忆体从想法到落地工程是一个充满挑战但也极具成就感的过程。选择Oracle DB作为基石看中的是其强大的数据管理能力和在企业环境中的可靠性。通过MCP协议进行封装则为这套记忆系统赋予了现代化的、标准化的接口使其能够灵活接入不同的AI Agent生态。这套“三位一体”的架构——工作记忆维持会话流情景记忆提供案例参考语义记忆存储常识偏好——在实践中显著提升了我所开发Agent的连贯性、准确性和用户体验。它不再是一个“金鱼脑”的对话机器人而是一个真正能够积累经验、持续学习的数字助手。当然这套架构仍在演进中。下一步我计划探索如何利用Oracle的Graph特性来更自然地表示记忆间的复杂关系以及如何将记忆的检索与生成过程更深度地融合到LLM的推理链条中。工程的世界没有银弹但每一次扎实的架构选择和技术深耕都能让我们的AI应用离真正的智能更近一步。