guinness-ai-v2 コーディングルール

---

基本原則

• Python 3.12 を使用する
• Pydantic でスキーマを定義し、dict や素の TypedDict を避ける
• PydanticAI でエージェントを構築し、LangChain / LangGraph は使わない
• 型ヒントをすべての関数に付ける（Any は原則禁止）
• ウォームスタートを活かす — Lambda コンテキスト外で初期化できるクライアントはモジュールトップレベルで生成する
• DB 分離を守る — AI ワーカーは PostgreSQL / MySQL に直接アクセスしない。バックエンドとの通信は Webhook のみ

---

ディレクトリ構成

各ワーカーアプリは以下のファイル構成に従う。

apps/<worker>/src/<worker>/
  handler.py      # Lambda エントリポイント — SQS パース・batchItemFailures
  service.py      # process_record() — メイン処理パイプライン
  schemas.py      # Pydantic モデル（SQS 入力・DocumentDB ドキュメント・Webhook ペイロード）
  repo.py         # I/O 操作（S3 ダウンロード・DocumentDB upsert・Webhook POST）
  config.py       # pydantic-settings による環境変数
  prompts.py      # LLM プロンプト定数・PROMPT_VERSION

packages/
  agentic/        # PydanticAI エージェント定義（ワーカー間で共有）
  models/         # DocumentDB コレクション定義・インデックス作成
  utils/          # S3・Webhook・ロギング共通ユーティリティ

責務の分離

ファイル	書いてよいもの	書いてはいけないもの
handler.py	SQS パース、batchItemFailures 構築、ログ初期化	ビジネスロジック、DB 操作
service.py	処理パイプライン全体の調整	直接 DB クエリ、プロンプト文字列
repo.py	S3 / DocumentDB / Webhook I/O	ビジネスルール
schemas.py	Pydantic モデル定義	ロジック
config.py	環境変数バリデーション（pydantic-settings）	ロジック
prompts.py	プロンプト文字列・バージョン定数	ロジック

---

命名規則

対象	規則	例
ファイル	snake_case	code_import/service.py
クラス	PascalCase	CodeImportMessage, CodeDocument
関数・変数	snake_case	process_record, vector_embedding
定数	SCREAMING_SNAKE_CASE	PROMPT_VERSION, VECTOR_DIMENSIONS
PydanticAI エージェント	build_<name>_agent()	build_vision_agent()
Pydantic 設定クラス	Config	config.py 内の class Config(BaseSettings)

---

Lambda ハンドラーパターン

ウォームスタート最適化

重いクライアントはモジュールトップレベルで初期化し、ハンドラー呼び出しごとに再生成しない。

# handler.py
from .config import Config
from .repo import DesignRepo
from packages.agentic.vision import build_vision_agent

config = Config()
repo = DesignRepo(config)        # ウォームスタートで再利用
agent = build_vision_agent(config)  # PydanticAI エージェントも同様

def lambda_handler(event: dict, context: object) -> dict:
    ...

SQS batchItemFailures

部分的な失敗はレコード単位で報告し、バッチ全体をリトライしない。

def lambda_handler(event: dict, context: object) -> dict:
    batch_item_failures = []

    for record in event["Records"]:
        try:
            message = CodeImportMessage.model_validate_json(record["body"])
            service.process_record(message, repo, agent)
        except Exception as e:
            logger.error("record failed", extra={"message_id": record["messageId"], "error": str(e)})
            batch_item_failures.append({"itemIdentifier": record["messageId"]})

    return {"batchItemFailures": batch_item_failures}

---

Pydantic スキーマ

SQS 入力スキーマ

SQS body の検証には model_validate_json を使う。必須フィールドは model_validator で強制する。

from pydantic import BaseModel, model_validator
import uuid

class CodeImportMessage(BaseModel):
    code_id: str
    organization_id: int
    project_id: int
    name: str
    source_code: str
    css_code: str | None = None
    img_url: str | None = None

    @model_validator(mode="after")
    def validate_code_id(self) -> "CodeImportMessage":
        uuid.UUID(self.code_id)  # UUID 形式を強制
        return self

DocumentDB ドキュメントスキーマ

DocumentDB に書き込むドキュメントも Pydantic で定義する。

class EmbeddingBlock(BaseModel):
    metadata: dict
    vector_embedding: list[float]

class CodeDocument(BaseModel):
    id: str  # _id にマップ
    organization_id: int
    project_id: int
    name: str
    source_code: str
    visual: EmbeddingBlock
    semantics: EmbeddingBlock
    index_schema_version: str

---

PydanticAI エージェント

エージェント定義

エージェントは packages/agentic/ に集約し、ワーカーアプリから import する。

# packages/agentic/src/agentic/vision.py
from pydantic import BaseModel
from pydantic_ai import Agent

class DesignDescription(BaseModel):
    layout: str
    component_types: list[str]
    color_palette: list[str]
    typography_style: str
    semantic_words: list[str]

def build_vision_agent(config: Config) -> Agent[None, DesignDescription]:
    return Agent(
        config.desc_model,  # "openai:gpt-5.4-nano" のようなプロバイダー付き形式
        output_type=DesignDescription,
        system_prompt=VISION_INSTRUCTIONS,
    )

モデル指定

モデルは環境変数で指定し、コードに直書きしない。形式は provider:model-name。

# 正しい
config.desc_model  # "openai:gpt-5.4-nano"

# 禁止
"gpt-5.4-nano"  # プロバイダーなし・直書き

run_sync の使用

Lambda 上では同期実行（run_sync）を使う。

result = agent.run_sync(user_prompt, message_history=None)
description: DesignDescription = result.output

---

埋め込み生成

バッチ呼び出しの徹底

複数の埋め込みは 1 回の embeddings.create で生成する。複数回に分けると次元ドリフトが発生する可能性がある。

# 正しい — 1 回のバッチ
response = openai_client.embeddings.create(
    model=config.embedding_model,
    input=[visual_text, semantic_text],   # まとめて渡す
    dimensions=config.embedding_dimensions,
)
visual_vector = response.data[0].embedding
semantic_vector = response.data[1].embedding

# 禁止 — 2 回に分けて呼ぶ
visual_vector = openai_client.embeddings.create(input=visual_text, ...).data[0].embedding
semantic_vector = openai_client.embeddings.create(input=semantic_text, ...).data[0].embedding

---

リポジトリ（I/O）パターン

DocumentDB upsert

_id にレコード ID を使い冪等に書き込む。

def upsert_code(self, doc: CodeDocument) -> None:
    self.collection.replace_one(
        {"_id": doc.id},
        doc.model_dump(by_alias=True),
        upsert=True,
    )

Webhook 通知

処理の成否をかならず Webhook で通知する。例外が発生しても failed を送るようにする。

def notify_success(self, record_id: str, **payload) -> None:
    self._post_webhook({"status": "success", "record_id": record_id, **payload})

def notify_failure(self, record_id: str, error: str) -> None:
    self._post_webhook({"status": "failed", "record_id": record_id, "error": error})

---

環境変数

環境変数は pydantic-settings の BaseSettings で管理し、バリデーションを強制する。

from pydantic_settings import BaseSettings

class Config(BaseSettings):
    desc_model: str
    embedding_model: str
    embedding_dimensions: int = 512
    openai_api_key: str
    documentdb_connection_string: str
    documentdb_name: str
    webhook_base_url: str

    class Config:
        env_file = ".env"

• 必須変数はデフォルト値なしで定義する（起動時に即座にエラー）
• 任意変数はデフォルト値を付ける
• シークレット（API キー等）はログに出力しない

---

エラーハンドリング

リトライ可能な失敗 vs 不可能な失敗

エラー種別	対応
不正な SQS ペイロード	batchItemFailures に追加せず破棄（リトライしても意味がない）
対象リソースが見つからない	Webhook failed を送信して終了
OpenAI タイムアウト / レート制限	batchItemFailures に追加して SQS リトライさせる
DocumentDB 接続エラー	batchItemFailures に追加して SQS リトライさせる
Webhook 配信失敗	batchItemFailures に追加して SQS リトライさせる

try/except の粒度

process_record 全体を囲み、handler.py でレコード単位のエラーを捕捉する。

# service.py — 業務エラーは Webhook failure として処理
def process_record(message: CodeImportMessage, repo: CodeRepo, agent: Agent) -> None:
    try:
        ...
    except DesignNotFoundError:
        repo.notify_failure(message.code_id, error="Design not found")
        return  # batchItemFailures に追加しない（リトライ不要）

# handler.py — インフラエラーはリトライに任せる
try:
    service.process_record(message, repo, agent)
except Exception as e:
    batch_item_failures.append({"itemIdentifier": record["messageId"]})

---

ロギング

packages/observability または Python 標準 logging で構造化ログを出力する。print() は禁止。

import logging

logger = logging.getLogger(__name__)

# 正しい
logger.info("process_record started", extra={"code_id": message.code_id, "project_id": message.project_id})
logger.error("webhook failed", extra={"code_id": message.code_id, "status_code": resp.status_code})

# 禁止
print(f"processing {message.code_id}")

---

テスト

テスト種別と配置

apps/<worker>/
  __tests__/
    unit/          # 外部依存をモックしたユニットテスト
    contract/      # SQS / Webhook スキーマの契約テスト
    integration/   # 実際の DocumentDB コンテナを使う統合テスト

ユニットテスト方針

• repo はモックして service.process_record をテストする
• PydanticAI エージェントは TestModel を使ってモックする
• 各テストは独立して実行できるようにする

from pydantic_ai.models.test import TestModel

def test_process_record_success(mocker):
    agent = build_vision_agent.__wrapped__(TestModel())
    repo = mocker.Mock(spec=CodeRepo)
    message = CodeImportMessage(code_id=str(uuid.uuid4()), ...)

    service.process_record(message, repo, agent)

    repo.upsert_code.assert_called_once()
    repo.notify_success.assert_called_once()

契約テスト

SQS メッセージと Webhook ペイロードのスキーマ変更を検出するために契約テストを書く。

def test_sqs_message_schema():
    raw = {"code_id": "660e...", "organization_id": 1, ...}
    msg = CodeImportMessage.model_validate(raw)
    assert msg.code_id == raw["code_id"]

---

コードレビューチェックリスト

• [ ] Lambda クライアントはモジュールトップレベルで初期化されているか
• [ ] SQS 入力は Pydantic モデルで検証されているか
• [ ] 複数の埋め込みを 1 回のバッチ呼び出しで生成しているか
• [ ] DocumentDB への書き込みは upsert（冪等）か
• [ ] 成功・失敗どちらも Webhook が送信されているか
• [ ] リトライ可能なエラーが batchItemFailures に追加されているか
• [ ] print() を使っていないか
• [ ] 環境変数を Config 経由で取得しているか（ハードコード禁止）
• [ ] モデル名に provider:model 形式を使っているか
• [ ] PostgreSQL / MySQL に直接アクセスしていないか