> ## Documentation Index
> Fetch the complete documentation index at: https://powercontext.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Source 与 Memory 实操

> 先采集 Source，再抽取 Memory。采集成功不等于留下知识。

上一页直接写下了你已经确认的 Memory。现在走另一条路：先采集 Source (记录发生过什么。采集不会自动写成长期知识。)，再让生成流程从证据里抽出以后还要用的知识。

两条路都可能得到 Memory (经过选择、以后还会用到的项目知识。)。前提、证据链和失败方式不一样。

![两条路都能留下 Memory：直接写下，或先采集再抽取。](https://mintcdn.com/powercontext/1t_ajifOEOG5YQjP/assets/basic-source-memory-zh.svg?fit=max\&auto=format\&n=1t_ajifOEOG5YQjP\&q=85\&s=37a21f81b7500ec552b8d4c01293dbef)

> [!NOTE]
>
> 生成模型和抽取配置以 [官方文档](https://oceanbase.github.io/powercontext/) 的最新发布版本为准。这页帮你看清两条路差在哪里，以及怎样核对证据链。

## 两条路先放在一起看

显式写入保存你已经确认要留的内容。抽取先保存 Source，而且可能不产生任何 Memory。

| 路径 | 需要模型 | 是否先创建 Source | 适合什么 |
| - | -: | -: | - |
| `remember_memory` | 否 | 否 | 你已经知道要保存的事实、决策或约束 |
| 采集 → 刷新/调度器 → 抽取 | 是 | 是 | 从一批工作证据中找出值得长期保留的内容 |

如果 Agent 已经得到明确指令“记住这条决策”，直接写下更简单。不要为了证据链好看，先人工造一条 Source 再绕回 Memory。

反过来，自动采集到的提示词通常只应该先成为 Source。要不要提炼成 Memory，是后面的判断。

## 先检查抽取能力

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
powercontext capabilities
```

必须看到：

```text theme={"theme":{"light":"github-light","dark":"github-dark"}}
Memory extraction: enabled
```

如果是 disabled：

* 采集仍可成功；
* Source 会留在日志；
* 待处理 Source 窗口无法执行模型抽取；
* 先运行配置流程，再用 `--env-file` 重启 Server。

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
powercontext config init --output .env
powercontext config show --env-file .env
powercontext config validate --env-file .env
powercontext server run --env-file .env
```

是否开启调度器是另一件事。本页主动调用 `flush_memory`，不依赖定时任务。

## 完整证据链脚本

新建 `source_memory.py`：

```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
import asyncio
import os
from uuid import uuid4

from powercontext.client import PowerContextClient
from powercontext.http import (
    CaptureContentSourceRequest,
    FlushMemoryRequest,
    ListMemoryEntriesRequest,
    PrepareContextRequest,
    SearchMemoryRequest,
)


SERVER_URL = os.getenv(
    "POWERCONTEXT_CLIENT_SERVER_URL",
    "http://127.0.0.1:8000",
)
API_TOKEN = os.getenv("POWERCONTEXT_CLIENT_API_TOKEN")


async def main() -> None:
    scope_id = f"project:powercontext-101:{uuid4().hex}"
    source_id = "openapi-task-1"

    async with PowerContextClient(
        SERVER_URL,
        token=API_TOKEN,
        timeout=30.0,
    ) as client:
        capabilities = await client.get_capabilities()
        if not capabilities.memory_extraction:
            raise RuntimeError(
                "Memory extraction is disabled. Configure a generation "
                "model and restart the Server with --env-file."
            )

        request = CaptureContentSourceRequest(
            scope_id=scope_id,
            source_id=source_id,
            content=(
                "Decision: after changing the OpenAPI contract, regenerate "
                "and inspect the checked-in client before contract tests."
            ),
            metadata={
                "origin": "powercontext-101",
                "chapter": "source-and-memory",
            },
        )

        captured = await client.capture_content_source(request)
        assert captured.status == "accepted"
        print("source:", captured.source)
        print("position:", captured.position)

        replayed = await client.capture_content_source(request)
        assert replayed.source == captured.source
        assert replayed.position == captured.position
        print("identical replay kept the original position")

        flushed = await client.flush_memory(
            FlushMemoryRequest(scope_id=scope_id)
        )
        print("flush status:", flushed.status)
        print(
            "cursor:",
            flushed.previous_cursor,
            "->",
            flushed.current_cursor,
            "high watermark:",
            flushed.high_watermark,
        )
        print("processed source count:", flushed.processed_source_count)
        print("memory:", flushed.memory)

        if flushed.current_cursor < captured.position:
            raise RuntimeError(
                "The cursor has not reached the captured Source yet. "
                "Flush again or inspect the generation failure."
            )

        page = await client.list_memory_entries(
            ListMemoryEntriesRequest(
                scope_id=scope_id,
                include_inactive=True,
            )
        )

        linked = [
            entry
            for entry in page.entries
            if captured.source in entry.source_refs
        ]

        if not linked:
            print(
                "The Source window was consumed, but extraction produced "
                "no Memory entry linked to this Source. This is a valid "
                "no-op; inspect the model output and content, then retry "
                "with a new Source ID if you need a positive lineage check."
            )

        for entry in linked:
            print("entry:", entry.kind, entry.text)
            print("citation:", entry.citation)
            print("source refs:", entry.source_refs)
            assert captured.source in entry.source_refs

        search = await client.search_memory(
            SearchMemoryRequest(
                scope_id=scope_id,
                query="What should happen before contract tests?",
                mode="auto",
                limit=10,
            )
        )
        print("search mode:", search.mode)
        print("hits:", len(search.hits))

        prepared = await client.prepare_context(
            PrepareContextRequest(
                scope_id=scope_id,
                query="What should happen before contract tests?",
            )
        )
        print("prepared status:", prepared.status)
        print(prepared.content)


if __name__ == "__main__":
    asyncio.run(main())
```

运行：

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
uv run python source_memory.py
```

生成结果本身不稳定。教程用一条明确的长期决策，提高抽出内容的机会，但不能把某段模型文本写成固定答案。真正稳定的检查是位置、游标、精确 SourceRef 和条目证据链。

## 每一步看什么

### 采集

成功状态是 HTTP 202，响应包含：

* `status="accepted"`；
* 精确内容 SourceRef；
* Scope 内部的数字 `position`。

同一个标识、同一串序列化字节再发一次，会回到原来的位置。

这次幂等比较看的是字节，不是语义是否相同。元数据键顺序变了，序列化字节也可能变，从而返回 `source_conflict`。真正重试时，复用原来那份输入。

同标识改内容一定不能覆盖旧 Source，应返回 HTTP 409。

### 刷新

判断 Source 是否被消费，看：

* `previous_cursor`；
* `current_cursor`；
* `processed_source_count`；
* 采集响应的 `position`。

当前游标达到或超过采集位置，才说明该窗口已被处理。

调度器可能抢先完成工作，所以手动刷新返回 `idle` 也可能正常。此时游标应已经到位。

### Memory 证据链

不要只看搜索是否出现相似文本。查找 `entry.source_refs` 是否包含采集响应的精确 SourceRef。

这才能证明当前 Memory 条目来自这条 Source。

### 抽取无变更

合法情况：

```text theme={"theme":{"light":"github-light","dark":"github-dark"}}
status = processed
processed_source_count > 0
current_cursor advanced
memory = null, or no linked entry
```

模型可能没有产生 Candidate，或者 Candidate 与启用状态内容重复后被去重。Source 已经消费，但 Memory 不一定改变。

## 验证向量，需要更严格的断言

如果 `capabilities.search_modes` 包含 `vector`，可以额外请求：

```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
vector = await client.search_memory(
    SearchMemoryRequest(
        scope_id=scope_id,
        query="generated client contract tests",
        mode="vector",
        limit=50,
    )
)
```

只有同时满足以下条件，才算验证这条证据链的嵌入往返验证：

* 响应 `mode == "vector"`；
* 命中 Citation 指向刚才的关联 Source 的条目；
* `matched_by` 包含 `vector`。

命中另一条历史条目、空命中或 `mode=None` 都不能证明本次 Source 已建立向量投影。

显式 `vector` 请求在能力不可用时返回 HTTP 422，不会自动切成 FTS。

## 跨会话召回靠 Scope，不靠会话

脚本为了可重复测试使用随机 Scope。要验证跨会话：

1. 改成稳定值，例如 `project:powercontext-101`；
2. 第一次运行采集/刷新；
3. 在新的 Python 进程或 Agent 会话中只运行搜索/准备；
4. 两边连接同一个 Server，并使用完全相同的 Scope。

如果第二次查不到，依次检查：

* Server URL；
* 数据库主目录/URL；
* 最终 `scope_id`；
* 条目是否启用状态；
* 查询是否相关。

会话 ID 不应该替代项目 Scope。

## 启用状态、停用状态与历史

抽取产生的条目与显式 Memory 使用同一生命周期：

* 修订创建新条目版本；
* 停用将当前条目设为停用状态；
* 默认列出/搜索/PreparedContext 只使用启用状态内容；
* `include_inactive=True` 和精确读取保留审计历史。

Source 证据链也随条目版本保存。修订时，新的证据会加入证据链，旧版本的精确证据不会消失。

## 失败排查

### 采集成功，刷新返回 503

检查生成能力和模型提供方凭证。Source 已经保存；修复配置后可以重试未消费窗口。

### 刷新是已处理，却没有 linked 条目

这是允许的无变更。先看 Source 内容是否确实包含可复用决策，再检查生成输出和去重。

### 游标一直追不上位置

查看就绪状态、生成诊断信息和调度器/刷新错误。不要把未处理窗口误报成已生成 Memory。

### 搜索能命中，但证据链不对

你命中了同 Scope 中旧的相似 Memory。以 `source_refs` 和精确条目 Citation 为准。

### 第二次采集变成冲突

同一 `source_id` 的内容或元数据字节发生变化。为新事件使用新 ID；真正的重试则复用原输入。

## 怎么知道跑通了

这一页完成后，你应能证明：

* 采集创建 Source，不会直接创建 Memory。
* 同样输入再发一次，位置不变。
* 刷新后，游标到达采集位置。
* 关联这条 Source 的条目，靠精确 `source_refs` 认出来。
* 窗口已经处理完，也可以合法没有留下 Memory。
* 换一个进程再找，靠的是稳定 Scope。
* 停用后的条目退出默认召回，历史还在。

对不上时，先对[排错](https://powercontext.mintlify.site/zh/start/troubleshooting.md)。

## 下一步

> [card] [Handoff 闭环](https://powercontext.mintlify.site/zh/basic-usage/handoff-loop.md)
>
> 准备就会写边界 Source。提交才留下里程碑。

> [card] [卡住时先对排错](https://powercontext.mintlify.site/zh/start/troubleshooting.md)
>
> 采集成功却搜不到 Memory，先打开排错。


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.