agent-memory-sdk

Agent Memory Python SDK 用于让 Python Agent 接入外部记忆服务。当前版本提供同步 Client 和轻量 AgentMemorySession 接入层。

安装

python -m pip install agent-memory-sdk

本地开发时可以从源码安装：

cd sdk/python
python -m pip install -e .

创建 Client

import os

from agent_memory import AgentMemoryClient

client = AgentMemoryClient(
    base_url=os.environ["AGENT_MEMORY_BASE_URL"],
    api_key=os.environ["AGENT_MEMORY_API_KEY"],
)

常用环境变量：

export AGENT_MEMORY_BASE_URL='http://localhost:8080'
export AGENT_MEMORY_API_KEY='<workspace api key>'
export AGENT_MEMORY_AGENT_INSTANCE_ID='<agent_instance_id>'

创建或复用 Session

每个 Agent 会话开始时先创建或复用 session。external_session_id 是调用方自己的会话 ID，同一个 Agent 下重复传入会复用同一个服务端 session。

import os

from agent_memory.types import CreateSessionReq

session = client.create_session(
    CreateSessionReq(
        agent_instance_id=os.environ["AGENT_MEMORY_AGENT_INSTANCE_ID"],
        external_session_id="python-task-20260616-001",
        metadata={"source": "python-agent"},
    )
)

print("session_id:", session.session_id)

写入 Memory Event

Agent 每轮推理和工具执行过程都应该写入 memory event。事件是记忆生成的事实来源。

from agent_memory.types import RecordEventReq

event = client.record_event(
    RecordEventReq(
        session_id=session.session_id,
        external_thread_id="main",
        event_id="evt-user-001",
        event_type="message",
        actor="user",
        status="success",
        summary="用户要求继续 SDK 接入任务。",
        payload={"text": "请继续 SDK 接入任务"},
    )
)

print("thread_id:", event.thread_id)

常用事件类型：

工具失败事件建议填写 status、error_code 和 retryable。

client.record_event(
    RecordEventReq(
        session_id=session.session_id,
        event_id="evt-tool-result-001",
        event_type="tool_result",
        actor="tool",
        status="error",
        tool_name="pytest",
        error_code="TEST_FAILED",
        retryable=True,
        summary="pytest 执行失败。",
        payload={"command": "pytest", "exit_code": 1},
    )
)

查询记忆

每轮 Agent 推理前建议调用一次记忆查询，获取少量可注入上下文。

from agent_memory.adapter import format_memory_context
from agent_memory.types import MemoryQueryReq

response = client.query_memories(
    MemoryQueryReq(
        session_id=session.session_id,
        thread_id=event.thread_id,
        query_text="继续 SDK 接入任务，需要恢复当前进展和注意事项",
        query_intent="execution_help",
        desired_memory_types=[
            "task_whiteboard",
            "tool_digest",
            "failure_digest",
            "procedure",
        ],
        limit=8,
        event_limit=3,
    )
)

prompt_context = format_memory_context(response)
print(prompt_context)

常用查询意图：

触发短记忆生成

短记忆用于整理当前 session/thread 的近期上下文，例如任务白板、工具结果摘要、失败摘要。

适合触发时机：

• 用户消息写入后，尤其是用户给出新目标或新约束。
• 工具调用或工具结果写入后。
• 一个阶段性步骤完成后。
• 出现错误、超时、取消、用户纠错后。

from agent_memory.types import GenerateMemoryReq

run = client.generate_short_memory(
    GenerateMemoryReq(
        session_id=session.session_id,
        thread_id=event.thread_id,
        trigger_source="thread_end",
        hints={"reason": "阶段完成后刷新任务白板"},
    )
)

print("short memory queued:", run.run_ids)

status=queued 表示任务已入队。需要 worker 消费完成后，新的短记忆才会出现在查询结果中。

触发长记忆整理

长记忆用于沉淀跨会话可复用的信息，例如事实、偏好、流程和规则。

适合触发时机：

• 会话结束时。
• 一个完整任务完成后。
• 用户明确表达稳定偏好时。
• 形成可复用流程、项目事实、团队规则时。
• 不建议对临时、未经确认、明显只对当前轮有效的信息触发长期整理。

run = client.consolidate_long_memory(
    GenerateMemoryReq(
        session_id=session.session_id,
        trigger_source="session_end",
        hints={"reason": "会话结束后沉淀长期记忆"},
    )
)

print("long memory queued:", run.run_ids)

使用 AgentMemorySession 简化接入

AgentMemorySession 会自动创建 session、写标准事件，并缓存服务端返回的 thread_id，后续查询和生成会自动带上线程上下文。

import os

from agent_memory import AgentMemoryClient, AgentMemorySession

client = AgentMemoryClient(
    base_url=os.environ["AGENT_MEMORY_BASE_URL"],
    api_key=os.environ["AGENT_MEMORY_API_KEY"],
)

agent_memory = AgentMemorySession(
    client,
    agent_instance_id=os.environ["AGENT_MEMORY_AGENT_INSTANCE_ID"],
    external_session_id="python-task-20260616-001",
    external_thread_id="main",
)

agent_memory.record_user_message("请继续 SDK 接入任务。")

short_run = agent_memory.generate_short_memory("thread_end")
print("short memory queued:", short_run.run_ids)

memory_context = agent_memory.query_context(
    query_text="继续 SDK 接入任务",
    query_intent="execution_help",
    desired_memory_types=["task_whiteboard", "tool_digest", "procedure"],
    limit=6,
    event_limit=3,
)

print(memory_context.text)

查看示例

cd sdk/python
python examples/basic.py
python examples/agent_loop.py