Skip to main content

TiMEM Python SDK - Time-based Memory Management and Rule Learning toolkit | TiMEM Python SDK - 时间记忆管理与规则学习工具包

Project description

TiMEM Python SDK

Time-based Memory Management and Rule Learning Toolkit

TiMEM SDK 是一个强大的 Python 客户端库,用于与 TiMEM Engine API 交互,提供记忆管理、通用规则学习和用户画像计算等功能。

特性

  • 记忆管理:支持 L1-L5 分层记忆的增删改查
  • 通用规则学习:从场景/结果中学习可复用规则,并在相似场景中召回
  • 用户画像:从 L5 级别记忆计算用户画像
  • 同步/异步支持:记忆管理和规则学习均提供同步/异步入口
  • 批量操作:支持批量记忆写入,适合高吞吐场景
  • 自动重试:内置请求重试机制保证可靠性

安装

pip install timem-ai

或从源码安装:

git clone https://github.com/TiMEM-AI/timem-sdk-python.git
cd timem-sdk-python
pip install -e .

快速开始

TiMEM SDK 的常用入口分为两类:TiMEMClient / AsyncTiMEMClient 用于记忆管理、用户画像等 TiMEM API;Rules / AsyncRules 用于通用规则学习。rules / async_rules 是等价的小写别名,可按项目风格选择。

from timem import TiMEMClient, AsyncTiMEMClient, Rules, AsyncRules

同步客户端

from timem import TiMEMClient, Rules

client = TiMEMClient(api_key="your-api-key", base_url="http://localhost:8001")
rules = Rules(api_key="your-api-key", base_url="http://localhost:8001")

# 1. 记忆管理:写入一条 L1 交互记忆
memory = client.add_memory(
    user_id=12345,
    domain="aicv",
    content={
        "type": "interaction",
        "action": "resume_analysis",
        "context": {"job": "软件工程师"},
    },
    layer_type="L1",
    tags=["resume", "analysis"],
)

# 2. 规则学习:从单条场景/结果学习通用规则
learned = rules.learn(
    situation_text="候选人项目经历影响力描述比较笼统",
    outcome_text="要求补充量化指标和个人贡献边界",
    attributes={"scene": "resume_review", "role": "backend"},
    suggested_tags=["resume", "project_impact"],
    user_id="user-123",
    agent_id="hire",
)
print(learned["rule_id"])

# 3. 规则召回:根据当前场景召回相关规则
recalled = rules.recall(
    situation_text="评估 Python 后端候选人的项目经历",
    context_text="候选人只写了负责系统开发,缺少指标和业务结果",
    mode="auto",
    filters={"role": "backend"},
    top_k=5,
    user_id="user-123",
    agent_id="hire",
)
print(f"Recalled {recalled['count']} rules")

异步客户端

import asyncio
from timem import AsyncTiMEMClient, AsyncRules

async def main():
    async with AsyncTiMEMClient(api_key="your-api-key", base_url="http://localhost:8001") as client:
        memories = [
            {"user_id": 12345, "domain": "aicv", "content": {"action": "view_job", "job_id": 1}},
            {"user_id": 12345, "domain": "aicv", "content": {"action": "apply_job", "job_id": 1}},
        ]
        results = await client.batch_add_memories(memories)

    async with AsyncRules(api_key="your-api-key", base_url="http://localhost:8001") as rules:
        learned = await rules.learn(
            situation_text="候选人项目经历影响力描述比较笼统",
            outcome_text="要求补充量化指标和个人贡献边界",
            suggested_tags=["resume", "project_impact"],
            user_id="user-123",
            agent_id="hire",
        )

        recalled = await rules.recall(
            situation_text="评估 Python 后端候选人的项目经历",
            context_text="候选人只写了负责系统开发,缺少指标和业务结果",
            mode="auto",
            top_k=5,
            user_id="user-123",
            agent_id="hire",
        )

asyncio.run(main())

API 文档

通用规则学习

Rules.learn() / TiMEMClient.learn_rule()

从单条场景/结果中学习一条可复用规则。

rules.learn(
    situation_text="场景描述",
    outcome_text="这次应沉淀下来的处理经验",
    attributes={"scene": "resume_review"},
    suggested_tags=["resume"],
    user_id="default",
    agent_id="default",
)

Rules.recall() / TiMEMClient.recall_rules()

根据当前场景召回相关规则。

rules.recall(
    situation_text="当前待判断场景",
    context_text="更长的上下文材料",
    mode="auto",
    tags_hint=["resume"],
    filters={"scene": "resume_review"},
    top_k=5,
)

记忆管理

对话生成(v0.2.0 异步)

POST /api/v1/memory/ 返回 HTTP 202 与 task_id;使用 wait_for_generation_task 轮询结果,或使用 Memory.add()(内部自动轮询)。

acceptance = client.generate_memory(
    character_id="agent_1",
    session_id="sess_abc",
    user_id="user_1",
    messages=[
        {"role": "user", "content": "你好"},
        {"role": "assistant", "content": "您好!"},
    ],
)
task = client.wait_for_generation_task(acceptance["task_id"])
print(task.get("memories", []))

记忆生成策略(Policy)

policy = client.get_memory_policy(user_id="user_1", expert_id="agent_1")
client.put_memory_policy(
    user_id="user_1",
    expert_id="agent_1",
    max_memory_layer="L5",
    profile_auto_recompute_enabled=False,
)

add_memory()

添加记忆到 TiMEM 系统

参数:

  • user_id (int): 用户ID
  • domain (str): 业务领域
  • content (Dict): 记忆内容
  • layer_type (str): 记忆层级(L1-L5),默认 "L1"
  • tags (List[str], optional): 标签
  • keywords (List[str], optional): 关键词

返回: 创建的记忆信息

search_memory()

搜索记忆

参数:

  • user_id (int, optional): 用户ID过滤
  • domain (str, optional): 领域过滤
  • layer_type (str, optional): 层级过滤
  • tags (List[str], optional): 标签过滤
  • keywords (List[str], optional): 关键词过滤
  • limit (int): 返回数量,默认 20
  • offset (int): 分页偏移,默认 0

返回: 搜索结果和记忆列表

用户画像

compute_profile()

从 L5 级别记忆计算用户画像

参数:

  • user_id (int): 用户ID
  • domain (str): 业务领域
  • source_memory_ids (List[str], optional): 指定记忆ID

返回: 计算的画像信息

get_profile()

获取用户画像

参数:

  • user_id (int): 用户ID
  • domain (str): 业务领域

返回: 用户画像信息

search_users()

根据画像特征搜索用户

参数:

  • criteria (Dict): 搜索条件(如偏好、行为模式)
  • domain (str, optional): 领域过滤
  • limit (int): 返回数量,默认 20

返回: 匹配的用户列表

异步批量操作

batch_add_memories()

并发添加多条记忆

async with AsyncTiMEMClient(api_key="your-key") as client:
    memories = [
        {"user_id": 1, "domain": "aicv", "content": {...}},
        {"user_id": 2, "domain": "aicv", "content": {...}}
    ]
    results = await client.batch_add_memories(memories)

异常处理

from timem import Rules, TiMEMError, AuthenticationError, APIError, ValidationError

try:
    rules = Rules(api_key="your-key")
    result = rules.learn(situation_text="场景", outcome_text="经验")
except ValidationError as e:
    print(f"验证错误: {e}")
except AuthenticationError as e:
    print(f"认证失败: {e}")
except APIError as e:
    print(f"API错误 [{e.status_code}]: {e}")
except TiMEMError as e:
    print(f"TiMEM错误: {e}")

配置选项

client = TiMEMClient(
    api_key="your-api-key",
    base_url="http://localhost:8001",  # TiMEM Engine URL
    timeout=60.0,                       # 请求超时(秒)
)

async_client = AsyncTiMEMClient(
    api_key="your-api-key",
    base_url="http://localhost:8001",
    timeout=60.0,
    max_retries=3,                      # 最大重试次数
    retry_delay=1.0,                    # 重试延迟(秒)
)

开发指南

安装开发依赖

pip install -r requirements-dev.txt

运行测试

pytest tests/

构建发布

python -m build
python -m twine upload dist/*

架构说明

TiMEM SDK 遵循以下设计原则:

  1. 模块化入口:记忆管理、规则学习和用户画像按能力拆分入口,可按需组合
  2. 分层记忆:支持 L1-L5 五层记忆管理
  3. 规则学习:支持从场景/结果中沉淀规则,并在相似场景中召回
  4. 同步/异步并行:核心能力均提供同步和异步调用方式
  5. 可靠调用:内置请求重试和统一异常处理

许可证

MIT License

联系我们

更新日志

v0.1.9 (2026-06-30)

  • 调整 README/PyPI 首屏文案,让记忆管理和规则学习作为同等级能力呈现
  • 快速开始同时展示 TiMEMClient / AsyncTiMEMClientRules / AsyncRules 常用入口
  • 更新架构原则,避免只突出规则学习路径

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

timem_ai-0.2.0.tar.gz (61.1 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

timem_ai-0.2.0-py3-none-any.whl (58.0 kB view details)

Uploaded Python 3

File details

Details for the file timem_ai-0.2.0.tar.gz.

File metadata

  • Download URL: timem_ai-0.2.0.tar.gz
  • Upload date:
  • Size: 61.1 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for timem_ai-0.2.0.tar.gz
Algorithm Hash digest
SHA256 d86a681069ff3a517cbcf7153e3f00c232980a1300ee1ad88eae7fb57c0be3ef
MD5 04809bcb1fc59ebd909432c7de7ac2f7
BLAKE2b-256 04222ebd7215395fc2978b80bfde50488687e142f97e124533e97f61bb457da6

See more details on using hashes here.

File details

Details for the file timem_ai-0.2.0-py3-none-any.whl.

File metadata

  • Download URL: timem_ai-0.2.0-py3-none-any.whl
  • Upload date:
  • Size: 58.0 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for timem_ai-0.2.0-py3-none-any.whl
Algorithm Hash digest
SHA256 1b6db8942f20c106247605db8d3aeccc8b318ed807bc30008f01de65160d538d
MD5 41225c4048c3ace2a8113c889d4aa89f
BLAKE2b-256 94c6d9653e7d299d73faf2a416b108b40adc551bf27ef46de0e93222528635a1

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page