目次

1. [LLMアプリケーション開発の概要](#1-overview)

2. [プロンプトエンジニアリングの基礎](#2-prompt-engineering)

3. [LLM APIとSDK](#3-apis-and-sdks)

4. [検索拡張生成（RAG）](#4-rag)

5. [ツール使用と関数呼び出し](#5-tool-use)

6. [ストリーミングと非同期パターン](#6-streaming)

7. [評価とテスト](#7-evaluation)

8. [コスト最適化](#8-cost-optimization)

9. [プロダクションデプロイ](#9-production-deployment)

10. [可観測性とモニタリング](#10-observability)

1. LLMアプリケーション開発の概要

1.1 LLMアプリケーションとは？

LLMアプリケーションとは、自然言語処理、コンテンツ生成、情報推論、またはアクション実行に大規模言語モデルをコアコンポーネントとして活用するあらゆるソフトウェアシステムです。すべての動作が明示的にプログラムされた従来のソフトウェアとは異なり、LLMアプリケーションはロジックの相当部分を事前学習済みモデルに委ねます。

一般的なLLMアプリケーションのカテゴリ：

| カテゴリ | 例 | 主な課題 |

| ---------------------------- | ------------------------------------------ | -------------------------------- |

| チャットボット・アシスタント | カスタマーサポート、パーソナルアシスタント | コンテキスト管理、トーンの一貫性 |

| ドキュメントQ&A | 契約書レビュー、社内検索 | 検索精度、ハルシネーション |

| コード生成 | オートコンプリート、PRレビュー、テスト作成 | 正確性、セキュリティ |

| コンテンツ生成 | マーケティングコピー、要約 | 品質管理、ブランドボイス |

| データ抽出 | フォームパース、構造化出力 | スキーマ準拠、堅牢性 |

| 自律エージェント | リサーチエージェント、タスク自動化 | 信頼性、コスト制御 |

1.2 開発スタック

モダンなLLMアプリケーションは一般的に以下のレイヤーで構成されます：

┌─────────────────────────────────────────┐

│ ユーザーインターフェース │

│ (Web, Mobile, API, Slack, CLI) │

├─────────────────────────────────────────┤

│ アプリケーションロジック │

│ (オーケストレーション、ビジネスルール) │

├─────────────────────────────────────────┤

│ LLMオーケストレーションレイヤー │

│ (LangChain, LlamaIndex, 生SDK) │

├─────────────────────────────────────────┤

│ LLMプロバイダー │

│ (OpenAI, Anthropic, Google, ローカル) │

├─────────────────────────────────────────┤

│ サポートサービス │

│ (ベクターDB、キャッシュ、検索、ツール) │

└─────────────────────────────────────────┘

1.3 基本原則

**1. シンプルに始め、必要な時だけ複雑さを加えてください。**

適切に作られたプロンプトを使った直接的なAPI呼び出しは、複雑なオーケストレーションフレームワークよりも高いパフォーマンスを発揮することがよくあります。明確なユースケースが証明された後に抽象化レイヤーを追加してください。

**2. プロンプトをコードのように扱ってください。**

プロンプトをバージョン管理し、テストを書き、変更を慎重に追跡してください。プロンプトのリグレッションはコードのリグレッションと同様に致命的です。

**3. リリース前に評価してください。**

LLMの出力は非決定論的です。体系的な評価なしには変更が品質を改善したか低下させたかを知ることができません。

**4. 失敗を考慮して設計してください。**

LLMはハルシネーションを起こし、タイムアウトが発生し、予期しないフォーマットを返します。最初からリトライロジック、フォールバック、バリデーションを組み込んでください。

2. プロンプトエンジニアリングの基礎

2.1 プロンプトの構造

プロダクションプロンプトは4つのオプションセクションで構成されます：

[システム指示]

あなたはAcme Corpの親切なカスタマーサポートエージェントです。

ユーザーが使う言語で応答してください。

常に丁寧かつ簡潔に。競合他社には絶対に言及しないでください。

[コンテキスト / 検索されたドキュメント]

注文 #12345、2026-03-10 受付。ステータス: 発送済み。

追跡番号: 1Z999AA10123456784

[例（few-shot）]

ユーザー: 私の注文はどこにありますか？

アシスタント: 注文 #99999が3月5日に発送され、配送中です。

予想到着日: 3月12日。

[ユーザーメッセージ]

先週注文したのですが、まだ受け取っていません。

2.2 システム指示のベストプラクティス

以下の特性を持つシステム指示を作成してください：

- **役割特化**: モデルが誰であり、目的は何かを正確に定義。

- **制約の明示**: モデルがすべきこととすべきでないことを明示。

- **フォーマット指定**: 出力フォーマットが重要な場合は明確に記述。

- **トーンの定義**: 丁寧さのレベル、言語、長さの期待値を指定。

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

責任:

- 正確性、セキュリティ脆弱性、パフォーマンスの問題をレビュー

- コード例を含む具体的な改善提案

- GDPR/CCPAに違反するPII処理の特定

出力フォーマット:

- 一文の全体評価から始める

- 深刻度とともに問題リスト: [CRITICAL], [WARNING], [SUGGESTION]

- 変更が必要な場合は修正されたコードブロックで終了

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

2.3 Few-Shotプロンプティング

Few-shotの例はモデルに期待する入出力パターンを示します。以下の状況で特に効果的です：

- カスタム出力フォーマット

- ドメイン特化のトーンや用語

- 非標準ラベルを使う分類

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}

アクションアイテム:"""

2.4 Chain-of-Thought（CoT）

複雑な推論タスクの場合、最終的な答えを提示する前に思考プロセスを示すよう要求してください。

COT_PROMPT = """以下の問題をステップバイステップで解いてください。

各ステップで推論プロセスを示してから最終的な答えを提示してください。

問題: 顧客のクレジットが500ドルあります。320ドルの注文をして、

80ドルの商品を1つ返品しました。残りのクレジットはいくらですか？

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

Zero-shot CoTトリガー: 例なしでプロンプトの最後に「ステップバイステップで考えてみましょう。」を追加するだけです。このシンプルな追加だけで多くのモデルでの多段階推論が大幅に向上します。

2.5 構造化出力

JSONモードまたはスキーマ制約を使ってパース可能な出力を強制してください。

from openai import OpenAI

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"}

PydanticとOpenAI SDKの構造化出力機能：

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

3. LLM APIとSDK

3.1 OpenAI SDK

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}")

3.2 Anthropic SDK

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}")

3.3 LiteLLMで統合インターフェースを構築

LiteLLMは100以上のLLMプロバイダーに単一のインターフェースを提供します：

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)

3.4 会話履歴管理

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

4. 検索拡張生成（RAG）

4.1 なぜRAGなのか？

LLMにはRAGが解決する2つの根本的な限界があります：

1. **知識カットオフ**: モデルはトレーニングデータにあったものしか知りません。

2. **コンテキストウィンドウの限界**: モデルはすべてのドキュメントを一度に「知る」ことができません。

RAGは推論時に関連情報を検索してプロンプトに注入することで両方を解決します。

ユーザークエリ

│

▼

[クエリ埋め込み] ──► [ベクター検索] ──► 上位K個の関連チャンク

│

▼

[拡張プロンプトの構築]

システム: 役立つアシスタントです。

コンテキスト: {検索されたチャンク}

ユーザー: {元のクエリ}

│

▼

[LLMが回答を生成]

4.2 ドキュメント収集パイプライン

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)}個のチャンクをインデックス化完了")

4.3 検索と生成

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')}")

4.4 検索品質の向上

**ハイブリッド検索**は密（セマンティック）と疎（キーワード）の検索を組み合わせます：

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]]

5. ツール使用と関数呼び出し

5.1 ツールの定義

ツールはLLMが外部APIを呼び出したり、データベースを検索したり、コードを実行したりできるようにします：

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"]

}

}

}

]

5.2 ツール呼び出しの処理

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)

5.3 並列ツール呼び出し

GPT-4oとClaude 3以上は並列ツール呼び出しをサポートしており、独立したタスクのレイテンシを大幅に削減します：

モデルが複数のツールを同時に呼び出せます

上記のループがすでにこれを処理します — 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(旅行ヒント)

6. ストリーミングと非同期パターン

6.1 応答のストリーミング

ストリーミングはテキストが生成されるとすぐに表示されるため、ユーザーが感じるレイテンシを大幅に改善します：

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)

6.2 FastAPIによる非同期ストリーミング

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")

6.3 非同期バッチ処理

多くのアイテムを処理する際、非同期並行処理はスループットを大幅に向上させます：

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))

7. 評価とテスト

7.1 LLM評価が難しい理由

従来のソフトウェアテストは決定論的アサーションを使用します：

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

LLMの出力は非決定論的であり、以下が必要です：

- **セマンティック等価性チェック**（文字列の同一性ではない）

- **ルーブリックベースのスコアリング**

- **参照なしの品質評価**

- **統計的サンプリング**（一度の実行では十分ではない）

7.2 LLM-as-Judge

有能なLLMを使って別のLLMの出力を評価します：

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:

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)

7.3 評価フレームワーク

**DeepEval**は包括的なLLM評価メトリクスを提供します：

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)

7.4 Promptfooによるリグレッションテスト

PromptfooはYAMLでテストケースを定義し、モデルバージョン間で実行できます：

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文

8. コスト最適化

8.1 トークン計算と予算策定

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

8.2 プロンプトキャッシング

AnthropicとOpenAIはどちらも繰り返しのシステムプロンプトや大容量コンテキストのためのプロンプトキャッシングを提供しています：

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%割引。

8.3 モデルルーティング

処理可能な最も安価なモデルにタスクをルーティングします：

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

9. プロダクションデプロイ

9.1 FastAPIバックエンド

from fastapi import FastAPI, HTTPException

from pydantic import BaseModel

from openai import AsyncOpenAI

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))

9.2 レート制限とリトライ

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

9.3 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

10. 可観測性とモニタリング

10.1 追跡すべき主要メトリクス

| メトリクス | 重要な理由 | 警告閾値 |

| --------------------------- | ------------ | ------------------------- |

| レイテンシ（p50, p95, p99） | ユーザー体験 | ストリーミングでp95 > 5秒 |

| トークン使用量 | コスト | 予算差異 > 20% |

| エラー率 | 信頼性 | リクエストの > 1% |

| キャッシュヒット率 | コスト効率 | < 30%（要調査） |

| 評価スコア | 品質 | 基準値から > 5%低下 |

10.2 LangSmithトレーシング

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でフルトレース（プロンプト、応答、レイテンシ、トークン）を確認可能

10.3 カスタムロギング

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))

10.4 ガードレールと安全性

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

まとめ

プロダクションレベルのLLMアプリケーション構築は複数の次元での習熟を必要とします：

| 領域 | 主な教訓 |

| -------------------------- | --------------------------------------------------------------------------- |

| プロンプトエンジニアリング | プロンプトをコードとして扱う; バージョン管理、テスト、反復 |

| RAG | ハイブリッド検索＋リランキングで検索品質を大幅向上 |

| ツール使用 | 並列ツール呼び出しで多段階タスクのレイテンシを削減 |

| ストリーミング | 対話型UXに必須; FastAPIとSSEの活用 |

| 評価 | LLM-as-judge＋自動化テストスイートでリグレッション検出 |

| コスト | キャッシング、ルーティング、プロンプトキャッシングでコストを80%以上削減可能 |

| モニタリング | 初日からレイテンシ、トークン、品質メトリクスを追跡 |

この分野は急速に進化していますが、これらの基本原則はどのモデルやフレームワークが来年主流になっても変わらず有用です。シンプルに始め、すべてを測定し、実際の使用データに基づいて反復してください。

**Q1. RAGとは何か、なぜ有用なのですか？**

**答え**: RAGはRetrieval-Augmented Generation（検索拡張生成）の略です。LLMが学習したことのないドキュメントに関する質問に答えられるよう、推論時に関連テキストを検索してプロンプトに注入します。これにより知識カットオフの問題とコンテキストウィンドウの限界の両方を解決します。

**Q2. LLMエージェントにおける並列ツール呼び出しの主なメリットは何ですか？**

**答え**: 並列ツール呼び出しにより、モデルは順次ではなく単一のターンで複数のツールを同時に呼び出せます。ツール呼び出しが互いに独立した多段階タスクの全体的なレイテンシを削減します。

**Q3. 単純な文字列マッチングよりLLM-as-judge評価が好まれる理由は何ですか？**

**答え**: LLMの出力はさまざまな表現方法でセマンティックに等価である可能性があり、文字列マッチングは偽陰性を生成します。LLMジャッジはルーブリックを使ってセマンティックな正確さ、有用性、品質を評価できるため、決定論的な比較よりはるかに正確な品質シグナルを提供します。

**Q4. 品質を低下させずにLLM APIのコストを削減する2つの手法を説明してください。**

**答え**:

1. プロンプトキャッシング: 繰り返しの大容量プレフィックス（システムプロンプト、参照ドキュメント）をキャッシュして、全額は最初の一度だけ課金されるようにします。

2. モデルルーティング: シンプルなタスク（分類、抽出）は安価な小型モデルで処理し、複雑な推論タスクにのみ高価な大型モデルを使用します。