MemOS 接入 DeepSeek Harness：让 Agent 获得可持续的长期记忆

预计阅读时间：9 分钟

Memmy 与 MemOS Local/Cloud Plugin 已完成对 DeepSeek Harness 的适配。通过 DeepSeek Harness 的 Cordis 插件机制，MemOS 可以在不修改 Harness 源码的前提下，为 Agent 接入长期记忆能力。

这意味着，Agent 不再只能依赖当前上下文窗口中的信息。任务执行前，它可以检索与当前请求相关的历史记忆；任务执行过程中，可以读取必要的用户偏好或项目背景；任务完成后，还可以把值得保留的信息写回记忆系统。

接入点：通过插件扩展运行时能力

DeepSeek Harness 会把模型、工具、界面和运行能力组织在 profile 中，并通过插件机制扩展运行时能力。对 MemOS 来说，这种架构的关键价值在于：记忆能力可以作为独立插件接入，而不是把记忆逻辑硬编码进 Harness。

一个典型的任务链路可以抽象为：

用户请求
   |
   v
Cordis 插件读取当前请求
   |
   v
MemOS 检索相关长期记忆
   |
   v
Agent 结合上下文、工具和记忆执行任务
   |
   v
MemOS 按策略保存新的长期记忆

这种方式减少了宿主系统与记忆服务之间的耦合。Harness 负责运行 Agent，插件负责把记忆能力暴露给运行时，MemOS 则负责记忆的存储、检索和管理。

长期记忆能解决什么问题

短期上下文适合处理当前对话，但它不适合长期保存稳定信息。例如：

用户习惯使用 Python 和 PostgreSQL；
某个项目采用 FastAPI、Docker 和 Kubernetes；
用户明确要求回答简洁，代码示例需要可直接运行；
某个任务已经完成，后续请求只需要继续处理未完成部分。

当这些信息被保存为可检索记忆后，Agent 可以在新任务中恢复必要背景，而不需要用户反复描述。

不过，长期记忆不等于“保存所有对话”。工程上更重要的是记忆筛选和边界控制：临时状态、敏感数据和低价值闲聊不应默认长期保存；能够影响未来决策的稳定事实，才更适合进入记忆系统。

可以这样验证插件式记忆流程

下面是一个可直接改造的 Python 示例。它使用本地 HTTP 服务模拟 MemOS 的检索和写入接口，展示 Harness 插件在任务前后如何调用记忆服务。真实环境中，只需要将 MEMOS_URL、请求路径和鉴权方式替换为部署环境的配置。

运行前准备一个本地记忆服务，或把示例中的 URL 改成 MemOS Local/Cloud Plugin 提供的地址。

import os
from typing import Any

import requests

MEMOS_URL = os.getenv("MEMOS_URL", "http://localhost:8000")
MEMOS_TOKEN = os.getenv("MEMOS_TOKEN", "")
USER_ID = os.getenv("MEMOS_USER_ID", "demo-user")


def memos_headers() -> dict[str, str]:
    headers = {"Content-Type": "application/json"}
    if MEMOS_TOKEN:
        headers["Authorization"] = f"Bearer {MEMOS_TOKEN}"
    return headers


def search_memory(query: str, limit: int = 5) -> list[dict[str, Any]]:
    response = requests.post(
        f"{MEMOS_URL}/v1/memories/search",
        headers=memos_headers(),
        json={"user_id": USER_ID, "query": query, "limit": limit},
        timeout=10,
    )
    response.raise_for_status()
    return response.json().get("memories", [])


def save_memory(content: str, importance: float = 0.7) -> None:
    response = requests.post(
        f"{MEMOS_URL}/v1/memories",
        headers=memos_headers(),
        json={
            "user_id": USER_ID,
            "content": content,
            "importance": importance,
            "source": "deepseek-harness-cordis",
        },
        timeout=10,
    )
    response.raise_for_status()


def build_agent_context(user_request: str) -> str:
    memories = search_memory(user_request)
    lines = ["相关长期记忆："]
    lines.extend(f"- {item['content']}" for item in memories)
    return "\n".join(lines) if memories else "暂无相关长期记忆。"


def run_task(user_request: str) -> None:
    memory_context = build_agent_context(user_request)
    prompt = f"""你是一个运行在 DeepSeek Harness 中的 Agent。

{memory_context}

当前请求：
{user_request}

只使用与当前请求相关的记忆；如果记忆与当前请求冲突，以用户当前明确说明为准。"""

    # 这里替换为实际的模型调用或 Harness 内部任务执行接口。
    print(prompt)

    # 只有稳定、对后续任务有帮助的信息才写入长期记忆。
    save_memory("用户偏好可直接运行的代码示例。")


if __name__ == "__main__":
    run_task("请为我的 Python 项目设计一个适合容器部署的 API 服务")

安装依赖并运行：

python -m venv .venv
source .venv/bin/activate
pip install requests
export MEMOS_URL="http://localhost:8000"
python harness_memory_demo.py

示例中的接口路径是用于说明插件调用边界的假设实现，实际部署时应以 MemOS Plugin 的接口定义为准。生产环境还需要补充重试、连接池、鉴权、超时和日志脱敏。

接入时需要留意的边界

记忆召回不应覆盖当前指令

长期记忆只是辅助上下文。用户当前明确提出的要求优先级更高，提示词和 Agent 策略应明确这一点，避免旧偏好覆盖新需求。

记忆写入需要筛选

可以在写入前增加规则，例如只保存稳定偏好、项目技术栈和明确确认的事实，并过滤密码、访问令牌、身份证号等敏感内容。对于高风险场景，还应支持用户查看、删除和纠正记忆。

Cloud 与 Local 需要分别评估

Local Plugin 更适合对数据边界、网络隔离和延迟有要求的环境；Cloud Plugin 便于集中管理和扩展。选择时应结合数据合规、部署成本、可用性和检索延迟，而不是只看接入速度。

运行时故障不能阻塞核心任务

记忆服务短暂不可用时，Agent 是否继续执行需要提前定义。对非关键记忆，可以采用降级策略，让任务在没有历史记忆的情况下继续；对必须依赖上下文的业务，则应返回明确错误并记录诊断信息。

一份落地检查清单

确认 DeepSeek Harness profile 已启用对应 Cordis 插件。
配置 MemOS Local 或 Cloud Plugin 的地址和鉴权信息。
定义任务前检索、任务后写入的触发条件。
对敏感信息、临时信息和低价值信息设置过滤规则。
记录检索耗时、召回数量、写入结果和失败原因。
测试旧记忆与当前指令冲突时的处理结果。
为记忆服务不可用设计降级或失败策略。

MemOS 与 DeepSeek Harness 的适配，把长期记忆从宿主应用逻辑中拆了出来，并通过 Cordis 插件机制接入运行时。对于需要持续理解用户和项目背景的 Agent，这种集成方式能保持 Harness 核心代码稳定，同时为记忆能力留下独立的演进空间。