AI エンジニアリング

本番レベルの LLM アプリケーション、RAG システム、インテリジェントエージェントを構築するためのパターン。

アクティベーションのタイミング

• RAG システム、LLM 機能、または AI エージェントワークフローを構築・改善する場合
• モデル、ベクターデータベース、埋め込み戦略を選定する場合
• 検索品質、レイテンシ、または推論コストを最適化する場合
• AI セーフティガードレール、コンテンツモデレーション、または PII 処理を実装する場合
• マルチモーダル入力(画像、音声、ドキュメント)を AI パイプラインに統合する場合
• マルチエージェント連携またはエージェントのツール使用ループを設計する場合
• AI 可観測性、評価、または A/B テストを設定する場合

モデル選択

モデル	最適用途	相対コスト
claude-opus-4-6	複雑な推論、アーキテクチャ、研究	高
claude-sonnet-4-6	バランスの取れたコーディング、ほとんどの開発タスク	中
claude-haiku-4-5	分類、抽出、高スループットタスク	低
GPT-4o	OpenAI ツールエコシステム、関数呼び出し	中〜高
Llama 3.1 70B (ローカル)	エアギャップ環境、コスト重視、PII リスクなし	なし(インフラコストのみ)

開発には Sonnet クラスのモデルを標準としてください。高スループットステップでは Haiku/ミニバリアントを使用してください。推論の多いタスクには Opus/GPT-4o を予約してください。

RAG アーキテクチャ

チャンキング戦略

# BAD: Fixed-size splits break semantic units
text_splitter = CharacterTextSplitter(chunk_size=500)

# GOOD: Semantic chunking preserves context
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    separators=["\n\n", "\n", ". ", " ", ""]
)

戦略	使用場面
再帰的文字分割	一般的な散文、コード
セマンティック(sentence-transformers)	可変長ドキュメント
ドキュメント構造認識	PDF、HTML、Markdown
スライディングウィンドウ	密度の高い技術的コンテンツ

ベクターデータベース選択

DB	ホスト版	セルフホスト	ハイブリッド検索	注記
Pinecone	あり	なし	あり	マネージド、サーバーレス
Qdrant	あり	あり	あり	Rust コア、高速フィルタリング
Weaviate	あり	あり	あり	GraphQL API
pgvector	Supabase 経由	あり	tsvector で対応	すでに Postgres を使用している場合に最適
Chroma	なし	あり	なし	ローカル開発のみ

ハイブリッド検索(ベクトル + キーワード)

from qdrant_client import QdrantClient
from qdrant_client.models import SparseVector, NamedSparseVector

# Dense vector (semantic) + sparse vector (BM25)
results = client.query_points(
    collection_name="docs",
    prefetch=[
        models.Prefetch(query=dense_embedding, using="dense", limit=20),
        models.Prefetch(query=SparseVector(indices=bm25_indices, values=bm25_values),
                        using="sparse", limit=20),
    ],
    query=models.FusionQuery(fusion=models.Fusion.RRF),  # Reciprocal Rank Fusion
    limit=10,
)

リランキング

# BAD: Return top-k by vector similarity alone
results = index.query(vector=embedding, top_k=5)

# GOOD: Over-fetch then rerank for precision
candidates = index.query(vector=embedding, top_k=20)

import cohere
co = cohere.Client()
reranked = co.rerank(
    model="rerank-english-v3.0",
    query=user_query,
    documents=[r.metadata["text"] for r in candidates.matches],
    top_n=5,
)

RAG パイプラインパターン

パターン	解決する問題
HyDE (Hypothetical Document Embeddings)	クエリ/ドキュメント埋め込みの不一致
RAG-Fusion	単一クエリが狭い — 複数のクエリバリアントを実行
Self-RAG	モデルが検索が必要かどうかを判断
GraphRAG	接続されたエンティティ間のマルチホップ推論
コンテキスト圧縮	取得したチャンクが雑音が多い; 関連スパンのみを抽出

# HyDE: generate a hypothetical answer, embed it, retrieve similar docs
hyde_prompt = f"Write a paragraph that would answer: {query}"
hypothetical_doc = llm.invoke(hyde_prompt)
hyde_embedding = embedder.embed(hypothetical_doc)
results = vector_store.similarity_search_by_vector(hyde_embedding)

エージェント オーケストレーション

エージェントループ(LangGraph)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    tool_calls_remaining: int

def agent_node(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: AgentState):
    last_message = state["messages"][-1]
    results = execute_tools(last_message.tool_calls)
    return {"messages": results, "tool_calls_remaining": state["tool_calls_remaining"] - 1}

def should_continue(state: AgentState):
    last = state["messages"][-1]
    if not last.tool_calls or state["tool_calls_remaining"] <= 0:
        return END
    return "tools"

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_edge("tools", "agent")
graph.add_conditional_edges("agent", should_continue)

エージェント メモリパターン

タイプ	ストレージ	用途
短期メモリ	メモリ内メッセージリスト	現在の会話コンテキスト
長期メモリ	ベクターストア + サマリー	セッション間の事実、好み
エピソード記憶	構造化 DB	過去のタスク結果
手続き記憶	プロンプト / ツール定義	スキル、ワークフロー

# Summarize + compress conversation memory
from langchain.memory import ConversationSummaryBufferMemory

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=1000,  # summarize once buffer exceeds limit
    return_messages=True,
)

マルチエージェント パターン(CrewAI)

from crewai import Agent, Task, Crew

researcher = Agent(
    role="Senior Researcher",
    goal="Find accurate information",
    tools=[search_tool, browse_tool],
    llm=llm,
)
writer = Agent(
    role="Technical Writer",
    goal="Synthesize research into clear prose",
    llm=llm,
)

research_task = Task(description="Research {topic}", agent=researcher)
write_task = Task(description="Write a report based on research", agent=writer)

crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
result = crew.kickoff(inputs={"topic": "vector databases"})

プロンプト エンジニアリング

構造化出力

# TypeScript: Zod schema → structured output
import Anthropic from "@anthropic-ai/sdk";
import { z } from "zod";
import zodToJsonSchema from "zod-to-json-schema";

const ExtractedData = z.object({
  entities: z.array(z.object({ name: z.string(), type: z.string() })),
  summary: z.string(),
});

const response = await client.messages.create({
  model: "claude-sonnet-4-6",
  max_tokens: 1024,
  tools: [{
    name: "extract_data",
    description: "Extract structured data from text",
    input_schema: zodToJsonSchema(ExtractedData),
  }],
  tool_choice: { type: "tool", name: "extract_data" },
  messages: [{ role: "user", content: `Extract from: ${text}` }],
});

プロンプティング テクニック

テクニック	使用場面
思考の鎖(Think step by step)	マルチステップ推論、数学、ロジック
思考の木	複数の解法パスの探索
自己整合性	複数の出力をサンプリング、多数決
Few-shot 例	一貫性のある形式、特殊なタスク
憲法的自己批判	セーフティチェック、トーン調整

# BAD: Vague instruction
"Summarize this document"

# GOOD: Structured, constrained prompt
"""Summarize the following document in exactly 3 bullet points.
Each bullet must start with a verb and be under 20 words.
Focus only on actionable findings.

Document:
{document}"""

本番環境パターン

セマンティック キャッシング

from semantic_router.encoders import OpenAIEncoder
from semantic_router.layer import RouteLayer

# Cache responses for semantically similar queries
cache = RedisSemanticCache(
    redis_url="redis://localhost:6379",
    embedding=OpenAIEmbeddings(),
    score_threshold=0.95,  # cosine similarity threshold
)

@cache
def get_answer(query: str) -> str:
    return llm.invoke(query)

FastAPI でのストリーミング

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import anthropic

app = FastAPI()
client = anthropic.Anthropic()

@app.post("/chat")
async def chat(query: str):
    async def generate():
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": query}],
        ) as stream:
            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")

コスト管理

# Estimate cost before execution
def estimate_cost(prompt: str, model: str = "claude-sonnet-4-6") -> float:
    input_tokens = len(prompt) // 4  # rough estimate
    # Sonnet 4.6: $3/M input, $15/M output
    return (input_tokens / 1_000_000) * 3.0

# Hard cap: reject if estimated cost exceeds threshold
if estimate_cost(prompt) > 0.10:
    raise ValueError("Prompt too large for single request — chunk it")

AI セーフティ & ガードレール

プロンプト インジェクション検出

INJECTION_PATTERNS = [
    r"ignore (previous|above|all) instructions",
    r"you are now",
    r"disregard your",
    r"new persona",
    r"act as (if you are|a)?",
]

def detect_injection(user_input: str) -> bool:
    import re
    return any(re.search(p, user_input, re.IGNORECASE) for p in INJECTION_PATTERNS)

# Wrap all user inputs
if detect_injection(user_message):
    return {"error": "Input rejected"}

PII 削除

import re

PII_PATTERNS = {
    "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
    "ssn": r"\b\d{3}-\d{2}-\d{4}\b",
    "credit_card": r"\b(?:\d{4}[- ]?){3}\d{4}\b",
    "phone": r"\b\+?1?\s?\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}\b",
}

def redact_pii(text: str) -> str:
    for label, pattern in PII_PATTERNS.items():
        text = re.sub(pattern, f"[{label.upper()}_REDACTED]", text)
    return text

コンテンツ モデレーション

# Use OpenAI Moderation API as a pre-filter (free)
import openai

def is_safe(text: str) -> bool:
    result = openai.moderations.create(input=text)
    return not result.results[0].flagged

# Gate all user inputs
if not is_safe(user_message):
    return {"error": "Message violates content policy"}

AI 可観測性

LangSmith トレーシング

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-key"
os.environ["LANGCHAIN_PROJECT"] = "my-rag-app"

# All LangChain calls now auto-traced — no code changes needed
chain = prompt | llm | output_parser
result = chain.invoke({"query": user_query})

カスタム メトリクス(Prometheus)

from prometheus_client import Counter, Histogram

llm_requests = Counter("llm_requests_total", "Total LLM API calls", ["model", "status"])
llm_latency = Histogram("llm_latency_seconds", "LLM response latency", ["model"])
retrieval_score = Histogram("retrieval_relevance_score", "RAG retrieval scores")

with llm_latency.labels(model="claude-sonnet-4-6").time():
    response = client.messages.create(...)
llm_requests.labels(model="claude-sonnet-4-6", status="success").inc()

RAG 評価

メトリクス	ツール	測定内容
コンテキスト精度	RAGAS	取得したチャンクは関連性があるか？
コンテキスト再現率	RAGAS	取得に必要なチャンクを逃していないか？
回答の誠実性	RAGAS	回答は取得したコンテキストと一致しているか？
回答の関連性	RAGAS	回答は質問に対応しているか？

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

dataset = Dataset.from_dict({
    "question": questions,
    "answer": answers,
    "contexts": retrieved_contexts,
    "ground_truth": expected_answers,
})
scores = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_precision])

注意信号

• 検索品質を評価せずに RAG を使用する — 高い埋め込み類似度は、取得したチャンクが質問に答えることを意味しません。生成品質とは別に検索精度/再現率を評価してください
• チャンクサイズを恣意的に選択する — 大きすぎると無関係なコンテンツで文脈が埋まり、小さすぎるとコヒーレンス が失われます。コミットする前に実際のクエリに対してチャンクサイズをベンチマークしてください
• プロンプト変更前に評価がない — 評価スイートなしで本番環境でプロンプトを変更することは、テストされていないコードをデプロイする AI 版です。まず評価を構築してください
• 制限なしのエージェントループ — 最大ターン上限のないエージェントは、曖昧なタスク上で無期限にループできます。常にハード制限を設定し、適切な停止動作を定義してください
• ユーザー提供テキストをシステムプロンプトに直接注入する — ユーザーコンテンツを介したプロンプト インジェクションは指示をオーバーライドできます。入力をサニタイズし、それをuserメッセージロールに保つ、決してsystemロールに入れないでください
• 起動後までコスト推定を延期する — LLM コストはトークン数 × リクエスト数でスケールします。アーキテクチャ段階でリクエストあたりのコストを推定し、起動後に推定してコストが変更できないほど高くならないようにしてください
• すべてのコンテンツタイプに単一の埋め込みモデルを使用する — コード、散文、テーブルは異なるセマンティック空間を持ちます。ドメイン適切なモデルをベンチマークするか、コンテンツタイプごとに別のインデックスを分離してください

チェックリスト

AI 機能をシップする前に:

• モデルが特定のバージョンに固定されている、latestではない
• max_tokensが明示的に設定されている; 生成バジェットがコストに対して検証されている
• プロンプト インジェクション検出がすべてのユーザー制御入力に適用されている
• PII 削除が外部モデルにデータを送信する前に実行されている
• コンテンツモデレーション ゲートがユーザー向けエンドポイントに設置されている
• 検索品質が測定されている(RAGAS または同等による コンテキスト精度/再現率)
• セマンティック キャッシングが繰り返しまたはほぼ重複するクエリに対して有効化されている
• ストリーミングが >200 トークンの応答に使用されている(クライアント タイムアウトを回避)
• レート制限および一時的なエラーに対して指数バックオフを使用して再試行している
• LLM 呼び出しがトレースされている(LangSmith、Phoenix、またはカスタム スパン)
• レイテンシおよびトークン使用メトリクスが監視スタックに発行されている
• フォールバックモデルまたはグレースフルデグラデーション パスが定義されている
• チャンキング戦略が代表的なドキュメントで検証されている
• 検索コーパスが 10K チャンク を超える場合、リランカーが設置されている

関連項目: claude-api、observability、security