Supermemory Pipecat SDK 集成指南为 Pipecat 语音 AI 流水线注入持久记忆【免费下载链接】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/supermemorySupermemory Pipecat SDKsupermemory-pipecat是 supermemory 项目为 Pipecat 语音对话框架提供的官方记忆集成包它作为 Pipecat 流水线中的一个FrameProcessor在每一轮对话中自动从 Supermemory 检索相关历史记忆注入 LLM 上下文同时将新产生的用户/助手消息异步回写存储让语音 AI 应用具备跨会话的持久记忆能力。读完本文你将掌握该 SDK 的安装、参数配置、三种记忆模式的选择以及如何把它无缝嵌入基于 FastAPI WebSocket 的实时语音流水线并理解其底层帧拦截、去重、注入与异步存储的实现原理。安装与环境要求SDK 已发布为 PyPI 包supermemory-pipecat当前仓库版本为0.1.3见 pyproject.toml。安装命令pip install supermemory-pipecat从 pyproject.toml 可以看到其依赖与运行环境约束Python3.103.10 / 3.11 / 3.12 均被声明支持pipecat-ai0.0.98,2.0.0兼容从 0.0.98 到 2.x 之前的 Pipecat 版本supermemory3.16.0底层 Supermemory Python SDK用于调用/v4/profile检索与消息写入pydantic2.10.0用于InputParams配置模型的声明与校验loguru0.7.3日志输出。另外建议配合 OpenAI 相关服务使用 STT/LLM/TTS 时设置OPENAI_API_KEY环境变量见 Agents.md。快速开始五分钟接入记忆服务在 Pipecat 流水线中接入记忆服务的核心模式是把SupermemoryPipecatService放在 context aggregator 与 LLM 之间。它监听流水线中的上下文帧检索记忆注入上下文再把新消息转发给下游 LLM。最小示例来自 README.mdimport os from pipecat.pipeline.pipeline import Pipeline from pipecat.processors.aggregators.llm_context import LLMContext from supermemory_pipecat import SupermemoryPipecatService # Create memory service memory SupermemoryPipecatService( api_keyos.getenv(SUPERMEMORY_API_KEY), user_iduser-123, # Required: used as container_tag session_idconversation-456, # Optional: groups memories by session ) # Use the universal LLM context supported by current Pipecat releases. context LLMContext([{role: system, content: You are a helpful assistant.}]) context_aggregator llm.create_context_aggregator(context) # Create pipeline with memory pipeline Pipeline([ transport.input(), stt, context_aggregator.user(), memory, # Automatically retrieves and injects relevant memories llm, transport.output(), context_aggregator.assistant(), ])需要特别说明的流水线排布要点见 Agents.mdmemory必须放在context_aggregator.user()之后、llm之前这样才能先拿到聚合好的用户上下文做记忆增强再把增强后的上下文交给 LLMuser_id是必填参数它被用作 Supermemory 的container_tag即记忆作用域的隔离标识对应服务端容器标签体系不要把 API Key 硬编码进代码应通过api_key参数或SUPERMEMORY_API_KEY环境变量提供。配置参数详解构造函数参数来自 service.py 与 README 的参数表参数类型必填说明user_idstr是用户标识直接作为container_tag用于记忆作用域隔离缺失会抛出ConfigurationErrorsession_idstr否会话/对话 ID用于对记忆分组存储时作为custom_id写入api_keystr否Supermemory API Key不传时回退读取SUPERMEMORY_API_KEY环境变量二者皆无则抛出ConfigurationErrorparamsInputParams否高级配置记忆检索与注入行为不传则使用默认值base_urlstr否自定义 Supermemory API 端点用于自托管self-hosted等场景从源码看构造时还会做两件关键初始化service.py其一尝试用supermemory.AsyncSupermemory(api_key..., base_url...)创建异步客户端其二初始化内部的查询去重状态、可存储上下文快照与一个 FIFO 存储队列为后续的按轮检索和异步回写做准备。若supermemory库未安装客户端初始化为None后续检索或存储操作会抛出MemoryRetrievalError/MemoryStorageError并提示安装supermemory。InputParams 高级配置from supermemory_pipecat import SupermemoryPipecatService memory SupermemoryPipecatService( user_iduser-123, session_idconv-456, paramsSupermemoryPipecatService.InputParams( search_limit10, # Max memories to retrieve search_threshold0.1, # Similarity threshold modefull, # profile, query, or full inject_modeauto, # auto, system, or user system_promptBased on previous conversations, I recall:\n\n, ), )这些字段在 service.py 中用 pydanticField声明并带默认值与约束含义如下字段默认值约束说明search_limit10ge1每次检索最多返回的记忆条数search_threshold0.1ge0.0, le1.0记忆相似度阈值仅query/full模式下作为检索参数传入modefullprofile/query/full记忆检索模式见下节inject_modeautoauto/system/user记忆注入方式system_promptBased on previous conversations, I recall:\n\n—注入记忆文本的引导前缀search_threshold只会在非profile模式下作为threshold参数传给 profile APIsearch_limit则对返回的检索结果做截断list(raw_search_results)[: self.params.search_limit]见 service.py。三种记忆模式Memory Modesmode决定每次对话检索哪些维度的记忆。README 中的对照表模式静态画像Static Profile动态画像Dynamic Profile搜索结果Search ResultsprofileYesYesNoqueryNoNoYesfullYesYesYesprofile只拉取用户画像静态 动态适合做轻量个性化而无需全文检索的场景此时 API 调用不携带q查询参数。query只做基于当前用户消息的相似度检索适合找到相关历史上下文的场景。full画像 检索全量启用是默认模式提供最完整的记忆上下文。这三种模式的实现逻辑在 service.pyinclude_profile mode in (profile, full)include_search mode in (query, full)随后按需对静态/动态画像与搜索结果做合并去重。测试 test_empty_profile.py 中还有一个值得注意的细节即使某条事实同时出现在 profile 与 search 结果中modequery时去重逻辑也会保证它不会重复注入。工作原理帧拦截 → 检索 → 注入 → 存储README 中给出的五步工作流程结合源码可以展开如下核心实现位于 service.py 的process_frame拦截上下文帧监听 Pipecat 的通用LLMContextFrame同时兼容旧版OpenAILLMContextFrame与LLMMessagesFrame。Pipecat 1.0 移除了旧版消息帧因此源码中通过try/except ImportError把旧帧类做成可选导入保证 0.0.98 到当前版本都能工作service.py。此外InputAudioRawFrame会被旁路直通同时标记检测到音频帧用于 speech-to-speech 模式下自动切换到系统消息注入service.py。追踪对话从上下文中提取真实对话消息区分普通消息与带记忆标签的注入消息。_snapshot_storable_messages会过滤掉非user/assistant角色消息以及注入的user_memories标签消息service.py保证存储的永远是干净的对话内容。检索记忆取出当前快照中最后一条用户文本消息作为查询词_latest_user_occurrence调用AsyncSupermemory.profile()/v4/profileAPI一次拿到 static/dynamic 画像并在非 profile 模式下携带q与threshold参数附带搜索service.py。同时通过记录_last_recalled_user_prefix做去重——只有当用户消息前缀发生变化即出现了新的用户输入才触发新一轮检索避免同一帧被反复查询service.py。注入记忆把去重格式化后的记忆文本用 XML 标签user_memories.../user_memories包裹。与旧式累加不同这里采用替换式策略每轮先清除上一轮注入的标签_clear_injected_memories再注入新记忆防止上下文无限膨胀和记忆串扰service.py。注入目标由inject_mode决定system注入系统消息user作为一条带标签的用户消息追加auto检测到音频帧speech-to-speech 场景时走系统消息否则走用户消息。检索失败或返回空时会先清除旧记忆再返回确保旧记忆不会泄漏到新一轮service.py。存储消息把新观察到的用户/助手消息序列化为 JSON 对话片段进入后台队列异步写入 Supermemory队列由单任务串行消费失败批次保留在队头等待重试并在cleanup()阶段统一排空service.py。增量判定使用_messages_after_overlap——通过上一快照后缀与当前快照前缀的最大重叠找出真正新增的消息天然兼容追加、替换、前截断等上下文变化service.py。存储了什么README 明确说明新产生的用户和助手消息会作为JSON 对话片段存储注入的user_memories消息在存储前被过滤且不会推进存储游标。例如User: Whats the weather like today? Assistant: Its sunny today.实际发送给 Supermemory 的载荷为{ content: [{\role\: \user\, \content\: \Whats the weather like today?\}, {\role\: \assistant\, \content\: \Its sunny today.\}], container_tags: [user-123], custom_id: conversation-456, metadata: { platform: pipecat } }对应源码中的_store_messagesservice.pycontent为 JSON 序列化的消息列表container_tags携带user_idcustom_id仅在提供session_id时设置metadata.platform固定为pipecat。记忆注入的文本格式与去重utils.py 提供了格式化与去重工具注入后的上下文大致呈现为Based on previous conversations, I recall: ## User Profile (Persistent) - 用户静态画像条目 ## Recent Context - 动态画像条目 ## Relevant Memories - [3d ago] 相关历史记忆条目值得注意的实现细节去重优先级为 静态 动态 搜索deduplicate_memories比较键会剥离动态条目上的[recent]/[YYYY-MM-DD]日期前缀并按大小写折叠、空白归一化因此同一事实以不同呈现形式出现只保留一份utils.py相对时间标注搜索结果条目通过updatedAt转换为[just now]、[Xmins ago]、[X hrs ago]、[Xd ago]、[X Jul]等人类可读的相对时间前缀让 LLM 感知记忆的新旧程度utils.py标签转义escape_memory_delimiters会把记忆文本中可能出现的user_memories字样转义为 HTML 实体防止注入内容破坏包裹标签的完整性utils.py。异常体系SDK 提供了完整的异常层级exceptions.py便于调用方精确捕获与降级SupermemoryPipecatError所有异常基类携带message与original_errorConfigurationErrorAPI Key 缺失或参数非法如缺少user_idMemoryRetrievalError记忆检索失败如客户端未初始化、API 调用异常MemoryStorageError记忆写入失败APIError带status_code与response_text的 API 层错误NetworkError网络层错误。检索失败不会让整条流水线崩溃——process_frame中捕获MemoryRetrievalError后仅记录 warning 并继续转发原始帧service.py体现了记忆增强是锦上添花、不能阻塞主对话的设计原则。完整实战示例FastAPI WebSocket 语音助手README 提供了一个开箱即用的完整示例——基于 FastAPI WebSocket 的实时语音助手使用 Google Gemini Live 实现 speech-to-speech并接入 Supermemory 记忆import asyncio import os from fastapi import FastAPI, WebSocket from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.task import PipelineTask from pipecat.pipeline.runner import PipelineRunner from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService from pipecat.transports.websocket.fastapi import ( FastAPIWebsocketTransport, FastAPIWebsocketParams, ) from supermemory_pipecat import SupermemoryPipecatService app FastAPI() app.websocket(/chat) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() transport FastAPIWebsocketTransport( websocketwebsocket, paramsFastAPIWebsocketParams(audio_in_enabledTrue, audio_out_enabledTrue), ) # Gemini Live for speech-to-speech llm GeminiLiveLLMService( api_keyos.getenv(GEMINI_API_KEY), modelmodels/gemini-2.5-flash-native-audio-preview-12-2025, ) context LLMContext([{role: system, content: You are a helpful assistant.}]) context_aggregator llm.create_context_aggregator(context) # Supermemory memory service memory SupermemoryPipecatService( user_idalice, session_idsession-123, ) pipeline Pipeline([ transport.input(), context_aggregator.user(), memory, llm, transport.output(), context_aggregator.assistant(), ]) runner PipelineRunner() task PipelineTask(pipeline) await runner.run(task)要点回顾每个 WebSocket 连接建立独立的Pipeline与记忆服务实例user_idalice保证记忆只属于该用户session_idsession-123用于把该次会话的消息聚合分组Gemini Live 属于语音直连模型流水线中出现InputAudioRawFrame此时inject_modeauto会自动切换为系统消息注入避免把记忆文本当成语音输出流水线结束后 Pipecat 会调用服务的cleanup()此时后台存储队列被排空确保最后一轮对话也被持久化service.py。进一步探索SDK 源码service.py帧处理与记忆读写核心、utils.py去重与格式化、exceptions.py异常体系测试用例test_empty_profile.py 覆盖了空 profile 的容错与modequery下的去重行为工程元数据pyproject.toml依赖、Python 版本、打包配置团队开发约定Agents.md集成模式、环境变量与边界约束如需了解底层/v4/profileAPI 的完整语义可进一步阅读仓库的 API 参考文档 与 用户画像文档。小结supermemory-pipecat用约五百行核心代码把检索 - 注入 - 回写三件事封装成一个符合 Pipecat 框架哲学的FrameProcessor既有对旧版/新版上下文帧的兼容又有基于重叠检测的增量存储、基于标签替换的记忆更新、基于队列的重试排空等健壮性设计。无论你是为语音助手补充跨会话记忆还是为实时对话系统做个性化增强都可以按本文的配置矩阵快速落地并结合 Agents.md 中的流水线排布约束避免踩坑。【免费下载链接】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），仅供参考