Few-shot の例を使って AI 出力を安定させ、自動で検証する

Convert each ticket into a JSON object.

Rules:
- Return one JSON array only.
- Do not add explanations or Markdown.
- Use exactly these keys in every object: ticket_id, priority, owner_team, action.
- priority must be LOW, MEDIUM, or HIGH.
- owner_team must be IT, FACILITIES, or FINANCE.

Example 1 input:
Ticket ID: E001
Text: Printer on floor 2 is out of paper.

Example 1 output:
{"ticket_id":"E001","priority":"LOW","owner_team":"IT","action":"Refill or check floor 2 printer"}

Example 2 input:
Ticket ID: E002
Text: Expense approval file is unavailable before today's reimbursement deadline.

Example 2 output:
{"ticket_id":"E002","priority":"HIGH","owner_team":"FINANCE","action":"Check expense approval file access"}

Now process these tickets:
T101: Payroll spreadsheet cannot be opened before today's payment run.
T102: Meeting room B light is flickering but the room is still usable.
T103: New employee laptop cannot connect to the office Wi-Fi.

[
  {
    "ticket_id": "T101",
    "priority": "HIGH",
    "owner_team": "FINANCE",
    "action": "Check payroll workbook access"
  },
  {
    "ticket_id": "T102",
    "priority": "LOW",
    "owner_team": "FACILITIES",
    "action": "Inspect meeting room B light"
  },
  {
    "ticket_id": "T103",
    "priority": "MEDIUM",
    "owner_team": "IT",
    "action": "Troubleshoot laptop Wi-Fi connection"
  }
]

import csv
import json
from pathlib import Path

SOURCE = Path("ai_output.json")
OUTPUT_DIR = Path("outputs") / "few_shot_check_result"
REPORT = OUTPUT_DIR / "format_check.csv"

EXPECTED = {
    "T101": ("HIGH", "FINANCE", "Check payroll workbook access"),
    "T102": ("LOW", "FACILITIES", "Inspect meeting room B light"),
    "T103": ("MEDIUM", "IT", "Troubleshoot laptop Wi-Fi connection"),
}
REQUIRED_KEYS = {"ticket_id", "priority", "owner_team", "action"}
PRIORITIES = {"LOW", "MEDIUM", "HIGH"}
TEAMS = {"IT", "FACILITIES", "FINANCE"}


def main() -> None:
    if not SOURCE.is_file():
        raise FileNotFoundError(f"AI output not found: {SOURCE}")
    if OUTPUT_DIR.exists():
        raise FileExistsError(f"Output folder already exists: {OUTPUT_DIR}")

    with SOURCE.open("r", encoding="utf-8") as stream:
        data = json.load(stream)

    if not isinstance(data, list):
        raise ValueError("Top-level JSON value must be an array.")
    if len(data) != len(EXPECTED):
        raise ValueError(f"Expected {len(EXPECTED)} objects, got {len(data)}.")

    rows = []
    seen = set()
    for index, item in enumerate(data, start=1):
        errors = []
        if not isinstance(item, dict):
            raise ValueError(f"Item {index} is not a JSON object.")

        if set(item) != REQUIRED_KEYS:
            errors.append("keys")

        for key in REQUIRED_KEYS:
            if key not in item or not isinstance(item.get(key), str) or not item.get(key).strip():
                errors.append(f"invalid_{key}")

        ticket_id = item.get("ticket_id", "")
        if ticket_id in seen:
            errors.append("duplicate_ticket_id")
        seen.add(ticket_id)

        if item.get("priority") not in PRIORITIES:
            errors.append("priority")
        if item.get("owner_team") not in TEAMS:
            errors.append("owner_team")

        expected = EXPECTED.get(ticket_id)
        if expected is None:
            errors.append("unexpected_ticket_id")
        else:
            actual = (
                item.get("priority"),
                item.get("owner_team"),
                item.get("action"),
            )
            if actual != expected:
                errors.append("content_mismatch")

        rows.append({
            "item": index,
            "ticket_id": ticket_id,
            "status": "PASS" if not errors else "FAIL",
            "errors": ";".join(errors),
        })

    missing_ids = set(EXPECTED) - seen
    if missing_ids:
        raise ValueError(f"Missing ticket IDs: {sorted(missing_ids)}")

    OUTPUT_DIR.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_DIR.mkdir()
    with REPORT.open("x", encoding="utf-8", newline="") as stream:
        writer = csv.DictWriter(
            stream,
            fieldnames=["item", "ticket_id", "status", "errors"],
        )
        writer.writeheader()
        writer.writerows(rows)

    failures = sum(row["status"] == "FAIL" for row in rows)
    print(f"Objects checked: {len(rows)}.")
    print(f"Passed: {len(rows) - failures}; failed: {failures}.")
    print(f"Report: {REPORT.as_posix()}")

    if failures:
        raise RuntimeError("AI output failed one or more checks.")


if __name__ == "__main__":
    main()

Objects checked: 3.
Passed: 3; failed: 0.
Report: outputs/few_shot_check_result/format_check.csv