Metadata-Version: 2.4
Name: agent-dreamer-sdk
Version: 0.0.3
Summary: Agent Memory HTTP SDK and lightweight Agent adapter
Author: Agent Memory Contributors
License: MIT
Keywords: agent,memory,sdk
Requires-Python: >=3.10
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Description-Content-Type: text/markdown

# agent-memory-sdk

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

## 安装

```bash
python -m pip install agent-memory-sdk
```

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

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

## 创建 Client

```python
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"],
)
```

常用环境变量：

```bash
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。

```python
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。事件是记忆生成的事实来源。

```python
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)
```

常用事件类型：

| event_type | 适合场景 |
| --- | --- |
| `message` | 用户消息、Agent 回复 |
| `tool_call` | 调用工具前记录工具名和参数摘要 |
| `tool_result` | 工具执行完成后记录结果、状态、错误码 |
| `planner_update` | 计划、待办、阶段性决策变化 |
| `task_state` | 任务开始、暂停、完成、失败、阻塞 |
| `artifact_change` | 文件、代码、文档、报告等产物变化 |
| `correction_signal` | 用户纠错或要求修正此前信息 |

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

```python
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 推理前建议调用一次记忆查询，获取少量可注入上下文。

```python
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)
```

常用查询意图：

| query_intent | 适合场景 |
| --- | --- |
| `execution_help` | 继续任务、写代码、调用工具、排错 |
| `policy_lookup` | 查询规则、约束、团队规范 |
| `preference_lookup` | 查询用户偏好、输出风格 |
| `factual_lookup` | 查询稳定事实、项目背景 |
| `broad_context` | 不确定意图时综合召回 |

## 触发短记忆生成

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

适合触发时机：

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

```python
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 消费完成后，新的短记忆才会出现在查询结果中。

## 触发长记忆整理

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

适合触发时机：

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

```python
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`，后续查询和生成会自动带上线程上下文。

```python
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)
```

## 查看示例

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