AI Design Import — 概要

design-import SQS キューからデザイン取り込みジョブを処理する Lambda ワーカー。S3 から Figma node 画像と Figma node JSON をダウンロードし、視覚的特徴、セマンティックキーワード、component functions、構造化 node evidence を抽出、3 本の 512 次元埋め込みベクトル（visual + semantic + structural）を生成し、DocumentDB の design コレクションにアップサートした後、Webhook でバックエンドにステータスを通知する。

置き換え対象: V1 design アプリ (guinness-ai-develop)
キュー: design-import
最大実行時間: 約 2〜5 分
ソース: guinness-ai-v2（新規リポジトリ）

1. 技術スタック

レイヤ	採用技術
ランタイム	Python 3.12 on AWS Lambda
AI フレームワーク	PydanticAI（エージェントオーケストレーション、プロバイダー非依存）
LLM	`DESC_MODEL` 環境変数（例: `openai:gpt-5.4-nano`）— 視覚的特徴抽出 + セマンティックキーワード
埋め込み	`EMBEDDING_MODEL` 環境変数（例: `text-embedding-3-small`、512 次元）— `openai` SDK 直呼び出し
データベース	Amazon DocumentDB（MongoDB 互換）— `design` コレクション、AI 専用
ストレージ	Amazon S3（デザイン画像、Figma JSON）
キュー	AWS SQS（`design-import` キュー）
ステータス通知	Webhook `POST /v1/webhooks/ai-status` → guinness-backend
RDB	アクセスなし — PostgreSQL / MySQL には接触しない（データベース分離を徹底）

2. V1 との主な変更点

項目	V1	V2
AI フレームワーク	LangGraph `StateGraph`	PydanticAI エージェント（`run_sync`）
デザインあたりの埋め込み数	1（視覚スタイルキーワード）	3（visual + semantic + structural）
データベース書き込み	MySQL 直接 + DocumentDB	DocumentDB のみ + Webhook（RDB アクセスなし）
Figma JSON	ドキュメントにそのまま保存	structural 埋め込み + コンポーネント ID + viewport にパース
バッチ処理	最初の失敗でバッチ全体を再試行	`batchItemFailures` レスポンス（レコード単位で再試行）
`design_id` 検証	説明的なのみ	Pydantic `model_validator` で強制
LLM クライアント	`ChatOpenAI`（langchain-openai）	PydanticAI（プロバイダー非依存）
埋め込みクライアント	`OpenAIEmbeddings`（langchain）— 単一	`openai` SDK 直呼び出し — visual + semantic + structural

3. 処理フロー

flowchart TD
    A["SQS: design-import メッセージ"] --> B["パース + SqsBody 検証\n(design_id 複合キー強制)"]
    B --> C["S3 からデザイン画像ダウンロード\n(最大 10 MB)"]
    C --> E["S3 から Figma JSON ダウンロード\nfigma.extract_structural(schema, node_id)"]
    E --> F["→ structural_text\n→ component_ids\n→ viewport"]
    F --> H
    H["Vision Agent — PydanticAI run_sync\nDesignDescription:\n  layout, component_types, color_palette\n  typography_style, semantic_words"]
    H --> I["埋め込み入力テキストの構築\nvisual_text, semantic_text\n+ structural_text\n+ query_terms"]
    I --> J["OpenAI Embeddings — 1 回バッチ呼び出し\n3 入力 → 512 次元ベクトル"]
    J --> K["DesignDocument を構築\n_id=design_id で DocumentDB に upsert"]
    K --> L["POST webhook success"]

    style A fill:#f9f,stroke:#333
    style H fill:#bbf,stroke:#333
    style K fill:#bfb,stroke:#333
    style L fill:#fdb,stroke:#333

ワーカーは各レコードに対して 8 ステップ を順番に実行します。どこかで失敗したメッセージはエラーハンドリングのパスに進み、失敗 Webhook が送出されてから batchItemFailures に追加されます。

4. 埋め込み戦略

1 デザインあたり最大 3 つの埋め込みが直交する検索シグナルをカバーします。ドリフト防止のため全て 1 回のバッチ呼び出しで生成します：

埋め込み	入力テキスト	用途
`visual`	`"layout: {layout} \\| components: {types} \\| palette: {colors} \\| typography: {style}"`	「同じように見える」デザインを探す
`semantics`	`" ".join(semantic_words)`	「同じドメインに関する」デザインを探す
`structural`	決定的 Figma ツリーレンダリング（`figma.py` 参照）	「同じコンポーネント構造を持つ」デザインを探す

structural は json_schema_url が必須のため、成功した import では必ず生成される。DocumentDB には Des2Code リランキング用の query_terms と Figma node evidence も永続化する。

5. Figma JSON パース（`figma.extract_structural`）

Figma JSON は LLM を使わず決定的にパースされます。パーサー（apps/design-import/src/design_import/figma.py）はノードツリーを走査し、3 つの出力を生成します：

structural_text — レイアウト、コンポーネント、ツリー形状をエンコードしたコンパクトな 1 行の文字列：

tree=FRAME[V,gap_md]>[FRAME[H,gap_none]>[TEXT_HEADING],INSTANCE[comp_input_email],INSTANCE[comp_input_password],INSTANCE[comp_button_primary]] | components=FRAME[H]x1 FRAME[V]x1 INSTANCE[comp_button_primary]x1 TEXT_HEADINGx1 | depth=2

設計上の選択： - レイアウト方向とスペーシングはバケット化（gap_none, gap_xs, gap_sm, gap_md, gap_lg, gap_xl） - INSTANCE ノードは Figma の componentId（正規 ID）をキー、デザイナー付与の名前は無視 - TEXT ノードはフォントサイズでバケット化（TEXT_HEADING ≥28px、TEXT_SUBHEADING ≥18px、TEXT_BODY） - ツリー深さは最大 12 レベル（max_depth）

component_ids — ツリー内の全 INSTANCE.componentId をソートしたリスト。des2code Stage 2 がコンポーネントマッピングのカバレッジを JSON 再パースなしに確認できます。

viewport — ルートノードの absoluteBoundingBox から取得した {width, height, aspect_ratio, device_class}。デバイスクラス: mobile（<600px）、tablet（<1024px）、desktop（<1600px）、wide（≥1600px）。

6. SQS メッセージスキーマ

バックエンド（apps/app/src/services/design.ts）が design-import キューにエンキューするメッセージ：

{
  "Records": [{
    "messageId": "0192a3b4-c5d6-7e8f-9a0b-1c2d3e4f5a6b",
    "body": "{\"organization_id\":1,\"project_id\":42,\"img_url\":\"s3://gnss-prod-inputs/designs/42_hDDA9B..._40002029:37033.png\",\"node_id\":\"40002029:37033\",\"file_id\":\"hDDA9BNori9OTXSClduXqR\",\"design_id\":\"42_hDDA9BNori9OTXSClduXqR_40002029:37033\",\"design_name\":\"Login Screen\",\"json_schema_url\":\"s3://gnss-prod-inputs/figma_json_schema/42_hDDA9BNori9OTXSClduXqR_40002029:37033.json\"}",
    "attributes": { "ApproximateReceiveCount": "1" },
    "eventSource": "aws:sqs"
  }]
}

Body フィールド（SqsBody Pydantic モデルで検証）：

フィールド	型	必須	備考
`organization_id`	integer	✓	tenant scope。DocumentDB document と webhook にもコピー
`project_id`	integer	✓	project scope。DocumentDB document と webhook にもコピー
`img_url`	string	✓	デザイン画像の S3 URL
`node_id`	string	✓	Figma node id（通常 `<int>:<int>` 形式）
`file_id`	string	✓	Figma file id
`design_id`	string	✓	`"{project_id}_{file_id}_{node_id}"` と一致必須 — Pydantic `model_validator` で強制。DocumentDB の `_id` になる
`design_name`	string	✓	Figma フレーム名
`json_schema_url`	string	✓	Figma node JSON エクスポートの S3 URL。必須。欠落、取得失敗、パース失敗は import 失敗

7. Webhook ペイロードスキーマ

処理完了後、WEBHOOK_BASE_URL に POST を送信：

成功時

{
  "type": "design-import",
  "recordId": "42_hDDA9BNori9OTXSClduXqR_40002029:37033",
  "projectId": 42,
  "organizationId": 1,
  "status": "success",
  "result": {
    "keywords": ["login", "authentication", "form", "submit", "credentials"],
    "embeddingDims": 512,
    "vectorsProduced": ["visual", "semantic", "structural"],
    "indexSchemaVersion": "2026.05.2"
  }
}

失敗時

{
  "type": "design-import",
  "recordId": "42_hDDA9BNori9OTXSClduXqR_40002029:37033",
  "projectId": 42,
  "organizationId": 1,
  "status": "failed"
}

失敗時は result ブロックはありません。パース前に失敗した場合は recordId、projectId、organizationId が null になることがあります。

8. PydanticAI — Vision エージェント

V1 の LangGraph StateGraph を PydanticAI の単一エージェントに置き換え。フレームワークはプロバイダー非依存 — モデル名は provider:model 形式で指定：

from pydantic_ai import Agent
from pydantic import BaseModel

class DesignDescription(BaseModel):
    layout: str                  # 11 種類のアーキタイプのうちひとつ
    component_types: list[str]   # 3〜15 個の正規化 snake_case 名
    color_palette: list[str]     # 支配的な hex カラー 2〜5 個（#RRGGBB）
    typography_style: str        # 8 種類のアーキタイプのうちひとつ
    semantic_words: list[str]    # ドメインキーワード 5〜10 個

vision_agent = Agent(
    DESC_MODEL,  # e.g., "openai:gpt-5.4-nano"
    output_type=DesignDescription,
)

入力: S3 からダウンロードしたデザイン画像（バイナリ）。
出力: DesignDescription — 構造化された視覚特徴とセマンティックキーワード。
実行モード: run_sync（同期）— レコードあたり 1 回。

9. 埋め込み生成

PydanticAI は埋め込みを扱わない。openai Python SDK を直接使用し、1 回のバッチ呼び出しで全ベクトルを生成（モデルドリフト防止）：

client.embeddings.create(
    model=config.embedding_model,
    input=[visual_text, semantic_text, structural_text],
    dimensions=config.embedding_dimensions,  # デフォルト 512
)
# data[0] → visual、data[1] → semantic、data[2] → structural

埋め込み	入力テキスト
Visual	`"layout: {layout} \\| components: {component_types} \\| palette: {color_palette} \\| typography: {typography_style}"`
Semantic	`" ".join(semantic_words)`
Structural	必須 Figma JSON から生成した `structural_text`

10. DocumentDB ドキュメント構造

{
  "_id": "42_hDDA9BNori9OTXSClduXqR_40002029:37033",
  "organization_id": 1,
  "project_id": 42,
  "name": "Login Screen",
  "type": "screen",
  "based_on": "figma_import",
  "index_schema_version": "2026.05.2",
  "generation_context": "Login Screen. Viewport: 375x812 (mobile). Layout: stacked_form. ...",

  "viewport": {
    "width": 375,
    "height": 812,
    "aspect_ratio": 0.4619,
    "device_class": "mobile"
  },

  "visual": {
    "metadata": {
      "name": "Login Screen",
      "image_url": "s3://gnss-prod-inputs/designs/...",
      "encoder": "text-embedding-3-small@512",
      "layout": "stacked_form",
      "component_types": ["input_email", "input_password", "button_primary"],
      "color_palette": ["#1A1A2E", "#FFFFFF", "#4A90E2"],
      "typography_style": "modern_sans"
    },
    "vector_embedding": [/* 512 個の float */]
  },

  "semantics": {
    "metadata": { "words": ["login", "authentication", "form", "submit", "credentials"] },
    "vector_embedding": [/* 512 個の float */]
  },

  "structural": {
    "metadata": {
      "structural_text": "tree=FRAME[V,gap_md]>[...] | components=... | depth=2"
    },
    "vector_embedding": [/* 512 個の float */]
  }
}

新しい成功 import では json_schema_url が必須のため、viewport と structural は必須です。

HNSW ベクトルインデックス

packages/models/documentdb/design.py がコールドスタート時に冪等に作成：

インデックス名	パス	次元	類似度	m	efConstruction
`visualVectorIndex`	`visual.vector_embedding`	512	cosine	16	64
`semanticVectorIndex`	`semantics.vector_embedding`	512	cosine	16	64
`structuralVectorIndex`	`structural.vector_embedding`	512	cosine	16	64

structuralVectorIndex はスパース動作 — structural ブロックを持たないドキュメントは構造化検索の結果から除外されます。

11. エラーハンドリング

部分バッチ失敗サポート — ハンドラは {"batchItemFailures": [...]} を返すため、1 件の失敗が残りのバッチを汚染しません。Lambda イベントソースマッピングで ReportBatchItemFailures を有効にする必要があります。

def lambda_handler(event, context):
    failures = []
    for record in event["Records"]:
        try:
            process_record(record, ...)
        except Exception:
            failures.append({"itemIdentifier": record["messageId"]})
    return {"batchItemFailures": failures}

シナリオ	挙動
無効な SQS イベントエンベロープ（`Records` なし）	`ValueError`。Webhook なし。呼び出し全体が失敗
`design_id` が複合キーと不一致	Pydantic バリデーションエラー。失敗 Webhook。`batchItemFailures` に追加
S3 GET 失敗（画像）	失敗 Webhook。`batchItemFailures` に追加
画像が 10 MB 超過	`ValueError`。失敗 Webhook。`batchItemFailures` に追加
Figma JSON 欠落/取得/パースエラー	失敗 Webhook。`batchItemFailures` に追加
Vision LLM 例外	失敗 Webhook。`batchItemFailures` に追加
Embedding API 例外	失敗 Webhook。`batchItemFailures` に追加
DocumentDB 書き込み例外	失敗 Webhook。`batchItemFailures` に追加（最大 2 回リトライ）
Webhook POST 自体が失敗	WARN ログ。SQS retry のため `batchItemFailures` に追加
SQS リトライ回数超過	メッセージは DLQ に移動

12. モジュール構成

apps/design-import/src/design_import/
  handler.py    # Lambda エントリ — SQS パース、batchItemFailures、ウォームスタートクライアント
  service.py    # process_record() — 8 ステップのパイプライン
  schemas.py    # SqsBody, DesignDocument, VisualBlock, SemanticsBlock, StructuralBlock,
                #   Viewport, WebhookSuccessPayload, WebhookFailurePayload
  repo.py       # DesignRepo — S3 ダウンロード、DocumentDB upsert、Webhook POST
  figma.py      # extract_structural() — Figma JSON → structural_text + component_ids + viewport
  prompts.py    # VISION_INSTRUCTIONS, VISION_USER_MESSAGE, PROMPT_VERSION
  config.py     # Config（pydantic-settings）

packages/agentic/src/agentic/
  vision.py     # build_vision_agent()、DesignDescription スキーマ

packages/models/src/models/documentdb/
  design.py     # コレクション名、HNSW インデックス作成、VECTOR_DIMENSIONS 定数

ファイル	責務	V1 の同等ファイル
`handler.py`	Lambda ハンドラー、batchItemFailures	`apps/design/src/main.py`
`service.py`	`process_record()` — 8 ステップパイプライン	`main.py` 内（V1 ではインライン）
`schemas.py`	Pydantic モデル（SQS / DocumentDB / Webhook）	`apps/design/src/schema.py`
`repo.py`	DocumentDB upsert、S3 ダウンロード	`main.py` 内（V1 ではインライン）
`figma.py`	Figma JSON パーサー	― （V2 新規）
`config.py`	環境変数（pydantic-settings）	`apps/design/src/config/env.py`

V1 から削除: config/mysql.py（V2 では MySQL アクセスなし）。

13. 環境変数

変数	説明	例 / デフォルト
`DESC_MODEL`	Vision LLM モデル（プロバイダー付き）	`openai:gpt-5.4-nano`
`EMBEDDING_MODEL`	OpenAI 埋め込みモデル	`text-embedding-3-small`
`EMBEDDING_DIMENSIONS`	ベクトル次元数（HNSW インデックスと一致させること）	`512`
`OPENAI_API_KEY`	OpenAI API キー	`sk-...`
`OPENAI_MAX_RETRIES`	SDK の 429/5xx リトライ回数	`3`
`DOCUMENTDB_CONNECTION_STRING`	DocumentDB 接続文字列	`mongodb://user:pass@host:27017/?tls=true`
`DOCUMENTDB_NAME`	DocumentDB データベース名	`guinness_ai`
`DESIGN_TABLE_NAME`	DocumentDB コレクション名	`design`
`DOCUMENTDB_CA_PATH`	TLS CA バンドルパス	`/var/task/global-bundle.pem`
`WEBHOOK_BASE_URL`	バックエンド Webhook エンドポイント（プライベート VPC）	`https://api.internal/v1/webhooks/ai-status`
`WEBHOOK_API_KEY`	`X-API-Key` として送る共有サービスキー	Secrets Manager / SOPS 値

すべての環境変数は AI インフラストラクチャ — 環境変数に記載。

14. 依存パッケージ

完全な依存パッケージリストは AI インフラストラクチャ — 共有依存パッケージを参照。