AIワークフローを業務に組み込もうとすると、「モデルは動くけれど、変更すると突然動かなくなる」「外部APIの遅延で処理が止まる」「実運用データでの想定外ケースに気づけない」といった悩みに直面しがちです。本記事では、Pythonで実装した日常的なAIワークフロー（CSV前処理、API/LLM呼び出し、推論バッチ）を対象に、実務で使えるテスト戦略とCIの構成を手順とコード例で解説します。現場で再現しやすい具体例に沿って進めますので、まずは小さく始め、運用に耐える品質を徐々に高めていきましょう。
本記事の対象と目的
想定読者：AIを仕事で活用したい実務担当者、個人事業主、中小企業の担当者。目的は、既存のPythonワークフローにテストとCIを導入して運用リスクを下げ、変更時の回帰を防ぐことです。
章立て（概要）
- テスト戦略の決め方（ユニット／統合／エンドツーエンド）
- pytestを使った単体テストの書き方（fixtures, parametrize）
- 外部API/LLM呼び出しのモック手法
- テストデータの生成と管理（faker, factory_boy, CSVサンプル）
- 統合テストと小型データセットでの実行方針
- GitHub ActionsでのCIパイプライン例
- 運用時の注意点と失敗パターン
- 導入チェックリストとテンプレート／30分クイックスタート
1. テスト戦略（どのテストをどこまで書くか）
まずは役割を明確にします。小さなチームや一人運用では、全てのテストを同時に充実させるのは非現実的です。優先度を決めて段階的に増やすと良いです。
| テスト種別 | 対象 | 速度 | 例 | 導入優先度（実務） |
|---|---|---|---|---|
| ユニットテスト | 関数単位、前処理ロジック | 速い | CSVの列変換、正規化関数 | 高 |
| 統合テスト | モジュール間連携、データパイプライン | 中 | CSV読み込み→前処理→推論ラッパー | 中 |
| エンドツーエンド（E2E） | 本番に近いフロー（API/外部サービス含む） | 遅い | 小さなデータセットでのバッチ実行 | 低→定期実行推奨 |
実務的な優先度の目安
- まずユニットテスト：データ前処理、ビジネスロジック、スコア計算。
- 次に統合テスト：モジュールの連携、不整合を早期検出。
- E2Eは定期実行（夜間／週次）で本番寄せの検証を行う。
2. pytestを使った単体テストの書き方（基本とコツ）
pytestはfixtureやparametrizeで可読性の高いテストが書けます。ここではCSV前処理関数の例を示します。
CSV前処理（例）
def clean_row(row):
# 入力: dict（CSVの1行）
# 戻り値: 正規化された dict
name = row.get("name", "").strip()
age = row.get("age")
try:
age = int(age)
except (TypeError, ValueError):
age = None
return {"name": name, "age": age}
pytestの例
import pytest
from myproject.csv_utils import clean_row
@pytest.mark.parametrize(
"input_row,expected",
[
({"name": " Alice ", "age": "30"}, {"name": "Alice", "age": 30}),
({"name": "", "age": "x"}, {"name": "", "age": None}),
],
)
def test_clean_row_param(input_row, expected):
assert clean_row(input_row) == expected
@pytest.fixture
def sample_row():
return {"name": " Bob ", "age": "45"}
def test_clean_row_fixture(sample_row):
out = clean_row(sample_row)
assert out["name"] == "Bob"
assert isinstance(out["age"], int)
assertion のコツ
- 等価性は == を使い、浮動小数点は pytest.approx を使う。
- 複雑なオブジェクトは必要最小限に比較（キーのみや主要値のみ）。
- 失敗時に読みやすいメッセージが出るよう、pytestのassertをご活用ください。
3. 外部API/LLM呼び出しのモック
外部依存はテストを不安定にするため、ユニット/統合レベルで適切にモックします。代表的な手法と使い分けを下表に示します。
| ライブラリ | 用途 | 向く場面 |
|---|---|---|
| unittest.mock | 関数/メソッドの置換（軽量） | 内部ラッパーの戻り値制御、エラー再現 |
| responses | requestsによるHTTPレスポンスのモック | 外部REST APIのユニットテスト |
| requests-mock | requestsのSession単位でのモック | 細かいHTTP振る舞いを制御したい場合 |
| VCR.py | 実際のHTTP交流を録画して再生 | 既知のAPI応答をキャッシュして再現性を得たいとき |
外部APIラッパーの例とテスト
# myproject/api_client.py
import requests
class APIClient:
def __init__(self, base_url):
self.base_url = base_url
def get_score(self, payload):
r = requests.post(f"{self.base_url}/score", json=payload, timeout=5)
r.raise_for_status()
return r.json()
# tests/test_api_client.py
from unittest.mock import patch
from myproject.api_client import APIClient
@patch("myproject.api_client.requests.post")
def test_get_score_mock(mock_post):
mock_resp = mock_post.return_value
mock_resp.json.return_value = {"score": 0.9}
mock_resp.raise_for_status.return_value = None
client = APIClient("https://api.example")
out = client.get_score({"x": 1})
assert out["score"] == 0.9
mock_post.assert_called_once()
よりHTTP層で試したい場合は responses を使います：
import responses
from myproject.api_client import APIClient
@responses.activate
def test_get_score_responses():
responses.add(
responses.POST,
"https://api.example/score",
json={"score": 0.8},
status=200,
)
client = APIClient("https://api.example")
assert client.get_score({"x": 1})["score"] == 0.8
4. テストデータの生成と管理
良いテストは良いデータから生まれます。現実の代表ケースと境界値、エッジケースを意図的に含めましょう。個人情報（PII）を含む実データは必ず合成化するか、マスクして使用してください。
| ルール | 説明 |
|---|---|
| 代表ケース抽出 | 現場データから頻度の高いパターンを抽出してサンプル化 |
| 異常値／境界値 | 欠損、極端値、異常文字列（例: 半角／全角混在）を含める |
| PIIの扱い | 本番データを使う場合は必ず匿名化または合成データで代替 |
| 小さなCSVサンプル | 数十行の代表的セットを用意して統合テストに利用 |
faker と factory_boy の簡単例
from faker import Faker
fake = Faker()
def make_user_record():
return {"name": fake.first_name(), "email": fake.email(), "age": fake.random_int(18, 80)}
生成したレコードはCSVに書き出して統合テスト用の小データとして利用できます。
小さなCSVサンプル設計（例）
| 行種別 | 説明 |
|---|---|
| 正常 | 代表的な1〜2行（標準的な列と値） |
| 欠損 | 必須列が欠けている行（ageが空など） |
| 境界 | 年齢=0、最大値、長い文字列など |
| 異常 | 数値が文字列、特殊文字を含む列 |
5. 統合テストの実行方針
統合テストは重くなりがちなので、次の方針を推奨します。
- ローカルではユニットテストを高速に回す（pre-commit / pre-pushで実行）。
- 統合テストは小さなデータセットで速く回るように設計し、CIでは長いE2Eは別ジョブ・夜間に実行。
- 外部APIはスタブやVCRで再現性を担保。定期的に実際のAPIで罹患テストを走らせる。
統合テスト（ワークフロー実行）の簡単な例
# myproject/workflow.py
import csv
from myproject.csv_utils import clean_row
from myproject.api_client import APIClient
def run_batch(csv_path, api_client):
results = []
with open(csv_path, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for r in reader:
r2 = clean_row(r)
res = api_client.get_score(r2)
results.append(res)
return results
# tests/test_workflow_integration.py
from myproject.workflow import run_batch
from types import SimpleNamespace
def test_run_batch(tmp_path):
csv_file = tmp_path / "sample.csv"
csv_file.write_text("name,age\nAlice,30\nBob,\n")
class DummyClient:
def get_score(self, payload):
return {"score": 0.5 if payload.get("age") else 0.0}
out = run_batch(str(csv_file), DummyClient())
assert len(out) == 2
assert out[0]["score"] == 0.5
assert out[1]["score"] == 0.0
6. GitHub ActionsでのCIパイプライン（テンプレート）
ここではPRで自動的にpytestを回し、カバレッジを計測して報告する最小テンプレートを示します。ファイル名は .github/workflows/ci.yml を想定してください。
name: CI
on:
pull_request:
push:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.10'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements-dev.txt
- name: Run tests with coverage
run: |
pytest --junitxml=reports/junit.xml --cov=myproject --cov-report=xml
- name: Upload coverage report
uses: actions/upload-artifact@v4
with:
name: coverage-report
path: coverage.xml
実務では coverage の結果を Codecov や Coveralls と連携したり、PR上で失敗を通知する仕組み（Reviewdog や GitHub Checks）を追加します。
7. 運用時の注意点と失敗パターン
| 問題 | 原因 | 対策 |
|---|---|---|
| フレークテスト（不安定なテスト） | 外部依存、時間依存、並列性の問題 | 依存のモック化、retry回避、テストの分離 |
| テストが遅い | 大きなデータセット、E2Eを頻繁に実行 | ユニットと統合を分離、夜間にE2Eを実行 |
| カバレッジ偏重 | 量だけ増えて中身が薄いテスト | 重要ロジックに対する深いアサーション、コードレビューでの品質担保 |
品質指標とゲート条件（目安）
| 指標 | 目安 |
|---|---|
| 全体テストカバレッジ | 70〜85%（業務重要度に応じて高める） |
| 重要モジュール | 90%前後を目指す |
| PRゲート | ユニットテスト全通＋カバレッジの変化が大きい場合は要確認 |
CIで失敗した時の簡単なエスカレーションフロー
- 1) PRに修正コミットを追加して再実行（まず試す）
- 2) それで直らない場合は失敗ログを貼って担当者に@メンション
- 3) 夜間のE2Eで発生した場合は運用チームにチケット登録し、翌稼働日までブロッキング回避策を適用
8. 導入チェックリストとテンプレート
| 項目 | 完了 | 注記 |
|---|---|---|
| requirements-dev.txt に pytest, pytest-cov を追加 | [ ] | 最低限のテスト実行に必要 |
| 少なくとも1つのユニットテストを追加 | [ ] | CSV前処理などビジネスロジック |
| CIワークフローを追加（.github/workflows/ci.yml） | [ ] | PRでテストが回るように |
| テストデータ（小さなCSV）をリポジトリに保存 | [ ] | 例: tests/data/sample.csv（PIIに注意） |
| 外部APIをモックするガイドを作成 | [ ] | responses や unittest.mock の例を含める |
9. 30分でできるクイックスタート（実践手順）
- requirements-dev.txt に次を追加：pytest, pytest-cov, responses、保存（5分）。
- リポジトリに tests/test_csv_utils.py を1つ作成し、上記の clean_row テストを貼る（10分）。
- .github/workflows/ci.yml を追加し、上記テンプレートを貼る（10分）。
- PRを作って動作を確認。失敗したらログを見て修正して再コミット（5分）。
補助リソースと推奨ライブラリ（導入順）
| 順 | ライブラリ | 目的 |
|---|---|---|
| 1 | pytest, pytest-cov | テスト実行とカバレッジ測定 |
| 2 | unittest.mock | 軽量なモック・パッチ |
| 3 | responses / requests-mock | HTTP APIのモック |
| 4 | faker, factory_boy | テスト用データ生成 |
| 5 | VCR.py | HTTP録画再生（必要時） |
| 6 | GitHub Actions | CIの実行環境 |
まとめ
テストとCIは、運用リスクを低減し、変更時の安心感を高めます。現場では小さく始めることが重要です。まずは主要な前処理やスコア計算に対するユニットテストを1つ書き、CIで自動実行する流れを作ってください。次に統合テストを小さなデータセットで追加し、外部依存はモックで切り離します。最終的には定期的に実運用に近いE2Eを走らせることで、運用上の齟齬を早期に発見できます。
Manage AI（https://manageai.online）では、今回のような実務寄りの手順を今後も扱っていきます。まずはこの記事のクイックスタートに従い、既存リポジトリに1つのユニットテストとCIを追加してみてください。
次の一歩（推奨）：ローカルで pytest が通ることを確認したら、PR を作成して GitHub Actions で自動化される流れを体験してください。困ったときはログ全文を保存してチームで共有すると原因特定が速くなります。