第5章: 実践的LLMアプリケーション

1. RAG（検索拡張生成）

RAG（Retrieval-Augmented Generation）は、外部知識でLLMを強化し、プライベートデータ、最新情報、ドメイン固有コンテンツについて正確な応答を可能にします。LLMの根本的な制限である、古くなる静的な訓練データの問題に対処します。

1.1 RAGアーキテクチャ

1.2 RAGシステムの構築

from typing import List, Optional
import numpy as np
from dataclasses import dataclass

@dataclass
class Document:
    """メタデータ付きドキュメントチャンクを表す"""
    content: str
    metadata: dict
    embedding: Optional[np.ndarray] = None

class RAGSystem:
    """
    現代のベストプラクティスを取り入れた本番対応RAG実装
    """

    def __init__(
        self,
        embedding_model: str = "text-embedding-3-large",
        llm_model: str = "gpt-4o",
        chunk_size: int = 512,
        chunk_overlap: int = 50,
        top_k: int = 5
    ):
        self.embedding_model = embedding_model
        self.llm_model = llm_model
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.top_k = top_k
        self.documents: List[Document] = []

    def retrieve(
        self,
        query: str,
        top_k: Optional[int] = None
    ) -> List[Document]:
        """
        コサイン類似度を使用して関連ドキュメントを検索
        """
        from openai import OpenAI
        client = OpenAI()

        top_k = top_k or self.top_k

        # クエリを埋め込み
        response = client.embeddings.create(
            model=self.embedding_model,
            input=[query]
        )
        query_embedding = np.array(response.data[0].embedding)

        # 類似度を計算
        similarities = []
        for doc in self.documents:
            if doc.embedding is not None:
                sim = np.dot(query_embedding, doc.embedding) / (
                    np.linalg.norm(query_embedding) * np.linalg.norm(doc.embedding)
                )
                similarities.append((sim, doc))

        # ソートしてtop-kを返す
        similarities.sort(key=lambda x: x[0], reverse=True)
        return [doc for _, doc in similarities[:top_k]]

    def generate(
        self,
        query: str,
        context_documents: List[Document]
    ) -> str:
        """
        検索されたコンテキストを使用して回答を生成
        """
        from openai import OpenAI
        client = OpenAI()

        # コンテキスト文字列を構築
        context = "\n\n---\n\n".join([
            f"[ソース: {doc.metadata.get('source', '不明')}]\n{doc.content}"
            for doc in context_documents
        ])

        system = """あなたは提供されたコンテキストに基づいて質問に答える有用なアシスタントです。
回答がコンテキスト内に見つからない場合は、明確にそう述べてください。
回答で使用したソースを常に引用してください。"""

        messages = [
            {"role": "system", "content": system},
            {"role": "user", "content": f"""コンテキスト:
{context}

質問: {query}

上記のコンテキストに基づいて回答してください。"""}
        ]

        response = client.chat.completions.create(
            model=self.llm_model,
            messages=messages,
            temperature=0.7
        )

        return response.choices[0].message.content

    def query(self, question: str) -> dict:
        """
        エンドツーエンドのRAGクエリパイプライン
        """
        retrieved_docs = self.retrieve(question)
        answer = self.generate(question, retrieved_docs)

        return {
            "question": question,
            "answer": answer,
            "sources": [
                {"content": doc.content[:200] + "...", "metadata": doc.metadata}
                for doc in retrieved_docs
            ]
        }

1.3 高度なRAG技術

2. Model Context Protocol（MCP）

MCPは、Anthropicが開発したLLMを外部ツールやデータソースに接続するためのオープン標準です。標準化されたインターフェースを通じて、AIモデルが訓練データを超えた機能にアクセスできるようにする統一プロトコルを提供します。

2.1 MCPアーキテクチャ

2.2 MCPサーバーの構築

"""
MCPサーバー実装例

ファイル操作とウェブ検索のツールを提供し、
MCPプロトコルのパターンを示す
"""

from mcp.server import Server
from mcp.types import Tool, TextContent
import asyncio
from pathlib import Path

server = Server("example-mcp-server")

@server.list_tools()
async def list_tools() -> list[Tool]:
    """利用可能なツールを定義"""
    return [
        Tool(
            name="read_file",
            description="ファイルシステムからファイルの内容を読み取る",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "読み取るファイルのパス"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="web_search",
            description="ウェブで情報を検索する",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "検索クエリ"
                    }
                },
                "required": ["query"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """ツールの実行を処理"""

    if name == "read_file":
        path = Path(arguments["path"])
        if not path.exists():
            return [TextContent(type="text", text=f"エラー: ファイルが見つかりません: {path}")]
        content = path.read_text(encoding="utf-8")
        return [TextContent(type="text", text=content)]

    elif name == "web_search":
        query = arguments["query"]
        # 検索APIを実装
        return [TextContent(type="text", text=f"'{query}'の検索結果...")]

    return [TextContent(type="text", text=f"不明なツール: {name}")]

3. AIエージェント

AIエージェントは、推論、計画、行動を行って目標を達成するLLM駆動システムです。LLMの推論能力とツール使用、メモリ、多段階実行を組み合わせます。

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

3.2 ReActエージェントパターン

"""
ReActエージェント: 推論 + 行動

ReActパターンは推論トレースと行動を交互に行い、
モデルが反復的に計画、行動、観察できるようにする
"""

from typing import List, Dict, Any
from dataclasses import dataclass, field
import json

@dataclass
class AgentState:
    """エージェント実行状態"""
    goal: str
    history: List[Dict[str, Any]] = field(default_factory=list)
    final_answer: str = None
    iterations: int = 0
    max_iterations: int = 10

class ReActAgent:
    """
    ReActエージェント実装

    パターン: 思考 -> 行動 -> 観察 -> 繰り返し
    """

    def __init__(
        self,
        llm_client,
        tools: Dict[str, callable],
        model: str = "claude-sonnet-4-20250514"
    ):
        self.llm = llm_client
        self.tools = tools
        self.model = model

        self.system_prompt = """あなたは問題を段階的に解決する有用なAIエージェントです。

各ステップで以下の形式で出力してください：
思考: <次に何をすべきかの推論>
行動: <ツール名>
行動入力: <ツールへのJSON入力>

または最終回答がある場合：
思考: <推論>
最終回答: <完全な回答>

利用可能なツール:
{tool_descriptions}

ルール:
1. 行動の前に必ず考える
2. 観察を次のステップに活用する
3. ツールが失敗したら別のアプローチを試す
4. 十分な情報が得られたら最終回答を提供する"""

    async def run(self, goal: str) -> str:
        """目標達成のためにエージェントを実行"""
        state = AgentState(goal=goal)
        messages = [{"role": "user", "content": f"目標: {goal}"}]

        while state.iterations < state.max_iterations:
            state.iterations += 1

            response = self.llm.messages.create(
                model=self.model,
                system=self.system_prompt.format(
                    tool_descriptions=self._format_tools()
                ),
                max_tokens=2048,
                messages=messages
            )

            parsed = self._parse_response(response.content[0].text)

            if 'final_answer' in parsed:
                return parsed['final_answer']

            if 'action' in parsed:
                observation = self._execute_tool(
                    parsed['action'],
                    parsed.get('action_input', '{}')
                )
                messages.append({"role": "assistant", "content": response.content[0].text})
                messages.append({"role": "user", "content": f"観察: {observation}"})

        return "エージェントは回答を見つけられませんでした。"

    def _format_tools(self) -> str:
        return "\n".join([f"- {name}: {tool.__doc__}" for name, tool in self.tools.items()])

    def _parse_response(self, text: str) -> dict:
        # 応答を構造化形式にパース
        result = {}
        for line in text.split('\n'):
            if line.startswith('思考:'):
                result['thought'] = line[3:].strip()
            elif line.startswith('行動:'):
                result['action'] = line[3:].strip()
            elif line.startswith('行動入力:'):
                result['action_input'] = line[5:].strip()
            elif line.startswith('最終回答:'):
                result['final_answer'] = line[5:].strip()
        return result

    def _execute_tool(self, name: str, input_str: str) -> str:
        if name not in self.tools:
            return f"エラー: 不明なツール '{name}'"
        try:
            return str(self.tools[name](**json.loads(input_str)))
        except Exception as e:
            return f"エラー: {e}"

4. マルチモーダルLLM

マルチモーダルLLMは、テキスト、画像、音声、動画など複数のタイプのコンテンツを処理・生成します。これにより、視覚的な質問応答、テキストからの画像生成、ドキュメント理解などの強力なアプリケーションが可能になります。

4.1 ビジョン言語モデル

4.2 画像の操作

"""
画像を使用したマルチモーダルLLMの使用例
"""

import anthropic
import base64
from pathlib import Path

def encode_image(image_path: str) -> tuple[str, str]:
    """画像をbase64エンコードしメディアタイプを検出"""
    path = Path(image_path)
    suffix = path.suffix.lower()

    media_types = {
        '.jpg': 'image/jpeg',
        '.jpeg': 'image/jpeg',
        '.png': 'image/png',
        '.gif': 'image/gif',
        '.webp': 'image/webp'
    }

    media_type = media_types.get(suffix, 'image/png')

    with open(path, 'rb') as f:
        data = base64.standard_b64encode(f.read()).decode('utf-8')

    return data, media_type

def analyze_image(image_path: str, question: str) -> str:
    """Claude Visionで画像を分析"""
    client = anthropic.Anthropic()

    image_data, media_type = encode_image(image_path)

    message = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": media_type,
                            "data": image_data
                        }
                    },
                    {
                        "type": "text",
                        "text": question
                    }
                ]
            }
        ]
    )

    return message.content[0].text

# 使用例
result = analyze_image(
    "screenshot.png",
    "このスクリーンショットにはどのようなUI問題がありますか？"
)

5. 本番デプロイメント

5.1 コスト最適化

5.2 本番チェックリスト

まとめ