本記事の目標

本シリーズの第5回では、校正エージェント全体の設計について解説します。これまでの記事で紹介した要素がどのように組み合わさっているか、プロジェクト構造やデータフローを含めて説明します。

プロジェクト構造

proofread-agent/
├── packages/agent/
│   ├── src/proofread_agent/
│   │   ├── __init__.py
│   │   ├── graph.py           # メイングラフ定義
│   │   ├── studio.py          # LangGraph Studio用
│   │   ├── config/
│   │   │   ├── constants.py   # 定数
│   │   │   └── prompts.py     # プロンプト読み込み
│   │   ├── nodes/
│   │   │   ├── aggregator.py  # Issue集約
│   │   │   ├── final_reporter.py  # レポート生成
│   │   │   └── parser.py      # 記事パース
│   │   ├── schemas/
│   │   │   ├── article.py     # 記事関連の型
│   │   │   ├── issue.py       # Issue型
│   │   │   └── state.py       # 状態定義
│   │   ├── subgraphs/
│   │   │   ├── base.py        # BaseChecker
│   │   │   ├── code_quality.py
│   │   │   ├── language_quality.py
│   │   │   ├── link.py
│   │   │   ├── markdown.py
│   │   │   ├── privacy.py
│   │   │   ├── structure.py
│   │   │   └── technical_accuracy.py
│   │   ├── tools/
│   │   │   ├── code_search.py
│   │   │   ├── technical_search.py
│   │   │   └── utils.py
│   │   └── utils/
│   │       ├── file_loader.py
│   │       └── logger.py
│   ├── prompts/               # プロンプトファイル群
│   ├── langgraph.json
│   ├── pyproject.toml
│   └── .env
└── docs/
    └── ARCHITECTURE.md

状態の設計

ProofreadState（メイン状態）

class ProofreadState(TypedDict):
    # 入力
    target_article: str
    article_type: Annotated[ArticleType, keep_value]

    # パース結果
    parsed_article: NotRequired[Annotated[ParsedArticle, keep_value]]

    # チェック結果
    issues: NotRequired[Annotated[list[Issue], add_values]]

    # 集約結果
    critical_count: NotRequired[int]
    warning_count: NotRequired[int]
    suggestion_count: NotRequired[int]

    # 出力
    overall_score: NotRequired[int]
    is_approved: NotRequired[bool]
    final_report: NotRequired[str]
    messages: NotRequired[Annotated[list[BaseMessage], add_messages]]

    # エラー
    errors: NotRequired[Annotated[list[ErrorInfo], add_values]]

SubgraphState（サブグラフ用状態）

サブグラフ（各チェッカー）は、メイン状態の一部のみを使用します。

class SubgraphState(TypedDict):
    # 入力（メイン状態から継承）
    article_type: Annotated[ArticleType, keep_value]
    parsed_article: Annotated[ParsedArticle, keep_value]

    # 内部
    messages: NotRequired[Annotated[list[BaseMessage], add_messages]]

    # 出力（メイン状態に戻る）
    issues: NotRequired[Annotated[list[Issue], add_values]]
    errors: NotRequired[Annotated[list[ErrorInfo], add_values]]

Reducer関数

状態の更新方法を制御するReducer関数を定義しています。

def keep_value(left: Any, right: Any) -> Any:
    """最初に設定された値を保持"""
    if left is not None:
        return left
    return right


def add_values(left: list | None, right: list | None) -> list:
    """リストを結合"""
    if left is None:
        left = []
    if right is None:
        right = []
    return left + right

keep_value: article_typeやparsed_articleなど、変更されない値に使用
add_values: issuesやerrorsなど、複数ノードから追加される値に使用
add_messages: LangChain提供。メッセージの重複排除と追加を処理

スキーマ設計

Issue

Category = Literal[
    "markdown_format",
    "technical_accuracy",
    "code_quality",
    "privacy",
    "structure",
    "language_quality",
    "link",
]

Severity = Literal["critical", "warning", "suggestion"]


class Issue(BaseModel):
    category: Category
    severity: Severity
    line_start: int | None = None
    line_end: int | None = None
    location_description: str
    problem: str
    suggestion: str
    reason: str
    source: str | None = None

CategoryとSeverityをLiteralで定義することで、型安全性を確保しています。

ParsedArticle

class ParsedArticle(BaseModel):
    front_matter: FrontMatter | None
    front_matter_line_count: int
    body: str
    body_start_line: int
    code_blocks: list[CodeBlock]
    links: list[tuple[str, int]]

    def get_numbered_body(self) -> str:
        """行番号付きの本文を返す"""
        lines = self.body.split("\n")
        numbered = []
        for i, line in enumerate(lines, start=self.body_start_line):
            numbered.append(f"{i}: {line}")
        return "\n".join(numbered)

get_numbered_body()メソッドは、LLMにIssueの行番号を正確に特定させるために使用します。

グラフ構築の設計

GraphConfigによる設定

@dataclass(frozen=True)
class GraphConfig:
    model_name: str = "gemini-2.0-flash"
    temperature: float = 0.0
    max_output_tokens: int = 16384

    enable_aws_mcp: bool = True
    enable_tavily: bool = True

    def __post_init__(self) -> None:
        if not 0.0 <= self.temperature <= 1.0:
            raise ValueError(f"temperature must be 0.0-1.0")

frozen=Trueで不変にし、__post_init__でバリデーションを行っています。

ProofreadGraphBuilder

@dataclass
class ProofreadGraphBuilder:
    config: GraphConfig = field(default_factory=GraphConfig)
    _llm: ChatGoogleGenerativeAI | None = field(default=None, repr=False)
    _graph: CompiledStateGraph | None = field(default=None, repr=False)

    @property
    def llm(self) -> ChatGoogleGenerativeAI:
        """LLMの遅延初期化"""
        if self._llm is None:
            self._llm = ChatGoogleGenerativeAI(
                model=self.config.model_name,
                temperature=self.config.temperature,
                max_output_tokens=self.config.max_output_tokens,
            )
        return self._llm

    async def _build_graph(self) -> CompiledStateGraph:
        """グラフの構築"""
        builder = StateGraph(ProofreadState)

        # ノードの追加
        builder.add_node("parse", parse_article_node)

        # 各チェッカーのサブグラフを追加
        for checker_cls, name in [
            (MarkdownChecker, "check_markdown_format"),
            (LinkChecker, "check_link"),
            # ...
        ]:
            checker = checker_cls(self.llm)
            builder.add_node(name, checker.create_graph())

        # ...
        return builder.compile()

LLMの遅延初期化（@property）により、不要なAPI呼び出しを防いでいます。

プロンプト管理

プロンプトはMarkdownファイルで管理し、動的に読み込みます。

# config/prompts.py
def load_prompt(name: str) -> str:
    """プロンプトファイルを読み込む"""
    prompt_path = PROMPTS_DIR / f"{name}.md"
    if not prompt_path.exists():
        raise FileNotFoundError(f"Prompt not found: {prompt_path}")
    return prompt_path.read_text(encoding="utf-8")

prompts/
├── check_markdown_format.md
├── check_structure.md
├── check_privacy.md
├── check_language_quality.md
├── extract_technical_claims.md
├── plan_technical_search.md
├── verify_and_generate_technically.md
└── ...

この設計により、プロンプトの更新がコード変更なしで可能です。

エラーハンドリング戦略

ErrorInfo

class ErrorInfo(BaseModel):
    step: str       # エラーが発生したステップ
    error_type: str # 例外の型名
    message: str    # エラーメッセージ

各ノードでのキャッチ

async def _check_node(self, state: SubgraphState) -> dict[str, Any]:
    try:
        # 処理...
        return {"issues": issues}
    except Exception as e:
        logger.exception(f"Error in {self.category} checker")
        return {
            "errors": [
                ErrorInfo(
                    step=f"check_{self.category}",
                    error_type=type(e).__name__,
                    message=str(e),
                )
            ],
        }

エラーが発生しても、他のチェッカーは継続実行されます。最終的なレポートでエラー情報も報告されます。

スコアリングロジック

def _calculate_score(critical: int, warning: int, suggestion: int) -> int:
    score = 100
    score -= critical * 20      # クリティカルは-20点
    score -= warning * 3        # 警告は-3点
    score -= suggestion // 2    # 提案は2個で-1点
    return max(0, score)


def _determine_approval(critical: int, score: int) -> bool:
    if critical > 0:
        return False  # クリティカルがあれば不承認
    if score < 50:
        return False  # 50点未満は不承認
    return True

今後の応用

この設計は以下の拡張に対応できます。

新しいチェッカーの追加：BaseCheckerを継承し、プロンプトを追加
スコアリングの調整：_calculate_scoreの重み付けを変更
出力形式の変更：final_reporter.pyを修正
対応プラットフォームの追加：article.pyにFrontMatter定義を追加

まとめ

本記事では、校正エージェント全体の設計について解説しました。

設計のポイントは以下の通りです。

状態の分離：メイン状態とサブグラフ状態を分離し、責務を明確化
Reducer関数：並列実行の結果を適切にマージ
型安全性：LiteralとTypedDictで型を厳密に定義
遅延初期化：LLMを必要時にのみ初期化
プロンプト分離：Markdownファイルで管理し、変更を容易に
エラーハンドリング：部分的な失敗を許容し、継続実行

次回は、サブグラフによるサブエージェント設計について解説します。