はじめに

AIに記事や商品説明を書かせていると、必ずぶつかる壁があります。

「同じプロンプトを投げているのに、出てくる品質が毎回バラバラ」

ある日は見出しが揃っていて読みやすいのに、翌日は箇条書きが崩れ、文字数も半分で、事実関係まで怪しい。結局すべて目視で直すことになり、「AIで自動化したはずが、確認作業で時間が溶ける」という状態になります。

原因はモデルの性能ではありません。出力の型を決めず、検証もせず、失敗したときのやり直しも自動化していないことです。

この記事では、Python + OpenAI API + Pydantic を使って「生成 → 検証 → 自動再生成」が回る最小構成のパイプラインを、コピペで動くコード付きで実装します。特別なフレームワークは使いません。標準的なライブラリだけで、100行未満に収めます。

対象読者は、AIでのコンテンツ量産を「一度きりの生成」から「継続的に回る仕組み」に引き上げたい人です。

第1章：AIコンテンツ生成における「品質のバラつき」をどう防ぐか

バラつきが起きる3つの原因

出力フォーマットの非決定性
自然言語での指示は、モデルにとって「お願い」でしかありません。「見出しを3つ、各300字で」と頼んでも、確率的に崩れる回数がゼロにはなりません。
前提の暗黙補完
書いていないことは、モデルが勝手に埋めます。文字数、トーン、禁止表現、事実の扱い。指示の外側は「モデルの常識」で塗りつぶされます。
検証工程の不在
生成物をチェックする仕組みがなければ、品質の乱れは人の目に触れるまで検知できません。そして人の目は、量産するほどボトルネックになります。

対策の方針：型・検証・再試行の3層で固める

レイヤ	役割	実装
出力形式	JSON Schemaで構造を固定する	Pydantic + Structured Outputs
内容の制約	機械的に判定できるルールで検査する	独自バリデータ
失敗時の処理	失敗を検知して自動でやり直す	指数バックオフ + フィードバック再生成

この3層を分けて実装すると、「どこで品質が落ちたか」がログから特定できるようになります。これが自動化を運用に乗せる上での分岐点です。

第2章：Pydanticを用いた出力スキーマの強制バリデーション

スキーマをコードで定義する

まず、生成してほしいコンテンツの形をPydanticモデルで定義します。ポイントは Field(description=...) に、その項目で守ってほしい制約を日本語で書き込むことです。JSON Schemaとしてモデルに渡るため、プロンプトの指示より強く効きます。

from typing import Literal
from pydantic import BaseModel, Field

class Section(BaseModel):
    heading: str = Field(description="節の見出し。30文字以内。体言止めにしない")
    body: str = Field(description="本文。200〜400文字。専門用語には必ず短い補足を付ける")

class ArticleDraft(BaseModel):
    title: str = Field(description="記事タイトル。32文字以内。煽り表現は使わない")
    summary: str = Field(description="120文字以内の要約")
    sections: list[Section] = Field(description="本文の節。3〜5個")
    tags: list[str] = Field(description="記事タグ。3〜5個")
    tone: Literal["formal", "casual"] = Field(description="文体。指定がなければ formal")

Literal を使うと、自由記述だった項目を選択式にできます。表記ゆれが構造的に消えるので、後段の分岐処理が安定します。

Structured Outputsで型を強制する

OpenAI Python SDK（1.40以降）には、Pydanticモデルをそのまま response_format に渡せるヘルパーがあります。JSONのパース失敗やキーの欠落を、API側で抑え込めます。

from openai import OpenAI

client = OpenAI()  # 環境変数 OPENAI_API_KEY を読む

def generate_draft(topic: str, model: str = "gpt-4o-mini") -> ArticleDraft:
    completion = client.beta.chat.completions.parse(
        model=model,
        messages=[
            {"role": "system", "content": "あなたは編集者です。与えられたテーマで、根拠のない断定を避けた記事構成を作成します。"},
            {"role": "user", "content": f"テーマ: {topic}"},
        ],
        response_format=ArticleDraft,
    )
    parsed = completion.choices[0].message.parsed
    if parsed is None:  # 安全側に倒した拒否が返るケース
        raise ValueError("モデルが構造化出力を返しませんでした")
    return parsed

parse は内部でスキーマ適合を検証し、適合しない場合は例外を送出します。自前で json.loads して欠損キーを埋める処理を書く必要がなくなるのが大きな利点です。

スキーマだけでは足りない部分

型が合っていても、中身が要件を満たすとは限りません。文字数、禁止語、見出しの重複、URLの有無といった「業務ルール」はコードで検証します。

import re

NG_WORDS = ["絶対に稼げる", "誰でも簡単に", "必ず成功"]

def validate_rules(draft: ArticleDraft) -> list[str]:
    errors: list[str] = []
    if len(draft.title) > 32:
        errors.append(f"タイトルが32文字を超えています ({len(draft.title)}文字)")
    if not 3 <= len(draft.sections) <= 5:
        errors.append(f"節の数が範囲外です ({len(draft.sections)}個)")
    for i, s in enumerate(draft.sections):
        if not 200 <= len(s.body) <= 400:
            errors.append(f"第{i + 1}節の本文が200〜400文字の範囲外です ({len(s.body)}文字)")
    joined = draft.title + draft.summary + "".join(s.body for s in draft.sections)
    for w in NG_WORDS:
        if w in joined:
            errors.append(f"禁止表現が含まれています: {w}")
    if re.search(r"https?://", joined) and "参考" not in joined:
        errors.append("URLを含む場合は出典表記が必要です")
    return errors

エラーは「文字列のリスト」で返すのがコツです。このまま次のプロンプトに渡して、修正点として使えるからです。

第3章：【コピペで動く】自動生成＆品質チェックパイプラインの最小実装コード

準備

pip install "openai>=1.40.0" "pydantic>=2.5" python-dotenv

.env にAPIキーを置きます（リポジトリにはコミットしないでください）。

OPENAI_API_KEY=sk-...

パイプライン本体

生成・検証・再生成を1ファイルにまとめた実装です。上記のモデル定義とあわせて pipeline.py に保存すれば、そのまま動きます。

import json
import logging
import os
import random
import time
from datetime import datetime, timezone

import openai
from dotenv import load_dotenv

load_dotenv()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("pipeline")

MAX_ATTEMPTS = 3

def build_messages(topic: str, errors: list[str], previous: ArticleDraft | None):
    system = (
        "あなたは編集者です。根拠のない断定や誇大表現を避け、"
        "読者が明日から実行できる具体性を優先して記事構成を作成します。"
    )
    if previous is None:
        return [
            {"role": "system", "content": system},
            {"role": "user", "content": f"テーマ: {topic}"},
        ]
    feedback = "前回の出力には次の問題がありました。すべて修正してください。\n" + "\n".join(f"- {e}" for e in errors)
    return [
        {"role": "system", "content": system},
        {"role": "user", "content": f"テーマ: {topic}"},
        {"role": "assistant", "content": previous.model_dump_json()},
        {"role": "user", "content": feedback},
    ]

def generate_with_retry(topic: str, model: str = "gpt-4o-mini") -> ArticleDraft:
    previous: ArticleDraft | None = None
    errors: list[str] = []
    for attempt in range(1, MAX_ATTEMPTS + 1):
        try:
            completion = client.beta.chat.completions.parse(
                model=model,
                messages=build_messages(topic, errors, previous),
                response_format=ArticleDraft,
                temperature=0.7,
            )
            draft = completion.choices[0].message.parsed
            if draft is None:
                raise ValueError("parsed is None")
        except (openai.RateLimitError, openai.APIConnectionError) as e:
            wait = min(2 ** attempt, 30) + random.uniform(0, 1)
            logger.warning("一時エラー(%s)。%.1f秒待って再試行します", type(e).__name__, wait)
            time.sleep(wait)
            continue
        except openai.APIError as e:
            logger.error("APIエラー: %s", e)
            raise

        errors = validate_rules(draft)
        if not errors:
            logger.info("検証OK (試行%d回目)", attempt)
            return draft

        logger.warning("検証NG (%d件): %s", len(errors), "; ".join(errors))
        previous = draft

    raise RuntimeError(f"{MAX_ATTEMPTS}回試行しても品質基準を満たしませんでした")

def to_markdown(draft: ArticleDraft) -> str:
    lines = [f"# {draft.title}", "", draft.summary, ""]
    for s in draft.sections:
        lines += [f"## {s.heading}", "", s.body, ""]
    lines.append("タグ: " + ", ".join(draft.tags))
    return "\n".join(lines)

if __name__ == "__main__":
    topic = os.environ.get("TOPIC", "AIで記事制作を効率化する具体的な手順")
    result = generate_with_retry(topic)
    out = to_markdown(result)
    path = f"output_{datetime.now(timezone.utc):%Y%m%d_%H%M%S}.md"
    with open(path, "w", encoding="utf-8") as f:
        f.write(out)
    logger.info("書き出し完了: %s (%d文字)", path, len(out))

出力例

2026-11-09 10:20:03 INFO 検証NG (1件): 第2節の本文が200〜400文字の範囲外です (183文字)
2026-11-09 10:20:11 INFO 検証OK (試行2回目)
2026-11-09 10:20:11 INFO 書き出し完了: output_20261109_102011.md (1842文字)

1回目で落ちても、2回目で基準を満たして自動的に抜けています。人が触ったのはテーマの指定だけです。

第4章：リトライループと例外処理のベストプラクティス

失敗を3種類に分けて扱う

同じ「失敗」でも、対処がまったく違います。

スキーマ不一致・内容NG → プロンプトを改善して再生成する（自動で回せる）
レート制限・通信エラー → 待って再試行する（指数バックオフ + ジッター）
認証エラー・残高不足・モデル指定ミス → 再試行しても無駄。即座に停止して人に渡す

上のコードでは1と2をループ内で処理し、3は openai.APIError 系のうち再試行不能なものを raise で外に出しています。ここを混ぜると「無駄なリトライで課金だけが増える」事故になります。

待ち時間は指数的に伸ばす

レート制限時のリトライ間隔は 2^試行回数 + ランダム 秒が基本です。固定間隔だと、同じタイミングで再試行が集中して再び弾かれます。ジッター（0〜1秒の乱数）はその衝突を避けるためのものです。

検証エラーは「次の指示」に変換する

リトライ時に同じプロンプトを投げ直すのは、ほぼ無意味です。モデルは同じ確率分布から引き直すだけだからです。前回の出力と「何がダメだったか」を渡すことで、修正すべき点が具体化し、2回目以降の合格率が大きく上がります。

上限を必ず設ける

MAX_ATTEMPTS = 3 のように上限を設け、超えたら例外で止めます。無限リトライは、品質が上がらないままコストだけが積み上がる最悪のパターンです。

ログはJSON Linesで残す

def log_event(path: str, **kwargs) -> None:
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps({"ts": datetime.now(timezone.utc).isoformat(), **kwargs}, ensure_ascii=False) + "\n")

試行回数・検証エラー・使用モデル・トークン数を残しておくと、「どのルールで何回詰まったか」が後から集計できます。プロンプト改善は、この集計を見てから行うのが最短です。

おわりに

この記事の要点は3つです。

出力はJSON Schemaで型を固定し、Pydanticで受ける
業務ルールはコードで検証し、エラーを文字列で返す
失敗は分類して、待つ・直す・止めるを使い分ける

この3層があるだけで、AIコンテンツ制作は「毎回人が確認する作業」から「基準を満たすまで自動で回る仕組み」に変わります。まずは1テーマで流してみて、どこで詰まるかをログで見るのがおすすめです。

実運用に組み込む際の考え方は、こちらの記事により詳しくまとめています。