Source 与 Memory 实操

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

上一页直接写下了你已经确认的 Memory。现在走另一条路：先采集 Source，再让生成流程从证据里抽出以后还要用的知识。

两条路都可能得到 Memory。前提、证据链和失败方式不一样。

生成模型和抽取配置以 官方文档 的最新发布版本为准。这页帮你看清两条路差在哪里，以及怎样核对证据链。

​

两条路先放在一起看

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

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

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

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

​

先检查抽取能力

powercontext capabilities

必须看到：

Memory extraction: enabled

如果是 disabled：

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

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：

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())

运行：

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。

​

抽取无变更

合法情况：

status = processed
processed_source_count > 0
current_cursor advanced
memory = null, or no linked entry

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

​

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

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

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。
• 停用后的条目退出默认召回，历史还在。

对不上时，先对排错。

​

下一步

Handoff 闭环

准备就会写边界 Source。提交才留下里程碑。

做一次交接

卡住时先对排错

采集成功却搜不到 Memory，先打开排错。

打开排错