Rezultat agenta može imati ispravan JSON oblik, a pogrešno pročitati izvor. Ovaj vodič daje Python programeru potpun put od jedne dozvoljene datoteke beležaka do lokalnog registra: tipizovan rezultat proizvođača, provere izvora i značenja, ljudsko odobrenje tačnih heševa i potrošač koji upisuje i ponovo čita grupisan Markdown izveštaj. Koristite svoje beleške sa izričitim obavezama. Za pripremu nisu korišćene stvarne beleške ni rezultat modela.

Velika promena

  • Šta se promenilo: OpenAI Agents SDK prihvata Pydantic output_type, pa aplikacija može tražiti definisan rezultat i proveriti ga pre upotrebe. Primer dodaje izvorne citate i verzionisan ugovor oko rezultata.
  • Zašto je važno: Programer koji povezuje agenta sa sledećim korakom može zaustaviti nepravilan, nepotpun ili nepotkrepljen izlaz pre nego što postane spisak radnji. Recenzent i dalje mora proceniti da li citirane beleške stvarno podržavaju svaki zadatak, odgovorno lice i datum i da li je model nešto propustio.
  • Na šta paziti: Odluka se donosi na granici potrošača. Postupak pravi samo lokalni registar za trijažu nakon odobrenja tačnog izvršavanja, izvora i kandidata. Kasniji uvoz u sistem zadataka traži svoja ovlašćenja, pravila ponavljanja i proveru čitanjem na odredištu.

Pripremite jedan odobren izvor

Potrebni su Python 3.12, OpenAI API nalog odobren za beleške, recenzent koji može čitati original i model čiju ste trenutnu podršku za JSON-schema strukturisan izlaz potvrdili na nalogu. Proverite uslove podataka i aktuelne cene pre pokretanja proizvođača. Ne stavljajte pristupne podatke u beleške ili odobrenje. Jedini poziv modelu ispod je produce; ostalo je lokalna obrada.

Sačuvajte originalne beleške kao UTF-8 u notes.txt. Moraju sadržati stvarne izričite obaveze iz vašeg rada. Dodelite stabilan source_id, na primer svoj identifikator zapisa, i ne menjajte nijedan bajt do pregleda i obrade. Napravite candidate/ pre proizvodnje; datoteka kandidata mora biti nova. Obezbedite odvojen izlazni direktorijum za kasniji lokalni izveštaj.

Pre izvršavanja zabeležite heš bajtova izvora:

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

U izolovanom Python 3.12 okruženju napravite recipe-requirements.txt sa ovim tačnim redovima i instalirajte:

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

Ove verzije su zajedno instalirane u lokalnom Python 3.12.7 okruženju 20. oktobra 2026. Potvrdite kompatibilnost i dostupnost modela pri korišćenju. Sačuvajte kompletan skript ispod kao action_register_recipe.py u istom direktorijumu. Ima dve komande: produce pravi kandidata i potvrdu izvršavanja; consume pravi izveštaj tek posle validacije i odobrenja.

Pokrenite tipizovanog proizvođača

Ugovor Extraction zahteva verziju action-register/v1, status ready, partial ili blocked, ID izvora i SHA-256 tačnih bajtova, radnje i gap_note. Svaka radnja zahteva tekst, tačan citat i raspon izvornih redova brojan od 1, odgovorno lice i datum koji mogu biti null, kao i complete ili needs_review kao status dokaza. Null odgovorno lice ili datum znači da nisu navedeni u beleškama. Model dobija nalog da ih ne zaključuje. Sva polja su obavezna i dodatna se odbijaju.

Izričito izaberite model za strukturisan izlaz. Dajte svoju dozvoljenu putanju i identifikator:

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

Pročitajte candidate/actions.json pored candidate/actions.run.json. Druga datoteka beleži UUID izvršavanja, heševe izvora i kandidata, završnog agenta i traženi model. Greška umesto toga pravi dijagnostički zapis .failure.json sa ID-jem izvršavanja u sufiksu. Prekid, None ili završni izlaz kao običan tekst, ili neočekivan završni agent, nikada nisu uspeh. Komanda ne čuva nastavljivo SDK stanje RunState; potvrda greške ne može nastaviti izvršavanje. Pregledajte zahtev, kandidata i nizvodno stanje pre odluke o novom pokretanju.

Pregledajte značenje i odobrite tačnog kandidata

Aplikacija proverava stroge tipove i obavezna polja, identitet izvora, spreman status, bar jednu potpunu radnju, važeće raspone, prisustvo citata u navedenom rasponu, neprazan zadatak ili vlasnika i kanonske datume YYYY-MM-DD. To dokazuje strukturu i mesto citata. Ne utvrđuje da je rečenica stvarna obaveza, da ime ili datum znače ono što kandidat tvrdi ili da su pronađene sve obaveze.

Pročitajte originalne beleške i svaku predloženu radnju. Poredite značenje, odgovorno lice, datum, citat i propuste. Nejasnoću izvora ili kandidata rešite novim pregledanim izvršavanjem; nepotkrepljene, delimične ili blokirane kandidate ostavite bez odobrenja. Kada proverite da nepromenjena datoteka i kandidat odgovaraju actions.run.json, napišite approval.json sa tačno ovih sedam ključeva. Kopirajte ID i heševe iz proverene potvrde. Koristite stvarni identifikator recenzenta i sadržajan opis stvarno obavljenih provera:

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 mora biti JSON logička vrednost true, ne tekst ili broj. Sve druge vrednosti moraju biti stringovi; recenzent i napomena ne smeju biti prazni. Dodatna polja padaju. Odobrenje vezuje jednog kandidata i reviziju izvora, ne buduća prepisivanja.

Obradite i ponovo pročitajte izveštaj

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

Potrošač grupiše prihvaćene radnje u Assigned with dates, Needs an owner i Needs a date. Svaka stavka sadrži zadatak, stanje odgovornog lica i datuma, citat i izvorne redove. Upisuje action-register-<payload-hash-prefix>.md i susednu datoteku .receipt.json sa heševima izvršavanja, izvora, kandidata i izveštaja. Otvorite obe, uporedite citirane radnje sa beleškama i proverite report_sha256 prema bajtovima Markdown datoteke. Važeća potvrda kaže samo da je lokalni rezultat napravljen i ponovo pročitan; ne odobrava upis u spoljni sistem zadataka.

Komanda obrade ispisuje stvarne putanje izveštaja i potvrde. Koristite ih za proveru bajtova:

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

Kada imate kandidata, proverite put odbijanja jasno sintetičkim strukturnim primerom. Kopirajte JSON u odvojenu datoteku, iz prve radnje uklonite bilo koje obavezno polje i pozovite consume sa neispravnom datotekom i novim praznim izlaznim direktorijumom. Mora pasti pre stvaranja rezultata. Primer nikada ne tretirajte kao činjenično izdvajanje niti ga odobravajte. Sledeća lokalna provera, na primer, iz kopije uklanja evidence_quote:

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

Druga komanda treba da padne, a izlazni direktorijum da ostane nepostojeći. Ne koristite primer ako pravi kandidat nema prvu radnju; već je nepotpun za ovog potrošača. Kod neizvesnog upisa pregledajte deterministički izveštaj i potvrdu pre ponavljanja. Isti postojeći bajtovi mogu se ponovo koristiti; različiti zaustavljaju radi usaglašavanja. Promenjen izvor ili verzija šeme traže novi pregled. Ne ponavljajte poziv modelu samo zato što je odgovor lokalnog potrošača bio nejasan.

Ceo Python postupak

Kopirajte ovaj skript tačno u 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()

Ograničenja i izvori

Ugovor i lokalni potrošač provereni su van mreže sa pet testova i sedam sintetičkih slučajeva simuliranog proizvođača ili ponavljanja. Obuhvaćeni su stroga šema, odbijanje pre stvaranja, grupisanje, heševi i greške. Nije bilo slanja šeme pružaocu, poziva modela, izdvajanja stvarnih beležaka, činjeničnog ljudskog odobrenja, izveštaja iz stvarnog izvora ni merenja računa ili kvaliteta. API naknade zavise od naloga; proverite kompatibilnost, cene i dozvolu za podatke pre slanja izvora.

Za srodan kontekst, BIG CHANGE vodič za izbor AI zadatka iz sopstvenog rada pokriva prethodnu odluku šta ispitati. Ovaj vodič počinje izabranim dozvoljenim izdvajanjem i vodi jedan tipizovan rezultat do proverenog lokalnog potrošača.

Izvori i dodatno čitanje