LLMアプリケーション開発ガイド：プロトタイプからプロダクションまで

┌─────────────────────────────────────────┐
│           ユーザーインターフェース        │
│  (Web, Mobile, API, Slack, CLI)         │
├─────────────────────────────────────────┤
│           アプリケーションロジック        │
│  (オーケストレーション、ビジネスルール)   │
├─────────────────────────────────────────┤
│        LLMオーケストレーションレイヤー    │
│  (LangChain, LlamaIndex, 生SDK)         │
├─────────────────────────────────────────┤
│           LLMプロバイダー                │
│  (OpenAI, Anthropic, Google, ローカル)  │
├─────────────────────────────────────────┤
│           サポートサービス               │
│  (ベクターDB、キャッシュ、検索、ツール)  │
└─────────────────────────────────────────┘

[システム指示]
あなたはAcme Corpの親切なカスタマーサポートエージェントです。
ユーザーが使う言語で応答してください。
常に丁寧かつ簡潔に。競合他社には絶対に言及しないでください。

[コンテキスト / 検索されたドキュメント]
注文 #12345、2026-03-10 受付。ステータス: 発送済み。
追跡番号: 1Z999AA10123456784

[例（few-shot）]
ユーザー: 私の注文はどこにありますか？
アシスタント: 注文 #99999が3月5日に発送され、配送中です。
予想到着日: 3月12日。

[ユーザーメッセージ]
先週注文したのですが、まだ受け取っていません。

SYSTEM_PROMPT = """あなたはフィンテック企業のシニアPythonコードレビュアーです。

責任:
- 正確性、セキュリティ脆弱性、パフォーマンスの問題をレビュー
- コード例を含む具体的な改善提案
- GDPR/CCPAに違反するPII処理の特定

出力フォーマット:
- 一文の全体評価から始める
- 深刻度とともに問題リスト: [CRITICAL], [WARNING], [SUGGESTION]
- 変更が必要な場合は修正されたコードブロックで終了

新機能は作成しません。与えられたものだけをレビューします。"""

FEW_SHOT_EXAMPLES = """
以下の議事録からアクションアイテムを抽出してください。
JSON配列で出力してください。

議事録: Johnが金曜日までにデプロイガイドを更新する予定です。
Sarahは取締役会の前にQ1予算をレビューする必要があります。
アクションアイテム: [
  {"owner": "John", "task": "デプロイガイドの更新", "due": "金曜日"},
  {"owner": "Sarah", "task": "Q1予算のレビュー", "due": "取締役会の前"}
]

議事録: APIチームが今スプリントにレート制限を追加することに合意しました。
ドキュメント更新は担当者が割り当てられていません。
アクションアイテム: [
  {"owner": "API team", "task": "レート制限の追加", "due": "今スプリント"},
  {"owner": null, "task": "ドキュメント更新", "due": null}
]

議事録: {meeting_text}
アクションアイテム:"""

COT_PROMPT = """以下の問題をステップバイステップで解いてください。
各ステップで推論プロセスを示してから最終的な答えを提示してください。

問題: 顧客のクレジットが500ドルあります。320ドルの注文をして、
80ドルの商品を1つ返品しました。残りのクレジットはいくらですか？

ステップバイステップで考えてみましょう:"""

from openai import OpenAI
import json

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    response_format={"type": "json_object"},
    messages=[
        {"role": "system", "content": "エンティティを抽出してください。有効なJSONのみ出力してください。"},
        {"role": "user", "content": "Appleが2024年9月9日にクパチーノでiPhone 16を発表しました。"}
    ]
)

data = json.loads(response.choices[0].message.content)
# {"company": "Apple", "product": "iPhone 16", "location": "クパチーノ", "date": "2024-09-09"}

from pydantic import BaseModel
from openai import OpenAI

class NewsEvent(BaseModel):
    company: str
    product: str
    location: str
    date: str

client = OpenAI()
response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "イベントの詳細を抽出してください。"},
        {"role": "user", "content": "Appleが2024年9月9日にクパチーノでiPhone 16を発表しました。"}
    ],
    response_format=NewsEvent,
)
event = response.choices[0].message.parsed
print(event.company)  # Apple

from openai import OpenAI

client = OpenAI(api_key="sk-...")  # またはOPENAI_API_KEY環境変数を設定

# 基本的なチャット補完
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "あなたは役立つアシスタントです。"},
        {"role": "user", "content": "トランスフォーマーアーキテクチャを3文で要約してください。"}
    ],
    temperature=0.7,
    max_tokens=200,
)

print(response.choices[0].message.content)
print(f"使用されたトークン: {response.usage.total_tokens}")

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-...")

message = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system="あなたは役立つアシスタントです。",
    messages=[
        {"role": "user", "content": "トランスフォーマーのアテンションメカニズムを説明してください。"}
    ]
)

print(message.content[0].text)
print(f"入力トークン: {message.usage.input_tokens}")
print(f"出力トークン: {message.usage.output_tokens}")

from litellm import completion

# OpenAI
response = completion(
    model="gpt-4o",
    messages=[{"role": "user", "content": "こんにちは"}]
)

# Anthropic（同じインターフェース）
response = completion(
    model="anthropic/claude-opus-4-5",
    messages=[{"role": "user", "content": "こんにちは"}]
)

# ローカルOllamaモデル（同じインターフェース）
response = completion(
    model="ollama/llama3",
    messages=[{"role": "user", "content": "こんにちは"}]
)

print(response.choices[0].message.content)

class ConversationManager:
    def __init__(self, system_prompt: str, max_history: int = 20):
        self.system_prompt = system_prompt
        self.max_history = max_history
        self.history: list[dict] = []
        self.client = OpenAI()

    def chat(self, user_message: str) -> str:
        self.history.append({"role": "user", "content": user_message})

        # コンテキストウィンドウの超過防止のために履歴をトリミング
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": self.system_prompt},
                *self.history
            ]
        )

        assistant_message = response.choices[0].message.content
        self.history.append({"role": "assistant", "content": assistant_message})
        return assistant_message

ユーザークエリ
    │
    ▼
[クエリ埋め込み] ──► [ベクター検索] ──► 上位K個の関連チャンク
                      │
                      ▼
            [拡張プロンプトの構築]
            システム: 役立つアシスタントです。
            コンテキスト: {検索されたチャンク}
            ユーザー: {元のクエリ}
                      │
                      ▼
               [LLMが回答を生成]

from langchain.document_loaders import PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 1. ドキュメントの読み込み
loader = DirectoryLoader("./docs", glob="**/*.pdf", loader_cls=PyPDFLoader)
documents = loader.load()

# 2. チャンクに分割
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""]
)
chunks = splitter.split_documents(documents)

# 3. 埋め込みと保存
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
print(f"{len(documents)}件のドキュメントから{len(chunks)}個のチャンクをインデックス化完了")

from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# 既存のベクターストアを読み込み
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small")
)

# リトリーバーの作成
retriever = vectorstore.as_retriever(
    search_type="mmr",          # 多様性のためのMaximal Marginal Relevance
    search_kwargs={"k": 5, "fetch_k": 20}
)

# カスタムプロンプト
QA_PROMPT = PromptTemplate(
    template="""以下のコンテキストを使って質問に答えてください。
コンテキストに答えがない場合は「その情報はありません。」と言ってください。
情報を作り出さないでください。

コンテキスト:
{context}

質問: {question}

回答:""",
    input_variables=["context", "question"]
)

# チェーン
llm = ChatOpenAI(model="gpt-4o", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs={"prompt": QA_PROMPT},
    return_source_documents=True
)

result = qa_chain.invoke({"query": "返金ポリシーは何ですか？"})
print(result["result"])
for doc in result["source_documents"]:
    print(f"出典: {doc.metadata['source']}, ページ {doc.metadata.get('page', 'N/A')}")

from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# 密リトリーバー（セマンティック）
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 疎リトリーバー（BM25キーワード）
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# 等しい重みでアンサンブル
ensemble_retriever = EnsembleRetriever(
    retrievers=[dense_retriever, bm25_retriever],
    weights=[0.6, 0.4]
)

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query: str, docs: list, top_k: int = 3) -> list:
    pairs = [(query, doc.page_content) for doc in docs]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, docs), key=lambda x: x[0], reverse=True)
    return [doc for _, doc in ranked[:top_k]]

import json
import requests
from openai import OpenAI

client = OpenAI()

# ツール定義
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "都市の現在の天気を取得します",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "都市名、例: '東京'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度の単位"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "最新情報のためにウェブを検索します",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "検索クエリ"}
                },
                "required": ["query"]
            }
        }
    }
]

def get_weather(city: str, unit: str = "celsius") -> dict:
    # 実際の実装では天気APIを呼び出す
    return {"city": city, "temp": 18, "unit": unit, "condition": "晴れ"}

def search_web(query: str) -> str:
    # 実際の実装では検索APIを呼び出す
    return f"検索結果: {query}"

TOOL_MAP = {
    "get_weather": get_weather,
    "search_web": search_web,
}

def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )

        message = response.choices[0].message

        # ツール呼び出しなし → 最終回答
        if not message.tool_calls:
            return message.content

        # 履歴にアシスタントの応答を追加
        messages.append(message)

        # 各ツール呼び出しを実行
        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            result = TOOL_MAP[func_name](**func_args)

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            })

answer = run_agent("東京の天気はどうですか？傘を持っていくべきですか？")
print(answer)

# モデルが複数のツールを同時に呼び出せます
# 上記のループがすでにこれを処理します — message.tool_callsはリストです
# 天気と検索を単一のモデルターンで両方呼び出し可能

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "東京とパリの天気を比較して、旅行のヒントも検索してください。"}
    ],
    tools=tools,
    parallel_tool_calls=True  # GPT-4oではデフォルトTrue
)

# 応答に3つのツール呼び出しが同時にある可能性: weather(東京), weather(パリ), search(旅行ヒント)

from openai import OpenAI

client = OpenAI()

# 同期ストリーミング
with client.chat.completions.stream(
    model="gpt-4o",
    messages=[{"role": "user", "content": "ロボットについての短い物語を書いてください。"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

@app.post("/chat")
async def chat(body: dict):
    async def generate():
        async with client.chat.completions.stream(
            model="gpt-4o",
            messages=body["messages"]
        ) as stream:
            async for text in stream.text_stream:
                yield f"data: {text}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def classify_one(text: str, semaphore: asyncio.Semaphore) -> str:
    async with semaphore:
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "POSITIVE、NEGATIVE、またはNEUTRALに分類してください。"},
                {"role": "user", "content": text}
            ],
            max_tokens=10
        )
        return response.choices[0].message.content.strip()

async def classify_batch(texts: list[str], max_concurrent: int = 20) -> list[str]:
    semaphore = asyncio.Semaphore(max_concurrent)
    tasks = [classify_one(text, semaphore) for text in texts]
    return await asyncio.gather(*tasks)

# 使用例
texts = ["この製品は本当に良いです！", "ひどい経験でした。", "まあまあでした。"] * 100
results = asyncio.run(classify_batch(texts))

assert add(2, 3) == 5  # 常に合格または失敗

from openai import OpenAI

client = OpenAI()

JUDGE_PROMPT = """AIアシスタントの応答を評価しています。
以下の基準で応答を評価してください（各1-5点）:
- 正確性: 情報は正しいですか？
- 有用性: 質問を完全にカバーしていますか？
- 簡潔性: 適切に簡潔ですか？

質問: {question}
応答: {response}
参照回答: {reference}

JSONで出力: {{"accuracy": X, "helpfulness": X, "conciseness": X, "reasoning": "..."}}"""

def evaluate(question: str, response: str, reference: str) -> dict:
    import json
    result = client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},
        messages=[{
            "role": "user",
            "content": JUDGE_PROMPT.format(
                question=question,
                response=response,
                reference=reference
            )
        }]
    )
    return json.loads(result.choices[0].message.content)

from deepeval import evaluate
from deepeval.metrics import (
    AnswerRelevancyMetric,
    FaithfulnessMetric,
    ContextualRecallMetric,
)
from deepeval.test_case import LLMTestCase

test_case = LLMTestCase(
    input="フランスの首都はどこですか？",
    actual_output="フランスの首都はパリです。",
    expected_output="パリ",
    retrieval_context=["フランスは西ヨーロッパの国です。首都はパリです。"]
)

metrics = [
    AnswerRelevancyMetric(threshold=0.8),
    FaithfulnessMetric(threshold=0.9),
    ContextualRecallMetric(threshold=0.8),
]

evaluate([test_case], metrics)

# promptfooconfig.yaml
prompts:
  - '以下のテキストを2文で要約してください: {{text}}'

providers:
  - openai:gpt-4o
  - openai:gpt-4o-mini

tests:
  - vars:
      text: 'エッフェル塔は1889年の万博のために建設されました...'
    assert:
      - type: llm-rubric
        value: '要約には1889年と万博が言及されている必要があります'
      - type: javascript
        value: "output.split('.').length <= 3" # 最大3文

import tiktoken

def count_tokens(text: str, model: str = "gpt-4o") -> int:
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

def estimate_cost(
    input_tokens: int,
    output_tokens: int,
    model: str = "gpt-4o"
) -> float:
    # 100万トークンあたりの価格（2026年3月時点の概算）
    PRICING = {
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "claude-opus-4-5": {"input": 15.00, "output": 75.00},
        "claude-haiku-3-5": {"input": 0.80, "output": 4.00},
    }
    p = PRICING.get(model, {"input": 5.0, "output": 15.0})
    return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000

# Anthropicプロンプトキャッシング
import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": very_long_system_prompt,  # 例: 50Kトークンのポリシー文書
            "cache_control": {"type": "ephemeral"}  # このプレフィックスをキャッシュ
        }
    ],
    messages=[{"role": "user", "content": user_question}]
)
# 最初の呼び出し: 全額。以降の呼び出し: キャッシュされたトークンは約90%割引。

def route_to_model(task: str, complexity: str) -> str:
    """タスクの複雑さに応じた適切なモデルへのルーティング。"""
    if complexity == "simple":
        return "gpt-4o-mini"          # シンプルな分類、抽出
    elif complexity == "medium":
        return "gpt-4o"               # 要約、Q&A
    else:
        return "claude-opus-4-5"      # 複雑な推論、コードレビュー

# 例: ルーティング前の複雑さ分類
def smart_complete(messages: list, task_description: str) -> str:
    from openai import OpenAI
    client = OpenAI()

    # 安価な分類ステップ
    complexity = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"このタスクの複雑さを'simple'、'medium'、'complex'のいずれかで評価してください: {task_description}"
        }],
        max_tokens=5
    ).choices[0].message.content.strip().lower()

    model = route_to_model(task_description, complexity)

    return client.chat.completions.create(
        model=model,
        messages=messages
    ).choices[0].message.content

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import AsyncOpenAI
import asyncio

app = FastAPI(title="LLM API")
client = AsyncOpenAI()

class ChatRequest(BaseModel):
    messages: list[dict]
    model: str = "gpt-4o"
    temperature: float = 0.7
    max_tokens: int = 1000

class ChatResponse(BaseModel):
    content: str
    usage: dict

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        response = await client.chat.completions.create(
            model=request.model,
            messages=request.messages,
            temperature=request.temperature,
            max_tokens=request.max_tokens,
        )
        return ChatResponse(
            content=response.choices[0].message.content,
            usage={
                "input_tokens": response.usage.prompt_tokens,
                "output_tokens": response.usage.completion_tokens
            }
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

import asyncio
import random
from functools import wraps

def with_retry(max_attempts: int = 3, base_delay: float = 1.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    # ジッターのある指数バックオフ
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                    await asyncio.sleep(delay)
        return wrapper
    return decorator

@with_retry(max_attempts=3)
async def robust_completion(messages: list) -> str:
    client = AsyncOpenAI()
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )
    return response.choices[0].message.content

import hashlib
import json
import redis.asyncio as redis

redis_client = redis.from_url("redis://localhost:6379")
CACHE_TTL = 3600  # 1時間

def cache_key(messages: list, model: str) -> str:
    payload = json.dumps({"messages": messages, "model": model}, sort_keys=True)
    return f"llm:{hashlib.md5(payload.encode()).hexdigest()}"

async def cached_completion(messages: list, model: str = "gpt-4o") -> str:
    key = cache_key(messages, model)

    # キャッシュを確認
    cached = await redis_client.get(key)
    if cached:
        return cached.decode()

    # 生成
    from openai import AsyncOpenAI
    client = AsyncOpenAI()
    response = await client.chat.completions.create(model=model, messages=messages)
    result = response.choices[0].message.content

    # TTLと共に保存
    await redis_client.setex(key, CACHE_TTL, result)
    return result

import os
from langchain_openai import ChatOpenAI
from langchain.callbacks.tracers import LangChainTracer

os.environ["LANGCHAIN_API_KEY"] = "ls__..."
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "my-llm-app"

# すべてのLangChain呼び出しが自動的にトレースされます
llm = ChatOpenAI(model="gpt-4o")
response = llm.invoke("RAGとは何ですか？")
# LangSmith UIでフルトレース（プロンプト、応答、レイテンシ、トークン）を確認可能

import time
import logging
from dataclasses import dataclass, field, asdict

logger = logging.getLogger(__name__)

@dataclass
class LLMCallLog:
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    success: bool
    error: str = ""
    metadata: dict = field(default_factory=dict)

async def traced_completion(messages: list, model: str = "gpt-4o", **metadata) -> str:
    from openai import AsyncOpenAI
    client = AsyncOpenAI()

    start = time.perf_counter()
    success = True
    error = ""
    input_tokens = output_tokens = 0

    try:
        response = await client.chat.completions.create(model=model, messages=messages)
        result = response.choices[0].message.content
        input_tokens = response.usage.prompt_tokens
        output_tokens = response.usage.completion_tokens
        return result
    except Exception as e:
        success = False
        error = str(e)
        raise
    finally:
        log = LLMCallLog(
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            latency_ms=(time.perf_counter() - start) * 1000,
            success=success,
            error=error,
            metadata=metadata
        )
        logger.info("llm_call", extra=asdict(log))

from guardrails import Guard
from guardrails.hub import ToxicLanguage, DetectPII

guard = Guard().use_many(
    ToxicLanguage(threshold=0.5, on_fail="exception"),
    DetectPII(pii_entities=["EMAIL_ADDRESS", "PHONE_NUMBER"], on_fail="fix"),
)

def safe_completion(user_input: str) -> str:
    from openai import OpenAI
    client = OpenAI()

    # 入力バリデーション
    guard.validate(user_input)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_input}]
    )
    output = response.choices[0].message.content

    # 出力バリデーションと修正
    validated = guard.validate(output)
    return validated.validated_output