AI-translated from English; not yet reviewed by a fluent editor.

# 把代理输出转换为经审核的行动登记表

> 用自己的笔记生成带类型的 OpenAI Agents SDK 结果，审核来源证据及含义，再把接受的结果交给本地 Markdown 行动登记表，并回读核对哈希。

By BIG CHANGE Editorial

Published: 2026-10-20T12:52:38.610Z
Updated: 2026-10-20T12:52:38.610Z
Canonical: https://bigchange.ai/blog/agents-sdk-reviewed-action-register-guide

![Conceptual charcoal illustration of one person holding a larger blank ruled source page and a smaller blank candidate page, with an orange reading strip across the source.](https://bigchange.ai/api/media/file/source-candidate-human-review-hero-v2.png)
AI-generated conceptual illustration by BIG CHANGE.

代理结果即使 JSON 结构正确，也可能误读来源。本指南为 Python 开发者提供从一份获准笔记文件到本地行动登记表的完整路径：带类型的生成端结果、来源与语义检查、绑定精确哈希的人工批准，以及写入并回读分组 Markdown 报告的消费端。请使用自己包含明确行动承诺的笔记。编写本指南时没有使用真实笔记或模型结果。

## 大变化

- **发生了什么变化：**OpenAI Agents SDK 接受 Pydantic 的 `output_type`，因此应用可以要求代理返回指定结果，并在使用前再次检查。本示例为结果增加来源引用及带版本的契约。
- **为什么重要：**把代理连接到下一工作步骤的开发者，可以在输出变成行动清单前拦下格式错误、不完整或无依据的内容。审核者仍须判断引用笔记是否真正支持每项任务、负责人及日期，以及模型有无遗漏。
- **需要注意：**决策点在消费端边界。本方案仅在批准精确运行、来源和候选后生成本地分流登记表。日后导入任务系统，还需独立权限、重试策略及目标端回读。

## 准备一份获准来源

你需要 Python 3.12、获准处理这些笔记的 OpenAI API 账户、能阅读原文件的审核者，以及已经确认在账户内支持 JSON Schema 结构化输出的当前模型。运行生成端前先检查账户数据条款和当前价格。不要在笔记或批准文件中放凭据。以下唯一模型调用是 `produce`；其余均为本地处理。

将原始笔记以 UTF-8 保存至 `notes.txt`。内容必须是自己工作中的真实、明确行动承诺。分配稳定的 `source_id`，例如自己的记录标识，并在审核和消费前保持文件每个字节不变。生成前创建 `candidate/`；候选文件必须是新文件。为之后的本地报告准备独立输出目录。

运行前记录来源字节哈希：

```sh
python -c "from pathlib import Path; import hashlib; print(hashlib.sha256(Path('notes.txt').read_bytes()).hexdigest())"
```

在隔离的 Python 3.12 环境中创建 `recipe-requirements.txt`，准确写入以下内容并安装：

```text
openai-agents==0.24.0
pydantic==2.14.1
```

```sh
python -m pip install -r recipe-requirements.txt
mkdir candidate
```

这些版本于 2026 年 10 月 20 日在本地 Python 3.12.7 环境中共同安装成功。实际使用时重新确认兼容性和模型可用性。将以下完整脚本保存为 `action_register_recipe.py`，放在同一目录。它有两个命令：`produce`生成候选及运行凭据；`consume`仅在验证和批准后构建报告。

## 运行带类型的生成端

这个 `Extraction` 契约要求版本 `action-register/v1`，状态为 `ready`、`partial`或`blocked`，以及来源 ID、精确文件字节的 SHA-256、行动和 `gap_note`。每项行动需要文本、精确引用及从 1 开始的来源行范围、允许为 null 的负责人和日期，以及 `complete`或`needs_review`证据状态。负责人或日期为 null 表示笔记未说明，模型被要求不得推断。所有字段均必填，额外字段会被拒绝。

明确选择支持结构化输出的模型。提供自己获准的路径及标识：

```sh
python action_register_recipe.py produce --source-id OWNED-ID --source notes.txt --candidate candidate/actions.json --model YOUR-STRUCTURED-OUTPUT-MODEL
```

阅读 `candidate/actions.json`及旁边的 `candidate/actions.run.json`。后者记录 UUID 运行 ID、来源及候选哈希、最终代理和请求模型。失败时则创建带运行 ID 后缀的 `.failure.json`诊断记录。中断、`None`或纯字符串最终输出，或由意外代理结束，都不算成功。此命令不保存 SDK 可恢复的 `RunState`；失败凭据无法恢复运行。决定是否重新运行前，先检查请求、候选及任何下游状态。

## 审核含义并批准精确候选

应用检查候选的严格类型和必填字段、来源身份、就绪状态、至少一项完整行动、有效行范围、引用出现在指定行内、非空任务或负责人，以及规范的 `YYYY-MM-DD`日期。这只能证明结构和引用位置，不能确定句子确实是承诺、姓名或日期的含义符合候选，或所有承诺均已找到。

阅读原笔记和每项拟议行动，对比任务含义、负责人、日期、引用及遗漏。通过新一次经审核的运行解决来源或候选歧义；无依据、部分完成或阻塞的候选不要批准。确认未变的文件和候选与 `actions.run.json`一致后，编写 `approval.json`，精确包含以下七个键。复制已检查凭据中的运行 ID 及哈希。使用审核者真实标识，并充分说明实际做过的检查：

```json
{
  "approved": true,
  "run_id": "COPY-RUN-ID",
  "source_id": "OWNED-ID",
  "source_sha256": "COPY-SOURCE-SHA256",
  "payload_sha256": "COPY-PAYLOAD-SHA256",
  "reviewer": "YOUR-REVIEWER-ID",
  "review_note": "Describe the source, task, owner, date, quotation and omission checks actually performed."
}
```

`approved`必须是 JSON 布尔值 `true`，不能是字符串或数字。其余值必须是字符串，审核者和说明不可为空。额外字段会失败。批准绑定的是一个候选及来源修订，不是未来任何改写。

## 消费并回读报告

```sh
python action_register_recipe.py consume --source-id OWNED-ID --source notes.txt --candidate candidate/actions.json --approval approval.json --out private-output
```

消费端把接受的行动分为 **Assigned with dates**、**Needs an owner**和**Needs a date**。每条记录包含任务、负责人及日期状态、引用和来源行。它写入 `action-register-<payload-hash-prefix>.md`及旁边的 `.receipt.json`，记录运行、来源、候选和报告哈希。打开两份文件，把报告引用的行动与笔记对比，并将 `report_sha256`与 Markdown 文件字节核对。有效凭据只说明本地文件已创建并回读，不批准写入外部任务系统。

消费命令会打印真实报告及凭据路径。使用这些路径进行字节检查：

```sh
python -c "from pathlib import Path; import hashlib,json,sys; report=Path(sys.argv[1]); receipt=json.loads(Path(sys.argv[2]).read_text(encoding='utf-8')); assert hashlib.sha256(report.read_bytes()).hexdigest()==receipt['report_sha256']; print('report hash matches receipt')" PATH-PRINTED-FOR-REPORT PATH-PRINTED-FOR-RECEIPT
```

有候选后，用明确标为合成的结构样例检查拒绝路径。将其 JSON 复制到另一文件，删除首项行动的任意必填字段，再调用 `consume`，使用格式错误的文件及新的空输出目录。必须在创建文件前失败。不要把样例当成事实提取，也不要批准它。例如，以下本地检查从副本中删除 `evidence_quote`：

```sh
python -c "import json; from pathlib import Path; p=json.loads(Path('candidate/actions.json').read_text(encoding='utf-8')); p['actions'][0].pop('evidence_quote'); Path('candidate/SYNTHETIC-malformed.json').write_text(json.dumps(p), encoding='utf-8')"
python action_register_recipe.py consume --source-id OWNED-ID --source notes.txt --candidate candidate/SYNTHETIC-malformed.json --approval approval.json --out synthetic-rejection-output
```

第二个命令应失败，输出目录仍不存在。真实候选若没有第一项行动，不要使用样例；它对该消费端已属不完整。消费写入结果不确定时，先检查确定性的报告和凭据再重试。相同的既有字节可以重用；不同字节会停止操作，等待核对。来源或模式版本改变需重新审核。不能仅因本地消费端答复不清就再次调用模型。

## 完整 Python 方案

将以下脚本原样复制到 `action_register_recipe.py`：

```python
"""Private, bounded recipe. No model call occurs unless `produce` is explicitly run."""

import argparse
import hashlib
import json
import uuid
from datetime import date
from pathlib import Path
from typing import Literal

from pydantic import BaseModel, ConfigDict, StrictInt, StrictStr, ValidationError


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)


class Action(StrictModel):
    action_text: StrictStr
    evidence_quote: StrictStr
    line_start: StrictInt
    line_end: StrictInt
    owner: StrictStr | None
    due_date: StrictStr | None
    evidence_status: Literal["complete", "needs_review"]


class Extraction(StrictModel):
    schema_version: Literal["action-register/v1"]
    status: Literal["ready", "partial", "blocked"]
    source_id: StrictStr
    source_sha256: StrictStr
    actions: list[Action]
    gap_note: StrictStr | None


def digest(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def canonical(payload: Extraction) -> bytes:
    return json.dumps(payload.model_dump(mode="json"), sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode("utf-8")


def check(payload: Extraction, source_id: str, source: bytes) -> None:
    if payload.source_id != source_id or payload.source_sha256 != digest(source):
        raise ValueError("source identity or revision mismatch")
    lines = source.decode("utf-8").splitlines()
    if payload.status == "ready" and (not payload.actions or payload.gap_note is not None):
        raise ValueError("ready requires actions and no gap")
    if payload.status != "ready":
        raise ValueError("partial or blocked result cannot reach consumer")
    for index, item in enumerate(payload.actions, 1):
        if item.evidence_status != "complete":
            raise ValueError(f"action {index}: evidence needs review")
        if not item.action_text.strip() or not item.evidence_quote.strip():
            raise ValueError(f"action {index}: empty task or evidence")
        if not 1 <= item.line_start <= item.line_end <= len(lines):
            raise ValueError(f"action {index}: invalid source line range")
        segment = "\n".join(lines[item.line_start - 1:item.line_end])
        if item.evidence_quote not in segment:
            raise ValueError(f"action {index}: quotation absent from source span")
        if item.owner is not None and not item.owner.strip():
            raise ValueError(f"action {index}: blank owner")
        if item.due_date is not None:
            try:
                if date.fromisoformat(item.due_date).isoformat() != item.due_date:
                    raise ValueError()
            except ValueError as exc:
                raise ValueError(f"action {index}: due date must be YYYY-MM-DD") from exc


def report(payload: Extraction) -> str:
    """Application consumer: a triage register grouped by missing ownership/date."""
    groups = [
        ("Assigned with dates", lambda x: x.owner is not None and x.due_date is not None),
        ("Needs an owner", lambda x: x.owner is None),
        ("Needs a date", lambda x: x.owner is not None and x.due_date is None),
    ]
    rows = ["# Action register for " + payload.source_id, "", "Review artifact; source meaning was approved separately.", ""]
    for heading, predicate in groups:
        rows += ["## " + heading, ""]
        matches = [(n, a) for n, a in enumerate(payload.actions, 1) if predicate(a)]
        for number, action in matches:
            rows += [f"{number}. {action.action_text}", f"   Owner: {action.owner or 'Unspecified'}; due: {action.due_date or 'Unspecified'}", f"   Source lines {action.line_start}-{action.line_end}: {action.evidence_quote}", ""]
        if not matches:
            rows += ["None.", ""]
    return "\n".join(rows)


def load_payload(path: Path) -> Extraction:
    return Extraction.model_validate_json(path.read_bytes(), strict=True)


def consume(source_id: str, source_path: Path, payload_path: Path, approval_path: Path, out_dir: Path) -> tuple[Path, Path]:
    source = source_path.read_bytes()
    payload = load_payload(payload_path)
    check(payload, source_id, source)
    run = json.loads(payload_path.with_suffix(".run.json").read_text(encoding="utf-8"))
    payload_hash = digest(canonical(payload))
    if run.get("source_id") != source_id or run.get("source_sha256") != digest(source) or run.get("payload_sha256") != payload_hash or run.get("last_agent") != "Owned-source action extractor" or not str(run.get("run_id", "")).strip():
        raise ValueError("producer receipt does not match accepted candidate")
    approval = json.loads(approval_path.read_text(encoding="utf-8"))
    if (type(approval) is not dict
            or set(approval) != {"approved", "run_id", "source_id", "source_sha256", "payload_sha256", "reviewer", "review_note"}
            or approval["approved"] is not True
            or any(type(approval[key]) is not str for key in ("run_id", "source_id", "source_sha256", "payload_sha256", "reviewer", "review_note"))
            or not approval["reviewer"].strip() or not approval["review_note"].strip()
            or approval["run_id"] != run["run_id"]
            or approval["source_id"] != source_id
            or approval["source_sha256"] != digest(source)
            or approval["payload_sha256"] != payload_hash):
        raise ValueError("human source/meaning approval must bind exact source and payload hashes")
    body = report(payload).encode("utf-8")
    stem = "action-register-" + payload_hash[:16]
    out_dir.mkdir(parents=True, exist_ok=True)
    target = out_dir / (stem + ".md")
    receipt = out_dir / (stem + ".receipt.json")
    record = {"run_id": run["run_id"], "source_id": source_id, "source_sha256": digest(source), "payload_sha256": payload_hash, "report_sha256": digest(body), "reviewer": approval["reviewer"], "operation": "local action-triage register"}
    encoded = (json.dumps(record, indent=2) + "\n").encode("utf-8")
    for path, data in [(target, body), (receipt, encoded)]:
        if path.exists():
            if path.read_bytes() != data:
                raise ValueError(f"uncertain prior output differs: {path}; inspect before retry")
        else:
            with path.open("xb") as handle:
                handle.write(data)
        if path.read_bytes() != data:
            raise IOError(f"readback failed: {path}")
    return target, receipt


def produce(source_id: str, source_path: Path, candidate_path: Path, model: str) -> None:
    from agents import Agent, Runner

    source = source_path.read_bytes()
    source_text = source.decode("utf-8")
    if not source_text.strip() or candidate_path.exists() or not candidate_path.parent.is_dir():
        raise ValueError("source must contain text; candidate must be new and its parent directory must already exist")
    run_id = str(uuid.uuid4())
    agent = Agent(
        name="Owned-source action extractor",
        model=model,
        instructions=("Extract only explicit action commitments from the supplied numbered source. "
                      "Return action-register/v1. Copy an exact quote with matching 1-based line span. "
                      "Use null for an unstated owner or due date; never infer them. "
                      "Use partial and needs_review for ambiguous evidence; blocked for unusable input. "
                      "Do not use tools or hand off."),
        output_type=Extraction,
    )
    numbered = "\n".join(f"{n}: {line}" for n, line in enumerate(source_text.splitlines(), 1))
    try:
        result = Runner.run_sync(agent, f"Source ID: {source_id}\nSHA-256: {digest(source)}\nNumbered source:\n{numbered}")
        if result.interruptions or result.last_agent is not agent or not isinstance(result.final_output, Extraction):
            raise ValueError("run interrupted, unfinished, or finished by unexpected agent; this CLI does not persist resumable RunState, so do not resume from its failure receipt")
        output = result.final_output
        if output.source_id != source_id or output.source_sha256 != digest(source):
            raise ValueError("producer returned wrong source identity")
        candidate = canonical(output)
        with candidate_path.open("xb") as handle:
            handle.write(candidate)  # Candidate only; consumer still requires review.
        receipt = {"run_id": run_id, "source_id": source_id, "source_sha256": digest(source), "payload_sha256": digest(candidate), "last_agent": result.last_agent.name, "model": model}
        with candidate_path.with_suffix(".run.json").open("x", encoding="utf-8") as handle:
            handle.write(json.dumps(receipt, indent=2) + "\n")
    except Exception as exc:
        failure = {"run_id": run_id, "source_id": source_id, "source_sha256": digest(source), "error_type": type(exc).__name__, "error": str(exc), "state": "unknown_or_blocked; inspect run and destination before retry"}
        with candidate_path.with_name(candidate_path.stem + "." + run_id + ".failure.json").open("x", encoding="utf-8") as handle:
            handle.write(json.dumps(failure, indent=2) + "\n")
        raise


def main() -> None:
    parser = argparse.ArgumentParser()
    sub = parser.add_subparsers(dest="command", required=True)
    for name in ("produce", "consume"):
        p = sub.add_parser(name)
        p.add_argument("--source-id", required=True)
        p.add_argument("--source", type=Path, required=True)
        if name == "produce":
            p.add_argument("--candidate", type=Path, required=True)
            p.add_argument("--model", required=True)
        else:
            p.add_argument("--candidate", type=Path, required=True)
            p.add_argument("--approval", type=Path, required=True)
            p.add_argument("--out", type=Path, required=True)
    args = parser.parse_args()
    if args.command == "produce":
        produce(args.source_id, args.source, args.candidate, args.model)
    else:
        print(*consume(args.source_id, args.source, args.candidate, args.approval, args.out), sep="\n")


if __name__ == "__main__":
    main()
```

## 限制及来源

脚本契约和本地消费端经过五项离线测试及七个模拟生成端或重试的合成案例检查。检查覆盖严格模式生成、创建前拒绝、分组输出、哈希和失败处理。它们没有向供应商提交模式、调用模型、提取真实笔记、获得事实性人工批准、生成真实来源报告或测量账单与质量。API 费用取决于账户；发送来源前检查当前模型兼容性、价格及数据许可。

相关背景可参见 BIG CHANGE 的[从自己工作中选择 AI 任务的指南](https://bigchange.ai/blog/map-work-choose-first-ai-task-guide)，它讨论更早的选题决策。本指南从已选择且获准的提取任务开始，把一个带类型的结果送入已检查的本地消费端。

### 来源与延伸阅读

- [OpenAI Agents SDK: Agents](https://openai.github.io/openai-agents-python/agents/)说明 Pydantic 的 `output_type`，但不提供本方案的人工批准或本地消费端。
- [OpenAI Agents SDK: Results](https://openai.github.io/openai-agents-python/results/)说明 `final_output`、`last_agent`、中断及 `to_state()`。此命令行程序不保存可恢复状态。
- [OpenAI Agents SDK: Quickstart](https://openai.github.io/openai-agents-python/quickstart/)介绍安装及 API 密钥配置。密钥应放在来源和凭据之外。
- [OpenAI Structured Outputs](https://developers.openai.com/api/docs/guides/structured-outputs)解释严格模式规则及模型支持。请确认所选模型的支持情况。
- [Pydantic ConfigDict](https://docs.pydantic.dev/latest/api/config/)说明严格验证和禁止额外字段。
- [OpenAI Agents SDK 0.24.0](https://pypi.org/project/openai-agents/0.24.0/)与[Pydantic 2.14.1](https://pypi.org/project/pydantic/2.14.1/)标明本地安装的固定版本。未测试供应商端接受情况。

## Sources

- [OpenAI Agents SDK: Agents](https://openai.github.io/openai-agents-python/agents/) — Pydantic output_type 官方手册。不实现本文的人工批准或本地消费端。
- [OpenAI Agents SDK: Results](https://openai.github.io/openai-agents-python/results/) — final_output、last_agent、中断和 to_state 的官方行为。本方案不持久化 RunState，也不提供恢复。
- [OpenAI Agents SDK: Quickstart](https://openai.github.io/openai-agents-python/quickstart/) — 官方安装及 API 密钥配置背景；准备时未使用密钥或真实供应商调用。
- [OpenAI Agents SDK：输出模式参考](https://openai.github.io/openai-agents-python/ref/agent_output/) — 官方严格输出模式生成参考；本地检查不证明供应商接受。
- [OpenAI Structured Outputs](https://developers.openai.com/api/docs/guides/structured-outputs) — 严格必填字段、additionalProperties 规则及模型兼容性；读者使用时必须核查所选模型。
- [Pydantic ConfigDict](https://docs.pydantic.dev/latest/api/config/) — 说明 strict 和 extra-forbid 配置；语义准确性仍需人工审核来源。
- [openai-agents 0.24.0 版本](https://pypi.org/project/openai-agents/0.24.0/) — 在隔离 Python 3.12.7 离线检查中安装的精确 SDK 固定版本；无模型调用或账单。
- [Pydantic 2.14.1 版本](https://pypi.org/project/pydantic/2.14.1/) — 与 SDK 0.24.0 一起在本地安装的精确验证库版本。
- [BIG CHANGE：梳理自己的工作，选择首个待测试的 AI 任务](https://bigchange.ai/blog/map-work-choose-first-ai-task-guide) — 本集成方案之前用于选择任务的相关指南；它不证明本方案正确。