为 OpenAI Python 应用接入持久记忆:supermemory-openai-sdk 中间件与 Function Calling 工具全指南
【免费下载链接】supermemoryMemory and context engine + app that is extremely fast, scalable, and can be run fully locally. The Memory API for the AI era.项目地址: https://gitcode.com/GitHub_Trending/su/supermemory
本篇技术指南围绕开源仓库supermemory中的packages/openai-sdk-python子包展开,系统讲解如何通过supermemory-openai-sdk为官方 OpenAI Python SDK 增加两种记忆能力:自动记忆注入中间件(with_supermemory)与手动记忆工具(SupermemoryTools七种 function calling 工具)。读者读完可以掌握完整的安装、配置、三种记忆模式的选择、后台记忆存储任务的管理、错误处理体系,以及从源码层面理解记忆是如何被检索、去重、注入到系统提示词中的。
包定位与核心能力
supermemory-openai-sdk是 Supermemory 生态中面向 Python 开发者的官方集成包,其核心定位是为 OpenAI 官方 Python SDK(openai>=1.102.0)提供"无限上下文"能力。包内主要包含两大模块:
- 自动记忆注入中间件:位于 middleware.py,通过
with_supermemory()包装 OpenAI 客户端,在每次chat.completions.create()调用前自动检索用户历史记忆并注入系统提示词,同时可自动将对话内容异步保存为记忆; - 手动记忆工具:位于 tools.py,暴露 7 个符合 OpenAI function calling 规范的工具,让模型在对话过程中自主决定何时搜索、添加、删除记忆。
包版本为1.0.8(见 pyproject.toml),要求 Python>=3.9,同时依赖supermemory>=3.50.0官方客户端与requests>=2.25.0作为 HTTP 回退方案。
安装与环境准备
推荐使用uv进行安装:
uv add supermemory-openai-sdk或使用pip:
pip install supermemory-openai-sdk由于中间件面向AsyncOpenAI客户端时依赖aiohttp发起异步 HTTP 请求,建议安装 async 扩展(对于异步客户端强烈推荐):
uv add supermemory-openai-sdk[async] # 或 pip install 'supermemory-openai-sdk[async]'依赖细节:根据 pyproject.toml 的声明,必选依赖为
openai>=1.102.0、supermemory>=3.50.0、typing-extensions>=4.0.0、requests>=2.25.0;可选依赖aiohttp>=3.8.0仅通过[async]扩展标记启用。从源码看(middleware.py),若未安装aiohttp,中间件会自动回退到requests同步实现,因此即使不装 async 扩展也可运行,只是异步场景下性能会受影响。
安装完成后,需要设置以下环境变量:
SUPERMEMORY_API_KEY:Supermemory API 密钥(若在中间件选项中显式传入api_key则可省略);OPENAI_API_KEY:OpenAI API 密钥(示例代码运行必需);SUPERMEMORY_BASE_URL:可选,自定义 Supermemory API 地址,默认https://api.supermemory.ai;MODEL_NAME:可选,测试用的模型名,默认"gpt-4"。
快速开始
方式一:自动记忆注入中间件(推荐)
with_supermemory()是接入记忆的最简路径:它包装 OpenAI 客户端,并在每次请求前自动完成"检索记忆 → 注入系统提示词"的全过程,业务代码几乎无需改动:
import asyncio from openai import AsyncOpenAI from supermemory_openai import with_supermemory, OpenAIMiddlewareOptions async def main(): # 创建 OpenAI 客户端 openai = AsyncOpenAI(api_key="your-openai-api-key") # 用 Supermemory 中间件包装 openai_with_memory = with_supermemory( openai, OpenAIMiddlewareOptions( container_tag="user-123", # 必填:用户/容器的唯一标识 custom_id="chat-123", # 必填:将多轮消息归组为同一文档 mode="full", # "profile"、"query" 或 "full" verbose=True, # 开启日志 add_memory="always", # 自动保存对话(默认值) api_key="your-supermemory-api-key", # 或使用 SUPERMEMORY_API_KEY 环境变量 # base_url="https://api.supermemory.ai", # 可选:自定义端点 ) ) # 正常使用即可——记忆会被自动注入! response = await openai_with_memory.chat.completions.create( model="gpt-4", messages=[ {"role": "user", "content": "What's my favorite programming language?"} ] ) print(response.choices[0].message.content) asyncio.run(main())从源码层面看,with_supermemory返回的并不是原始客户端,而是一个SupermemoryOpenAIWrapper包装器(middleware.py)。该包装器在构造时完成三件事:
- 解析 API key 与 base URL——显式传入的
api_key优先,其次读取SUPERMEMORY_API_KEY环境变量,两者皆无则直接抛出SupermemoryConfigurationError; - 实例化内部的
supermemory.Supermemory客户端; - 通过
setattr替换client.chat.completions.create方法,使其先执行记忆注入逻辑再调用原始方法(middleware.py)。
包装器通过__getattr__将所有其他属性(如models)委托给原始客户端(middleware.py),所以包装后客户端的其余 API 不受影响。
方式二:使用记忆工具(Function Calling)
若不希望记忆注入是"全自动"的,而是让模型按需调用记忆能力,可以使用SupermemoryTools:
import asyncio import openai from supermemory_openai import SupermemoryTools, execute_memory_tool_calls async def main(): # 初始化 OpenAI 客户端 client = openai.AsyncOpenAI(api_key="your-openai-api-key") # 初始化 Supermemory 工具 tools = SupermemoryTools( api_key="your-supermemory-api-key", config={"project_id": "my-project"} ) # 携带记忆工具进行对话 response = await client.chat.completions.create( model="gpt-5", messages=[ { "role": "system", "content": "You are a helpful assistant with access to user memories." }, { "role": "user", "content": "Remember that I prefer tea over coffee" } ], tools=tools.get_tool_definitions() ) # 处理模型发起的工具调用 if response.choices[0].message.tool_calls: tool_results = await execute_memory_tool_calls( api_key="your-supermemory-api-key", tool_calls=response.choices[0].message.tool_calls, config={"project_id": "my-project"} ) print("Tool results:", tool_results) print(response.choices[0].message.content) asyncio.run(main())同步客户端支持
中间件同样支持同步OpenAI客户端,用法完全一致:
from openai import OpenAI from supermemory_openai import with_supermemory # 同步客户端 openai = OpenAI(api_key="your-openai-api-key") openai_with_memory = with_supermemory( openai, OpenAIMiddlewareOptions( container_tag="user-123", custom_id="session-456" ) ) # 用法相同 response = openai_with_memory.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": "Hello!"}] )事件循环管理:同步路径下,中间件内部使用asyncio.run()驱动记忆检索与注入;若同步客户端恰好在已有的异步上下文中被调用(例如在 async 函数内误用了OpenAI而非AsyncOpenAI),asyncio.run()会抛出RuntimeError: cannot be called from a running event loop,此时中间件会自动将任务提交到独立的ThreadPoolExecutor线程中执行以避免冲突(middleware.py)。测试用例test_sync_client_in_async_context专门覆盖了这一场景(见 tests/test_middleware.py)。
后台任务管理:当add_memory="always"时,对话内容会通过asyncio.create_task在后台异步保存,不阻塞主请求。因此建议使用上下文管理器确保退出时后台任务完成,或手动调用wait_for_background_tasks():
from supermemory_openai import with_supermemory, OpenAIMiddlewareOptions # 异步上下文管理器(推荐) async with with_supermemory( openai, OpenAIMiddlewareOptions(container_tag="user-123", custom_id="session-456") ) as client: response = await client.chat.completions.create(...) # 退出时自动等待后台任务完成 # 手动清理 client = with_supermemory( openai, OpenAIMiddlewareOptions(container_tag="user-123", custom_id="session-456") ) response = await client.chat.completions.create(...) await client.wait_for_background_tasks() # 确保记忆已保存wait_for_background_tasks()的默认超时为 10 秒(middleware.py):超时后它会取消所有未完成任务并抛出asyncio.TimeoutError;异步上下文管理器退出时使用 5 秒超时。后台任务保存失败(网络错误等)只会记录日志,不会影响主请求的返回——这正是"记忆是增强而非依赖"的设计理念。
中间件配置详解
记忆注入模式(Memory Modes)
中间件通过mode参数控制记忆检索策略,对应三种模式:
| 模式 | 行为 | 适用场景 |
|---|---|---|
"profile"(默认) | 仅注入静态+动态用户画像记忆,不针对当前消息做检索 | 需要每轮请求都携带稳定用户上下文的场景 |
"query" | 仅检索与当前用户消息相关的记忆 | 记忆库较大、追求检索效率的场景 |
"full" | 画像记忆 + 相关检索结果两者结合 | 既要用户画像又要即时相关性的场景 |
# profile 模式:注入全部静态与动态画像记忆 openai_with_memory = with_supermemory( openai, OpenAIMiddlewareOptions(container_tag="user-123", custom_id="session-456", mode="profile") ) # query 模式:仅检索与当前消息相关的记忆 openai_with_memory = with_supermemory( openai, OpenAIMiddlewareOptions(container_tag="user-123", custom_id="session-456", mode="query") ) # full 模式:画像 + 相关检索 openai_with_memory = with_supermemory( openai, OpenAIMiddlewareOptions(container_tag="user-123", custom_id="session-456", mode="full") )从 middleware.py 的实现可以看到三种模式的本质差异:
- 中间件统一调用
/v4/profile端点获取profile(含static/dynamic)与searchResults; mode != "profile"时,会用get_last_user_message()提取最后一条用户消息作为检索查询词q;mode == "query"时,static/dynamic画像会被丢弃(传入空列表),只保留检索结果;mode != "query"时,画像记忆会经convert_profile_to_markdown转换为 Markdown 格式(## Static Profile、## Dynamic Profile两个小节);mode != "profile"且存在检索结果时,会追加一段Search results for user's recent message:前缀的列表。
另外,当mode为query/full但消息列表中没有 user 消息时,中间件会跳过记忆检索直接透传请求(middleware.py)。
记忆存储策略(Memory Storage)
通过add_memory控制对话是否自动保存为记忆:
# 总是保存对话为记忆(v2.0.0+ 默认行为) OpenAIMiddlewareOptions(container_tag="user-123", custom_id="session-456", add_memory="always") # 从不保存对话 OpenAIMiddlewareOptions(container_tag="user-123", custom_id="session-456", add_memory="never")底层行为(middleware.py)值得注意:
add_memory="always"时,中间件提取最后一条用户消息,若配置了custom_id,还会用get_conversation_content()将整段对话格式化为User: ...\n\nAssistant: ...的文本,并以conversation:{custom_id}作为custom_id提交给 Supermemory——这意味着同一custom_id的多轮对话会被归组进同一个记忆文档,实现"会话级记忆";- 同步客户端下保存是同步执行的,且网络错误仅记录 warning 而不中断主流程。
完整配置示例
from supermemory_openai import with_supermemory, OpenAIMiddlewareOptions openai_with_memory = with_supermemory( openai_client, OpenAIMiddlewareOptions( container_tag="user-123", # 必填:用户/容器唯一标识 custom_id="chat-session-456", # 必填:将消息归组为同一文档 verbose=True, # 开启详细日志 mode="full", # 同时使用画像与检索 add_memory="always" # 自动保存对话(默认) ) )记忆注入的底层机制:幂等的标签包裹与去重
注入并不是简单地把记忆拼到 system prompt 后面。工具函数wrap_memory_context(utils.py)会把检索结果包裹进一个带标记的只读块:
<supermemory context="user-memories" readonly> ...记忆内容... </supermemory>中间件在注入前会先通过strip_memory_context()正则移除上一轮遗留的旧记忆块,再注入新记忆,从而保证:
- 多轮对话中记忆内容始终是最新一次检索结果,不会累积膨胀;
- 模型输出内容中若出现
</supermemory>等标签,会被转义为</>,防止记忆文本意外"逃逸"出块边界(_escape_memory_context_delimiters,utils.py); - 记忆块优先注入
developer消息,其次system消息;若两者皆不存在,则自动在消息列表头部创建一条 system 消息(middleware.py)。
此外,deduplicate_memories()(utils.py)会对 static、dynamic、search results 三类来源的记忆做去重:优先级为 Static > Dynamic > Search Results,同一事实只在最高优先级来源保留一次。去重时会剥离[recent]前缀和[2026-01-01]这类日期前缀再比较归一化文本,避免同一条事实因时间戳差异被重复注入。
测试 tests/test_middleware.py 中的test_existing_system_prompt_enhancement验证了上述机制:预置的旧<supermemory>块被新记忆替换、且块数量保持为 1;test_empty_memories_do_not_modify_messages则验证了记忆为空时不会向消息列表追加空 system 消息或空白字符,避免污染上下文。
手动记忆工具(SupermemoryTools)
SupermemoryTools共暴露七个 OpenAI function calling 工具:
search_memories与add_memoryget_profiledocument_list、document_add与document_deletememory_forget
工具调用的作用域由project_id或container_tags决定:配置的第一个容器标签用于 profile、list、search、forget 等单空间操作;而所有配置的标签都会应用于添加操作,并限定document_delete允许删除的文档范围——模型无法自行选择不同的标签(源码见 tools.py 的SupermemoryToolsConfig文档字符串)。
SupermemoryTools 类用法
from supermemory_openai import SupermemoryTools tools = SupermemoryTools( api_key="your-supermemory-api-key", config={ "project_id": "my-project", # 或使用 container_tags "base_url": "https://custom-endpoint.com", # 可选 } ) # 搜索记忆 result = await tools.search_memories( information_to_get="user preferences", limit=10 ) # 添加记忆 result = await tools.add_memory( memory="User prefers tea over coffee" ) # 获取配置用户的画像 result = await tools.get_profile(query="favorite drinks") # 列出、添加或删除源文档 documents = await tools.document_list(limit=10, page=1) document = await tools.document_add( content="Meeting notes...", title="Weekly meeting" ) deleted = await tools.document_delete(document_id="document-id-here") # 软遗忘一条抽取出的记忆 forgotten = await tools.memory_forget( memory_id="memory-entry-id-here", reason="outdated" )作用域解析规则(
_resolve_container_tags,tools.py):
- 同时传入
project_id与container_tags会抛出SupermemoryConfigurationError;- 传入
project_id时自动映射为标签sm_project_{project_id};- 传入
container_tags时要求至少一个非空标签;- 两者皆不传时使用默认标签
sm_project_default。
关于
include_full_docs:该参数作为兼容性参数保留(Python 层仍可传入,会触发DeprecationWarning),但 v4 搜索只返回相关记忆与文本块(chunk),不再返回完整源文档,因此它已不再出现在 OpenAI 工具 schema 中(tools.py)。测试test_search_memories_uses_search_memories_hybrid明确断言include_full_docs不会进入底层调用参数。
单独创建工具
SupermemoryTools之外,包还提供 7 个独立工具类及对应的工厂函数,适合只注册单个工具的轻量场景:
from supermemory_openai import ( create_search_memories_tool, create_add_memory_tool, create_get_profile_tool, create_document_list_tool, create_document_delete_tool, create_document_add_tool, create_memory_forget_tool, ) search_tool = create_search_memories_tool("your-api-key") add_tool = create_add_memory_tool("your-api-key") profile_tool = create_get_profile_tool("your-api-key") list_tool = create_document_list_tool("your-api-key") delete_tool = create_document_delete_tool("your-api-key") document_add_tool = create_document_add_tool("your-api-key") forget_tool = create_memory_forget_tool("your-api-key")每个独立工具类(如SearchMemoriesTool)都持有definition(即 OpenAI 工具 schema)与execute()方法(tools.py),可直接作为 OpenAI 的tools参数使用。
Function Calling 集成
from supermemory_openai import execute_memory_tool_calls # 拿到 OpenAI 返回的 tool_calls 之后 if response.choices[0].message.tool_calls: tool_results = await execute_memory_tool_calls( api_key="your-supermemory-api-key", tool_calls=response.choices[0].message.tool_calls, config={"project_id": "my-project"} ) # 将工具结果追加回对话 messages.append(response.choices[0].message) messages.extend(tool_results)execute_memory_tool_calls内部会为每个 tool call 创建独立的SupermemoryTools实例并通过asyncio.gather并行执行(tools.py),返回格式为 OpenAI 标准的ChatCompletionToolMessageParam(role="tool"、tool_call_id与 JSON 字符串形式的content)。
七个工具的参数 Schema 速查
工具的 JSON Schema 集中定义在MEMORY_TOOL_SCHEMAS(tools.py),关键约束如下:
| 工具 | 必填参数 | 其他参数与约束 |
|---|---|---|
search_memories | information_to_get | limit:默认 10,范围 1–100 |
add_memory | memory | 建议单句或短段落 |
get_profile | 无 | query:可选,附带检索结果 |
document_list | 无 | limit:默认 10,最大 1100;page:1 起 |
document_delete | document_id | 拒绝删除作用域外/共享/处理中的文档 |
document_add | content | title、description可选;内容排队异步处理,自动抽取记忆 |
memory_forget | 无(二选一) | memory_id或memory_content至少一个;reason可选 |
document_add与add_memory的分工是设计重点:add_memory用于保存单条可泛化的事实;document_add用于一次性摄入大段原始文本(粘贴的文本、对话转录、笔记、URL 等),Supermemory 会在后台完成分块、向量化、索引,并自动抽取画像记忆——模型无需再对文档内的事实逐条调用add_memory。
API 参考
中间件函数与配置
def with_supermemory( openai_client: Union[OpenAI, AsyncOpenAI], options: OpenAIMiddlewareOptions ) -> Union[OpenAI, AsyncOpenAI]参数说明:
openai_client:OpenAI或AsyncOpenAI客户端实例;options:配置选项(见下)。
@dataclass class OpenAIMiddlewareOptions: container_tag: str # 必填:记忆存储的唯一标识 custom_id: str # 必填:将消息归组为同一文档 verbose: bool = False # 是否输出详细日志 mode: Literal["profile", "query", "full"] = "profile" # 记忆注入模式 add_memory: Literal["always", "never"] = "always" # 自动保存行为 api_key: Optional[str] = None # 缺省时回退到 SUPERMEMORY_API_KEY base_url: Optional[str] = None # 缺省时回退到 SUPERMEMORY_BASE_URL补充说明:从源码 middleware.py 看,
api_key与base_url的解析顺序均为"选项显式值 → 环境变量 → 默认值",其中base_url的最终默认值为常量DEFAULT_SUPERMEMORY_BASE_URL = "https://api.supermemory.ai",并会去除尾部/。
SupermemoryTools 方法
get_tool_definitions()— 获取全部 OpenAI function 定义(7 个工具);search_memories()— 搜索用户记忆;add_memory()— 添加新记忆;get_profile()— 获取配置用户的画像;document_list()— 列出源文档元数据(含分页);document_add()— 排队处理一个源文档;document_delete()— 删除作用域内的源文档(删除前会校验文档的全部标签均在配置的作用域内,tools.py);memory_forget()— 软遗忘一条抽取的记忆;execute_tool_call()— 执行单个工具调用(含参数校验,tools.py)。
错误处理体系
包内定义了层次化的异常体系(exceptions.py):
SupermemoryError:所有 Supermemory 异常的基类,持有original_error便于追溯;SupermemoryConfigurationError:配置问题(如缺失 API key、project_id与container_tags冲突);SupermemoryAPIError:API 请求失败,携带status_code与response_text;SupermemoryNetworkError:网络连通性问题(OSError/ConnectionError的包装);SupermemoryMemoryOperationError:记忆搜索/添加操作失败;SupermemoryTimeoutError:操作超时。
典型用法:
from supermemory_openai import ( with_supermemory, OpenAIMiddlewareOptions, SupermemoryConfigurationError, SupermemoryAPIError, SupermemoryNetworkError, SupermemoryMemoryOperationError, ) try: # API key 缺失时这里会抛出 SupermemoryConfigurationError client = with_supermemory( openai_client, OpenAIMiddlewareOptions(container_tag="user-123", custom_id="session-456") ) response = await client.chat.completions.create( messages=[{"role": "user", "content": "Hello"}], model="gpt-4" ) except SupermemoryConfigurationError as e: print(f"Configuration issue: {e}") except SupermemoryAPIError as e: print(f"Supermemory API error: {e} (Status: {e.status_code})") except SupermemoryNetworkError as e: print(f"Network error: {e}") except SupermemoryMemoryOperationError as e: print(f"Memory operation failed: {e}") except Exception as e: print(f"Unexpected error: {e}")所有异常在构造时都保留了原始错误对象,并生成包含上下文的描述性错误信息;SupermemoryAPIError的__str__还会拼出状态码与响应体文本(exceptions.py)。需要注意:记忆检索失败不会静默吞掉——supermemory_profile_search中的非 2xx 响应会直接抛出SupermemoryAPIError并中断本次请求(middleware.py),而后台记忆保存失败则只记日志、不影响主请求。
底层调用链:一次请求发生了什么
综合源码(middleware.py),以AsyncOpenAI+mode="full"+add_memory="always"为例,一次chat.completions.create()的完整链路为:
- 包装:
with_supermemory构造SupermemoryOpenAIWrapper,替换chat.completions.create; - 触发:调用包装后的
create(),进入_create_with_memory_async; - 后台保存(若
add_memory="always"):提取最后一条 user 消息,格式化整段对话,通过asyncio.create_task提交add_memory_tool异步保存,任务被登记到_background_tasks集合; - 记忆检索:
add_system_prompt调用supermemory_profile_search,向{base_url}/v4/profile发起 POST,请求体为{"containerTag": ..., "include": ["static", "dynamic"], "q": ...}(q仅在非 profile 模式携带),携带Authorization: Bearer {api_key}头;aiohttp不可用时回退requests;单次请求超时 30 秒; - 去重与格式化:
deduplicate_memories按 Static > Dynamic > Search 优先级去重,convert_profile_to_markdown将画像转为 Markdown; - 注入:
_update_chat_memory_contexts将记忆块注入 developer/system 消息或新建 system 消息,同时清除上一轮旧记忆块; - 透传:调用原始
create()并返回结果。
对应测试可参考 tests/test_middleware.py(内存注入、去重替换、空记忆、后台任务、超时取消等场景)与 tests/test_tools.py(7 工具 schema、作用域冲突校验、client.add/search.memories底层调用参数断言)。仓库根目录还提供了真实 API 联调脚本 test_integration.py,设置OPENAI_API_KEY与SUPERMEMORY_API_KEY后可直接运行验证。
开发与测试
仓库内该包使用uv管理依赖与开发环境:
# 安装 uv curl -LsSf https://astral.sh/uv/install.sh | sh # 克隆仓库后进入子包目录 cd packages/openai-sdk-python uv sync --dev运行测试与代码检查:
# 运行全部测试 uv run pytest # 带覆盖率运行 uv run pytest --cov=supermemory_openai # 运行单个测试文件 uv run pytest tests/test_infinite_chat.py # 类型检查 uv run mypy src/supermemory_openai # 格式化 uv run black src/ tests/ uv run isort src/ tests/测试配置见 pyproject.toml(pytest-asyncio的asyncio_mode = "auto",mypy 采用严格模式配置);注意部分集成测试(如tests/test_tools.py中的TestMemoryOperations)需要真实SUPERMEMORY_API_KEY,未设置时会自动跳过。
总结
supermemory-openai-sdk为 OpenAI Python 应用提供了一条低侵入的记忆接入路径:中间件路线适合"开箱即用",通过with_supermemory+OpenAIMiddlewareOptions三个关键决策点(container_tag/custom_id标识、mode检索策略、add_memory存储策略)即可获得自动记忆注入与保存;工具路线适合对记忆行为有精细控制需求的应用,7 个 function calling 工具覆盖了从记忆搜索、画像读取到文档管理与遗忘的完整闭环。结合其幂等的记忆块注入、跨来源去重、作用域安全校验与完善的异常体系,开发者可以在几行代码内为任意 OpenAI 对话应用赋予真正的长期记忆能力。
【免费下载链接】supermemoryMemory and context engine + app that is extremely fast, scalable, and can be run fully locally. The Memory API for the AI era.项目地址: https://gitcode.com/GitHub_Trending/su/supermemory
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考