MemOS 入门：把 AI 长期记忆和知识库分开管理

https://github.com/MemTensor/MemOS

python -m venv .venv

# macOS / Linux
source .venv/bin/activate

# Windows PowerShell
.venv\Scripts\Activate.ps1

pip install openai requests python-dotenv

MEMOS_API_KEY=你的_MemOS_Key
OPENAI_API_KEY=你的_OpenAI_Key
MEMOS_BASE_URL=https://memos.memtensor.cn/api/openmem/v1
KNOWLEDGE_BASE_IDS=["你的_知识库ID"]

import json
import os
from dataclasses import dataclass
from typing import Any, Dict, List

import requests
from dotenv import load_dotenv
from openai import OpenAI


load_dotenv()


@dataclass
class MemOSConfig:
    api_key: str
    base_url: str
    knowledge_base_ids: List[str]


class MemOSClient:
    """
    应用层只依赖这三个动作：
    - search_memory：搜索相关记忆和知识库内容
    - add_message：写入用户消息和模型回复
    - get_messages：读取近期会话消息

    具体 HTTP 路径以 MemOS 控制台或仓库 README 的 API 说明为准。
    """

    def __init__(self, config: MemOSConfig):
        self.config = config
        self.headers = {
            "Authorization": f"Bearer {config.api_key}",
            "Content-Type": "application/json",
        }

    def _post(self, path: str, payload: Dict[str, Any]) -> Dict[str, Any]:
        url = f"{self.config.base_url.rstrip('/')}/{path.lstrip('/')}"
        response = requests.post(url, headers=self.headers, json=payload, timeout=30)
        response.raise_for_status()
        return response.json()

    def search_memory(self, user_id: str, query: str) -> List[Dict[str, Any]]:
        payload = {
            "user_id": user_id,
            "query": query,
            "knowledge_base_ids": self.config.knowledge_base_ids,
        }

        # 按 MemOS API 文档替换为搜索记忆接口路径
        data = self._post("REPLACE_WITH_SEARCH_MEMORY_PATH", payload)
        return data.get("items", [])

    def add_message(self, user_id: str, role: str, content: str) -> None:
        payload = {
            "user_id": user_id,
            "role": role,
            "content": content,
        }

        # 按 MemOS API 文档替换为添加消息接口路径
        self._post("REPLACE_WITH_ADD_MESSAGE_PATH", payload)

    def get_messages(self, user_id: str, limit: int = 20) -> List[Dict[str, Any]]:
        payload = {
            "user_id": user_id,
            "limit": limit,
        }

        # 按 MemOS API 文档替换为获取消息接口路径
        data = self._post("REPLACE_WITH_GET_MESSAGES_PATH", payload)
        return data.get("messages", [])

class KnowledgeBaseAssistant:
    def __init__(self, memos: MemOSClient):
        self.memos = memos
        self.llm = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

    def chat(self, user_id: str, user_input: str) -> str:
        memories = self.memos.search_memory(user_id=user_id, query=user_input)
        history = self.memos.get_messages(user_id=user_id, limit=10)

        memory_context = self._format_memories(memories)
        history_context = self._format_history(history)

        messages = [
            {
                "role": "system",
                "content": (
                    "你是一个有长期记忆能力的助手。"
                    "回答时优先参考给定的知识库内容、用户记忆和近期对话。"
                    "如果上下文不足，不要编造事实。"
                ),
            },
            {
                "role": "system",
                "content": f"相关记忆和知识库内容：\n{memory_context}",
            },
            {
                "role": "system",
                "content": f"近期对话：\n{history_context}",
            },
            {
                "role": "user",
                "content": user_input,
            },
        ]

        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
        )

        answer = response.choices[0].message.content

        self.memos.add_message(user_id=user_id, role="user", content=user_input)
        self.memos.add_message(user_id=user_id, role="assistant", content=answer)

        return answer

    @staticmethod
    def _format_memories(memories: List[Dict[str, Any]]) -> str:
        if not memories:
            return "暂无相关记忆。"

        lines = []
        for index, item in enumerate(memories, start=1):
            content = item.get("content") or item.get("text") or str(item)
            lines.append(f"{index}. {content}")
        return "\n".join(lines)

    @staticmethod
    def _format_history(messages: List[Dict[str, Any]]) -> str:
        if not messages:
            return "暂无近期对话。"

        lines = []
        for item in messages:
            role = item.get("role", "unknown")
            content = item.get("content", "")
            lines.append(f"{role}: {content}")
        return "\n".join(lines)

def load_config() -> MemOSConfig:
    return MemOSConfig(
        api_key=os.environ["MEMOS_API_KEY"],
        base_url=os.environ["MEMOS_BASE_URL"],
        knowledge_base_ids=json.loads(os.environ["KNOWLEDGE_BASE_IDS"]),
    )


if __name__ == "__main__":
    config = load_config()
    memos = MemOSClient(config)
    assistant = KnowledgeBaseAssistant(memos)

    user_id = "test_user_001"

    while True:
        user_input = input("你：").strip()

        if user_input.lower() in {"exit", "quit"}:
            break

        answer = assistant.chat(user_id=user_id, user_input=user_input)
        print(f"AI：{answer}")

tenant_id / organization_id
project_id
user_id
agent_id
knowledge_base_id

{
  "content": "用户偏好回答中包含示例代码。",
  "type": "preference",
  "confidence": 0.86,
  "updated_at": "2026-06-07T10:00:00Z"
}