# Agent 框架接入方案

MemPlumb 是 Agent 的长期 MemoryOps 和发布证据层，不是 Agent 编排器，也不替代 OpenAI Agents、LangGraph 或业务自己的工作流。接入时只需要把三个明确的挂载点接到现有 Agent 生命周期：

> **版本边界**：本文的 `agentTurn(...)` / `agent_turn(...)` 和
> `demo-agent` 目前位于 source-only v0.14 迭代线。公开的 v0.13.2 SDK
> 压缩包尚未包含它；v0.13.2 请手动调用 `ingest`、`context` 和
> `recordOutcome`。发布前必须通过归档内容检查，不能用同一版本号覆盖旧 SDK
> 资产。

`demo-agent` 只用于验证本地接线。它只有在本轮 ingestion 产生或更新的 Memory
确实被本轮 Context 选中时才将演示任务标为成功，并把 Outcome 明确标记为
`unverified`、`demo_only`。它不能替代受管 evaluator，也不能进入生产质量门禁。

```text
用户输入
  -> 可选：写入 Event（同一 Actor ID）
  -> 创建持久化 Context Run
  -> 将 Context 交给 Agent
  -> 业务结果完成后记录 Outcome / Feedback
  -> 线上故障进入 Review -> Case -> Replay -> Release
```

## 最短可用路径

```javascript
import { MemPlumbClient } from "@memplumb/sdk";

const memory = new MemPlumbClient({
  baseUrl: process.env.MEMPLUMB_URL || "http://127.0.0.1:6060",
  apiKey: process.env.MEMPLUMB_API_KEY,
});

const actorId = "user_123";
const turnId = "order-2026-08-15-001";

// 1. 只在业务允许写入时发送 Event。跨进程重试必须复用这个键。
await memory.ingest(
  { actor_id: actorId, text: "用户刚完成一次杭州出差" },
  { idempotencyKey: `agent-event-${turnId}` },
);

// 2. Context 会创建持久化 Context Run，返回 context_id 和检索证据。
const context = await memory.context({
  actor_id: actorId,
  query: "这次对话需要知道用户的哪些长期偏好？",
  budget_tokens: 1200,
});

// 3. 将 context.data.context 交给任意 Agent 框架。
const answer = await runAgent({
  input: userMessage,
  memoryContext: context.data.context,
});

// 4. 只有业务评估器知道任务是否成功；不要从 HTTP 200 推断成功。
await memory.recordOutcome(
  context.data.context_id,
  {
    task_success: true,
    safety_pass: true,
    score: 0.92,
    rubric_version: "agent-task-v1",
    evaluator_type: "programmatic",
    evaluator_id: "agent-runtime",
  },
  { idempotencyKey: `agent-outcome-${turnId}` },
);
```

`runAgent` 是业务或框架自己的调用。MemPlumb 只负责可追溯的写入、检索和结果证据；如果 Context 创建失败，应用必须按自己的降级策略处理，不能把空 Context 当作“没有记忆”或“记忆服务正常”。

## SDK 一步接入

如果不需要把生命周期节点分别放进自己的工作流，可以使用 SDK 提供的框架无关 helper。它不依赖 OpenAI Agents、LangGraph 或任何模型库，只把四个阶段按明确顺序串起来：

```javascript
const turn = await memory.agentTurn(
  {
    actor_id: "user_123",
    query: "这次任务需要知道哪些长期偏好？",
    ingest: { actor_id: "user_123", text: userMessage },
    budget_tokens: 1200,
  },
  async ({ context, context_id, ingest }) => ({
    output: await runAgent({
      input: userMessage,
      memoryContext: context.context,
      contextId: context_id,
      ingestedEvent: ingest?.event.id,
    }),
    // 只有业务评估器能判断时才返回 outcome；省略则不会写 Outcome。
    outcome: {
      task_success: true,
      safety_pass: true,
      score: 0.92,
      rubric_version: "agent-task-v1",
      evaluator_type: "programmatic",
      evaluator_id: "agent-runtime",
    },
  }),
  { turnId: `conversation-${conversationId}` },
);
```

### 将 Outcome 写入交给独立评估身份

默认情况下，helper 会使用发起 `agentTurn` 的同一个客户端写入 Outcome。这适合
开发环境或应用凭据本身已经具备 `evaluation:write` 的情况。生产环境通常应把
“运行 Agent”和“裁决结果”分成两个 API Key：应用 Key 只负责 `memory:read` /
`memory:write`，评估 Key 绑定受管的 evaluator 身份，并通过
`outcomeRecorder`（Python 为 `outcome_recorder`）注入。服务端以 Key 绑定的身份
为准，请求体不能冒充另一个评估者；只有经过认证的 Outcome 才能进入质量门禁。

```javascript
const appMemory = new MemPlumbClient({
  baseUrl: process.env.MEMPLUMB_URL,
  apiKey: process.env.MEMPLUMB_AGENT_KEY,
});
const evaluatorMemory = new MemPlumbClient({
  baseUrl: process.env.MEMPLUMB_URL,
  apiKey: process.env.MEMPLUMB_EVALUATOR_KEY,
});

const turn = await appMemory.agentTurn(
  {
    actor_id: "user_123",
    query: "这次任务需要哪些长期偏好？",
  },
  async ({ context }) => ({
    output: await runAgent({ memoryContext: context.context }),
    outcome: {
      task_success: true,
      safety_pass: true,
      score: 0.92,
      rubric_version: "agent-task-v1",
    },
  }),
  {
    turnId: `conversation-${conversationId}`,
    outcomeRecorder: evaluatorMemory,
  },
);
```

Python 使用同一边界；异步客户端可以注入另一个 `AsyncMemPlumbClient`，其
`record_outcome` 可以是协程：

```python
async with (
    AsyncMemPlumbClient(
        base_url=os.environ["MEMPLUMB_URL"],
        api_key=os.environ["MEMPLUMB_AGENT_KEY"],
    ) as app_memory,
    AsyncMemPlumbClient(
        base_url=os.environ["MEMPLUMB_URL"],
        api_key=os.environ["MEMPLUMB_EVALUATOR_KEY"],
    ) as evaluator_memory,
):
    turn = await app_memory.agent_turn(
        "user_123",
        run_agent,
        query="这次任务需要哪些长期偏好？",
        turn_id=f"conversation-{conversation_id}",
        outcome_recorder=evaluator_memory,
    )
```

评估 Key 应由服务端创建并绑定 evaluator ID/type，且具备
`evaluation:write`；不要让终端用户或 Agent 自己填写
`evaluator_id`。如果省略 recorder，行为保持向后兼容，Outcome 仍写入当前 client；
如果当前身份没有评估写权限，应省略 Outcome、稍后由受管评估流水线补写，而不是
把 HTTP 成功当成业务成功。

Python 同步客户端使用同一契约，异步客户端的 `run_agent` 可以是普通函数或 async 函数：

```python
def run_agent(turn):
    return {
        "output": run_business_agent(
            user_message,
            memory_context=turn["context"]["context"],
        ),
        "outcome": {
            "task_success": True,
            "safety_pass": True,
            "score": 0.92,
            "rubric_version": "agent-task-v1",
        },
    }

turn = memory.agent_turn(
    "user_123",
    run_agent,
    query="这次任务需要知道哪些长期偏好？",
    ingest={"actor_id": "user_123", "text": user_message},
    turn_id=f"conversation-{conversation_id}",
)
```

`turnId`/`turn_id` 会先做 SHA-256，再派生两个不同的稳定键：`...-event` 和
`...-outcome`，避免把业务会话标识放进请求头。也可以显式传
`eventIdempotencyKey`、`outcomeIdempotencyKey`（Python 为下划线命名）。
Actor ID、turn ID 和显式幂等键会先去除首尾空白，必须是非空字符串且不超过
200 个字符；`ingest` 必须是对象（Python 为 mapping），其中的 `actor_id` 必须
与请求 Actor 相同。校验失败会在本地抛出异常，不会产生网络请求。不要把 API Key、
原始凭据或隐私内容放进这些标识。
Context Run 本身仍是带证据的非幂等创建，不会被 helper 自动重试；网络不确定时
不要从头重跑整个 helper。应保留 `AgentTurnError` 中已经完成的阶段，并针对已知
`context_id` 单独重试 `recordOutcome`。

四个阶段中的任何一步失败都会抛出带 `stage` 的 `AgentTurnError`（Python 同名异常），并在可用时附带已完成的 `ingest`、`context` 和 `agent` 结果。尤其是 Context 失败不会返回空 Context，也不会调用 Agent 回调；Outcome 只有回调明确返回时才会写入。

## OpenAI Agents 挂载点

使用 OpenAI Agents 时，把 MemPlumb 放在 Agent 运行前后的应用层：

| 生命周期          | MemPlumb 动作                | 传给 Agent 的内容                               |
| ----------------- | ---------------------------- | ----------------------------------------------- |
| turn start        | `ingest`（可选）             | 不把 API Key 或原始凭据写入 Event               |
| before model/tool | `context`                    | `context.data.context`，以及必要的 `context_id` |
| after task        | `recordOutcome` / `feedback` | 只提交业务已确认的结果和评分                    |
| incident          | `contextTrace` + Console     | 进入 Review、Memory Case 和 Replay              |

不要把 Context Run ID 当作 OpenAI 的 Session ID；前者是 MemPlumb 的证据锚点，后者由 Agent 框架管理。两者可以在业务 trace metadata 中关联，但不应互相替代。

## LangGraph 挂载点

在 LangGraph 中建议把 Memory 节点和 Outcome 节点显式放进图，而不是隐藏在模型节点内部：

```text
START
  -> memory_ingest (可选，幂等)
  -> memory_context (创建 Context Run，写入 context_id)
  -> model / tools
  -> outcome_record (由业务评估器决定)
  -> END
```

图状态至少保留 `actor_id`、`context_id` 和本次 turn 的幂等键。重试 `memory_ingest` 或 `outcome_record` 时复用同一个键；重试 `memory_context` 前先确认应用是否愿意创建新的 Context Run，因为 Context 是带证据的非幂等读取写入组合。

## 首个 Memory Case 如何产生

普通应用不应直接上传 Case、Oracle 或评审身份。正确路径是：

1. Agent 产生真实 Context Run。
2. 业务反馈或失败信号进入质量队列。
3. 具备 `evaluation:write` 的评审身份提交精确裁决。
4. 具备 `evaluation:execute` 的执行者从认证裁决创建 Memory Case。
5. Replay Lab 比较候选策略；发布时使用 Cohort Plan 和 freshness gate。

这样得到的 Case 绑定真实生产状态，不会因为客户端拼接数据而失去可信度。

## 安全与隐私边界

- 应用只给 Agent 运行时 `memory:read`/`memory:write`；评审、导出和发布使用独立凭据。
- Bearer/API Key 只放在服务端密钥管理器，不进入 Prompt、Event metadata、日志或截图。
- Actor ID 在业务侧保持稳定；Memory 值、查询和评审证据按最小权限读取。
- 生产发布必须使用 Runtime 返回的 `passed`、`fresh` 和 `blocked` 等状态，不根据前端颜色或 HTTP 200 猜测。
- 适配器本身不改变 Memory Policy、Context Policy 或 Release Artifact 契约；框架升级应先跑完整发布门禁。

## 验收清单

- [ ] 空 Workspace 能完成一次 Event -> Context -> Agent -> Outcome。
- [ ] Event 和 Outcome 的重试复用同一幂等键，并能在 Console 中看到 `context_id`。
- [ ] Context 失败时应用有明确降级，不把空结果当作成功。
- [ ] 线上失败能从 Context 进入 Review，再生成 Case 和 Replay。
- [ ] 候选策略通过完整质量门禁后才进入 Release；发布工作流先执行 `npm run verify:release`。