🤖 LLM アプリケーションパターン

LLM アプリケーション構築のための本番環境対応パターン。Dify と業界のベストプラクティスに触発されています。

このスキルを使う場合

このスキルは以下の場合に使用してください：

• LLM を活用したアプリケーションを設計する
• RAG（Retrieval-Augmented Generation）を実装する
• ツールを備えた AI エージェントを構築する
• LLMOps 監視を設定する
• エージェント アーキテクチャを選択する

---

1. RAG パイプライン アーキテクチャ

概要

RAG（Retrieval-Augmented Generation）は、LLM の応答をあなたのデータに基づかせます。

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Ingest    │────▶│   Retrieve  │────▶│   Generate  │
│  Documents  │     │   Context   │     │   Response  │
└─────────────┘     └─────────────┘     └─────────────┘
      │                   │                   │
      ▼                   ▼                   ▼
 ┌─────────┐       ┌───────────┐       ┌───────────┐
 │ Chunking│       │  Vector   │       │    LLM    │
 │Embedding│       │  Search   │       │  + Context│
 └─────────┘       └───────────┘       └───────────┘

1.1 ドキュメント取り込み

# チャンキング戦略
class ChunkingStrategy:
    # 固定サイズのチャンク（シンプルだが文脈が途切れる可能性あり）
    FIXED_SIZE = "fixed_size"  # 例：512 トークン

    # セマンティック チャンキング（意味を保持）
    SEMANTIC = "semantic"      # 段落 / セクションで分割

    # 再帰的分割（複数の区切り文字を試す）
    RECURSIVE = "recursive"    # ["\n\n", "\n", " ", ""]

    # ドキュメント認識（構造を尊重）
    DOCUMENT_AWARE = "document_aware"  # ヘッダー、リストなど

# 推奨される設定
CHUNK_CONFIG = {
    "chunk_size": 512,       # トークン
    "chunk_overlap": 50,     # チャンク間のトークンオーバーラップ
    "separators": ["\n\n", "\n", ". ", " "],
}

1.2 埋め込み & ストレージ

# ベクトル データベースの選択
VECTOR_DB_OPTIONS = {
    "pinecone": {
        "use_case": "本番環境、マネージドサービス",
        "scale": "数十億のベクトル",
        "features": ["ハイブリッド検索", "メタデータフィルタリング"]
    },
    "weaviate": {
        "use_case": "自己ホスト、マルチモーダル",
        "scale": "数百万のベクトル",
        "features": ["GraphQL API", "モジュール"]
    },
    "chromadb": {
        "use_case": "開発、プロトタイピング",
        "scale": "数千のベクトル",
        "features": ["シンプル API", "インメモリオプション"]
    },
    "pgvector": {
        "use_case": "既存の Postgres インフラストラクチャ",
        "scale": "数百万のベクトル",
        "features": ["SQL 統合", "ACID コンプライアンス"]
    }
}

# 埋め込みモデルの選択
EMBEDDING_MODELS = {
    "openai/text-embedding-3-small": {
        "dimensions": 1536,
        "cost": "$0.02/100万トークン",
        "quality": "ほとんどのユースケースに適切"
    },
    "openai/text-embedding-3-large": {
        "dimensions": 3072,
        "cost": "$0.13/100万トークン",
        "quality": "複雑なクエリに最適"
    },
    "local/bge-large": {
        "dimensions": 1024,
        "cost": "無料（計算のみ）",
        "quality": "OpenAI small に相当"
    }
}

1.3 検索戦略

# 基本的なセマンティック検索
def semantic_search(query: str, top_k: int = 5):
    query_embedding = embed(query)
    results = vector_db.similarity_search(
        query_embedding,
        top_k=top_k
    )
    return results

# ハイブリッド検索（セマンティック + キーワード）
def hybrid_search(query: str, top_k: int = 5, alpha: float = 0.5):
    """
    alpha=1.0: 純セマンティック
    alpha=0.0: 純キーワード（BM25）
    alpha=0.5: バランス型
    """
    semantic_results = vector_db.similarity_search(query)
    keyword_results = bm25_search(query)

    # 相互ランク融合
    return rrf_merge(semantic_results, keyword_results, alpha)

# 複数クエリ検索
def multi_query_retrieval(query: str):
    """複数のクエリバリエーションを生成して再現率を向上させる"""
    queries = llm.generate_query_variations(query, n=3)
    all_results = []
    for q in queries:
        all_results.extend(semantic_search(q))
    return deduplicate(all_results)

# 文脈圧縮
def compressed_retrieval(query: str):
    """検索してから関連部分のみに圧縮"""
    docs = semantic_search(query, top_k=10)
    compressed = llm.extract_relevant_parts(docs, query)
    return compressed

1.4 文脈を用いた生成

RAG_PROMPT_TEMPLATE = """
以下の文脈に基づいてのみ、ユーザーの質問に答えてください。
文脈に十分な情報がない場合は、「その質問に答えるための十分な情報がありません。」と答えてください。

文脈：
{context}

質問：{question}

答え："""

def generate_with_rag(question: str):
    # 検索
    context_docs = hybrid_search(question, top_k=5)
    context = "\n\n".join([doc.content for doc in context_docs])

    # 生成
    prompt = RAG_PROMPT_TEMPLATE.format(
        context=context,
        question=question
    )

    response = llm.generate(prompt)

    # 引用付きで返す
    return {
        "answer": response,
        "sources": [doc.metadata for doc in context_docs]
    }

---

2. エージェント アーキテクチャ

2.1 ReAct パターン（推論 + 行動）

Thought: X に関する情報を検索する必要があります
Action: search("X")
Observation: [検索結果]
Thought: 結果に基づいて、私は...すべきです
Action: calculate(...)
Observation: [計算結果]
Thought: これで十分な情報があります
Action: final_answer("答えは...")

REACT_PROMPT = """
あなたは、ツールを使用して質問に答えることができる AI アシスタントです。

利用可能なツール：
{tools_description}

このフォーマットを使用してください：
Thought: [次に何をするかについての推論]
Action: [tool_name(arguments)]
Observation: [ツール結果 - 自動的に入力されます]
... (必要に応じて Thought/Action/Observation を繰り返す)
Thought: 回答するのに十分な情報があります
Final Answer: [最終回答]

質問：{question}
"""

class ReActAgent:
    def __init__(self, tools: list, llm):
        self.tools = {t.name: t for t in tools}
        self.llm = llm
        self.max_iterations = 10

    def run(self, question: str) -> str:
        prompt = REACT_PROMPT.format(
            tools_description=self._format_tools(),
            question=question
        )

        for _ in range(self.max_iterations):
            response = self.llm.generate(prompt)

            if "Final Answer:" in response:
                return self._extract_final_answer(response)

            action = self._parse_action(response)
            observation = self._execute_tool(action)
            prompt += f"\nObservation: {observation}\n"

        return "最大反復回数に達しました"

2.2 関数呼び出しパターン

# スキーマを備えた関数としてツールを定義
TOOLS = [
    {
        "name": "search_web",
        "description": "Web から最新情報を検索する",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "検索クエリ"
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "calculate",
        "description": "数学計算を実行する",
        "parameters": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "評価する数式"
                }
            },
            "required": ["expression"]
        }
    }
]

class FunctionCallingAgent:
    def run(self, question: str) -> str:
        messages = [{"role": "user", "content": question}]

        while True:
            response = self.llm.chat(
                messages=messages,
                tools=TOOLS,
                tool_choice="auto"
            )

            if response.tool_calls:
                for tool_call in response.tool_calls:
                    result = self._execute_tool(
                        tool_call.name,
                        tool_call.arguments
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": str(result)
                    })
            else:
                return response.content

2.3 計画と実行パターン

class PlanAndExecuteAgent:
    """
    1. 計画を作成（ステップのリスト）
    2. 各ステップを実行
    3. 必要に応じて再計画
    """

    def run(self, task: str) -> str:
        # 計画フェーズ
        plan = self.planner.create_plan(task)
        # 返り値：["Step 1: ...", "Step 2: ...", ...]

        results = []
        for step in plan:
            # 各ステップを実行
            result = self.executor.execute(step, context=results)
            results.append(result)

            # 再計画が必要かチェック
            if self._needs_replan(task, results):
                new_plan = self.planner.replan(
                    task,
                    completed=results,
                    remaining=plan[len(results):]
                )
                plan = new_plan

        # 最終答えを合成
        return self.synthesizer.summarize(task, results)

2.4 複数エージェントの協働

class AgentTeam:
    """
    複雑なタスクで協働する専門エージェント
    """

    def __init__(self):
        self.agents = {
            "researcher": ResearchAgent(),
            "analyst": AnalystAgent(),
            "writer": WriterAgent(),
            "critic": CriticAgent()
        }
        self.coordinator = CoordinatorAgent()

    def solve(self, task: str) -> str:
        # コーディネーターがサブタスクを割り当て
        assignments = self.coordinator.decompose(task)

        results = {}
        for assignment in assignments:
            agent = self.agents[assignment.agent]
            result = agent.execute(
                assignment.subtask,
                context=results
            )
            results[assignment.id] = result

        # クリティックが評価
        critique = self.agents["critic"].review(results)

        if critique.needs_revision:
            # フィードバック付きで反復
            return self.solve_with_feedback(task, results, critique)

        return self.coordinator.synthesize(results)

---

3. プロンプト IDE パターン

3.1 変数付きプロンプト テンプレート

class PromptTemplate:
    def __init__(self, template: str, variables: list[str]):
        self.template = template
        self.variables = variables

    def format(self, **kwargs) -> str:
        # すべての変数が提供されたか検証
        missing = set(self.variables) - set(kwargs.keys())
        if missing:
            raise ValueError(f"不足している変数：{missing}")

        return self.template.format(**kwargs)

    def with_examples(self, examples: list[dict]) -> str:
        """少数ショット例を追加"""
        example_text = "\n\n".join([
            f"入力：{ex['input']}\n出力：{ex['output']}"
            for ex in examples
        ])
        return f"{example_text}\n\n{self.template}"

# 使用例
summarizer = PromptTemplate(
    template="以下のテキストを {style} スタイルで要約してください：\n\n{text}",
    variables=["style", "text"]
)

prompt = summarizer.format(
    style="professional",
    text="長い記事の内容..."
)

3.2 プロンプトのバージョン管理 & A/B テスト

class PromptRegistry:
    def __init__(self, db):
        self.db = db

    def register(self, name: str, template: str, version: str):
        """バージョン付きのプロンプトを保存"""
        self.db.save({
            "name": name,
            "template": template,
            "version": version,
            "created_at": datetime.now(),
            "metrics": {}
        })

    def get(self, name: str, version: str = "latest") -> str:
        """特定バージョンを取得"""
        return self.db.get(name, version)

    def ab_test(self, name: str, user_id: str) -> str:
        """ユーザーバケットに基づいてバリアントを返す"""
        variants = self.db.get_all_versions(name)
        bucket = hash(user_id) % len(variants)
        return variants[bucket]

    def record_outcome(self, prompt_id: str, outcome: dict):
        """プロンプトのパフォーマンスを追跡"""
        self.db.update_metrics(prompt_id, outcome)

3.3 プロンプト チェーニング

class PromptChain:
    """
    プロンプトをチェーンして、出力を次の入力として渡す
    """

    def __init__(self, steps: list[dict]):
        self.steps = steps

    def run(self, initial_input: str) -> dict:
        context = {"input": initial_input}
        results = []

        for step in self.steps:
            prompt = step["prompt"].format(**context)
            output = llm.generate(prompt)

            # 必要に応じて出力をパース
            if step.get("parser"):
                output = step["parser"](output)

            context[step["output_key"]] = output
            results.append({
                "step": step["name"],
                "output": output
            })

        return {
            "final_output": context[self.steps[-1]["output_key"]],
            "intermediate_results": results
        }

# 例：調査 → 分析 → 要約
chain = PromptChain([
    {
        "name": "research",
        "prompt": "トピックについて調査してください：{input}",
        "output_key": "research"
    },
    {
        "name": "analyze",
        "prompt": "これらの知見を分析してください：\n{research}",
        "output_key": "analysis"
    },
    {
        "name": "summarize",
        "prompt": "この分析を 3 つの要点で要約してください：\n{analysis}",
        "output_key": "summary"
    }
])

---

4. LLMOps & 可観測性

4.1 追跡すべきメトリクス

LLM_METRICS = {
    # パフォーマンス
    "latency_p50": "50 パーセンタイルのレスポンス時間",
    "latency_p99": "99 パーセンタイルのレスポンス時間",
    "tokens_per_second": "生成速度",

    # 品質
    "user_satisfaction": "高い評価 / 低い評価の比率",
    "task_completion": "完了したタスクのパーセンテージ",
    "hallucination_rate": "事実誤認を含む応答のパーセンテージ",

    # コスト
    "cost_per_request": "API 呼び出しあたりの平均ドル",
    "tokens_per_request": "使用される平均トークン数",
    "cache_hit_rate": "キャッシュから提供されたリクエストのパーセンテージ",

    # 信頼性
    "error_rate": "失敗したリクエストのパーセンテージ",
    "timeout_rate": "タイムアウトしたリクエストのパーセンテージ",
    "retry_rate": "再試行が必要なリクエストのパーセンテージ"
}

4.2 ログ & トレース

import logging
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

class LLMLogger:
    def log_request(self, request_id: str, data: dict):
        """デバッグと分析のために LLM リクエストをログ"""
        log_entry = {
            "request_id": request_id,
            "timestamp": datetime.now().isoformat(),
            "model": data["model"],
            "prompt": data["prompt"][:500],  # 保存用に切り詰め
            "prompt_tokens": data["prompt_tokens"],
            "temperature": data.get("temperature", 1.0),
            "user_id": data.get("user_id"),
        }
        logging.info(f"LLM_REQUEST: {json.dumps(log_entry)}")

    def log_response(self, request_id: str, data: dict):
        """LLM レスポンスをログ"""
        log_entry = {
            "request_id": request_id,
            "completion_tokens": data["completion_tokens"],
            "total_tokens": data["total_tokens"],
            "latency_ms": data["latency_ms"],
            "finish_reason": data["finish_reason"],
            "cost_usd": self._calculate_cost(data),
        }
        logging.info(f"LLM_RESPONSE: {json.dumps(log_entry)}")

# 分散トレース
@tracer.start_as_current_span("llm_call")
def call_llm(prompt: str) -> str:
    span = trace.get_current_span()
    span.set_attribute("prompt.length", len(prompt))

    response = llm.generate(prompt)

    span.set_attribute("response.length", len(response))
    span.set_attribute("tokens.total", response.usage.total_tokens)

    return response.content

4.3 評価フレームワーク

class LLMEvaluator:
    """
    LLM 出力の品質を評価
    """

    def evaluate_response(self,
                          question: str,
                          response: str,
                          ground_truth: str = None) -> dict:
        scores = {}

        # 関連性：質問に答えているか？
        scores["relevance"] = self._score_relevance(question, response)

        # 一貫性：構造は適切か？
        scores["coherence"] = self._score_coherence(response)

        # 根拠性：提供された文脈に基づいているか？
        scores["groundedness"] = self._score_groundedness(response)

        # 正確性：正解と一致するか？
        if ground_truth:
            scores["accuracy"] = self._score_accuracy(response, ground_truth)

        # 安全性：安全か？
        scores["safety"] = self._score_safety(response)

        return scores

    def run_benchmark(self, test_cases: list[dict]) -> dict:
        """テストセットで評価を実行"""
        results = []
        for case in test_cases:
            response = llm.generate(case["prompt"])
            scores = self.evaluate_response(
                question=case["prompt"],
                response=response,
                ground_truth=case.get("expected")
            )
            results.append(scores)

        return self._aggregate_scores(results)

---

5. 本番環境パターン

5.1 キャッシング戦略

import hashlib
from functools import lru_cache

class LLMCache:
    def __init__(self, redis_client, ttl_seconds=3600):
        self.redis = redis_client
        self.ttl = ttl_seconds

    def _cache_key(self, prompt: str, model: str, **kwargs) -> str:
        """決定論的なキャッシュキーを生成"""
        content = f"{model}:{prompt}:{json.dumps(kwargs, sort_keys=True)}"
        return hashlib.sha256(content.encode()).hexdigest()

    def get_or_generate(self, prompt: str, model: str, **kwargs) -> str:
        key = self._cache_key(prompt, model, **kwargs)

        # キャッシュをチェック
        cached = self.redis.get(key)
        if cached:
            return cached.decode()

        # 生成
        response = llm.generate(prompt, model=model, **kwargs)

        # キャッシュ（決定論的な出力のみをキャッシュ）
        if kwargs.get("temperature", 1.0) == 0:
            self.redis.setex(key, self.ttl, response)

        return response

5.2 レート制限 & リトライ

import time
from tenacity import retry, wait_exponential, stop_after_attempt

class RateLimiter:
    def __init__(self, requests_per_minute: int):
        self.rpm = requests_per_minute
        self.timestamps = []

    def acquire(self):
        """レート制限を超える場合は待機"""
        now = time.time()

        # 古いタイムスタンプを削除
        self.timestamps = [t for t in self.timestamps if now - t < 60]

        if len(self.timestamps) >= self.rpm:
            sleep_time = 60 - (now - self.timestamps[0])
            time.sleep(sleep_time)

        self.timestamps.append(time.time())

# 指数バックオフでリトライ
@retry(
    wait=wait_exponential(multiplier=1, min=4, max=60),
    stop=stop_after_attempt(5)
)
def call_llm_with_retry(prompt: str) -> str:
    try:
        return llm.generate(prompt)
    except RateLimitError:
        raise  # リトライをトリガー
    except APIError as e:
        if e.status_code >= 500:
            raise  # サーバーエラーをリトライ
        raise  # クライアントエラーはリトライしない

5.3 フォールバック戦略

class LLMWithFallback:
    def __init__(self, primary: str, fallbacks: list[str]):
        self.primary = primary
        self.fallbacks = fallbacks

    def generate(self, prompt: str, **kwargs) -> str:
        models = [self.primary] + self.fallbacks

        for model in models:
            try:
                return llm.generate(prompt, model=model, **kwargs)
            except (RateLimitError, APIError) as e:
                logging.warning(f"モデル {model} が失敗しました：{e}")
                continue

        raise AllModelsFailedError("すべてのモデルが枯渇しました")

# 使用例
llm_client = LLMWithFallback(
    primary="gpt-4-turbo",
    fallbacks=["gpt-3.5-turbo", "claude-3-sonnet"]
)

---

アーキテクチャ決定マトリックス

パターン	使用場面	複雑性	コスト
シンプル RAG	FAQ、ドキュメント検索	低	低
ハイブリッド RAG	混合クエリ	中	中
ReAct エージェント	マルチステップタスク	中	中
関数呼び出し	構造化されたツール	低	低
計画と実行	複雑なタスク	高	高
マルチエージェント	調査タスク	非常に高	非常に高

---

リソース

• Dify Platform
• LangChain Docs
• LlamaIndex
• Anthropic Cookbook

制限事項

• このスキルは、タスクが上述の範囲に明確に適合する場合にのみ使用してください。
• 出力は、環境固有の検証、テスト、またはエキスパートレビューの代替品として扱わないでください。
• 必要な入力、権限、安全境界、または成功基準が不足している場合は、停止して説明を求めてください。