Hasil agen bisa memiliki bentuk JSON yang benar tetapi salah membaca sumber. Panduan ini memberi pengembang Python jalur lengkap dari satu berkas catatan yang diizinkan ke register lokal: hasil produsen bertipe, pemeriksaan sumber dan makna, persetujuan manusia atas hash persis, serta konsumen yang menulis dan membaca ulang laporan Markdown berkelompok. Gunakan catatan sendiri dengan komitmen tindakan eksplisit. Tidak ada catatan nyata atau hasil model yang dipakai dalam persiapan.

Perubahan besar

  • Yang berubah: OpenAI Agents SDK menerima output_type Pydantic, sehingga aplikasi dapat meminta hasil yang terdefinisi lalu memeriksanya sebelum dipakai. Contoh ini menambahkan kutipan sumber dan kontrak berversi pada hasil.
  • Mengapa penting: Pengembang yang menghubungkan agen ke langkah berikutnya dapat menghentikan keluaran salah format, tidak lengkap, atau tanpa dukungan sebelum menjadi daftar tindakan. Peninjau tetap harus menilai apakah catatan yang dikutip benar-benar mendukung setiap tugas, penanggung jawab, dan tanggal, serta apakah model melewatkan sesuatu.
  • Yang perlu diperhatikan: Titik keputusan ada pada batas konsumen. Resep ini hanya membuat register pemilahan lokal setelah persetujuan atas proses, sumber, dan kandidat yang persis. Impor nanti ke sistem tugas memerlukan kewenangan, kebijakan percobaan ulang, dan pembacaan ulang tujuan tersendiri.

Siapkan satu sumber yang diizinkan

Anda memerlukan Python 3.12, akun OpenAI API yang diizinkan untuk catatan, peninjau yang dapat membaca berkas asli, dan model yang telah dipastikan saat ini mendukung keluaran terstruktur JSON-schema dalam akun. Periksa ketentuan data serta harga terkini sebelum menjalankan produsen. Jangan masukkan kredensial ke catatan atau persetujuan. Satu-satunya pemanggilan model di bawah adalah produce; sisanya pemrosesan lokal.

Simpan catatan asli sebagai UTF-8 di notes.txt. Catatan harus memuat komitmen tindakan nyata dan eksplisit dari pekerjaan Anda. Tetapkan source_id yang stabil, misalnya pengenal rekaman sendiri, dan pertahankan setiap byte tanpa perubahan sampai tinjauan dan konsumsi. Buat candidate/ sebelum produksi; berkas kandidat harus baru. Siapkan direktori keluaran terpisah untuk laporan lokal nanti.

Catat hash byte sumber sebelum proses:

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

Dalam lingkungan Python 3.12 terisolasi, buat recipe-requirements.txt dengan baris persis berikut dan pasang:

Text
openai-agents==0.24.0
pydantic==2.14.1
Terminal
python -m pip install -r recipe-requirements.txt
mkdir candidate

Versi ini berhasil dipasang bersama dalam lingkungan Python 3.12.7 lokal pada 20 Oktober 2026. Konfirmasikan kompatibilitas dan ketersediaan model saat menjalankan panduan. Simpan skrip lengkap berikut sebagai action_register_recipe.py di direktori yang sama. Ada dua perintah: produce membuat kandidat dan tanda terima proses; consume membangun laporan hanya setelah validasi dan persetujuan.

Jalankan produsen bertipe

Kontrak Extraction mengharuskan versi action-register/v1, status ready, partial atau blocked, ID sumber dan SHA-256 byte berkas persis, tindakan, serta gap_note. Setiap tindakan membutuhkan teks, kutipan persis dan rentang baris sumber mulai dari 1, penanggung jawab dan tanggal yang boleh null, serta complete atau needs_review sebagai status bukti. Penanggung jawab atau tanggal null berarti tidak dinyatakan dalam catatan. Model diperintahkan tidak menyimpulkannya. Semua kolom wajib dan kolom tambahan ditolak.

Pilih model keluaran terstruktur secara eksplisit. Berikan jalur dan pengenal yang diizinkan:

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

Baca candidate/actions.json di samping candidate/actions.run.json. Berkas kedua mencatat UUID proses, hash sumber dan kandidat, agen terakhir, serta model yang diminta. Kegagalan membuat catatan diagnostik .failure.json dengan akhiran ID proses. Interupsi, None atau keluaran akhir berupa string biasa, atau agen akhir yang tidak diharapkan, tidak pernah dihitung sebagai sukses. Perintah ini tidak menyimpan status SDK yang dapat dilanjutkan, RunState; tanda terima kegagalannya tidak dapat melanjutkan proses. Periksa permintaan, kandidat, dan status lanjutan sebelum memutuskan apakah proses baru layak.

Tinjau makna dan setujui kandidat yang persis

Aplikasi memeriksa tipe ketat dan kolom wajib kandidat, identitas sumber, status siap, setidaknya satu tindakan lengkap, rentang baris valid, keberadaan kutipan dalam rentang, tugas atau penanggung jawab tidak kosong, serta tanggal baku YYYY-MM-DD. Ini membuktikan struktur dan letak kutipan. Tidak membuktikan kalimat benar-benar komitmen, nama atau tanggal bermakna seperti klaim kandidat, atau semua komitmen sudah ditemukan.

Baca catatan asli dan setiap tindakan usulan. Bandingkan makna tugas, penanggung jawab, tanggal, kutipan, dan kelalaian. Selesaikan ambiguitas melalui proses baru yang ditinjau; biarkan kandidat tanpa dukungan, parsial, atau terblokir tidak disetujui. Setelah memastikan berkas tanpa perubahan dan kandidat cocok dengan actions.run.json, tulis approval.json dengan tepat tujuh kunci berikut. Salin ID proses dan hash dari tanda terima yang diperiksa. Gunakan pengenal peninjau sebenarnya dan uraian substansial tentang pemeriksaan yang sungguh dilakukan:

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 harus boolean JSON true, bukan string atau angka. Semua nilai lain harus string; peninjau dan catatan tidak boleh kosong. Kolom tambahan gagal. Persetujuan mengikat satu kandidat dan revisi sumber, bukan penulisan ulang di masa depan.

Konsumsi dan baca ulang laporan

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

Konsumen mengelompokkan tindakan yang diterima ke Assigned with dates, Needs an owner dan Needs a date. Setiap entri memuat tugas, status penanggung jawab/tanggal, kutipan, dan baris sumber. Konsumen menulis action-register-<payload-hash-prefix>.md serta berkas .receipt.json di sebelahnya, dengan hash proses, sumber, kandidat, dan laporan. Buka keduanya, bandingkan tindakan yang dikutip dengan catatan, dan verifikasi report_sha256 terhadap byte berkas Markdown. Tanda terima valid hanya menyatakan artefak lokal telah dibuat dan dibaca ulang; tidak menyetujui penulisan ke sistem tugas eksternal.

Perintah konsumsi mencetak jalur laporan dan tanda terima sebenarnya. Gunakan jalur itu untuk pemeriksaan byte:

Terminal
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

Setelah ada kandidat, periksa jalur penolakan dengan contoh struktural yang jelas sintetis. Salin JSON ke berkas terpisah, hapus satu kolom wajib dari tindakan pertama, lalu panggil consume dengan berkas salah format itu dan direktori keluaran baru yang kosong. Harus gagal sebelum membuat artefak. Jangan anggap contoh ini ekstraksi faktual atau menyetujuinya. Misalnya, pemeriksaan lokal berikut menghapus evidence_quote dari salinan:

Terminal
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

Perintah kedua harus gagal dan direktori keluaran tetap tidak ada. Jangan gunakan contoh jika kandidat nyata tidak memiliki tindakan pertama; kandidat itu sudah tidak lengkap bagi konsumen. Jika penulisan tidak pasti, periksa laporan deterministik dan tanda terima sebelum mengulang. Byte lama yang identik dapat dipakai lagi; byte berbeda menghentikan operasi untuk rekonsiliasi. Sumber atau versi skema berubah memerlukan tinjauan baru. Jangan ulangi panggilan model hanya karena respons konsumen lokal tidak jelas.

Resep Python lengkap

Salin skrip ini persis ke 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()

Batasan dan sumber

Kontrak dan konsumen lokal diperiksa secara offline dengan lima pengujian dan tujuh kasus sintetis produsen tiruan atau percobaan ulang. Pemeriksaan mencakup pembuatan skema ketat, penolakan sebelum artefak dibuat, pengelompokan, hash, dan penanganan kegagalan. Tidak ada pengiriman skema ke penyedia, panggilan model, ekstraksi catatan nyata, persetujuan manusia faktual, laporan dari sumber nyata, atau pengukuran tagihan maupun kualitas. Biaya API bergantung pada akun; periksa kompatibilitas, harga, dan izin data sebelum mengirim sumber.

Untuk konteks terkait, panduan BIG CHANGE memilih tugas AI dari pekerjaan sendiri membahas keputusan sebelumnya tentang pekerjaan yang diuji. Panduan ini berawal dari ekstraksi terpilih yang diizinkan dan membawa satu hasil bertipe ke konsumen lokal yang diperiksa.

Sumber dan bacaan lanjutan