简体中文 | English

matterloop-core

可恢复、可审计的 Loop Engineering 内核。它负责把规划、执行、验证、人工反馈和停止条件串成一个 确定性控制流，不包含模型 SDK、工具、默认策略或数据库驱动。

架构说明 · 企业集成指南

pip install matterloop-core

Python 导入名固定为 matterloop_core。本包没有第三方运行时依赖，也不提供旧的 core 兼容包。

最小可运行装配

Core 不偷偷选择 Planner、审批规则或存储。下面的示例给每个端口放入一个最小实现，并借用 matterloop-memory 的进程内检查点；代码可直接运行，生产环境只需 替换这些端口，不需要改 Loop。

pip install matterloop-core matterloop-memory

import asyncio

from matterloop_core import (
    AgentLoop,
    ApprovalDecision,
    ComponentRegistry,
    ExecutionResult,
    Executor,
    LocalEventPublisher,
    LoopContext,
    LoopRequest,
    LoopStatus,
    Plan,
    Planner,
    PlanStep,
    RetryAction,
    RetryDecision,
    VerificationResult,
    Verifier,
)
from matterloop_memory import InMemoryCheckpointStore


class OneStepPlanner:
    async def plan(self, context: LoopContext) -> Plan:
        return Plan((PlanStep(context.request.goal, executor="echo"),))


class EchoExecutor:
    async def execute(self, step: PlanStep, context: LoopContext) -> ExecutionResult:
        return ExecutionResult(output=f"done: {step.description}")


class PassVerifier:
    async def verify(
        self,
        step: PlanStep,
        result: ExecutionResult,
        context: LoopContext,
    ) -> VerificationResult:
        return VerificationResult(passed=True, evidence=(result.output,))


class AlwaysContinue:
    def can_continue(self, context: LoopContext) -> bool:
        return True


class ApproveSteps:
    async def decide(self, step: PlanStep, context: LoopContext) -> ApprovalDecision:
        return ApprovalDecision.APPROVED


class FailFast:
    def decide(self, error: Exception, attempt: int, context: LoopContext) -> RetryDecision:
        return RetryDecision(RetryAction.FAIL)


async def main() -> None:
    planners = ComponentRegistry[Planner]()
    executors = ComponentRegistry[Executor]()
    verifiers = ComponentRegistry[Verifier]()
    planners.register("default", OneStepPlanner())
    executors.register("echo", EchoExecutor())
    verifiers.register("default", PassVerifier())

    loop = AgentLoop(
        planners=planners,
        executors=executors,
        verifiers=verifiers,
        checkpoint_store=InMemoryCheckpointStore(),
        policy=AlwaysContinue(),
        events=LocalEventPublisher(),
        approval_gate=ApproveSteps(),
        retry_policy=FailFast(),
    )
    result = await loop.run(LoopRequest(goal="生成一份交付说明"))
    assert result.status is LoopStatus.COMPLETED
    print(result.output)


asyncio.run(main())

AgentLoop 只借用这些组件，不负责关闭它们。持有连接池、模型客户端或后台任务的组件应交给应用 组合根或 matterloop-runtime 管理生命周期。

闭环到底做了什么

flowchart LR
    P["Planner"] --> Plan["有序 Plan"]
    Plan --> A{"步骤需要审批?"}
    A -->|通过或无需审批| E["Executor"]
    A -->|延期| H["PAUSED / 等待人"]
    E --> V["Verifier"]
    V -->|未通过| P
    V -->|通过且还有步骤| A
    V -->|全部通过| C["CompletionEvaluator（可选）"]
    C -->|REPLAN| P
    C -->|REQUEST_HUMAN| H
    C -->|ACCEPT| D["COMPLETED"]

Planner 每轮返回完整、有序的 Plan。每个 PlanStep 自己指定 executor，所以同一计划可以把步骤 交给不同执行器。Executor 只产生 ExecutionResult；是否正确由 Verifier 判断。验证失败同样会保存 IterationRecord，然后把 feedback 带入下一轮规划。

所有步骤通过并不必然等于目标完成。需要整体交付验收时注入 CompletionEvaluator；它可以接受、 要求重规划、请求人工复核或停止。未配置时，全部步骤通过即完成。

cycle、attempt 和 step 是三把不同的闸

IterationRecord.attempt 是某一步内部从 1 开始的尝试序号，LoopContext.total_attempts 才是全局 Executor 调用数。completed_steps 表示已经形成执行和验证记录的数量；验证失败也会增加它，不能把 这个字段显示成“成功步骤数”。

timeout_seconds 统计活跃时间，包括组件调用、重试等待、检查点和事件发布。人工暂停期间停止计时； 恢复不会重置已经消耗的时间、cycle 或 attempt。长组件调用由控制器监督，并按 heartbeat_interval_seconds 写入 LOOP_HEARTBEAT；取消和总超时会先回收组件协程，再提交终态。

审批和人工反馈

只有 requires_approval=True 的步骤才调用 ApprovalGate：

• APPROVED：进入执行；
• REJECTED：返回 BLOCKED / APPROVAL_REJECTED，没有待处理人工请求；
• DEFERRED：创建 HumanInteractionRequest，保存当前步骤游标并返回 PAUSED。

人工响应与恢复是两个操作。这样 API 或队列可以安全重试“提交决定”，不会把它和长时间运行绑在 同一个请求中。

from matterloop_core import HumanAction, HumanResponse, ResumeMode

interaction = paused.pending_interaction
assert interaction is not None

await loop.submit_human_response(
    paused.run_id,
    HumanResponse(
        interaction_id=interaction.interaction_id,
        action=HumanAction.APPROVE,
        idempotency_key="approval-ticket-42",
    ),
)
resumed = await loop.resume(paused.run_id, mode=ResumeMode.CONTINUE)

调用方应持久化 idempotency_key 并在网络重试时复用。相同键、相同响应是 no-op；相同键却改变 interaction、动作、正文或 metadata 会抛 HumanResponseConflictError。interaction_id 或 action 不匹配当前请求时，Core 不会猜测用户意图。

整体完成验收也可以返回 REQUEST_HUMAN。此时批准的是当前完成草稿，而不是某个普通步骤；恢复逻辑 仍由检查点保存，不需要在 UI 中重建控制器状态。

精确恢复、checkpoint CAS 与事件顺序

ResumeMode.CONTINUE 使用检查点中的 current_plan 和 current_step_index，不会重放已经形成 record 的步骤。缺少未完成计划时会抛 LoopNotResumableError，不会悄悄改成重新规划。ResumeMode.REPLAN 丢弃当前计划和本轮审批缓存，但保留 records、反馈历史以及已经消耗的 cycle/attempt 计数。Token、 费用和工具额度属于外部策略账本，其跨进程恢复能力由账本实现决定。

resume() 只接受 PAUSED 和 BLOCKED，且 pending_interaction 必须先被响应。recover() 用于进程 崩溃遗留的活跃检查点：VERIFYING 会复用已经保存的 pending_execution，不会再次调用 Executor； EXECUTING 没有已保存结果时进入 BLOCKED / RECOVERY_REQUIRED，由宿主按 active_operation_id 对账， 不会盲目重放。显式 run_id 也是提交幂等键：相同请求重复提交返回已有快照，不同请求抛 LoopRequestConflictError。

CheckpointStore 的关键契约不是“能保存对象”，而是 revision CAS：

new_revision = await store.save(context, expected_revision=context.revision)
assert new_revision > context.revision

两个恢复请求或两个人工响应从同一 revision 写入时，只能有一个成功。另一个得到 CheckpointConflictError 并重新读取；它不能覆盖胜者。LoopCheckpointCodec 提供严格的当前 JSON 结构编解码：

from matterloop_core import LoopCheckpointCodec

codec = LoopCheckpointCodec()
payload = codec.dumps(context)
restored = codec.loads(payload)

当前结构保存计划游标、records、人工历史、批准步骤、执行操作、待验证结果、心跳、活跃计时、event sequence、revision、propagation_context 和 external_state_refs。v2 顶层包含 schema_version 与 context，同时兼容读取旧的无版本 v1；错误字段类型、无时区时间或非 JSON metadata 会抛 CheckpointSchemaError。

一次状态提交的顺序固定为：

1. 更新上下文并分配连续的 event_sequence；
2. 可选 CheckpointPreparer 把本次提交所需的关联信息写入上下文；
3. 用旧 revision CAS 保存检查点；
4. 保存成功后，按 sequence 调用 EventPublisher。

所以事件对应的状态一定已经保存，但检查点成功不代表事件一定送达。Core 没有跨存储和消息系统的 事务型 outbox；生产端应检测 sequence 缺口并做补偿，不能把 LoopEvent.sequence 当作恰好一次投递。 LocalEventPublisher 只按订阅顺序串行调用 handler，handler 失败会停止后续发布并向控制器传播。

扩展协议与注册表

所有端口都是 runtime_checkable Protocol，实现类无需继承 MatterLoop：

Core 会给组件传入隔离的 LoopContext snapshot。修改 snapshot 不会推进真实运行；组件应通过返回值表达 结果。LoopPolicy 和 RetryPolicy 是同步决策点，不应在其中执行阻塞 I/O。

ComponentRegistry 按稳定名称解析 Planner、Executor 和 Verifier，替换只影响下一次查询。它保证名称 映射原子更新，但不启动组件、不等待旧调用排空，也不调用 aclose()。需要资源级无中断替换时使用 matterloop-runtime 的容器或对应组件自己的租约注册表。

ComponentSpec、PluginDefinition 和 FactoryCatalog 用于延迟创建和 Python Entry Point 发现。插件 扫描只有调用 discover() 时发生；批量安装先创建所有实例，再一次性更新注册表，工厂失败不会留下 半套组件。

AgentLoop 构造参数

常用方法是 run()、resume()、recover()、submit_human_response() 和 cancel()。cancel() 会在 安全边界或受监督长调用的下一次轮询生效；要在启动前拿到关联 ID，可先调用 create_run_id()。

失败边界

调用方不能只看“有没有抛异常”。可预期的控制结果通常放在 LoopResult，组件或一致性错误通常抛出：

BLOCKED 不是终态，但不代表“重试一定有用”：耗尽的额度和次数不会因 resume 重置。LoopResult.output 取最后一条 record 的 Executor 输出，不会聚合所有步骤；完整证据在 records。error 可能包含业务 组件的原始异常文本，供应商适配器和持久化层必须先处理敏感信息。

开发与验证

在 workspace 根目录执行：

uv run ruff check matterloop-core
uv run mypy matterloop-core/src/python matterloop-core/tests
uv run pytest matterloop-core/tests

公共 API 由 matterloop_core.__all__ 导出并附带 py.typed。修改 DTO、构造参数或协议时，同时更新 checkpoint schema、恢复测试和文档契约；不要只让 workspace 源码导入通过，还要验证构建后的 wheel。