Lindorm长期记忆最佳实践指南

安装SDK

pip install lindorm-memobase

最小化配置示例

创建.env文件，填入 Lindorm 实例的连接信息，如何获取，请参见查看连接地址。

# Lindorm 认证
MEMOBASE_LINDORM_USERNAME=root
MEMOBASE_LINDORM_PASSWORD=your_password

# Lindorm 宽表（MySQL 协议）
MEMOBASE_LINDORM_TABLE_HOST=your-host.lindorm.aliyuncs.com
MEMOBASE_LINDORM_TABLE_PORT=33060
MEMOBASE_LINDORM_TABLE_USERNAME=root
MEMOBASE_LINDORM_TABLE_PASSWORD=your_table_password
MEMOBASE_LINDORM_TABLE_DATABASE=memobase

# Lindorm 搜索引擎
MEMOBASE_LINDORM_SEARCH_HOST=your-search-host.lindorm.aliyuncs.com
MEMOBASE_LINDORM_SEARCH_PORT=30070
MEMOBASE_LINDORM_SEARCH_USERNAME=root
MEMOBASE_LINDORM_SEARCH_PASSWORD=your_search_password

# Lindorm AI（LLM 和 Embedding）
MEMOBASE_LLM_STYLE=lindormai
MEMOBASE_LLM_BASE_URL=http://your-lindorm-ai-url:9002
MEMOBASE_BEST_LLM_MODEL=qwen-max-latest
MEMOBASE_EXTRACT_LLM_MODEL=qwen-max

MEMOBASE_EMBEDDING_PROVIDER=lindormai
MEMOBASE_EMBEDDING_BASE_URL=http://your-lindorm-ai-url:9002
MEMOBASE_EMBEDDING_MODEL=text-embedding-v4
MEMOBASE_EMBEDDING_DIM=1024

# Rerank
MEMOBASE_RERANK_PROVIDER=lindormai
MEMOBASE_RERANK_BASE_URL=http://your-lindorm-ai-url:9002/dashscope/api/v1/services/rerank/text-rerank/text-rerank
MEMOBASE_RERANK_MODEL=qwen3-rerank

初始化SDK

SDK 支持三种初始化方式（.env、YAML文件、代码参数）。以下使用.env初始化并完成第一次记忆提取：

from lindormmemobase import LindormMemobase
from lindormmemobase.models.blob import ChatBlob, OpenAICompatibleMessage, BlobType
import asyncio

async def main():
    # 方式一：使用默认配置（从 .env 或 config.yaml 加载）
    memobase = LindormMemobase()

    # 方式二：从 YAML 文件加载
    # memobase = LindormMemobase.from_yaml_file("config.yaml")

    # 方式三：通过参数配置
    # memobase = LindormMemobase.from_config(
    #     llm_api_key="your-key",
    #     language="zh"
    # )

    # 提取记忆
    user_id = "user123"
    blob = ChatBlob(
        messages=[
            OpenAICompatibleMessage(role="user", content="我喜欢吃辣的食物，特别是川菜"),
            OpenAICompatibleMessage(role="assistant", content="记住了，您偏爱川菜"),
        ]
    )

    result = await memobase.extract_memories(user_id, [blob])
    print(f"提取完成: {result}")

asyncio.run(main())

配置优先级

SDK 按以下优先级加载配置：

1. 代码参数：LindormMemobase.from_config(key=value)；

2. 环境变量：MEMOBASE_* 前缀的环境变量；

3. YAML 文件：config.yaml 文件；

4. 默认值：内置默认配置。

Lindorm 宽表引擎配置

Lindorm 搜索引擎配置

LLM 模型配置

SDK 支持为不同任务配置不同的模型：

Embedding 配置

合并阈值配置

控制画像批处理合并策略：

# config.yaml
merge_thresholds:
  "interests::hobbies": 5      # 累积 5 条后合并
  "preferences::dietary": 3    # 累积 3 条后合并
max_pending_profiles: 1000     # 单个 subtopic 最大缓存数

项目配置（ProfileConfig）

SDK 支持为不同项目配置不同的画像提取策略，通过 ProfileConfig 实现：

画像主题结构：

{
    "topic": "interests",           # 主主题
    "description": "用户兴趣偏好",   # 主题描述
    "sub_topics": [                 # 子主题列表
        {
            "name": "hobbies",      # 子主题名称
            "description": "用户的兴趣爱好",
            "update_description": "用户的兴趣爱好变化",
            "validate_value": false,  # 是否需要验证提取值
            "merge_threshold": 5     # 子主题级别的合并阈值
        }
    ]
}

配置优先级：

1. Project-level ProfileConfig（通过 set_project_config 设置）；

2. Global config.yaml 或环境变量；

3. SDK 内置默认配置。

输入数据类型（Blob）

SDK 通过 Blob 对象接收输入数据，支持以下类型：

提取内容结构

提取结果包含两类数据：

• 画像结构，记录用户的长期特征和偏好。

• 时间结构，记录用户行为的变化。

{
    "topic": "interests",           # 主主题
    "sub_topic": "hobbies",         # 子主题
    "content": "用户喜欢阅读科幻小说和编程",  # 画像内容
    "created_at": "2024-01-01T00:00:00Z",
    "updated_at": "2024-01-01T00:00:00Z",
    "update_hits": 5                # 更新次数
}

{
    "event_data": {
        "profile_delta": [          # 画像变化
            {
                "topic": "preferences",
                "sub_topic": "dietary",
                "old_value": "无特殊偏好",
                "new_value": "偏爱素食"
            }
        ],
        "event_tip": "用户表达了饮食偏好变化",  # 事件摘要
        "event_tags": [             # 事件标签
            {"tag": "interest", "value": "food"},
            {"tag": "preference", "value": "dietary"}
        ]
    },
    "created_at": "2024-01-01T00:00:00Z"
}

使用 Buffer 批处理

当需要累积多个 Blob后统一处理时，使用 Buffer 机制提高效率。

from lindormmemobase.models.blob import ChatBlob, OpenAICompatibleMessage, BlobType

# 添加多个 Blob 到 Buffer
blob1 = ChatBlob(messages=[OpenAICompatibleMessage(role="user", content="我喜欢川菜")])
blob2 = ChatBlob(messages=[OpenAICompatibleMessage(role="user", content="我不吃香菜")])

await memobase.add_blob_to_buffer("user123", blob1, blob_id="msg1")
await memobase.add_blob_to_buffer("user123", blob2, blob_id="msg2")

# 检查 Buffer 是否已满，满则自动处理
status = await memobase.detect_buffer_full_or_not("user123", BlobType.chat)
if status["is_full"]:
    # 处理 Buffer 中的所有 Blob
    result = await memobase.process_buffer("user123", BlobType.chat)

# 也可以手动触发处理指定的 Blob
result = await memobase.process_buffer(
    "user123",
    BlobType.chat,
    blob_ids=["msg1", "msg2"]  # 只处理指定的 Blob
)

合并阈值策略

对于频繁更新的画像（如饮食偏好），可设置合并阈值，将多条零散更新合并为一条完整画像，减少数据库写入：

# 设置合并阈值
config = {
    "merge_thresholds": {
        "preferences::dietary": 3,  # 累积 3 条后合并
        "interests::hobbies": 5     # 累积 5 条后合并
    }
}
memobase = LindormMemobase.from_config(**config)

# 提取记忆时自动应用阈值策略
await memobase.extract_memories("user123", blobs)

# 也可以手动触发某个 subtopic 的合并
from lindormmemobase.models.response import MergeOperationResult
result: MergeOperationResult = await memobase.trigger_merge(
    user_id="user123",
    topic="interests",
    subtopic="hobbies"
)
print(f"合并了 {result.merged_count} 条画像")

检索模式对比

SDK 提供多种检索模式，适用于不同场景：

向量检索

profiles = await memobase.search_profiles_by_embedding(
    user_id="user123",
    query="旅游目的地推荐",
    max_results=5,
    min_score=0.5,        # 最低相似度阈值
    topics=["travel"]     # 可选：限定主题
)

for profile in profiles:
    print(f"{profile.topic}::{profile.sub_topic}: {profile.content}")

Rerank 检索

profiles = await memobase.search_profiles_with_rerank(
    user_id="user123",
    query="我的饮食偏好是什么？",
    max_results=5,
    combine_by_topic=True  # 按 topic::subtopic 合并后重排序
)

混合检索

profiles = await memobase.search_profiles_hybrid(
    user_id="user123",
    query="假期计划",
    max_results=5,
    embedding_candidates=30,   # 向量检索候选数
    min_embedding_score=0.3    # 向量检索最低分数
)

Lindorm filter_rrf 检索

profiles = await memobase.search_profiles_by_filter_rrf(
    user_id="user123",
    query="旅行偏好",
    topics=["travel"],
    subtopics=["destinations", "preferences"],
    max_results=5,
    min_score=0.5
)

事件检索

from lindormmemobase.models.response import EventSearchFilters

# 高级事件检索
filters = EventSearchFilters(
    topics=["life_plan"],
    subtopics=["travel"],
    time_range_in_days=30
)

events = await memobase.search_events_advanced(
    user_id="user123",
    query="欧洲旅行计划",
    limit=20,
    similarity_threshold=0.3,
    filters=filters
)

for event in events:
    print(f"{event.created_at}: {event.event_data}")

对话上下文生成

conversation = [
    OpenAICompatibleMessage(role="user", content="我想计划一次旅行")
]

context = await memobase.get_conversation_context(
    user_id="user123",
    conversation=conversation,
    max_token_size=2000,
    prefer_topics=["travel", "preferences"],  # 优先考虑的主题
    time_range_in_days=30,
    profile_event_ratio=0.6  # 60% Profile，40% Event
)

print(context)  # 可直接用于 LLM 对话

智能客服场景

以下示例展示完整流程——从用户对话中提取画像，在后续对话中检索上下文，生成个性化回复。

from lindormmemobase import LindormMemobase
from lindormmemobase.models.blob import ChatBlob, OpenAICompatibleMessage, BlobType

async def customer_service_demo():
    # 初始化
    memobase = LindormMemobase()

    user_id = "customer_001"

    # 用户对话
    conversation = [
        OpenAICompatibleMessage(role="user", content="我想买一台笔记本，平时用来做视频剪辑"),
        OpenAICompatibleMessage(role="assistant", content="好的，视频剪辑需要较高的配置。您的预算大概是多少？"),
        OpenAICompatibleMessage(role="user", content="大概一万左右，希望屏幕色彩好一些"),
    ]

    # 提取记忆
    blob = ChatBlob(messages=conversation)
    await memobase.extract_memories(user_id, [blob])

    # 后续对话：获取相关上下文
    new_query = OpenAICompatibleMessage(role="user", content="有什么推荐的吗？")

    context = await memobase.get_conversation_context(
        user_id=user_id,
        conversation=[new_query],
        prefer_topics=["shopping", "preferences"],
        max_token_size=1500
    )

    # 将 context 传给 LLM 生成个性化回复
    prompt = f"""
用户画像:
{context}

用户问题: {new_query.content}

请根据用户画像给出推荐:
"""
    # ... 调用 LLM

asyncio.run(customer_service_demo())

多项目配置

SDK 支持为不同业务场景配置独立的提取策略。通过 project_id 参数隔离不同项目的画像配置。

from lindormmemobase.models.profile_topic import ProfileConfig

# 为不同项目设置不同的画像配置
education_config = ProfileConfig(
    language="zh",
    overwrite_user_profiles=[
        {
            "topic": "学习偏好",
            "sub_topics": [
                {"name": "学习时间", "description": "用户偏好的学习时段"},
                {"name": "学习方式", "description": "线上/线下偏好"}
            ]
        }
    ]
)

ecommerce_config = ProfileConfig(
    language="zh",
    overwrite_user_profiles=[
        {
            "topic": "购物偏好",
            "sub_topics": [
                {"name": "价格区间", "description": "用户偏好的价格范围"},
                {"name": "品牌偏好", "description": "用户喜欢的品牌"}
            ]
        }
    ]
)

# 存储项目配置
await memobase.set_project_config("education_app", education_config)
await memobase.set_project_config("ecommerce_app", ecommerce_config)

# 提取时会自动使用对应项目的配置
await memobase.extract_memories("user123", blobs, project_id="education_app")

# overwrite: 完全覆盖默认画像主题
strict_config = ProfileConfig(
    profile_strict_mode=True,
    overwrite_user_profiles=[
        {
            "topic": "特定主题",
            "sub_topics": [{"name": "特定字段"}]
        }
    ]
)
# 提取时只会识别 "特定主题::特定字段"，其他主题会被忽略

# additional: 在默认主题基础上追加
flexible_config = ProfileConfig(
    additional_user_profiles=[
        {
            "topic": "扩展主题",
            "sub_topics": [{"name": "扩展字段"}]
        }
    ]
)
# 提取时会识别默认主题 + "扩展主题::扩展字段"

# 方式一：通过 ProfileConfig 配置
config = ProfileConfig(
    overwrite_user_profiles=[
        {
            "topic": "preferences",
            "sub_topics": [
                {
                    "name": "dietary",
                    "merge_threshold": 3  # 该 subtopic 累积 3 条后合并
                }
            ]
        }
    ]
)

# 方式二：通过 merge_thresholds 字段配置（优先级更高）
config = ProfileConfig(
    merge_thresholds={
        "preferences::dietary": 3,  # 全局配置
        "interests::hobbies": 5
    }
)

config = ProfileConfig(
    event_theme_requirement="用户行为变化",
    event_tags=[
        {"name": "行为改变", "description": "用户行为发生显著变化"},
        {"name": "兴趣转移", "description": "用户兴趣点发生变化"}
    ]
)

# education_config.yaml
language: zh
profile_strict_mode: false
overwrite_user_profiles:
  - topic: 学习偏好
    sub_topics:
      - name: 学习时间
        description: 用户偏好的学习时段
      - name: 学习方式
        description: 线上/线下偏好
merge_thresholds:
  "学习偏好::学习方式": 3

# 加载并使用
config = ProfileConfig.load_from_file("education_config.yaml")
await memobase.set_project_config("education_app", config)