从零构建一个有记忆的AI Agent：Cortex Memory实战指南

s***k35

1159人浏览 · 2025-12-28 10:24:11

s***k35 · 2025-12-28 10:24:11 发布

请添加图片描述

摘要

随着AI技术的发展，越来越多的开发者开始构建自己的AI Agent。然而，如何让AI Agent记住用户的信息、历史对话和上下文，成为了一个普遍的难题。本文从实战角度出发，详细介绍如何使用Cortex Memory为AI Agent添加记忆能力，包括架构设计、代码实现、部署上线等完整流程。

1. 为什么AI Agent需要记忆

1.1 无状态AI的局限

传统的AI应用（如ChatGPT）是无状态的，每次对话都是独立的：

这种无状态特性导致：

重复询问：用户需要反复提供相同信息
缺乏个性化：AI无法学习用户偏好
上下文断裂：无法理解跨会话的对话
决策受限：缺乏历史数据支持

1.2 有记忆AI的优势

为AI Agent添加记忆能力后，可以实现：

有记忆的AI Agent能够：

记住用户信息：姓名、偏好、习惯等
理解上下文：跨会话的对话历史
提供个性化服务：根据用户偏好调整回复
持续学习：从交互中不断学习

2. Cortex Memory简介

2.1 什么是Cortex Memory

Cortex Memory是一个开源的AI Agent记忆管理系统，提供：

智能记忆管理：自动提取、去重、优化记忆
语义搜索：基于向量嵌入的自然语言检索
多模态访问：REST API、CLI、MCP协议等
高性能实现：Rust语言，低延迟、高吞吐量

项目地址：https://github.com/sopaco/cortex-mem

2.2 核心概念

记忆（Memory）

记忆是Cortex Memory的基本单位，包含：

pub struct Memory {
    pub id: String,              // 唯一标识
    pub content: String,         // 记忆内容
    pub embedding: Vec<f32>,     // 向量嵌入
    pub metadata: MemoryMetadata, // 元数据
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

记忆类型（MemoryType）

Cortex Memory支持六种记忆类型：

类型	说明	示例
Conversational	对话记录	“用户询问了退款流程”
Procedural	过程性知识	“如何重置密码”
Factual	事实性信息	“用户住在上海”
Semantic	语义概念	“什么是机器学习”
Episodic	情节记忆	“用户上周参加了培训”
Personal	个人化信息	“用户喜欢编程”

语义搜索

Cortex Memory使用向量嵌入实现语义搜索：

3. 架构设计

3.1 整体架构

一个有记忆的AI Agent通常包含以下组件：

3.2 技术栈选择

组件	技术选择	说明
AI Agent框架	LangChain / AutoGPT	提供Agent编排能力
大语言模型	OpenAI GPT-4 / Claude	提供智能对话能力
记忆系统	Cortex Memory	提供记忆管理能力
向量数据库	Qdrant	存储向量嵌入
后端服务	Python FastAPI / Node.js	提供API服务
前端界面	React / Vue	提供用户界面

4. 快速开始

4.1 安装Cortex Memory

方式1：使用Cargo安装

# 安装CLI工具
cargo install cortex-mem-cli

# 安装REST API服务
cargo install cortex-mem-service

# 安装MCP服务器
cargo install cortex-mem-mcp

方式2：从源码编译

# 克隆仓库
git clone https://github.com/sopaco/cortex-mem.git
cd cortex-mem

# 编译安装
cargo install --path cortex-mem-service
cargo install --path cortex-mem-cli

4.2 配置Cortex Memory

创建配置文件 config.toml：

# HTTP服务器配置
[server]
host = "127.0.0.1"
port = 8000
cors_origins = ["*"]

# Qdrant向量数据库配置
[qdrant]
url = "http://localhost:6333"
collection_name = "cortex-memory"
timeout_secs = 5

# LLM配置
[llm]
api_base_url = "https://api.openai.com/v1"
api_key = "sk-your-openai-api-key"
model_efficient = "gpt-3.5-turbo"
temperature = 0.7
max_tokens = 8192

# 嵌入服务配置
[embedding]
api_base_url = "https://api.openai.com/v1"
api_key = "sk-your-openai-api-key"
model_name = "text-embedding-3-small"
batch_size = 16
timeout_secs = 10

# 记忆管理配置
[memory]
max_memories = 10000
similarity_threshold = 0.65
max_search_results = 50
auto_summary_threshold = 32768
auto_enhance = true
deduplicate = true
merge_threshold = 0.75
search_similarity_threshold = 0.50

# 日志配置
[logging]
enabled = true
log_directory = "logs"
level = "info"

4.3 启动服务

# 启动Qdrant向量数据库
docker run -p 6333:6333 -p 6334:6334 qdrant/qdrant

# 启动Cortex Memory服务
cortex-mem-service --config config.toml

5. 代码实现

5.1 Python集成示例

安装依赖

pip install requests openai

基础集成

import requests
import json

class CortexMemoryClient:
    def __init__(self, base_url="http://localhost:8000"):
        self.base_url = base_url

    def create_memory(self, content, user_id=None, agent_id=None):
        """创建记忆"""
        url = f"{self.base_url}/memories"
        data = {
            "content": content,
            "metadata": {}
        }

        if user_id:
            data["metadata"]["user_id"] = user_id
        if agent_id:
            data["metadata"]["agent_id"] = agent_id

        response = requests.post(url, json=data)
        response.raise_for_status()
        return response.json()

    def search_memories(self, query, user_id=None, limit=10):
        """搜索记忆"""
        url = f"{self.base_url}/memories/search"
        data = {
            "query": query,
            "limit": limit
        }

        if user_id:
            data["filters"] = {"user_id": user_id}

        response = requests.post(url, json=data)
        response.raise_for_status()
        return response.json()

# 使用示例
client = CortexMemoryClient()

# 创建记忆
memory = client.create_memory(
    content="用户喜欢编程，特别是Rust语言",
    user_id="user123"
)
print(f"创建记忆: {memory['id']}")

# 搜索记忆
results = client.search_memories(
    query="用户的爱好",
    user_id="user123",
    limit=5
)
print(f"搜索结果: {len(results['results'])} 条")

5.2 LangChain集成

安装依赖

pip install langchain openai

自定义Memory类

from langchain.memory import BaseMemory
from langchain.schema import BaseMessage, get_buffer_string
from typing import List, Dict, Any
import requests

class CortexMemory(BaseMemory):
    """Cortex Memory集成"""

    def __init__(self, base_url: str = "http://localhost:8000", user_id: str = "default"):
        self.base_url = base_url
        self.user_id = user_id

    @property
    def memory_variables(self) -> List[str]:
        return ["context"]

    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, str]:
        """加载记忆变量"""
        # 从输入中获取查询
        query = inputs.get("query", "")

        # 搜索相关记忆
        results = self._search_memories(query)

        # 格式化为上下文
        context = self._format_context(results)

        return {"context": context}

    def save_context(self, inputs: Dict[str, str], outputs: Dict[str, str]) -> None:
        """保存上下文"""
        # 从输入和输出中提取信息
        content = self._extract_content(inputs, outputs)

        # 创建记忆
        self._create_memory(content)

    def _create_memory(self, content: str) -> None:
        """创建记忆"""
        url = f"{self.base_url}/memories"
        data = {
            "content": content,
            "metadata": {"user_id": self.user_id}
        }

        requests.post(url, json=data)

    def _search_memories(self, query: str, limit: int = 5) -> List[Dict]:
        """搜索记忆"""
        url = f"{self.base_url}/memories/search"
        data = {
            "query": query,
            "filters": {"user_id": self.user_id},
            "limit": limit
        }

        response = requests.post(url, json=data)
        results = response.json()
        return results.get("results", [])

    def _format_context(self, results: List[Dict]) -> str:
        """格式化上下文"""
        if not results:
            return "无相关记忆"

        context_parts = []
        for i, result in enumerate(results, 1):
            memory = result["memory"]
            context_parts.append(f"{i}. {memory['content']}")

        return "\n".join(context_parts)

    def _extract_content(self, inputs: Dict[str, str], outputs: Dict[str, str]) -> str:
        """提取内容"""
        # 简单实现：合并输入和输出
        input_text = inputs.get("query", "")
        output_text = outputs.get("response", "")

        return f"用户: {input_text}\n助手: {output_text}"

# 使用示例
from langchain.chains import ConversationChain
from langchain.llms import OpenAI

# 创建Cortex Memory
memory = CortexMemory(user_id="user123")

# 创建对话链
llm = OpenAI(temperature=0)
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 进行对话
response = conversation.predict(query="我叫小明")
print(response)

response = conversation.predict(query="我叫什么名字？")
print(response)

5.3 Node.js集成

安装依赖

npm install axios

基础集成

const axios = require('axios');

class CortexMemoryClient {
    constructor(baseUrl = 'http://localhost:8000') {
        this.baseUrl = baseUrl;
    }

    async createMemory(content, userId = null, agentId = null) {
        const url = `${this.baseUrl}/memories`;
        const data = {
            content: content,
            metadata: {}
        };

        if (userId) {
            data.metadata.user_id = userId;
        }
        if (agentId) {
            data.metadata.agent_id = agentId;
        }

        const response = await axios.post(url, data);
        return response.data;
    }

    async searchMemories(query, userId = null, limit = 10) {
        const url = `${this.baseUrl}/memories/search`;
        const data = {
            query: query,
            limit: limit
        };

        if (userId) {
            data.filters = { user_id: userId };
        }

        const response = await axios.post(url, data);
        return response.data;
    }
}

// 使用示例
const client = new CortexMemoryClient();

async function main() {
    // 创建记忆
    const memory = await client.createMemory(
        '用户喜欢编程，特别是Rust语言',
        'user123'
    );
    console.log(`创建记忆: ${memory.id}`);

    // 搜索记忆
    const results = await client.searchMemories(
        '用户的爱好',
        'user123',
        5
    );
    console.log(`搜索结果: ${results.results.length} 条`);
}

main().catch(console.error);

6. 高级功能

6.1 记忆优化

Cortex Memory提供智能记忆优化功能：

class CortexMemoryOptimizer:
    def __init__(self, client):
        self.client = client

    def optimize_memories(self, user_id):
        """优化记忆"""
        url = f"{self.client.base_url}/optimization"
        data = {
            "filters": {"user_id": user_id},
            "strategy": "full"
        }

        response = requests.post(url, json=data)
        return response.json()

    def get_optimization_status(self, job_id):
        """获取优化状态"""
        url = f"{self.client.base_url}/optimization/{job_id}"
        response = requests.get(url)
        return response.json()

# 使用示例
optimizer = CortexMemoryOptimizer(client)

# 启动优化
job = optimizer.optimize_memories("user123")
print(f"优化任务ID: {job['job_id']}")

# 检查状态
status = optimizer.get_optimization_status(job['job_id'])
print(f"优化状态: {status['status']}")

6.2 批量操作

class CortexMemoryBatch:
    def __init__(self, client):
        self.client = client

    def batch_create_memories(self, memories):
        """批量创建记忆"""
        url = f"{self.client.base_url}/memories/batch/create"
        data = {"memories": memories}

        response = requests.post(url, json=data)
        return response.json()

    def batch_delete_memories(self, memory_ids):
        """批量删除记忆"""
        url = f"{self.client.base_url}/memories/batch/delete"
        data = {"memory_ids": memory_ids}

        response = requests.post(url, json=data)
        return response.json()

# 使用示例
batch = CortexMemoryBatch(client)

# 批量创建
memories = [
    {"content": "记忆1", "metadata": {"user_id": "user123"}},
    {"content": "记忆2", "metadata": {"user_id": "user123"}},
    {"content": "记忆3", "metadata": {"user_id": "user123"}},
]

result = batch.batch_create_memories(memories)
print(f"创建 {len(result['created_ids'])} 条记忆")

6.3 实时监控

class CortexMemoryMonitor:
    def __init__(self, client):
        self.client = client

    def get_memory_stats(self, user_id=None):
        """获取记忆统计"""
        url = f"{self.client.base_url}/stats"
        params = {}

        if user_id:
            params["user_id"] = user_id

        response = requests.get(url, params=params)
        return response.json()

    def get_system_health(self):
        """获取系统健康状态"""
        url = f"{self.client.base_url}/health"
        response = requests.get(url)
        return response.json()

# 使用示例
monitor = CortexMemoryMonitor(client)

# 获取统计
stats = monitor.get_memory_stats("user123")
print(f"总记忆数: {stats['total_count']}")
print(f"平均质量: {stats['avg_importance']}")

# 健康检查
health = monitor.get_system_health()
print(f"系统状态: {health['status']}")

7. 部署上线

7.1 Docker部署

Dockerfile

FROM rust:1.75 as builder

WORKDIR /app
COPY . .

# 编译
RUN cargo build --release

# 运行时镜像
FROM debian:bookworm-slim

RUN apt-get update && apt-get install -y \
    ca-certificates \
    && rm -rf /var/lib/apt/lists/*

COPY --from=builder /app/target/release/cortex-mem-service /usr/local/bin/

EXPOSE 8000

CMD ["cortex-mem-service", "--config", "/config/config.toml"]

docker-compose.yml

version: '3.8'

services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_data:/qdrant/storage

  cortex-mem:
    build: .
    ports:
      - "8000:8000"
    volumes:
      - ./config.toml:/config/config.toml
      - ./logs:/logs
    depends_on:
      - qdrant
    environment:
      - RUST_LOG=info

volumes:
  qdrant_data:

启动服务

# 构建并启动
docker-compose up -d

# 查看日志
docker-compose logs -f cortex-mem

7.2 Kubernetes部署

Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: cortex-mem
spec:
  replicas: 3
  selector:
    matchLabels:
      app: cortex-mem
  template:
    metadata:
      labels:
        app: cortex-mem
    spec:
      containers:
      - name: cortex-mem
        image: cortex-mem:latest
        ports:
        - containerPort: 8000
        env:
        - name: RUST_LOG
          value: "info"
        volumeMounts:
        - name: config
          mountPath: /config
        - name: logs
          mountPath: /logs
      volumes:
      - name: config
        configMap:
          name: cortex-mem-config
      - name: logs
        emptyDir: {}
---
apiVersion: v1
kind: Service
metadata:
  name: cortex-mem
spec:
  selector:
    app: cortex-mem
  ports:
  - port: 8000
    targetPort: 8000
  type: LoadBalancer

ConfigMap

apiVersion: v1
kind: ConfigMap
metadata:
  name: cortex-mem-config
data:
  config.toml: |
    [server]
    host = "0.0.0.0"
    port = 8000

    [qdrant]
    url = "http://qdrant-service:6333"
    collection_name = "cortex-memory"

    [llm]
    api_base_url = "https://api.openai.com/v1"
    api_key = "${OPENAI_API_KEY}"
    model_efficient = "gpt-3.5-turbo"

    [embedding]
    api_base_url = "https://api.openai.com/v1"
    api_key = "${OPENAI_API_KEY}"
    model_name = "text-embedding-3-small"

8. 最佳实践

8.1 记忆管理

原则1：只存储有价值的信息

# 好的做法
memory = client.create_memory(
    "用户喜欢编程，特别是Rust语言",
    user_id="user123"
)

# 不好的做法
memory = client.create_memory(
    "用户说：你好",  # 无价值的问候
    user_id="user123"
)

原则2：使用合适的记忆类型

# 个人化信息
memory = client.create_memory(
    "用户喜欢编程",
    user_id="user123",
    memory_type="Personal"
)

# 过程性知识
memory = client.create_memory(
    "如何重置密码：步骤1、步骤2、步骤3",
    user_id="user123",
    memory_type="Procedural"
)

原则3：定期优化记忆

# 每周优化一次
import schedule
import time

def optimize_user_memories():
    optimizer = CortexMemoryOptimizer(client)
    job = optimizer.optimize_memories("user123")
    print(f"优化任务: {job['job_id']}")

# 每周一凌晨2点执行
schedule.every().monday.at("02:00").do(optimize_user_memories)

while True:
    schedule.run_pending()
    time.sleep(60)

8.2 性能优化

优化1：使用批量操作

# 不好的做法：逐条创建
for content in contents:
    client.create_memory(content, user_id="user123")

# 好的做法：批量创建
memories = [
    {"content": c, "metadata": {"user_id": "user123"}}
    for c in contents
]
batch.batch_create_memories(memories)

优化2：缓存搜索结果

from functools import lru_cache
import hashlib

class CachedCortexMemoryClient:
    def __init__(self, client):
        self.client = client
        self.cache = {}

    def search_memories(self, query, user_id=None, limit=10):
        # 生成缓存键
        cache_key = f"{query}:{user_id}:{limit}"

        # 检查缓存
        if cache_key in self.cache:
            return self.cache[cache_key]

        # 执行搜索
        results = self.client.search_memories(query, user_id, limit)

        # 存入缓存
        self.cache[cache_key] = results

        return results

优化3：异步请求

import asyncio
import aiohttp

class AsyncCortexMemoryClient:
    def __init__(self, base_url="http://localhost:8000"):
        self.base_url = base_url

    async def create_memory(self, content, user_id=None):
        url = f"{self.base_url}/memories"
        data = {
            "content": content,
            "metadata": {"user_id": user_id} if user_id else {}
        }

        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=data) as response:
                return await response.json()

    async def batch_create_memories(self, memories):
        tasks = [
            self.create_memory(m["content"], m["metadata"].get("user_id"))
            for m in memories
        ]
        return await asyncio.gather(*tasks)

# 使用示例
async def main():
    client = AsyncCortexMemoryClient()

    memories = [
        {"content": f"记忆{i}", "metadata": {"user_id": "user123"}}
        for i in range(10)
    ]

    results = await client.batch_create_memories(memories)
    print(f"创建 {len(results)} 条记忆")

asyncio.run(main())

9. 常见问题

9.1 性能问题

问题：搜索延迟过高

解决方案：

检查Qdrant配置，确保索引已创建
减少返回结果数量
使用缓存减少重复查询
考虑增加服务器资源

9.2 记忆质量问题

问题：搜索结果不相关

解决方案：

调整相似度阈值
使用更合适的嵌入模型
优化查询语句
定期运行记忆优化

9.3 部署问题

问题：服务无法启动

解决方案：

检查配置文件格式
确保Qdrant服务正常运行
检查端口是否被占用
查看日志文件排查问题

10. 总结

通过Cortex Memory，我们可以轻松为AI Agent添加记忆能力，实现：

持久化记忆：记住用户信息和历史对话
智能检索：基于语义理解的自然语言搜索
自动优化：持续优化记忆质量
高性能：低延迟、高吞吐量的服务

Cortex Memory提供了一个完整、易用的记忆管理解决方案，让开发者能够专注于业务逻辑，而不是记忆系统的底层实现。

项目地址：https://github.com/sopaco/cortex-mem

如果你正在构建AI Agent，不妨试试Cortex Memory，相信它会为你的项目带来质的飞跃。

参考资料

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

AI Agent 工程化实践：从 ReAct 循环到多智能体编排

更务实的做法是使用**消息队列模式**：每个 Agent 有一个独立的输入队列，通过一个统一的消息句柄（AgentHandle）进行收发，上层调用者（CLI / UI / API）通过这个句柄与 Agent 系统交互。一个常见的模式是使用 **事件总线（EventBus）** 进行异步通信——Agent 产生事件，订阅者消费事件。一个健壮的工具执行系统需要解决三个问题：**能不能调**（权限）、*