目次
- 注意喚起！！
- はじめに
- 本題
- まとめ
- 今後の改善点
注意喚起！！
Xのアカウント乗っ取りにはくれぐれも気を付けてください。
私は投資用アフィリエイトのフィッシング詐欺に合い、ハッカーから送られたリンクにアクセスし、パスワードを入力してしまいました（愚の極み）。
結果、Xのパスワード、電話番号、メールアドレスが流出する＆乗っ取られてしまいました。
今すぐ二段階認証を設定してください！！！
一連のやり取りは↓を参照。世の中上手い話は、無いと猛省しています💦
-
乗っ取られたアカウント
マルチンゲール@生成AI❎産業応用
※同じ被害を受けないためにくれぐれもフォローしない様にお願いします
- 作り直したアカウント
マルチンゲール@生成AI❎産業応用
※表示名が同じですがこちらが本家のアカウントです。フォロワーが激減したので、フォローいただけると嬉しいです😣🙏
ちなみに、Xサポートによる救済は充てにしない方が良いです。
運が良ければ、アカウントが戻ってくるくらいに考えておいた方が良いです（笑）
はじめに
さて、そんな散々な目に合った私ですが、先月幸いにもXのバックアックデータを取得していました。過去のツイートを基に自動ツイートするRAGシステムを作成するためです。
-
ツイートバックアップデータの情報：
項目 値 最古のツイート 2019年5月4日 最新のツイート 2025年10月12日 ツイート総件数 6,011件
アカウントの乗っ取りにより、フォロワーは一旦0に戻りましたが、同じ内容、観点でツイートをすれば賛同（フォロー）してくれる人が増えるはずです。
なので、アカウントが消滅したこの状況だからこそRAGシステムが活きそうなのです！
バックアップデータはアカウント設定の以下の箇所から、Xサポートに申請すれば送ってもらえます。私は1週間くらいかかりました。
元々のモチベーションは以下
- AzureとRAGの勉強を兼ねて何か作りたかった
- 自分ぽいことをつぶやいてくれるbotがあれば楽だと感じた
本題
本記事では、Azure上に構築したRAGベースの自動ツイートシステムについて、以下の3つの技術ポイントに焦点を当てて解説します：
- Azure AI SearchのSemantic Ranker - 2段階検索による高精度なRAG実装
- GPT-4のFunction Calling - 文字数制限を確実に守る3層防御アーキテクチャ
- Managed Identity + Key Vault - シークレットレスなセキュリティ設計
システムアーキテクチャ
全体構成
Timer Triggersが定期的にAzure Functionsを起動し、AI SearchとAzure OpenAIを使ってRAG検索・ツイート生成を行います。Key Vaultが全サービスのシークレットを一元管理し、Managed IdentityでRBAC認証を行います。
使用サービス
| サービス | SKU | 役割 | 月額目安 |
|---|---|---|---|
| Azure AI Search | Standard S1 | ベクトル検索 + Semantic Ranker | 約18,000円 |
| Azure OpenAI | Standard | GPT-4.1 + text-embedding-3-large | 約4,000円 |
| Cosmos DB | Serverless | ツイート履歴管理 | 約500円 |
| Azure Functions | Flex Consumption | サーバーレス実行 | 約500円 |
| Key Vault | Standard | シークレット管理 | 約500円 |
注目ポイント: AI Search S1は月額約18,000円と高額ですが、Semantic Rankerを使うにはStandard以上のSKUが必須です。
会社から（学習用に）付与されているVisual Studio Enterprise サブスクリプションの月額クレジット（$150 ≒ 約17,000円）の範囲で実施しています。このクレジットを使い切るとシステムが自動停止します。結果、月の半分くらいしか動かせません。
RAG + Semantic Rankerの実装
なぜSemantic Rankerが必要か
ベクトル検索は「意味的に近い」文書を高速に取得できますが、必ずしも最も関連性の高い結果が上位に来るとは限りません。
例えば「プログラミングの学習方法」で検索した場合：
- ベクトル検索結果: 「プログラミング」という単語を含む様々なツイート
- Semantic Ranker後: 実際に「学習方法」について言及しているツイートが上位に
Semantic Rankerは、Microsoftの自然言語理解モデルを使って検索結果を意味的な関連性で再ランキングします。
2段階検索の仕組み
Stage 1: ベクトル検索（高速）
→ 50件の候補を取得（HNSW: O(log n)）
Stage 2: Semantic Ranker（高精度）
→ 50件を意味的関連性で再ランキング
→ 上位5件を返却
インデックス設計
# src/services/search_service.py
# ベクトル検索の設定（HNSW: 高速近似最近傍検索アルゴリズム）
vector_search = VectorSearch(
algorithms=[
HnswAlgorithmConfiguration(
name="my-hnsw-algorithm",
parameters=HnswParameters(
metric=VectorSearchAlgorithmMetric.COSINE,
m=4, # ノードあたりの接続数
ef_construction=400, # 構築時の精度
ef_search=500 # 検索時の精度
)
)
],
profiles=[
VectorSearchProfile(
name="my-vector-profile",
algorithm_configuration_name="my-hnsw-algorithm"
)
]
)
# セマンティック検索の設定
semantic_config = SemanticConfiguration(
name="semantic-config",
prioritized_fields=SemanticPrioritizedFields(
content_fields=[
SemanticField(field_name="text") # ツイート本文を主要コンテンツに
]
)
)
ポイント:
-
m=4: 各ノードの接続数（メモリ使用量とのトレードオフ） -
ef_search=500: 検索時の精度（大きいほど精度↑、速度↓） -
content_fields: Semantic Rankerが分析するフィールドを指定
2段階検索クエリの実装
def vector_search_with_semantic_ranker(
self,
query_text: str,
query_vector: List[float],
top: int = 5,
filter_expression: str = None
) -> List[Dict[str, Any]]:
"""
セマンティックランカーを使用したベクトル検索
"""
# ベクトル検索クエリの作成
vector_query = VectorizedQuery(
vector=query_vector,
k_nearest_neighbors=50, # セマンティックランカー用に多めに取得
fields="embedding"
)
# セマンティック検索の実行
results = self.search_client.search(
search_text=query_text, # セマンティックランカーがこのテキストを使用
vector_queries=[vector_query], # ベクトル検索用
filter=filter_expression,
top=top,
query_type="semantic", # セマンティックランカーを有効化
semantic_configuration_name="semantic-config",
query_caption="extractive", # 関連部分の抽出
query_answer="extractive"
)
documents = []
for result in results:
documents.append({
"id": result["id"],
"text": result["text"],
"vector_score": result.get("@search.score", 0),
"reranker_score": result.get("@search.reranker_score", 0),
"captions": result.get("@search.captions", [])
})
return documents
重要なパラメータ:
-
query_type="semantic": Semantic Rankerを有効化 -
k_nearest_neighbors=50: Stage 1で50件取得（Stage 2で絞り込み） -
@search.reranker_score: Semantic Rankerのスコア（最終的な関連性）
インデックス自動更新の仕組み
RAGシステムでは、検索対象のインデックスを最新に保つことが重要です。本システムではIndexUpdater（Timer Trigger）が毎日自動でインデックスを更新します。
更新フローの6ステップ
- IndexUpdater起動（03:00 JST）- Timer TriggerがDataCollectorをHTTP呼び出し
-
新規ツイート取得 - X APIから
since_id以降のツイートをフェッチ - AI生成ID取得 - Cosmos DBから過去に投稿したAI生成ツイートのIDを取得
- フィルタリング - 人間が書いたツイートのみを抽出（詳細は次節）
- Embedding生成 - Azure OpenAIでtext-embedding-3-largeを使用
- インデックス登録 - AI Searchにバッチアップロード
スケジュール設計のポイント
TweetScheduler: 08:00, 12:00, 20:00 JST（ツイート生成）
IndexUpdater: 03:00 JST（インデックス更新）
↑
最小5時間のギャップを確保
IndexUpdaterをTweetSchedulerと5時間以上離すことで、インデックス更新中にRAG検索が走ることを防止しています。
AI生成ツイートの除外（モデル崩壊防止）
上記フローのStep 3-4が本システムの重要なポイントです。
RAGシステムで見落としがちなのがモデル崩壊の問題です。システムが生成したツイートを再度インデックスに追加すると、「AI→AI」の循環参照が発生し、参照データの質が徐々に劣化します。
なので、インデックスに追加するのは人間（私）のツイートのみにしています。
Cosmos DBへの履歴保存
TweetGeneratorで投稿成功後、ツイートIDをCosmos DBに保存します。
# src/services/cosmos_service.py
class CosmosService:
def __init__(self):
# Managed Identityで認証
endpoint = os.environ["COSMOS_DB_ENDPOINT"]
credential = DefaultAzureCredential()
self.client = CosmosClient(endpoint, credential)
self.database = self.client.get_database_client(database_name)
self.container = self.database.get_container_client(container_name)
def save_tweet_history(self, tweet_data: Dict[str, Any]) -> str:
"""ツイート履歴をCosmos DBに保存"""
document = {
"id": str(uuid.uuid4()),
"type": "tweet_history",
"tweet_id": tweet_data.get("tweet_id"), # X APIから返されるID
"text": tweet_data.get("text"),
"category": tweet_data.get("category"),
"hashtags": tweet_data.get("hashtags", []),
"status": tweet_data.get("status"), # "success" or "failed"
"created_at": datetime.utcnow().isoformat(),
"rag_source_ids": tweet_data.get("rag_source_ids", []) # 参照した過去ツイートのID
}
self.container.create_item(body=document)
return document["id"]
def get_all_generated_tweet_ids(self) -> set:
"""全てのAI生成ツイートIDを取得（インデックス更新時のフィルタリング用）"""
query = "SELECT c.tweet_id FROM c WHERE c.type = 'tweet_history'"
results = self.container.query_items(
query=query,
enable_cross_partition_query=True
)
return {item['tweet_id'] for item in results if item.get('tweet_id')}
DataCollectorでのフィルタリング
# DataCollector/__init__.py
# Cosmos DBからAI生成ツイートのIDを取得
generated_ids = cosmos_service.get_all_generated_tweet_ids()
# 人間が書いたツイートのみをフィルタリング
human_tweets = [t for t in tweets if str(t.id) not in generated_ids]
logger.info(
f"Total fetched: {len(tweets)}, "
f"Human tweets: {len(human_tweets)}, "
f"AI-generated (excluded): {len(tweets) - len(human_tweets)}"
)
# 人間のツイートのみをインデックスに追加
for tweet in human_tweets:
embedding = openai_service.create_embedding(tweet.text)
# ...インデックス登録処理
設計ポイント:
-
TweetGenerator: 投稿成功時にCosmos DBへ
tweet_idを保存 -
DataCollector: インデックス更新前に
get_all_generated_tweet_ids()で除外リストを取得 - 結果: 常に「人間が書いたツイートのみ」からFew-shot学習
- rag_source_ids: どの過去ツイートを参考にしたかをトレーサビリティとして記録
Function Callingによる文字数検証
問題: GPT-4.1は文字数を正確に数えられない
X（Twitter）API無料版には280カウントの文字数制限があり、日本語は全角1文字=2カウントで計算されます。
しかし、GPT-4に「280文字以内で」と指示しても、正確な文字数を守ることは困難です。トークナイザーと文字数カウントは異なる概念だからです。
解決策: Function Callingによる3層防御
Layer 1: tool_choice="required"
→ 初回は必ず関数呼び出しを強制
Layer 2: システムプロンプトでの絶対遵守事項
→ 「280カウント以内になるまで繰り返す」を明記
Layer 3: truncate_to_tweet_limit()
→ 最終フォールバックとしてプログラム側で切り詰め
図の様にFunction Callingのレスポンスが推論結果に含まれている事例が見られます。改善ポイントです。
Function Callingの実装
# src/services/openai_service.py
def _get_tweet_length_tool_definition(self) -> Dict[str, Any]:
"""文字数カウント関数の定義"""
return {
"type": "function",
"function": {
"name": "count_tweet_length",
"description": "Count tweet length according to X (Twitter) API rules. "
"Full-width characters (Japanese) count as 2, half-width as 1. "
"Maximum allowed is 280 count.",
"parameters": {
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "The tweet text to count"
}
},
"required": ["text"]
}
}
}
検証ループの実装
def generate_tweet(self, category: str, past_tweets: List[str]) -> str:
"""GPT-4 Function Callingによるツイート生成（3層防御）"""
system_prompt = """あなたは、ユーザーの投稿スタイルを学習してツイートを生成するアシスタントです。
【絶対遵守事項】
1. ツイートを生成したら、count_tweet_length 関数を必ず呼び出す
2. 280カウントを超えている場合は、必ず短くして再生成する
3. 再生成後も必ず count_tweet_length で確認する
4. 280カウント以内になるまで繰り返す
この手順を省略することは許されません。"""
tools = [self._get_tweet_length_tool_definition()]
# Function Callingループ（最大5回）
max_iterations = 5
for iteration in range(max_iterations):
# Layer 1: 初回は関数呼び出しを強制
tool_choice = "required" if iteration == 0 else "auto"
response = self.client.chat.completions.create(
model=self.gpt_deployment,
messages=messages,
tools=tools,
tool_choice=tool_choice,
temperature=0.8
)
message = response.choices[0].message
messages.append(message)
# 関数呼び出しがあれば実行
if message.tool_calls:
for tool_call in message.tool_calls:
function_result = self._handle_function_call(
tool_call.function.name,
tool_call.function.arguments
)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": function_result
})
continue # GPTの応答を待つ
# 関数呼び出しなし → 最終回答
if message.content:
generated_text = self._clean_gpt_output(message.content.strip())
# Layer 3: 最終フォールバック
if count_tweet_length(generated_text) > 280:
logger.warning("GPT-4 validation failed. Truncating...")
generated_text = truncate_to_tweet_limit(generated_text)
return generated_text
関数呼び出しのハンドリング
GPT-4がcount_tweet_length関数を呼び出した際の処理実装です。
# src/services/openai_service.py
def _handle_function_call(self, function_name: str, arguments: str) -> str:
"""
GPT-4からの関数呼び出しを処理
Args:
function_name: 呼び出された関数名
arguments: 関数引数（JSON文字列）
Returns:
関数実行結果（JSON文字列）
"""
if function_name == "count_tweet_length":
args = json.loads(arguments)
text = args["text"]
count = count_tweet_length(text)
result = {
"count": count,
"is_valid": count <= 280,
"max_allowed": 280
}
logger.debug(f"count_tweet_length called: {count} chars")
return json.dumps(result, ensure_ascii=False)
raise ValueError(f"Unknown function: {function_name}")
設計ポイント:
-
is_validフラグでGPT-4に検証結果を明確に伝達 - GPT-4.1は
is_valid: falseを受け取ると、自動的に短いバージョンを再生成
GPT-4.1出力のクリーンアップ
GPT-4.1がFunction Calling使用時に混入させる不要な文字列を除去します。
def _clean_gpt_output(self, text: str) -> str:
"""
GPT-4出力から関数実行記録などのアーティファクトを除去
GPT-4はFunction Calling使用時に以下のような文字列を出力に
混入させることがある:
- 「（count_tweet_length 実行済...）」
- 「（ツイートカウント：XXX／280）」
"""
# 「（実行...）」パターンを除去
text = re.sub(r'[（\(][^）\)]*?実行[^）\)]*?[）\)]', '', text)
# 「（count_tweet_length...）」パターンを除去
text = re.sub(r'[（\(]count_tweet_length[^）\)]*?[）\)]', '', text)
# 「（ツイートカウント：XXX／280）」パターンを除去
text = re.sub(r'[（\(][^）\)]*?ツイートカウント[^）\)]*?[）\)]', '', text)
# 単独の「count_tweet_length」を除去
text = re.sub(r'\s*count_tweet_le\w*\s*', ' ', text)
return text.strip()
文字数カウントの実装
# src/utils/text_utils.py
def count_tweet_length(text: str) -> int:
"""X API準拠の文字数カウント"""
count = 0
for char in text:
char_code = ord(char)
if (0x3000 <= char_code <= 0x9FFF or # CJK (日本語)
0xFF00 <= char_code <= 0xFFEF or # 全角形式
0x1F000 <= char_code <= 0x1F9FF): # 絵文字
count += 2
else:
count += 1
return count
def truncate_to_tweet_limit(text: str, max_length: int = 280) -> str:
"""280カウント以内に切り詰め"""
if count_tweet_length(text) <= max_length:
return text
suffix = "..."
suffix_length = count_tweet_length(suffix)
target_length = max_length - suffix_length
truncated = ""
current_length = 0
for char in text:
char_length = 2 if is_full_width(char) else 1
if current_length + char_length > target_length:
break
truncated += char
current_length += char_length
return truncated + suffix
3層防御の設計ポイント:
-
Layer 1:
tool_choice="required"で初回は必ず文字数チェックを強制 - Layer 2: システムプロンプトで「280カウント以内になるまで繰り返す」を明記
-
Layer 3:
truncate_to_tweet_limit()でプログラム側の最終フォールバック - 最大5回のループで収束を保証
セキュリティ設計
フィッシング詐欺にかかっておいて何ですが、クラウドのセキュリティ設定は重要です。ローカルに認証情報を持たないことが大切です。
Managed Identityによるシークレットレス認証
Azure Functions には System-assigned Managed Identity を有効化し、各サービスへはRBACで認証します。
| サービス | 割り当てRBACロール |
|---|---|
| Key Vault | Key Vault Secrets User |
| Cosmos DB | Cosmos DB Data Contributor |
| AI Search | Search Service Contributor |
| Azure OpenAI | Cognitive Services User |
Key Vault RBAC設計
// infrastructure/bicep/keyvault.bicep
resource keyVault 'Microsoft.KeyVault/vaults@2023-07-01' = {
name: keyVaultName
location: location
properties: {
// RBAC-only authorization (best practice)
enableRbacAuthorization: true
// Soft delete (誤削除対策)
enableSoftDelete: true
softDeleteRetentionInDays: 90
// Purge protection (完全削除防止)
enablePurgeProtection: true
// Access Policyは使用しない（RBACのみ）
accessPolicies: []
}
}
RBACを選ぶ理由:
- Access Policyより細かい権限制御が可能
- Azure AD統合による監査ログ
- Infrastructure as Codeとの親和性
KeyVaultServiceの実装
# src/services/keyvault_service.py
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
class KeyVaultService:
"""
DefaultAzureCredential を使用して、以下の順序で認証を試みる:
1. 環境変数（ローカル開発）
2. マネージドID（Azure Functions）
3. Azure CLI（ローカル開発）
"""
def __init__(self):
key_vault_url = os.environ["KEY_VAULT_URL"]
credential = DefaultAzureCredential()
self.client = SecretClient(vault_url=key_vault_url, credential=credential)
def get_secret(self, secret_name: str) -> str:
return self.client.get_secret(secret_name).value
DefaultAzureCredentialの利点:
- 環境に応じて自動的に認証方法を選択
- ローカル開発時はAzure CLI認証
- Azure上ではManaged Identity
- コードの変更なしで両方に対応
Key Vaultに保存するシークレット
| シークレット名 | 用途 |
|---|---|
twitter-bearer-token |
X API v2認証 |
twitter-api-key / twitter-api-secret
|
OAuth 1.0a |
twitter-access-token / twitter-access-secret
|
OAuth 1.0a |
openai-api-key |
Azure OpenAI認証 |
search-admin-key |
AI Search管理操作 |
cosmos-connection-string |
Cosmos DB接続 |
TweetGeneratorの完全な処理フロー
TweetGeneratorは、以下の7ステップでツイートを自動生成・投稿します。
# TweetGenerator/__init__.py
def main(req: func.HttpRequest) -> func.HttpResponse:
logger.info('TweetGenerator function triggered')
# 1. カテゴリをランダム選択
categories = os.environ.get("TWEET_CATEGORIES", "技術,日常,趣味").split(",")
category = random.choice(categories)
# サービス初期化
twitter_service = TwitterService()
search_service = SearchService()
openai_service = OpenAIService()
cosmos_service = CosmosService()
# 2. 検索クエリ生成（GPT-4）
query = openai_service.generate_search_query(category)
logger.info(f"Generated search query: {query}")
# 3. クエリの埋め込みベクトル生成
query_embedding = openai_service.create_embedding(query)
# 4. RAG検索（Semantic Ranker使用）
search_results = search_service.vector_search_with_semantic_ranker(
query_text=query, # セマンティックランカー用
query_vector=query_embedding, # ベクトル検索用
top=5,
filter_expression=f"category eq '{category}'"
)
# 5. Few-shot学習でツイート生成
past_tweets = [result["text"] for result in search_results]
generated_text = openai_service.generate_tweet(
category=category,
past_tweets=past_tweets
)
# 6. ハッシュタグ追加（過去ツイートから抽出 or GPT-4生成）
hashtags = extract_common_hashtags(search_results, max_count=3)
if not hashtags:
hashtags = openai_service.generate_hashtags(category=category, text=generated_text)
if hashtags:
hashtag_text = " " + " ".join([f"#{tag}" for tag in hashtags])
if count_tweet_length(generated_text) + count_tweet_length(hashtag_text) <= 280:
generated_text += hashtag_text
# 7. X APIに投稿（リトライ付き）
tweet_id = post_tweet_with_retry(twitter_service, generated_text)
# 8. システム情報リプライを投稿（透明性確保）
if os.environ.get("ENABLE_SYSTEM_INFO_REPLY", "true").lower() == "true":
reply_text = f"""🤖 このツイートはRAG（検索拡張生成）を使って過去のツイートを基に自動生成されています
📊 参照元: {len(search_results)}件の過去ツイート
🏷️ カテゴリ: {category}
🔍 検索クエリ: {query}"""
try:
twitter_service.post_reply(tweet_id, reply_text)
except Exception as e:
logger.warning(f"Failed to post system info reply: {e}")
# 9. Cosmos DBに履歴保存（AI生成ツイートとしてマーク）
cosmos_service.save_tweet_history({
"tweet_id": tweet_id,
"text": generated_text,
"category": category,
"hashtags": hashtags,
"status": "success",
"rag_source_ids": [r["id"] for r in search_results]
})
return func.HttpResponse(json.dumps({"tweet_id": tweet_id}), status_code=200)
処理フローのポイント:
-
Step 4:
vector_search_with_semantic_rankerでベクトル検索とセマンティック再ランキングを一度に実行 - Step 5: RAGで取得した過去ツイートをFew-shot学習の参考例として使用
- Step 6: ハッシュタグは過去ツイートからの頻度抽出を優先、無ければGPT-4で生成
- Step 8: 生成AIによる自動投稿であることを明示するリプライを投稿（透明性確保）
- Step 9: 投稿したツイートIDをCosmos DBに保存（後のインデックス更新時に除外するため）
システム情報リプライ機能
生成AIによるツイートであることをフォロワーに明示するため、投稿後に自動でリプライを追加します。
リプライ例:
🤖 このツイートはRAG（検索拡張生成）を使って過去のツイートを基に自動生成されています
📊 参照元: 5件の過去ツイート
🏷️ カテゴリ: DX
🔍 検索クエリ: 企業のDX推進事例やデジタル化の取り組み
各項目の説明:
| 項目 | 説明 |
|---|---|
| 📊 参照元 | RAG検索でヒットした過去ツイートの件数。Few-shot学習の参考例として使用 |
| 🏷️ カテゴリ | ランダム選択されたツイートカテゴリ（技術、日常、趣味など） |
| 🔍 検索クエリ | GPT-4.1が生成した検索クエリ。このクエリでベクトル検索+セマンティック検索を実行 |
特にステマって訳でもないので、外しても良いかなと思っています。
設計ポイント:
-
ENABLE_SYSTEM_INFO_REPLY環境変数でON/OFF切り替え可能（デフォルト: ON） - リプライ失敗時は警告ログのみ（メインツイートの投稿は成功扱い）
- 参照元件数、カテゴリ、検索クエリを表示してRAGの動作を可視化
Timer Triggerの設定
Azure Functions の Timer Trigger は CRON 式で設定します。追加のAzureサービスは不要です。
// TweetScheduler/function.json
{
"scriptFile": "__init__.py",
"bindings": [
{
"name": "mytimer",
"type": "timerTrigger",
"direction": "in",
"schedule": "0 0 23,3,11 * * *"
}
]
}
スケジュール設定:
-
0 0 23,3,11 * * *= UTC 23:00, 03:00, 11:00（JST 08:00, 12:00, 20:00） - 形式:
{秒} {分} {時} {日} {月} {曜日}
# TweetScheduler/__init__.py
import azure.functions as func
import requests
def main(mytimer: func.TimerRequest) -> None:
# TweetGeneratorのHTTPエンドポイントを呼び出し
response = requests.post(
f"{os.environ['FUNCTION_APP_URL']}/api/TweetGenerator",
headers={"x-functions-key": os.environ["FUNCTION_KEY"]}
)
logger.info(f"TweetGenerator response: {response.status_code}")
その他の工夫
Few-shot学習によるスタイル模倣
RAG検索で取得した過去ツイートをプロンプトに含めることで、GPT-4.1が文体を学習します。
system_prompt = """過去の投稿を参考にして、同じトーン・スタイルで新しいツイートを生成してください。
要件:
- 過去の投稿と同じ文体（です・ます調、だ・である調など）
- 絵文字の使用パターンを踏襲
- 自然で違和感のない内容"""
past_tweets_text = "\n".join([f"- {tweet}" for tweet in past_tweets])
user_prompt = f"""参考にする過去の投稿:
{past_tweets_text}
上記を参考に新しいツイートを生成してください。"""
指数バックオフによるリトライ
外部APIコール（OpenAI、X API）には指数バックオフを適用しています。
# src/utils/retry.py
import time
import functools
from typing import Callable, Any
def retry_with_exponential_backoff(
max_retries: int = 3,
initial_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0
):
"""指数バックオフ付きリトライデコレータ"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
delay = initial_delay
last_exception = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < max_retries:
logger.warning(
f"Attempt {attempt + 1} failed: {str(e)}. "
f"Retrying in {delay} seconds..."
)
time.sleep(delay)
delay = min(delay * exponential_base, max_delay)
else:
logger.error(f"All {max_retries + 1} attempts failed.")
raise last_exception
return wrapper
return decorator
使用例:
# OpenAI API呼び出し（2秒 → 4秒 → 8秒の間隔でリトライ）
@retry_with_exponential_backoff(max_retries=3, initial_delay=2.0)
def create_embedding(self, text: str) -> List[float]:
return self.client.embeddings.create(model=self.embedding_deployment, input=text)
# X API投稿（1秒 → 2秒 → 4秒の間隔でリトライ）
@retry_with_exponential_backoff(max_retries=3, initial_delay=1.0)
def post_tweet_with_retry(twitter_service, text):
return twitter_service.post_tweet(text)
設計ポイント:
-
exponential_base=2.0: 遅延が2倍ずつ増加 -
max_delay=60.0: 遅延の上限を60秒に制限 - レートリミットや一時的な障害に対応
IndexUpdaterのスケジュール設計
TweetScheduler（ツイート生成）とIndexUpdater（インデックス更新）が同時実行されないよう、時間をずらしています。
- TweetScheduler: 08:00, 12:00, 20:00 JST
- IndexUpdater: 03:00 JST（次の日）
最小5時間のギャップを確保し、インデックス更新中のRAG検索を防止。
この様にしたのは、ツイート生成中にインデックス更新が動き、Azure Functionsへの負荷が高まることを避けたかったからです。
コストと運用
使用サブスクリプション
本システムはVisual Studio Enterprise サブスクリプションの月額クレジット（$150 ≒ 約17,000円）の範囲内で運用することを想定して設計しています。
Visual Studio Enterprise サブスクリプション特典
- 月額 $150 の Azure クレジット（個人開発・検証用途）
- 本番環境での使用は不可
- クレジットは翌月に繰り越し不可
月額コスト内訳（約25,000円）
AI Search S1: 18,000円（72%）
Azure OpenAI: 4,000円（16%）
Cosmos DB: 500円（ 2%）
Functions: 500円（ 2%）
Key Vault: 500円（ 2%）
Storage/AppInsights: 1,500円（ 6%）
AI Search S1がコストの大半を占めます。Semantic Rankerが不要であれば、Basic SKUで約3,000円に削減可能です。
VS Enterpriseクレジット内での運用制約
月額約25,000円はクレジット（約17,000円）を超過するため、現在の構成では月の半分程度しか稼働できません（クレジット上限到達のため自動停止）。
クレジット内でフル稼働させたい場合の選択肢：
- AI Search を Basic SKU に変更: 約10,000円に削減（Semantic Ranker非対応）
- Cosmos DB を Table Storage に置換: 約200円に削減
- 開発・検証時のみ起動: 使用しない時間帯はAI Searchを停止（手動操作が必要）
CLIベースのモニタリング
Azure Portalを使わず、CLIで完結する運用を設計しました。
# 最近のエラー確認
az monitor app-insights query \
--app appi-tweet-auto-posting-dev \
--analytics-query "exceptions | where timestamp > ago(24h) | take 20"
# ツイート投稿成功率
az monitor app-insights query \
--app appi-tweet-auto-posting-dev \
--analytics-query "customEvents
| where name == 'TweetPosted'
| summarize SuccessRate = countif(customDimensions.status == 'success') * 100.0 / count()"
まとめ
本記事では、Azure上にRAGベースの自動ツイート生成システムを構築する際の3つの技術ポイントを解説しました。
- Semantic Ranker: ベクトル検索 + 意味的再ランキングの2段階検索で、真に関連性の高いツイートを取得
- Function Calling: GPT-4に文字数カウント関数を実行させる3層防御で、確実に280カウント以内を実現
- Managed Identity + RBAC: シークレットをコードに含めない、現代的なAzureセキュリティ設計
今後の改善点
- GitHub ActionsによるCI/CDパイプライン構築
- GithubへソースをPUSHしたら、自動的にAzure Web Appへデプロイされる
- 画像の含めた投稿
- ツイート文からNano Banana Proで画像生成し、ツイートへ付加
- VNet統合によるネットワークセキュリティ強化
- カテゴリ自動分類機能の追加
データラーニングギルドについて
データラーニングギルド は、株式会社データラーニングが運営する、
データサイエンスを中心とした学習者・現役データサイエンティスト・エンジニアのためのコミュニティです。
学びの共有・キャリア形成・横のつながりを大切にし、
勉強会、LT会、技術相談、キャリア支援、案件紹介など、
「データ領域で挑戦したい人を応援する活動」を幅広く行っています。
初心者から実務者まで、誰もが成長できる場づくりを目指しています。
興味のある方はコミュニティに参加してみてください！
🔗公式サイト：https://data-learning.com/guild
タグ: Azure, RAG, OpenAI, SemanticSearch, Python, AzureFunctions, ServerlessArchitecture