Jinja2で始めるAIプロンプトテンプレートエンジン設計：動的生成の実用的アプローチ

AIアプリケーションを本番環境にデプロイする際、最大の問題の一つがプロンプトの管理です。私のプロジェクトでは当初、プロンプトを文字列結合で構築していましたが、保守性が著しく低下し、多くのバグを引き起こしました。

本稿では、PythonのJinja2テンプレートエンジンを活用したプロンプトテンプレートエンジンの設計と実装を解説します。

実体験から始まった課題

ある重要なリリース前夕、私は以下のようなエラーに直面しました：

ConnectionError: HTTPSConnectionPool(host='api.openai.com', port=443): 
Max retries exceeded with url: /v1/chat/completions (Caused by 
NewConnectionError('<urllib3.connection.HTTPSConnection object at 
0x7f8a2c1a3d90>: Failed to establish a new connection: [Errno 110] 
Connection timed out'))

requests.exceptions.ConnectionError: connection error

このタイムアウト問題と並行して、プロンプトのバージョン管理も崩壊寸前でした。複数のプロンプトがハードコードされ、条件分岐が複雑化していたのです。

Jinja2テンプレートエンジンの基本設計

Jinja2はPython製の高速で柔軟なテンプレートエンジンです。プロンプトの変数埋め込み、条件分岐、ループ処理を直感的に記述できます。

"""
Jinja2 プロンプトテンプレートエンジン - 基本実装
"""
from jinja2 import Environment, BaseLoader, TemplateNotFound
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from enum import Enum
import hashlib
import json


class ModelProvider(Enum):
    """対応モデルプロバイダー"""
    HOLYSHEEP = "holysheep"
    OPENAI = "openai"
    ANTHROPIC = "anthropic"


@dataclass
class PromptTemplate:
    """プロンプトテンプレート定義"""
    name: str
    template: str
    description: str = ""
    variables: List[str] = field(default_factory=list)
    model: str = "gpt-4o-mini"
    temperature: float = 0.7
    max_tokens: int = 2048


@dataclass
class RenderedPrompt:
    """レンダリング済みプロンプト"""
    template_name: str
    rendered_content: str
    variables: Dict[str, Any]
    model: str
    metadata: Dict[str, Any] = field(default_factory=dict)


class PromptTemplateEngine:
    """Jinja2驅動のプロンプトテンプレートエンジン"""
    
    def __init__(self, cache_enabled: bool = True):
        self.env = Environment(loader=BaseLoader())
        self.env.globals['len'] = len
        self.env.globals['str'] = str
        self.templates: Dict[str, PromptTemplate] = {}
        self._cache: Dict[str, str] = {} if cache_enabled else None
    
    def register_template(
        self,
        name: str,
        template_str: str,
        description: str = "",
        model: str = "gpt-4o-mini",
        **kwargs
    ) -> None:
        """テンプレートを登録"""
        from jinja2 import meta
        ast = self.env.parse(template_str)
        variables = list(meta.find_undeclared_variables(ast))
        
        self.templates[name] = PromptTemplate(
            name=name,
            template=template_str,
            description=description,
            variables=variables,
            model=model,
            **kwargs
        )
    
    def render(
        self,
        template_name: str,
        variables: Dict[str, Any],
        use_cache: bool = True
    ) -> RenderedPrompt:
        """テンプレートをレンダリング"""
        if template_name not in self.templates:
            raise TemplateNotFound(f"Template '{template_name}' not found")
        
        template = self.templates[template_name]
        
        # キャッシュキーの生成
        if use_cache and self._cache is not None:
            cache_key = self._generate_cache_key(template_name, variables)
            if cache_key in self._cache:
                cached = self._cache[cache_key]
                return RenderedPrompt(
                    template_name=template_name,
                    rendered_content=cached,
                    variables=variables,
                    model=template.model,
                    metadata={"cached": True, "cache_key": cache_key}
                )
        
        # Jinja2でレンダリング
        jinja_template = self.env.from_string(template.template)
        rendered = jinja_template.render(**variables)
        
        # キャッシュに保存
        if use_cache and self._cache is not None:
            self._cache[cache_key] = rendered
        
        return RenderedPrompt(
            template_name=template_name,
            rendered_content=rendered,
            variables=variables,
            model=template.model,
            metadata={"cached": False}
        )
    
    def _generate_cache_key(self, template_name: str, variables: Dict) -> str:
        """キャッシュキーを生成"""
        content = json.dumps({"template": template_name, "vars": variables}, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()
    
    def clear_cache(self) -> int:
        """キャッシュをクリア"""
        if self._cache is not None:
            count = len(self._cache)
            self._cache.clear()
            return count
        return 0


使用例
if __name__ == "__main__":
    engine = PromptTemplateEngine(cache_enabled=True)
    
    # システムプロンプトテンプレートの登録
    engine.register_template(
        name="customer_support",
        template_str="""あなたは高度なカスタマーサポートAIです。
{% if customer_tier == "premium" %}
顧客タイプ: Premium会員様
対応方針: 最高水準的服务を提供し、優先的に問題を解決します。
{% elif customer_tier == "standard" %}
顧客タイプ: 一般会員様
対応方針: 迅速かつ正確なサポートを提供します。
{% else %}
顧客タイプ: ゲスト
対応方針: 基本情報を提供し、必要に応じて登録案内を行います。
{% endif %}

対応言語: {{ language | default("日本語") }}
会話番号: {{ ticket_id }}

customer_message: {{ message }}
""",
        description="カスタマーサポート対応プロンプト",
        model="gpt-4o-mini"
    )
    
    # 変数をレンダリング
    result = engine.render(
        template_name="customer_support",
        variables={
            "customer_tier": "premium",
            "language": "日本語",
            "ticket_id": "TKT-2024-001234",
            "message": "請求書に誤りがあります。確認お願いします。"
        }
    )
    
    print("=== レンダリング結果 ===")
    print(result.rendered_content)
    print(f"\n使用モデル: {result.model}")
    print(f"キャッシュ: {result.metadata.get('cached', False)}")

HolySheep AI APIとの連携実装

私のプロジェクトではHolySheep AIを採用しています。理由は明白です：レートが¥1=$1という圧倒的なコスト効率（公式¥7.3=$1比85%節約）で、WeChat PayやAlipayにも対応しており、レイテンシも<50msという高速応答を実現しています。

"""
HolySheep AI API驅動のLLM呼び出しクライアント
"""
import requests
from typing import Optional, Dict, Any, List, Union
from dataclasses import dataclass
import time


@dataclass
class LLMResponse:
    """LLM応答"""
    content: str
    model: str
    usage: Dict[str, int]
    latency_ms: float
    finish_reason: str


class HolySheepAIClient:
    """HolySheep AI APIクライアント（OpenAI互換）"""
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1"
    ):
        self.api_key = api_key
        self.base_url = base_url.rstrip('/')
        self.chat_endpoint = f"{self.base_url}/chat/completions"
        self._session = requests.Session()
        self._session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def chat_completion(
        self,
        messages: List[Dict[str, str]],
        model: str = "gpt-4o-mini",
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
        timeout: float = 30.0
    ) -> LLMResponse:
        """チャット補完リクエストを送信"""
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature
        }
        
        if max_tokens:
            payload["max_tokens"] = max_tokens
        
        start_time = time.perf_counter()
        
        try:
            response = self._session.post(
                self.chat_endpoint,
                json=payload,
                timeout=timeout
            )
            
            # ステータスコードチェック
            response.raise_for_status()
            
            data = response.json()
            latency = (time.perf_counter() - start_time) * 1000
            
            return LLMResponse(
                content=data['choices'][0]['message']['content'],
                model=data.get('model', model),
                usage=data.get('usage', {}),
                latency_ms=latency,
                finish_reason=data['choices'][0].get('finish_reason', 'stop')
            )
            
        except requests.exceptions.Timeout:
            raise TimeoutError(f"リクエストが{timeout}秒でタイムアウトしました")
        except requests.exceptions.HTTPError as e:
            status = e.response.status_code
            if status == 401:
                raise PermissionError("APIキーが無効です。HolySheep AIで新しいキーを発行してください")
            elif status == 429:
                raise RuntimeError("レート制限に達しました。しばらくお待ちください")
            elif status == 500:
                raise RuntimeError("サーバーエラーが発生しました。再試行してください")
            else:
                raise RuntimeError(f"HTTPエラー {status}: {str(e)}")
        except requests.exceptions.ConnectionError as e:
            raise ConnectionError(f"接続エラー: {str(e)}。ネットワーク状態を確認してください")


def create_prompt_workflow():
    """プロンプトテンプレート + LLM呼び出しのワークフロー"""
    
    # 初期化
    api_key = "YOUR_HOLYSHEEP_API_KEY"  # HolySheep AIのAPIキーに置換
    client = HolySheepAIClient(api_key=api_key)
    engine = PromptTemplateEngine(cache_enabled=True)
    
    # テンプレートの登録
    engine.register_template(
        name="code_review",
        template_str="""【コードレビュー指示】

対象リポジトリ: {{ repository_name }}
ブランチ: {{ branch_name }}
PR番号: #{{ pr_number }}

{% if focus_areas %}
確認重点領域:
{% for area in focus_areas %}
  - {{ area }}
{% endfor %}
{% else %}
確認重点領域: 全般的な品質チェック
{% endif %}

レビュー対象コード:
```{{ language }}
{{ code_snippet }}

あなたは経験豊富なコードレビューアーです。以下の観点を考慮してレビューを実施:
1. セキュリティ上の脆弱性
2. パフォーマンス問題
3. コードの可読性と保守性
4. ベストプラクティスとの整合性

JSON形式でレビュー結果を返答:
{
  "summary": "レビュー概要",
  "issues": [
    {
      "severity": "critical|major|minor",
      "line": 行番号,
      "description": "問題の説明",
      "suggestion": "修正提案"
    }
  ],
  "approval": true|false
}
""",
        description="コードレビュー用プロンプト",
        model="gpt-4o-mini"
    )
    
    return client, engine


if __name__ == "__main__":
    client, engine = create_prompt_workflow()
    
    # プロンプトをレンダリング
    rendered = engine.render(
        template_name="code_review",
        variables={
            "repository_name": "holy-service/core-api",
            "branch_name": "feature/user-auth",
            "pr_number": 234,
            "language": "python",
            "code_snippet": '''def authenticate_user(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return {"user_id": payload["sub"], "valid": True}
    except jwt.ExpiredSignatureError:
        return {"error": "Token expired", "valid": False}''',
            "focus_areas": ["セキュリティ", "認証フロー"]
        }
    )
    
    print("=== レンダリング済みプロンプト ===")
    print(rendered.rendered_content)
    
    # API呼び出し（コメントアウト状态下）
    # messages = [{"role": "user", "content": rendered.rendered_content}]
    # response = client.chat_completion(messages, model=rendered.model)
    # print(f"\n応答: {response.content}")
    # print(f"レイテンシ: {response.latency_ms:.2f}ms")
    # print(f"コスト確認: ${response.usage['total_tokens'] / 1_000_000 * 0.15:.6f}")

Advanced: テンプレートヘルパーとフィルター

より複雑なプロンプトに対応するため、Jinja2のカスタムフィルターとヘルパー関数を追加します。

"""
Jinja2 カスタムフィルターとヘルパー関数
"""
from jinja2 import Environment, BaseLoader, pass_context
from datetime import datetime
import json
import re


def create_advanced_engine() -> Environment:
    """高度なJinja2エンジンを作成"""
    env = Environment(loader=BaseLoader())
    
    # カスタムフィルターの登録
    @env.filter(name='truncate_words')
    def truncate_words(text: str, num_words: int = 50) -> str:
        """指定語数でテキストを切り詰め"""
        words = text.split()
        if len(words) <= num_words:
            return text
        return ' '.join(words[:num_words]) + '...'
    
    @env.filter(name='json_minify')
    def json_minify(data: Any) -> str:
        """JSONを圧縮"""
        return json.dumps(data, ensure_ascii=False, separators=(',', ':'))
    
    @env.filter(name='to_code_block')
    def to_code_block(text: str, language: str = "") -> str:
        """Markdownコードブロックに変換"""
        return f"
{language}\n{text}\n```"
    
    @env.filter(name='list_to_text')
    def list_to_text(items: list, sep: str = "、") -> str:
        """リストを日本語区切りテキストに変換"""
        if not items:
            return ""
        return sep.join(str(item) for item in items)
    
    @env.filter(name='number_to_kanji')
    def number_to_kanji(num: int) -> str:
        """数字を漢数字に変換"""
        kanji_nums = ["零", "一", "二", "三", "四", "五", "六", "七", "八", "九", "十"]
        if 0 <= num <= 10:
            return kanji_nums[num]
        return str(num)
    
    # カスタムコンテキスト関数の登録
    @pass_context
    @env.global_function
    def current_date(ctx):
        """現在日付を返す"""
        return datetime.now().strftime("%Y年%m月%d日")
    
    @pass_context
    @env.global_function
    def word_count(ctx, text: str) -> int:
        """文字数をカウント"""
        return len(text.strip())
    
    @pass_context
    @env.global_function
    def repeat_text(ctx, text: str, times: int = 2) -> str:
        """テキストを指定回数繰り返す"""
        return text * times
    
    return env


使用例
if __name__ == "__main__":
    env = create_advanced_engine()
    
    template_str = """
文章サマリー生成 - {{ current_date() }}

対象テキスト:
{{ original_text | to_code_block("markdown") }}

文字数: {{ original_text | word_count }}

{% if keywords %}
キーワード: {{ keywords | list_to_text }}
{% endif %}

生成されるサマリー:
{{ summary | truncate_words(30) }}

{% if metadata %}
メタデータ: {{ metadata | json_minify }}
{% endif %}
"""
    
    template = env.from_string(template_str)
    result = template.render(
        original_text="この文章はプロンプトエンジニアリングにおけるJinja2テンプレートの応用例を示しています。カスタムフィルターを使用することで、テキスト処理をより柔軟に行うことができます。",
        summary="Jinja2のカスタムフィルターを活用すれば、テキスト処理が驚くほど簡単になります。本稿では具体的な実装例を紹介します。",
        keywords=["Jinja2", "テンプレート", "プロンプト", "フィルター"],
        metadata={"version": "1.0", "author": "HolySheep AI Team"}
    )
    
    print(result)

料金比較とコスト最適化

私のプロジェクトでHolySheep AIを採用した決め手は料金体系です。2026年現在のOutput価格(/MTok)を比較すると：

• GPT-4.1: $8.00（HolySheep利用時：同等品質を大幅に低コストで）
• Claude Sonnet 4.5: $15.00（高性能だがコスト高）
• Gemini 2.5 Flash: $2.50（コストパフォーマンス優れる）
• DeepSeek V3.2: $0.42（最安値、需要増加中）

DeepSeek V3.2の$0.42/MTokという価格は、大量処理が必要な場合に極めて経済的です。登録すると無料クレジットが付与されるため、実質的なテスト環境としても優れています。

# 症状
PermissionError: APIキーが無効です。HolySheep AIで新しいキーを発行してください

原因
- APIキーの有効期限切れ
- キーの入力ミス
- 環境変数の読み込み失敗

解決策
import os

api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
    # 環境変数から読み込み
    api_key = "YOUR_HOLYSHEEP_API_KEY"

client = HolySheepAIClient(api_key=api_key)

キーの有効性を確認
try:
    test_response = client.chat_completion(
        messages=[{"role": "user", "content": "test"}],
        model="gpt-4o-mini",
        max_tokens=5
    )
    print("API認証成功")
except PermissionError as e:
    print(f"認証エラー: {e}")
    # HolySheep AIダッシュボードで新しいキーを発行

# 症状
ConnectionError: Failed to establish a new connection: [Errno 110] Connection timed out

原因
- ネットワーク不安定
- ファイアウォールによるブロック
- DNS解決失敗
- タイムアウト設定が短すぎる

解決策
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import socket

def create_resilient_client(api_key: str) -> HolySheepAIClient:
    """リトライ機構付きクライアントを作成"""
    
    # タイムアウトを延長
    client = HolySheepAIClient(api_key=api_key)
    
    # リトライ策略を設定
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["POST"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    client._session.mount("https://", adapter)
    client._session.mount("http://", adapter)
    
    return client

使用
client = create_resilient_client("YOUR_HOLYSHEEP_API_KEY")

try:
    response = client.chat_completion(
        messages=[{"role": "user", "content": "Hello"}],
        timeout=60.0  # タイムアウト60秒
    )
except ConnectionError:
    # 代替エンドポイントにフォールバック
    print("接続エラー: ネットワーク状態を確認してください")

# 症状
jinja2.exceptions.TemplateNotFound: Template 'customer_support' not found

原因
- テンプレート登録忘れ
- 名前タイポ
- テンプレートの読み込み順序問題

解決策
class SafePromptEngine(PromptTemplateEngine):
    """安全なレンダリングを提供する拡張エンジン"""
    
    def safe_render(
        self,
        template_name: str,
        variables: Dict[str, Any],
        raise_on_missing: bool = True
    ) -> Optional[RenderedPrompt]:
        """テンプレート存在を碓認してからレンダリング"""
        
        # 登録済みテンプレート一覧
        available = list(self.templates.keys())
        
        if template_name not in self.templates:
            if raise_on_missing:
                available_str = ", ".join(available) if available else "なし"
                raise ValueError(
                    f"テンプレート '{template_name}' が見つかりません。\n"
                    f"利用可能なテンプレート: {available_str}"
                )
            return None
        
        return self.render(template_name, variables)

使用例
engine = SafePromptEngine()

登録確認
print(f"登録済みテンプレート: {list(engine.templates.keys())}")

安全レンダリング
try:
    result = engine.safe_render(
        template_name="customer_support",
        variables={"message": "Hello"}
    )
except ValueError as e:
    print(f"エラー: {e}")
    # フォールバックテンプレートを使用
    result = engine.safe_render(
        template_name="fallback",
        variables={"message": "Hello"}
    )

# 症状
RuntimeError: レート制限に達しました。しばらくお待ちください

原因
- 短時間的大量リクエスト
- プランの制限超過
- 瞬間的なトラフィック増加

解決策
import time
from threading import Lock

class RateLimitedClient:
    """レート制限対応のラッパークラス"""
    
    def __init__(self, api_key: str, requests_per_minute: int = 60):
        self.client = HolySheepAIClient(api_key=api_key)
        self.rpm = requests_per_minute
        self.min_interval = 60.0 / requests_per_minute
        self.last_request_time = 0.0
        self._lock = Lock()
    
    def chat_completion(self, *args, **kwargs) -> LLMResponse:
        """レート制限を考慮したリクエスト"""
        
        with self._lock:
            current_time = time.time()
            elapsed = current_time - self.last_request_time
            
            # 間隔を確保
            if elapsed < self.min_interval:
                wait_time = self.min_interval - elapsed
                print(f"レート制限対策: {wait_time:.2f}秒待機")
                time.sleep(wait_time)
            
            self.last_request_time = time.time()
        
        # 本来のリクエスト
        try:
            return self.client.chat_completion(*args, **kwargs)
        except RuntimeError as e:
            if "レート制限" in str(e):
                # 指数バックオフでリトライ
                for attempt in range(3):
                    wait = 2 ** attempt
                    print(f"リトライ {attempt + 1}/3: {wait}秒待機")
                    time.sleep(wait)
                    try:
                        return self.client.chat_completion(*args, **kwargs)
                    except RuntimeError:
                        continue
                raise RuntimeError("レート制限超過: リトライ上限に達しました")
            raise

使用
limited_client = RateLimitedClient(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    requests_per_minute=30  # 1分あたり30リクエスト
)

まずは今すぐ登録して無料クレジットを試用し、テンプレートエンジンの効果を体験してみてください。