Contents
iFLYTEK 音声認識 API の全体概要と選定ポイント
iFLYTEK OpenPlatform が提供する音声認識は、利用シーンに合わせて 3 種類 に大別されます。本セクションではそれぞれの特徴を把握し、導入時に「バッチ vs リアルタイム」「汎用 vs 業界特化」の観点で最適な API を選択できるように解説します。実際のプロジェクトでどちらが適合するかを判断する材料として活用してください。
標準音声認識
標準 API はバッチ処理向けで、ファイル単位の高精度文字起こしに最適です。
- 用途例:会議録音の事後文字起こし、ドキュメント自動生成
- 選定基準:遅延が許容でき、音声データをまとめて送信できる場合
主な特徴と注意点
標準認識は 非リアルタイム であるため、リクエストごとのサイズ上限(10 MB)や、1 回の呼び出しで処理可能な最大音声長(5 分程度)に留意してください。
リアルタイム字幕
リアルタイム字幕はストリーミング対応で、音声入力と同時に文字列を取得できます。
- 用途例:ライブ配信の自動字幕、スマートスピーカーの音声コマンド認識
- 選定基準:低レイテンシが必須で、WebSocket もしくは HTTP/2 のストリーミングに対応できる環境
実装上のポイント
リアルタイム字幕では 音声チャンク(20 ms〜100 ms)単位で送信し、サーバーから返ってくる result.text を逐次描画する必要があります。ネットワーク遅延が大きい環境ではバッファリング戦略を導入すると安定します。
カスタムモデル
カスタムモデルは業界固有語彙やアクセントに最適化された認識を提供します。
- 用途例:医療現場の専門用語、金融機関の固有表現
- 選定基準:既存汎用モデルで精度が不足する場合にカスタム辞書・学習データを投入して改善
カスタマイズ手順概要
- コンソールで「辞書管理」→ CSV(UTF‑8)形式で語彙リストをアップロード
- モデル作成画面で
engine_typeにfinance_iat、medical_iatなどの業界モデルを選択 - 承認が完了したら API 呼び出し時に該当
engine_typeを指定
ポイント:利用シーン(バッチ vs リアルタイム)と認識精度要件(汎用 vs 業界特化)をまず整理し、対応 API を選択します。
法人向け OpenPlatform 登録手順と API キー・シークレット取得方法
このセクションでは iFLYTEK OpenPlatform のアカウント作成から API 認証情報取得までの流れを解説します。法人アカウントが必須かどうかは公式サイトで最新情報をご確認ください(個人向けプランが提供されている場合もあります)。また、取得したキーは機密情報ですので安全な保管方法についても併せて説明します。
アカウント種別の確認
- 現時点(2026 年 4 月)では公式サイト上で「法人向けプラン」のみが明示されています。
- 個人利用やスタートアップ向けの無料トライアルがあるかは、iFLYTEK OpenPlatform の料金・プランページ を随時確認してください。
登録手順(法人アカウント前提)
-
公式サイトへアクセス
https://iflytek.co.jp/pages/openplatform にアクセスし、右上の「新規登録」ボタンをクリック。 -
法人情報入力
- 会社名・所在地・担当者メールアドレスなど必須項目を正確に入力します。
-
「法人アカウント作成」を選択すると、次のステップへ進みます。
-
メール認証
-
送信された確認メール内のリンクをクリックし、画面表示される認証コードを入力して完了させます。
-
利用規約への同意
-
API 利用契約書・プライバシーポリシーに目を通し、チェックボックスへチェックします。
-
ダッシュボードでキー取得
- ログイン後の「API 管理」タブから「新規アプリケーション」を作成すると
AppKeyとAppSecretが自動発行されます。
重要:
AppSecretはサーバー側だけに保持し、クライアントコード(ブラウザやモバイルアプリ)へ埋め込まないでください。環境変数やシークレットマネージャーを利用して安全に管理します。
OAuth2.0 と HMAC 認証方式の実装例(Python・Node.js)
iFLYTEK OpenPlatform では OAuth 2.0 と HMAC の二種類の認証が公式にサポートされています。以下ではそれぞれのフロー、必須ヘッダー、および実装時に注意すべきエラーハンドリング・リトライロジックを具体的に示します。
OAuth2.0 認証
- 仕様書:https://doc.iflytek.com/openplatform/oauth2(トークン取得エンドポイント、スコープ定義など)
- フロー概要
AppKeyとAppSecretを用いて client_credentials グラントでトークンを取得- 有効期限(デフォルト 24 h)切れ前に再取得、失敗時は指数バックオフでリトライ
Python(requests)実装例(エラー処理・リトライ付き)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 |
import requests, time, logging from urllib3.util import Retry from requests.adapters import HTTPAdapter APP_KEY = "YOUR_APPKEY" APP_SECRET = "YOUR_APPSECRET" TOKEN_URL = "https://openapi.iflytek.com/v1/oauth2/token" # ---------- ロガー設定 ---------- logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def _session_with_retry(): session = requests.Session() retries = Retry( total=3, backoff_factor=1.5, # 1.5秒, 3秒, 4.5秒 と指数バックオフ status_forcelist=[500, 502, 503, 504], method_whitelist=["POST"] ) session.mount("https://", HTTPAdapter(max_retries=retries)) return session def get_token(): payload = { "grant_type": "client_credentials", "client_id": APP_KEY, "client_secret": APP_SECRET } sess = _session_with_retry() try: resp = sess.post(TOKEN_URL, data=payload, timeout=10) resp.raise_for_status() token = resp.json().get("access_token") if not token: raise ValueError("Token field missing in response") logger.info("OAuth2 token obtained successfully.") return token except Exception as e: logger.error(f"Failed to obtain OAuth2 token: {e}") raise # 実行例 if __name__ == "__main__": access_token = get_token() # 取得した token は以降の API 呼び出しに使用 |
Node.js(axios)実装例(エラー処理・リトライ付き)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 |
const axios = require('axios'); const retryAxios = require('retry-axios'); const APP_KEY = 'YOUR_APPKEY'; const APP_SECRET = 'YOUR_APPSECRET'; const TOKEN_URL = 'https://openapi.iflytek.com/v1/oauth2/token'; // Axios にリトライ機能を組み込む const client = axios.create(); client.defaults.raxConfig = { instance: client, retry: 3, // 最大リトライ回数 noResponseRetries: 2, backoffType: 'exponential', delay: 1500 // 初回遅延 1.5 秒 }; retryAxios.attach(client); async function getToken() { const params = new URLSearchParams(); params.append('grant_type', 'client_credentials'); params.append('client_id', APP_KEY); params.append('client_secret', APP_SECRET); try { const response = await client.post(TOKEN_URL, params); if (response.data && response.data.access_token) { console.log('OAuth2 token obtained.'); return response.data.access_token; } throw new Error('access_token not found in response'); } catch (err) { console.error('Failed to obtain OAuth2 token:', err.message); throw err; } } // 実行例 (async () => { const token = await getToken(); // 以降の API 呼び出しはこの token を使用 })(); |
HMAC 認証
- 仕様書:https://doc.iflytek.com/openplatform/hmac(署名生成ルール、必須ヘッダー一覧)
- フロー概要
- リクエストパラメータをキー順に連結し、
AppKey + timestamp + rawStringを元に SHA‑256 HMAC を算出 X-App-Key,X-Timestamp,X-Auth-Signatureヘッダーをリクエストに付与
Python(hashlib)実装例(エラーハンドリング付き)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 |
import hashlib, hmac, base64, time, json, logging, requests from urllib3.util import Retry from requests.adapters import HTTPAdapter APP_KEY = "YOUR_APPKEY" APP_SECRET = "YOUR_APPSECRET" logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def _session(): s = requests.Session() retries = Retry( total=2, backoff_factor=2, status_forcelist=[429, 500, 502, 503], method_whitelist=["POST"] ) s.mount("https://", HTTPAdapter(max_retries=retries)) return s def sign(params: dict) -> str: """HMAC‑SHA256 署名を生成して Base64 エンコードした文字列を返す""" raw = ''.join(f"{k}{v}" for k, v in sorted(params.items())) timestamp = str(int(time.time() * 1000)) message = f"{APP_KEY}{timestamp}{raw}" signature = hmac.new(APP_SECRET.encode(), message.encode(), hashlib.sha256).digest() return base64.b64encode(signature).decode(), timestamp def post_with_hmac(url: str, payload: dict): sig, ts = sign(payload) headers = { "X-App-Key": APP_KEY, "X-Timestamp": ts, "X-Auth-Signature": sig, "Content-Type": "application/json" } sess = _session() try: resp = sess.post(url, json=payload, headers=headers, timeout=10) resp.raise_for_status() logger.info("HMAC request succeeded.") return resp.json() except Exception as e: logger.error(f"HMAC request failed: {e}") raise # 使用例 if __name__ == "__main__": api_url = "https://openapi.iflytek.com/v1/iat" data = {"audio": "BASE64_AUDIO_DATA", "engine_type": "iat"} result = post_with_hmac(api_url, data) print(json.dumps(result, ensure_ascii=False, indent=2)) |
Node.js(crypto)実装例(エラーハンドリング・リトライ)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 |
const crypto = require('crypto'); const axios = require('axios'); const rax = require('retry-axios'); const APP_KEY = 'YOUR_APPKEY'; const APP_SECRET = 'YOUR_APPSECRET'; function sign(params) { const sortedKeys = Object.keys(params).sort(); const raw = sortedKeys.map(k => `${k}${params[k]}`).join(''); const timestamp = Date.now().toString(); const message = APP_KEY + timestamp + raw; const hmac = crypto.createHmac('sha256', APP_SECRET); hmac.update(message); return { signature: hmac.digest('base64'), timestamp }; } // axios にリトライ設定を付与 const client = axios.create(); client.defaults.raxConfig = { instance: client, retry: 2, backoffType: 'exponential', delay: 2000 }; rax.attach(client); async function postWithHmac(url, payload) { const { signature, timestamp } = sign(payload); const headers = { 'X-App-Key': APP_KEY, 'X-Timestamp': timestamp, 'X-Auth-Signature': signature, 'Content-Type': 'application/json' }; try { const resp = await client.post(url, payload, { headers }); console.log('HMAC request succeeded.'); return resp.data; } catch (err) { console.error('HMAC request failed:', err.message); throw err; } } // 使用例 (async () => { const apiUrl = 'https://openapi.iflytek.com/v1/iat'; const payload = { audio: 'BASE64_AUDIO_DATA', engine_type: 'iat' }; const result = await postWithHmac(apiUrl, payload); console.log(JSON.stringify(result, null, 2)); })(); |
ポイント:OAuth2.0 はトークンの有効期限管理が必須です。失効エラー (
1001) が返ってきたら即座にget_token()を再呼び出ししてリトライしましょう。一方 HMAC は毎回署名を生成するため、シークレット漏洩リスクを最小化できますが、タイムスタンプのズレ に注意してください。
リクエストパラメータとサンプル JSON の作り方
音声データのフォーマット・エンコーディングは認識精度に直結します。本節では公式ドキュメント(API パラメータリファレンス)を元に、必須項目と推奨設定を一覧化し、実装者がすぐにコピペできるサンプル JSON を提示します。
音声データ形式・エンコーディング
- 対応フォーマット:PCM(16bit, 16 kHz 推奨)、WAV、MP3
- リアルタイム字幕向け:必ず PCM (16bit, 16 kHz) に変換し、ヘッダー情報は除去した純粋なバイナリを Base64 化してください。
- バッチ認識向け:WAV/MP3 でも可。ただしサンプリングレートが 8 kHz 以下の場合は自動で 16 kHz にアップサンプリングされますが、精度が低下する可能性があります。
言語・ノイズ除去オプション
| パラメータ | 説明 | 推奨設定 |
|---|---|---|
language |
認識対象言語。例: zh-CN, en-US, ja-JP |
日本語の場合は "ja-JP" |
accent |
方言・アクセント指定(日本語は standard がデフォルト) |
必要に応じて "cantonese" など |
noise_reduction |
ノイズ抑制レベル(0=無効、1=標準、2=強化) | 一般的なノイズ環境では 1 推奨 |
標準音声認識向けサンプル JSON
|
1 2 3 4 5 6 7 8 9 |
{ "engine_type": "iat", "audio_format": "wav", "language": "ja-JP", "accent": "standard", "noise_reduction": 1, "audio": "<Base64 エンコード済み音声データ>" } |
リアルタイム字幕向けサンプル JSON(WebSocket 初回メッセージ)
|
1 2 3 4 5 6 7 |
{ "engine_type": "real_time_subtitle", "sample_rate": 16000, "language": "ja-JP", "authorization": "Bearer <OAuth2 Token>" } |
音声チャンク送信用 JSON(ストリーミング中)
|
1 2 3 4 |
{ "audio_chunk": "<Base64 エンコードした PCM データ>" } |
ポイント:音声ファイルは送信前にサンプリングレートを統一し、WAV ヘッダーなど余計なメタ情報を除去すると認識精度が向上します。
レスポンス構造とリアルタイム文字起こし取得、エラーコード対策
API の返却 JSON は 結果 と メタ情報 に分かれます。ここでは主要フィールドの意味と、リアルタイム字幕取得時に留意すべき点、さらに実運用で必ずハンドリングすべきエラーコード一覧を示します。
基本的なレスポンス項目
| フィールド | 説明 |
|---|---|
code |
0 が成功。非 0 はエラーコード(整数) |
message |
エラーメッセージまたは成功時の補足情報 |
session_id |
リクエスト単位で付与される識別子。リアルタイム取得に必須 |
result |
認識結果オブジェクト |
result.text |
文字起こしテキスト(UTF‑8) |
result.confidence |
信頼度(0〜1) |
result.timestamps |
単語単位の開始・終了時刻配列(リアルタイム字幕で利用) |
リアルタイム字幕取得フロー
- WebSocket 接続:エンドポイント
wss://openapi.iflytek.com/v1/real_time_subtitleに接続。 - 認証メッセージ送信:OAuth2 の Bearer トークンまたは HMAC 署名付き JSON を最初に送信し、
session_idを受領。 - 音声チャンク送信:20 ms〜100 ms 切り出した PCM データを Base64 化して
audio_chunkフィールドで逐次送信。 - サーバーからの結果受信:サーバーはリアルタイムで
result.textとtimestampsを返すので、画面表示や字幕生成に利用。
Python(websockets)実装例(エラーハンドリング・リトライ付き)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 |
import asyncio, websockets, json, base64, logging, time from aiohttp import ClientSession logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) TOKEN_URL = "https://openapi.iflytek.com/v1/oauth2/token" WS_ENDPOINT = "wss://openapi.iflytek.com/v1/real_time_subtitle" async def get_oauth_token(): # 簡易実装(本番ではリトライロジックを追加) async with ClientSession() as sess: payload = { "grant_type": "client_credentials", "client_id": "YOUR_APPKEY", "client_secret": "YOUR_APPSECRET" } async with sess.post(TOKEN_URL, data=payload) as resp: data = await resp.json() return data["access_token"] async def realtime_subtitle(audio_path): token = await get_oauth_token() async with websockets.connect(WS_ENDPOINT) as ws: # ① 認証メッセージ送信 init_msg = { "engine_type": "real_time_subtitle", "language": "ja-JP", "authorization": f"Bearer {token}" } await ws.send(json.dumps(init_msg)) logger.info("Auth message sent, waiting for session_id...") # 受信はブロッキングなのでタイムアウト設定 try: reply = await asyncio.wait_for(ws.recv(), timeout=5) sess_info = json.loads(reply) session_id = sess_info.get("session_id") logger.info(f"Session ID: {session_id}") except asyncio.TimeoutError: logger.error("Failed to receive session_id within timeout.") return # ② 音声チャンク送信 with open(audio_path, "rb") as f: while chunk := f.read(3200): # 20ms PCM (16kHz * 2bytes) payload = {"audio_chunk": base64.b64encode(chunk).decode()} await ws.send(json.dumps(payload)) await asyncio.sleep(0.02) # 実際の送信間隔に合わせる # ③ 結果受信ループ(最大 30 秒まで待機) try: async for message in asyncio.wait_for(ws, timeout=30): data = json.loads(message) if "result" in data: logger.info(f"字幕: {data['result']['text']}") except asyncio.TimeoutError: logger.warning("No more subtitle data received (timeout).") if __name__ == "__main__": asyncio.run(realtime_subtitle("sample.pcm")) |
主なエラーコードと対処法
| コード | 意味 | 推奨対策 |
|---|---|---|
| 1001 | 認証失敗(トークン無効、シグネチャ不正) | トークンを再取得/署名ロジックを見直す。リトライ前に 30 秒待機でレート制限回避 |
| 2003 | パラメータエラー(フォーマット・必須項目欠如) | audio_format、Base64 エンコード、必須フィールドの有無を検証 |
| 3001 | 音声データサイズ上限超過 | ファイルを 5 分以下に分割し、バッチ送信またはストリーミングへ切り替える |
| 4002 | 利用上限オーバー(月間クォータ) | プランのアップグレード/利用時間モニタリングとアラート設定 |
| 5000 | サーバ内部エラー(一時的障害) | 指数バックオフで最大 3 回リトライ。障害が続く場合はサポートへ問い合わせ |
実装上のベストプラクティス:全ての API 呼び出しを
try / catch(Python のexcept、Node.js のcatch) で包み、エラーコードに応じたリトライ/通知フローを統一的に扱うヘルパー関数を作成すると保守性が向上します。
カスタマイズ機能・料金シミュレーション・実装ベストプラクティス
本章では語彙追加や業界モデルの利用方法、最新料金情報の取得手順、そして実際にプロダクションへ組み込む際の注意点をまとめます。料金は変動する可能性があるため必ず公式サイト(2026 年版)で最新プランを確認してください。
語彙追加・業界モデル選択
| 項目 | 手順概要 | 注意点 |
|---|---|---|
| 語彙追加 | コンソール → 「辞書管理」→ CSV(UTF‑8、1 行 1 語)をアップロード。最大 10,000 件まで登録可。 | アップロード後は数分で即時反映されますが、キャッシュが残る場合は再起動または一定時間待機してください。 |
| 業界モデル | 「モデル選択」パラメータ engine_type に finance_iat, medical_iat 等を指定。利用には事前審査が必要です。 |
審査通過後に API キーの権限が自動付与されます。未承認状態でリクエストすると 403 エラーが返ります。 |
最新料金プランと従量課金シミュレーション
- 公式料金ページ:https://iflytek.com/openplatform/price(2026 年 4 月時点)
- 標準音声認識:1,000 秒=0.04 USD(※為替レートは変動します)
- リアルタイム字幕:1,000 秒=0.05 USD
- カスタムモデル:1,000 秒=0.07 USD
月間利用時間別概算シミュレーション(2026 年 4 月レート)
| 月間利用時間 | 標準音声認識費用 (USD) | リアルタイム字幕費用 (USD) | カスタムモデル費用 (USD) |
|---|---|---|---|
| 10 時間 (36,000 秒) | $1.44 | $1.80 | $2.52 |
| 100 時間 (360,000 秒) | $14.40 | $18.00 | $25.20 |
| 1,000 時間 (3,600,000 秒) | $144.00 | $180.00 | $252.00 |
注意:上記は純粋な従量課金のみのシミュレーションです。法人向けボリュームディスカウントや年間契約割引は別途問い合わせが必要です。
実装ベストプラクティス
| 項目 | 推奨手法 |
|---|---|
| 音声前処理 | 16 kHz に統一、サイレンス除去、noise_reduction=1 で軽度ノイズ抑制 |
| バッチ送信 | 5 分以下に分割し、非同期リクエストで同時実行数を制御(例:Python の asyncio.gather) |
| ロギング・トレーシング | session_id, タイムスタンプ, リクエスト/レスポンスサイズ を必ず記録し、ELK スタック等で可視化 |
| リトライ戦略 | 5xx 系は指数バックオフ(初回 1 s、2 回目 2 s、3 回目 4 s)で最大 3 回再送。認証エラーは新トークン取得後即リトライ |
| モニタリング | Grafana + Prometheus で「利用秒数」「エラーレート」「平均レイテンシ」を監視し、閾値超過時に Slack/メール通知 |
| セキュリティ | AppSecret は環境変数またはクラウド KMS に保存。通信は必ず TLS(HTTPS/WSS)で暗号化 |
Python でのロギング例
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 |
import logging, json, uuid, time logger = logging.getLogger("iflytek") handler = logging.StreamHandler() formatter = logging.Formatter( '[%(asctime)s] %(levelname)s %(message)s' ) handler.setFormatter(formatter) logger.addHandler(handler) logger.setLevel(logging.INFO) def log_request(session_id: str, payload: dict): logger.info(json.dumps({ "event": "request", "session_id": session_id, "timestamp": int(time.time()), "payload_size": len(json.dumps(payload)), "payload_snippet": json.dumps(payload)[:200] # 長すぎる場合は省略 })) def log_response(session_id: str, response: dict): logger.info(json.dumps({ "event": "response", "session_id": session_id, "timestamp": int(time.time()), "code": response.get("code"), "message": response.get("message") })) |
まとめ
- アカウント種別は公式サイトで最新情報を必ず確認し、必要に応じて個人向けプランも検討
- 料金は変動するため、公式価格ページのリンクを添えて都度チェック
- 認証方式は OAuth2.0 と HMAC の両方が利用可能。エラーハンドリングとリトライロジックを実装して安定運用を確保
- 音声前処理・パラメータ設定は精度に直結するため、公式ドキュメントの推奨設定を遵守
- リアルタイム字幕は WebSocket でのストリーミング実装が必須。接続切れや遅延に備えて再接続ロジックを組み込むこと
以上のポイントを踏まえれば、iFLYTEK 音声認識 API を安全・効率的にプロダクションへ組み込めます。ぜひ本稿を実装ガイドとして活用し、継続的なモニタリングと改善サイクルを回してください。